feat: steppable aggr fn

poc: step aggr query

feat: mvp poc stuff

test: sqlness

chore: import missing

feat: support first/last_value

fix: check also include first/last value

Cargo.lock

@@ -5133,7 +5133,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=454c52634c3bac27de10bf0d85d5533eed1cf03f#454c52634c3bac27de10bf0d85d5533eed1cf03f"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=2dca1dc67862d7b410838aef81232274c019b3f6#2dca1dc67862d7b410838aef81232274c019b3f6"
 dependencies = [
  "prost 0.13.5",
  "serde",

Cargo.toml

@@ -133,7 +133,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "454c52634c3bac27de10bf0d85d5533eed1cf03f" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "2dca1dc67862d7b410838aef81232274c019b3f6" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

src/common/function/src/adjust_flow.rs

@@ -0,0 +1,90 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_macro::admin_fn;
+use common_query::error::{
+    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
+};
+use common_query::prelude::Signature;
+use datafusion::logical_expr::Volatility;
+use datatypes::value::{Value, ValueRef};
+use session::context::QueryContextRef;
+use snafu::ensure;
+use store_api::storage::ConcreteDataType;
+
+use crate::handlers::FlowServiceHandlerRef;
+use crate::helper::parse_catalog_flow;
+
+fn adjust_signature() -> Signature {
+    Signature::exact(
+        vec![
+            ConcreteDataType::string_datatype(), // flow name
+            ConcreteDataType::uint64_datatype(), // min_run_interval in seconds
+            ConcreteDataType::uint64_datatype(), // max filter number per query
+        ],
+        Volatility::Immutable,
+    )
+}
+
+#[admin_fn(
+    name = AdjustFlowFunction,
+    display_name = adjust_flow,
+    sig_fn = adjust_signature,
+    ret = uint64
+)]
+pub(crate) async fn adjust_flow(
+    flow_service_handler: &FlowServiceHandlerRef,
+    query_ctx: &QueryContextRef,
+    params: &[ValueRef<'_>],
+) -> Result<Value> {
+    ensure!(
+        params.len() == 3,
+        InvalidFuncArgsSnafu {
+            err_msg: format!(
+                "The length of the args is not correct, expect 3, have: {}",
+                params.len()
+            ),
+        }
+    );
+
+    let (flow_name, min_run_interval, max_filter_num) = match (params[0], params[1], params[2]) {
+        (
+            ValueRef::String(flow_name),
+            ValueRef::UInt64(min_run_interval),
+            ValueRef::UInt64(max_filter_num),
+        ) => (flow_name, min_run_interval, max_filter_num),
+        _ => {
+            return UnsupportedInputDataTypeSnafu {
+                function: "adjust_flow",
+                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
+            }
+            .fail();
+        }
+    };
+
+    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;
+
+    let res = flow_service_handler
+        .adjust(
+            &catalog_name,
+            &flow_name,
+            min_run_interval,
+            max_filter_num as usize,
+            query_ctx.clone(),
+        )
+        .await?;
+    let affected_rows = res.affected_rows;
+
+    Ok(Value::from(affected_rows))
+}

src/common/function/src/admin.rs

@@ -26,6 +26,7 @@ use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;
 
+use crate::adjust_flow::AdjustFlowFunction;
 use crate::flush_flow::FlushFlowFunction;
 use crate::function_registry::FunctionRegistry;
 
@@ -43,5 +44,6 @@ impl AdminFunction {
         registry.register_async(Arc::new(FlushTableFunction));
         registry.register_async(Arc::new(CompactTableFunction));
         registry.register_async(Arc::new(FlushFlowFunction));
+        registry.register_async(Arc::new(AdjustFlowFunction));
     }
 }

src/common/function/src/flush_flow.rs

@@ -12,21 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_error::ext::BoxedError;
 use common_macro::admin_fn;
 use common_query::error::{
-    ExecuteSnafu, InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
+    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
 };
 use common_query::prelude::Signature;
 use datafusion::logical_expr::Volatility;
 use datatypes::value::{Value, ValueRef};
 use session::context::QueryContextRef;
-use snafu::{ensure, ResultExt};
-use sql::parser::ParserContext;
+use snafu::ensure;
 use store_api::storage::ConcreteDataType;
 
 use crate::handlers::FlowServiceHandlerRef;
+use crate::helper::parse_catalog_flow;
 
 fn flush_signature() -> Signature {
     Signature::uniform(
@@ -47,20 +45,6 @@ pub(crate) async fn flush_flow(
     query_ctx: &QueryContextRef,
     params: &[ValueRef<'_>],
 ) -> Result<Value> {
-    let (catalog_name, flow_name) = parse_flush_flow(params, query_ctx)?;
-
-    let res = flow_service_handler
-        .flush(&catalog_name, &flow_name, query_ctx.clone())
-        .await?;
-    let affected_rows = res.affected_rows;
-
-    Ok(Value::from(affected_rows))
-}
-
-fn parse_flush_flow(
-    params: &[ValueRef<'_>],
-    query_ctx: &QueryContextRef,
-) -> Result<(String, String)> {
     ensure!(
         params.len() == 1,
         InvalidFuncArgsSnafu {
@@ -70,7 +54,6 @@ fn parse_flush_flow(
             ),
         }
     );
-
     let ValueRef::String(flow_name) = params[0] else {
         return UnsupportedInputDataTypeSnafu {
             function: "flush_flow",
@@ -78,27 +61,14 @@ fn parse_flush_flow(
         }
         .fail();
     };
-    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
-        .map_err(BoxedError::new)
-        .context(ExecuteSnafu)?;
+    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;
 
-    let (catalog_name, flow_name) = match &obj_name.0[..] {
-        [flow_name] => (
-            query_ctx.current_catalog().to_string(),
-            flow_name.value.clone(),
-        ),
-        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
-        _ => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
-                    obj_name
-                ),
-            }
-            .fail()
-        }
-    };
-    Ok((catalog_name, flow_name))
+    let res = flow_service_handler
+        .flush(&catalog_name, &flow_name, query_ctx.clone())
+        .await?;
+    let affected_rows = res.affected_rows;
+
+    Ok(Value::from(affected_rows))
 }
 
 #[cfg(test)]
@@ -154,10 +124,7 @@ mod test {
             ("catalog.flow_name", ("catalog", "flow_name")),
         ];
         for (input, expected) in testcases.iter() {
-            let args = vec![*input];
-            let args = args.into_iter().map(ValueRef::String).collect::<Vec<_>>();
-
-            let result = parse_flush_flow(&args, &QueryContext::arc()).unwrap();
+            let result = parse_catalog_flow(input, &QueryContext::arc()).unwrap();
             assert_eq!(*expected, (result.0.as_str(), result.1.as_str()));
         }
     }

src/common/function/src/handlers.rs

@@ -87,6 +87,15 @@ pub trait FlowServiceHandler: Send + Sync {
         flow: &str,
         ctx: QueryContextRef,
     ) -> Result<api::v1::flow::FlowResponse>;
+
+    async fn adjust(
+        &self,
+        catalog: &str,
+        flow: &str,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+        ctx: QueryContextRef,
+    ) -> Result<api::v1::flow::FlowResponse>;
 }
 
 pub type TableMutationHandlerRef = Arc<dyn TableMutationHandler>;

src/common/function/src/helper.rs

@@ -12,12 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_query::error::{InvalidInputTypeSnafu, Result};
+use common_error::ext::BoxedError;
+use common_query::error::{ExecuteSnafu, InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
+use session::context::QueryContextRef;
 use snafu::ResultExt;
+use sql::parser::ParserContext;
 
 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -43,3 +46,30 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
         })
         .map(|v| v.as_u64())
 }
+
+pub fn parse_catalog_flow(
+    flow_name: &str,
+    query_ctx: &QueryContextRef,
+) -> Result<(String, String)> {
+    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
+        .map_err(BoxedError::new)
+        .context(ExecuteSnafu)?;
+
+    let (catalog_name, flow_name) = match &obj_name.0[..] {
+        [flow_name] => (
+            query_ctx.current_catalog().to_string(),
+            flow_name.value.clone(),
+        ),
+        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
+        _ => {
+            return InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
+                    obj_name
+                ),
+            }
+            .fail()
+        }
+    };
+    Ok((catalog_name, flow_name))
+}

src/common/function/src/lib.rs

@@ -15,6 +15,7 @@
 #![feature(let_chains)]
 #![feature(try_blocks)]
 
+mod adjust_flow;
 mod admin;
 mod flush_flow;
 mod macros;

src/common/function/src/state.rs

@@ -148,6 +148,17 @@ impl FunctionState {
             ) -> Result<api::v1::flow::FlowResponse> {
                 todo!()
             }
+
+            async fn adjust(
+                &self,
+                _catalog: &str,
+                _flow: &str,
+                _min_run_interval_secs: u64,
+                _max_filter_num_per_query: usize,
+                _ctx: QueryContextRef,
+            ) -> Result<api::v1::flow::FlowResponse> {
+                todo!()
+            }
         }
 
         Self {

src/flow/Cargo.toml

@@ -61,6 +61,7 @@ prost.workspace = true
 query.workspace = true
 rand.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 servers.workspace = true
 session.workspace = true
 smallvec.workspace = true

src/flow/src/adapter/flownode_impl.rs

@@ -18,7 +18,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 
 use api::v1::flow::{
-    flow_request, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
+    flow_request, AdjustFlow, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
 };
 use api::v1::region::InsertRequests;
 use catalog::CatalogManager;
@@ -32,6 +32,7 @@ use common_telemetry::{error, info, trace, warn};
 use datatypes::value::Value;
 use futures::TryStreamExt;
 use itertools::Itertools;
+use serde::{Deserialize, Serialize};
 use session::context::QueryContextBuilder;
 use snafu::{ensure, IntoError, OptionExt, ResultExt};
 use store_api::storage::{RegionId, TableId};
@@ -809,6 +810,25 @@ impl common_meta::node_manager::Flownode for FlowDualEngine {
                     ..Default::default()
                 })
             }
+            Some(flow_request::Body::Adjust(AdjustFlow { flow_id, options })) => {
+                #[derive(Debug, Serialize, Deserialize)]
+                struct Options {
+                    min_run_interval_secs: u64,
+                    max_filter_num_per_query: usize,
+                }
+                let options: Options = serde_json::from_str(&options).with_context(|_| {
+                    common_meta::error::DeserializeFromJsonSnafu { input: options }
+                })?;
+                self.batching_engine
+                    .adjust_flow(
+                        flow_id.unwrap().id as u64,
+                        options.min_run_interval_secs,
+                        options.max_filter_num_per_query,
+                    )
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                Ok(Default::default())
+            }
             other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
         }
     }
@@ -841,93 +861,6 @@ fn to_meta_err(
     }
 }
 
-#[async_trait::async_trait]
-impl common_meta::node_manager::Flownode for StreamingEngine {
-    async fn handle(&self, request: FlowRequest) -> MetaResult<FlowResponse> {
-        let query_ctx = request
-            .header
-            .and_then(|h| h.query_context)
-            .map(|ctx| ctx.into());
-        match request.body {
-            Some(flow_request::Body::Create(CreateRequest {
-                flow_id: Some(task_id),
-                source_table_ids,
-                sink_table_name: Some(sink_table_name),
-                create_if_not_exists,
-                expire_after,
-                comment,
-                sql,
-                flow_options,
-                or_replace,
-            })) => {
-                let source_table_ids = source_table_ids.into_iter().map(|id| id.id).collect_vec();
-                let sink_table_name = [
-                    sink_table_name.catalog_name,
-                    sink_table_name.schema_name,
-                    sink_table_name.table_name,
-                ];
-                let expire_after = expire_after.map(|e| e.value);
-                let args = CreateFlowArgs {
-                    flow_id: task_id.id as u64,
-                    sink_table_name,
-                    source_table_ids,
-                    create_if_not_exists,
-                    or_replace,
-                    expire_after,
-                    comment: Some(comment),
-                    sql: sql.clone(),
-                    flow_options,
-                    query_ctx,
-                };
-                let ret = self
-                    .create_flow(args)
-                    .await
-                    .map_err(BoxedError::new)
-                    .with_context(|_| CreateFlowSnafu { sql: sql.clone() })
-                    .map_err(to_meta_err(snafu::location!()))?;
-                METRIC_FLOW_TASK_COUNT.inc();
-                Ok(FlowResponse {
-                    affected_flows: ret
-                        .map(|id| greptime_proto::v1::FlowId { id: id as u32 })
-                        .into_iter()
-                        .collect_vec(),
-                    ..Default::default()
-                })
-            }
-            Some(flow_request::Body::Drop(DropRequest {
-                flow_id: Some(flow_id),
-            })) => {
-                self.remove_flow(flow_id.id as u64)
-                    .await
-                    .map_err(to_meta_err(snafu::location!()))?;
-                METRIC_FLOW_TASK_COUNT.dec();
-                Ok(Default::default())
-            }
-            Some(flow_request::Body::Flush(FlushFlow {
-                flow_id: Some(flow_id),
-            })) => {
-                let row = self
-                    .flush_flow_inner(flow_id.id as u64)
-                    .await
-                    .map_err(to_meta_err(snafu::location!()))?;
-                Ok(FlowResponse {
-                    affected_flows: vec![flow_id],
-                    affected_rows: row as u64,
-                    ..Default::default()
-                })
-            }
-            other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
-        }
-    }
-
-    async fn handle_inserts(&self, request: InsertRequests) -> MetaResult<FlowResponse> {
-        self.handle_inserts_inner(request)
-            .await
-            .map(|_| Default::default())
-            .map_err(to_meta_err(snafu::location!()))
-    }
-}
-
 impl FlowEngine for StreamingEngine {
     async fn create_flow(&self, args: CreateFlowArgs) -> Result<Option<FlowId>, Error> {
         self.create_flow_inner(args).await

src/flow/src/batching_mode/engine.rs

@@ -388,6 +388,20 @@ impl BatchingEngine {
     pub async fn flow_exist_inner(&self, flow_id: FlowId) -> bool {
         self.tasks.read().await.contains_key(&flow_id)
     }
+
+    pub async fn adjust_flow(
+        &self,
+        flow_id: FlowId,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+    ) -> Result<(), Error> {
+        let task = self.tasks.read().await.get(&flow_id).cloned();
+        let task = task.with_context(|| FlowNotFoundSnafu { id: flow_id })?;
+        debug!("Adjusting flow {flow_id} with min_run_interval_secs={} and max_filter_num_per_query={}", min_run_interval_secs, max_filter_num_per_query);
+        task.adjust(min_run_interval_secs, max_filter_num_per_query);
+
+        Ok(())
+    }
 }
 
 impl FlowEngine for BatchingEngine {

src/flow/src/batching_mode/frontend_client.rs

@@ -14,8 +14,9 @@
 
 //! Frontend client to run flow as batching task which is time-window-aware normal query triggered every tick set by user
 
-use std::sync::{Arc, Weak};
-use std::time::SystemTime;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant, SystemTime};
 
 use api::v1::greptime_request::Request;
 use api::v1::CreateTableExpr;
@@ -26,20 +27,21 @@ use common_meta::cluster::{NodeInfo, NodeInfoKey, Role};
 use common_meta::peer::Peer;
 use common_meta::rpc::store::RangeRequest;
 use common_query::Output;
-use common_telemetry::warn;
+use common_telemetry::{debug, warn};
+use itertools::Itertools;
 use meta_client::client::MetaClient;
-use rand::rng;
-use rand::seq::SliceRandom;
 use servers::query_handler::grpc::GrpcQueryHandler;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use snafu::{OptionExt, ResultExt};
 
+use crate::batching_mode::task::BatchingTask;
 use crate::batching_mode::{
     DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT, FRONTEND_ACTIVITY_TIMEOUT, GRPC_CONN_TIMEOUT,
     GRPC_MAX_RETRIES,
 };
 use crate::error::{ExternalSnafu, InvalidRequestSnafu, NoAvailableFrontendSnafu, UnexpectedSnafu};
-use crate::{Error, FlowAuthHeader};
+use crate::metrics::METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD;
+use crate::{Error, FlowAuthHeader, FlowId};
 
 /// Just like [`GrpcQueryHandler`] but use BoxedError
 ///
@@ -74,6 +76,105 @@ impl<
 
 type HandlerMutable = Arc<std::sync::Mutex<Option<Weak<dyn GrpcQueryHandlerWithBoxedError>>>>;
 
+/// Statistics about running query on this frontend from flownode
+#[derive(Debug, Default, Clone)]
+struct FrontendStat {
+    /// The query for flow id has been running since this timestamp
+    since: HashMap<FlowId, Instant>,
+    /// The average query time for each flow id
+    /// This is used to calculate the average query time for each flow id
+    past_query_avg: HashMap<FlowId, (usize, Duration)>,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct FrontendStats {
+    /// The statistics for each flow id
+    stats: Arc<Mutex<HashMap<String, FrontendStat>>>,
+}
+
+impl FrontendStats {
+    pub fn observe(&self, frontend_addr: &str, flow_id: FlowId) -> FrontendStatsGuard {
+        let mut stats = self.stats.lock().expect("Failed to lock frontend stats");
+        let stat = stats.entry(frontend_addr.to_string()).or_default();
+        stat.since.insert(flow_id, Instant::now());
+
+        FrontendStatsGuard {
+            stats: self.stats.clone(),
+            frontend_addr: frontend_addr.to_string(),
+            cur: flow_id,
+        }
+    }
+
+    /// return frontend addrs sorted by load, from lightest to heaviest
+    /// The load is calculated as the total average query time for each flow id plus running query's total running time elapsed
+    pub fn sort_by_load(&self) -> Vec<String> {
+        let stats = self.stats.lock().expect("Failed to lock frontend stats");
+        let fe_load_factor = stats
+            .iter()
+            .map(|(node_addr, stat)| {
+                // total expected avg running time for all currently running queries
+                let total_expect_avg_run_time = stat
+                    .since
+                    .keys()
+                    .map(|f| {
+                        let (count, total_duration) =
+                            stat.past_query_avg.get(f).unwrap_or(&(0, Duration::ZERO));
+                        if *count == 0 {
+                            0.0
+                        } else {
+                            total_duration.as_secs_f64() / *count as f64
+                        }
+                    })
+                    .sum::<f64>();
+                let total_cur_running_time = stat
+                    .since
+                    .values()
+                    .map(|since| since.elapsed().as_secs_f64())
+                    .sum::<f64>();
+                (
+                    node_addr.to_string(),
+                    total_expect_avg_run_time + total_cur_running_time,
+                )
+            })
+            .sorted_by(|(_, load_a), (_, load_b)| {
+                load_a
+                    .partial_cmp(load_b)
+                    .unwrap_or(std::cmp::Ordering::Equal)
+            })
+            .collect::<Vec<_>>();
+        debug!("Frontend load factor: {:?}", fe_load_factor);
+        for (node_addr, load) in &fe_load_factor {
+            METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD
+                .with_label_values(&[&node_addr.to_string()])
+                .observe(*load);
+        }
+        fe_load_factor
+            .into_iter()
+            .map(|(addr, _)| addr)
+            .collect::<Vec<_>>()
+    }
+}
+
+pub struct FrontendStatsGuard {
+    stats: Arc<Mutex<HashMap<String, FrontendStat>>>,
+    frontend_addr: String,
+    cur: FlowId,
+}
+
+impl Drop for FrontendStatsGuard {
+    fn drop(&mut self) {
+        let mut stats = self.stats.lock().expect("Failed to lock frontend stats");
+        if let Some(stat) = stats.get_mut(&self.frontend_addr) {
+            if let Some(since) = stat.since.remove(&self.cur) {
+                let elapsed = since.elapsed();
+                let (count, total_duration) = stat.past_query_avg.entry(self.cur).or_default();
+                *count += 1;
+                *total_duration += elapsed;
+            }
+        }
+    }
+}
+
 /// A simple frontend client able to execute sql using grpc protocol
 ///
 /// This is for computation-heavy query which need to offload computation to frontend, lifting the load from flownode
@@ -83,6 +184,7 @@ pub enum FrontendClient {
         meta_client: Arc<MetaClient>,
         chnl_mgr: ChannelManager,
         auth: Option<FlowAuthHeader>,
+        fe_stats: FrontendStats,
     },
     Standalone {
         /// for the sake of simplicity still use grpc even in standalone mode
@@ -114,6 +216,7 @@ impl FrontendClient {
                 ChannelManager::with_config(cfg)
             },
             auth,
+            fe_stats: Default::default(),
         }
     }
 
@@ -192,6 +295,7 @@ impl FrontendClient {
             meta_client: _,
             chnl_mgr,
             auth,
+            fe_stats,
         } = self
         else {
             return UnexpectedSnafu {
@@ -208,8 +312,21 @@ impl FrontendClient {
                 .duration_since(SystemTime::UNIX_EPOCH)
                 .unwrap()
                 .as_millis() as i64;
-            // shuffle the frontends to avoid always pick the same one
-            frontends.shuffle(&mut rng());
+            let node_addrs_by_load = fe_stats.sort_by_load();
+            // index+1 to load order asc, so that the lightest node has load 1 and non-existent node has load 0
+            let addr2load = node_addrs_by_load
+                .iter()
+                .enumerate()
+                .map(|(i, id)| (id.clone(), i + 1))
+                .collect::<HashMap<_, _>>();
+            // sort frontends by load, from lightest to heaviest
+            frontends.sort_by(|(_, a), (_, b)| {
+                // if not even in stats, treat as 0 load since never been queried
+                let load_a = addr2load.get(&a.peer.addr).unwrap_or(&0);
+                let load_b = addr2load.get(&b.peer.addr).unwrap_or(&0);
+                load_a.cmp(load_b)
+            });
+            debug!("Frontend nodes sorted by load: {:?}", frontends);
 
             // found node with maximum last_activity_ts
             for (_, node_info) in frontends
@@ -257,6 +374,7 @@ impl FrontendClient {
         create: CreateTableExpr,
         catalog: &str,
         schema: &str,
+        task: Option<&BatchingTask>,
     ) -> Result<u32, Error> {
         self.handle(
             Request::Ddl(api::v1::DdlRequest {
@@ -265,6 +383,7 @@ impl FrontendClient {
             catalog,
             schema,
             &mut None,
+            task,
         )
         .await
     }
@@ -276,15 +395,19 @@ impl FrontendClient {
         catalog: &str,
         schema: &str,
         peer_desc: &mut Option<PeerDesc>,
+        task: Option<&BatchingTask>,
     ) -> Result<u32, Error> {
         match self {
-            FrontendClient::Distributed { .. } => {
+            FrontendClient::Distributed { fe_stats, .. } => {
                 let db = self.get_random_active_frontend(catalog, schema).await?;
 
                 *peer_desc = Some(PeerDesc::Dist {
                     peer: db.peer.clone(),
                 });
 
+                let flow_id = task.map(|t| t.config.flow_id).unwrap_or_default();
+                let _guard = fe_stats.observe(&db.peer.addr, flow_id);
+
                 db.database
                     .handle_with_retry(req.clone(), GRPC_MAX_RETRIES)
                     .await

src/flow/src/batching_mode/state.rs

@@ -32,6 +32,7 @@ use crate::batching_mode::MIN_REFRESH_DURATION;
 use crate::error::{DatatypesSnafu, InternalSnafu, TimeSnafu, UnexpectedSnafu};
 use crate::metrics::{
     METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE, METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT,
+    METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT,
 };
 use crate::{Error, FlowId};
 
@@ -52,6 +53,11 @@ pub struct TaskState {
     pub(crate) shutdown_rx: oneshot::Receiver<()>,
     /// Task handle
     pub(crate) task_handle: Option<tokio::task::JoinHandle<()>>,
+
+    /// min run interval in seconds
+    pub(crate) min_run_interval: Option<u64>,
+    /// max filter number per query
+    pub(crate) max_filter_num: Option<usize>,
 }
 impl TaskState {
     pub fn new(query_ctx: QueryContextRef, shutdown_rx: oneshot::Receiver<()>) -> Self {
@@ -63,6 +69,8 @@ impl TaskState {
             exec_state: ExecState::Idle,
             shutdown_rx,
             task_handle: None,
+            min_run_interval: None,
+            max_filter_num: None,
         }
     }
 
@@ -87,20 +95,17 @@ impl TaskState {
     pub fn get_next_start_query_time(
         &self,
         flow_id: FlowId,
-        time_window_size: &Option<Duration>,
+        _time_window_size: &Option<Duration>,
         max_timeout: Option<Duration>,
     ) -> Instant {
-        let last_duration = max_timeout
+        let next_duration = max_timeout
             .unwrap_or(self.last_query_duration)
             .min(self.last_query_duration)
-            .max(MIN_REFRESH_DURATION);
-
-        let next_duration = time_window_size
-            .map(|t| {
-                let half = t / 2;
-                half.max(last_duration)
-            })
-            .unwrap_or(last_duration);
+            .max(
+                self.min_run_interval
+                    .map(Duration::from_secs)
+                    .unwrap_or(MIN_REFRESH_DURATION),
+            );
 
         // if have dirty time window, execute immediately to clean dirty time window
         if self.dirty_time_windows.windows.is_empty() {
@@ -246,6 +251,10 @@ impl DirtyTimeWindows {
             .with_label_values(&[flow_id.to_string().as_str()])
             .observe(first_nth.len() as f64);
 
+        METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT
+            .with_label_values(&[flow_id.to_string().as_str()])
+            .observe(self.windows.len() as f64);
+
         let full_time_range = first_nth
             .iter()
             .fold(chrono::Duration::zero(), |acc, (start, end)| {

src/flow/src/batching_mode/task.rs

@@ -144,6 +144,12 @@ impl BatchingTask {
         })
     }
 
+    pub fn adjust(&self, min_run_interval_secs: u64, max_filter_num_per_query: usize) {
+        let mut state = self.state.write().unwrap();
+        state.min_run_interval = Some(min_run_interval_secs);
+        state.max_filter_num = Some(max_filter_num_per_query);
+    }
+
     /// mark time window range (now - expire_after, now) as dirty (or (0, now) if expire_after not set)
     ///
     /// useful for flush_flow to flush dirty time windows range
@@ -280,7 +286,7 @@ impl BatchingTask {
         let catalog = &self.config.sink_table_name[0];
         let schema = &self.config.sink_table_name[1];
         frontend_client
-            .create(expr.clone(), catalog, schema)
+            .create(expr.clone(), catalog, schema, Some(self))
             .await?;
         Ok(())
     }
@@ -361,7 +367,7 @@ impl BatchingTask {
             };
 
             frontend_client
-                .handle(req, catalog, schema, &mut peer_desc)
+                .handle(req, catalog, schema, &mut peer_desc, Some(self))
                 .await
         };
 
@@ -580,19 +586,20 @@ impl BatchingTask {
                 ),
             })?;
 
-        let expr = self
-            .state
-            .write()
-            .unwrap()
-            .dirty_time_windows
-            .gen_filter_exprs(
+        let expr = {
+            let mut state = self.state.write().unwrap();
+            let max_window_cnt = state
+                .max_filter_num
+                .unwrap_or(DirtyTimeWindows::MAX_FILTER_NUM);
+            state.dirty_time_windows.gen_filter_exprs(
                 &col_name,
                 Some(l),
                 window_size,
-                DirtyTimeWindows::MAX_FILTER_NUM,
+                max_window_cnt,
                 self.config.flow_id,
                 Some(self),
-            )?;
+            )?
+        };
 
         debug!(
             "Flow id={:?}, Generated filter expr: {:?}",

src/flow/src/metrics.rs

@@ -42,6 +42,14 @@ lazy_static! {
         vec![60., 2. * 60., 3. * 60., 5. * 60., 10. * 60.]
     )
     .unwrap();
+    pub static ref METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT: HistogramVec =
+        register_histogram_vec!(
+            "greptime_flow_batching_engine_stalled_query_window_cnt",
+            "flow batching engine stalled query time window count",
+            &["flow_id"],
+            vec![0.0, 5., 10., 20., 40.]
+        )
+        .unwrap();
     pub static ref METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT: HistogramVec =
         register_histogram_vec!(
             "greptime_flow_batching_engine_query_window_cnt",
@@ -58,6 +66,14 @@ lazy_static! {
             vec![60., 4. * 60., 16. * 60., 64. * 60., 256. * 60.]
         )
         .unwrap();
+    pub static ref METRIC_FLOW_BATCHING_ENGINE_GUESS_FE_LOAD: HistogramVec =
+        register_histogram_vec!(
+            "greptime_flow_batching_engine_guess_fe_load",
+            "flow batching engine guessed frontend load",
+            &["fe_addr"],
+            vec![60., 4. * 60., 16. * 60., 64. * 60., 256. * 60.]
+        )
+        .unwrap();
     pub static ref METRIC_FLOW_RUN_INTERVAL_MS: IntGauge =
         register_int_gauge!("greptime_flow_run_interval_ms", "flow run interval in ms").unwrap();
     pub static ref METRIC_FLOW_ROWS: IntCounterVec = register_int_counter_vec!(

src/operator/src/flow.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use api::v1::flow::FlowRequestHeader;
+use api::v1::flow::{AdjustFlow, FlowRequestHeader};
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use common_function::handlers::FlowServiceHandler;
@@ -22,6 +22,7 @@ use common_query::error::Result;
 use common_telemetry::tracing_context::TracingContext;
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
+use serde_json::json;
 use session::context::QueryContextRef;
 use snafu::{OptionExt, ResultExt};
 
@@ -57,9 +58,96 @@ impl FlowServiceHandler for FlowServiceOperator {
     ) -> Result<api::v1::flow::FlowResponse> {
         self.flush_inner(catalog, flow, ctx).await
     }
+
+    async fn adjust(
+        &self,
+        catalog: &str,
+        flow: &str,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+        ctx: QueryContextRef,
+    ) -> Result<api::v1::flow::FlowResponse> {
+        self.adjust_inner(
+            catalog,
+            flow,
+            min_run_interval_secs,
+            max_filter_num_per_query,
+            ctx,
+        )
+        .await
+    }
 }
 
 impl FlowServiceOperator {
+    async fn adjust_inner(
+        &self,
+        catalog: &str,
+        flow: &str,
+        min_run_interval_secs: u64,
+        max_filter_num_per_query: usize,
+        ctx: QueryContextRef,
+    ) -> Result<api::v1::flow::FlowResponse> {
+        let id = self
+            .flow_metadata_manager
+            .flow_name_manager()
+            .get(catalog, flow)
+            .await
+            .map_err(BoxedError::new)
+            .context(common_query::error::ExecuteSnafu)?
+            .context(common_meta::error::FlowNotFoundSnafu {
+                flow_name: format!("{}.{}", catalog, flow),
+            })
+            .map_err(BoxedError::new)
+            .context(common_query::error::ExecuteSnafu)?
+            .flow_id();
+
+        let all_flownode_peers = self
+            .flow_metadata_manager
+            .flow_route_manager()
+            .routes(id)
+            .await
+            .map_err(BoxedError::new)
+            .context(common_query::error::ExecuteSnafu)?;
+
+        // order of flownodes doesn't matter here
+        let all_flow_nodes = FuturesUnordered::from_iter(
+            all_flownode_peers
+                .iter()
+                .map(|(_key, peer)| self.node_manager.flownode(peer.peer())),
+        )
+        .collect::<Vec<_>>()
+        .await;
+
+        // TODO(discord9): use proper type for flow options
+        let options = json!({
+            "min_run_interval_secs": min_run_interval_secs,
+            "max_filter_num_per_query": max_filter_num_per_query,
+        });
+
+        for node in all_flow_nodes {
+            let _res = {
+                use api::v1::flow::{flow_request, FlowRequest};
+                let flush_req = FlowRequest {
+                    header: Some(FlowRequestHeader {
+                        tracing_context: TracingContext::from_current_span().to_w3c(),
+                        query_context: Some(
+                            common_meta::rpc::ddl::QueryContext::from(ctx.clone()).into(),
+                        ),
+                    }),
+                    body: Some(flow_request::Body::Adjust(AdjustFlow {
+                        flow_id: Some(api::v1::FlowId { id }),
+                        options: options.to_string(),
+                    })),
+                };
+                node.handle(flush_req)
+                    .await
+                    .map_err(BoxedError::new)
+                    .context(common_query::error::ExecuteSnafu)?
+            };
+        }
+        Ok(Default::default())
+    }
+
     /// Flush the flownodes according to the flow id.
     async fn flush_inner(
         &self,

src/query/src/dist_plan/analyzer.rs

@@ -15,6 +15,7 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 
+use common_telemetry::debug;
 use datafusion::datasource::DefaultTableSource;
 use datafusion::error::Result as DfResult;
 use datafusion_common::config::ConfigOptions;
@@ -154,6 +155,7 @@ struct PlanRewriter {
     /// Partition columns of the table in current pass
     partition_cols: Option<Vec<String>>,
     column_requirements: HashSet<Column>,
+    expand_on_next_call: bool,
 }
 
 impl PlanRewriter {
@@ -174,6 +176,10 @@ impl PlanRewriter {
         {
             return true;
         }
+        if self.expand_on_next_call {
+            self.expand_on_next_call = false;
+            return true;
+        }
         match Categorizer::check_plan(plan, self.partition_cols.clone()) {
             Commutativity::Commutative => {}
             Commutativity::PartialCommutative => {
@@ -190,12 +196,17 @@ impl PlanRewriter {
                     self.stage.push(plan)
                 }
             }
-            Commutativity::TransformedCommutative(transformer) => {
+            Commutativity::TransformedCommutative {
+                transformer,
+                expand_on_parent,
+            } => {
                 if let Some(transformer) = transformer
-                    && let Some(plan) = transformer(plan)
+                    && let Some(changed_plan) = transformer(plan)
                 {
-                    self.update_column_requirements(&plan);
-                    self.stage.push(plan)
+                    debug!("PlanRewriter: transformed plan: {changed_plan} from {plan}");
+                    self.update_column_requirements(&changed_plan);
+                    self.stage.push(changed_plan);
+                    self.expand_on_next_call = expand_on_parent;
                 }
             }
             Commutativity::NonCommutative
@@ -391,10 +402,21 @@ impl TreeNodeRewriter for PlanRewriter {
             return Ok(Transformed::yes(node));
         };
 
+        let parent = parent.clone();
+
         // TODO(ruihang): avoid this clone
-        if self.should_expand(&parent.clone()) {
+        if self.should_expand(&parent) {
             // TODO(ruihang): does this work for nodes with multiple children?;
-            let node = self.expand(node)?;
+            debug!("PlanRewriter: should expand child:\n {node}\n Of Parent: {parent}");
+            let node = self.expand(node);
+            debug!(
+                "PlanRewriter: expanded plan: {}",
+                match &node {
+                    Ok(n) => n.to_string(),
+                    Err(e) => format!("Error expanding plan: {e}"),
+                }
+            );
+            let node = node?;
             self.pop_stack();
             return Ok(Transformed::yes(node));
         }

src/query/src/dist_plan/commutativity.rs

@@ -15,6 +15,9 @@
 use std::collections::HashSet;
 use std::sync::Arc;
 
+use common_function::aggr::{HllState, UddSketchState, HLL_NAME, UDDSKETCH_STATE_NAME};
+use common_telemetry::debug;
+use datafusion::functions_aggregate::sum::sum_udaf;
 use datafusion_expr::{Expr, LogicalPlan, UserDefinedLogicalNode};
 use promql::extension_plan::{
     EmptyMetric, InstantManipulate, RangeManipulate, SeriesDivide, SeriesNormalize,
@@ -23,12 +26,157 @@ use promql::extension_plan::{
 use crate::dist_plan::merge_sort::{merge_sort_transformer, MergeSortLogicalPlan};
 use crate::dist_plan::MergeScanLogicalPlan;
 
+/// generate the upper aggregation plan that will execute on the frontend.
+pub fn step_aggr_to_upper_aggr(aggr_plan: &LogicalPlan) -> datafusion_common::Result<LogicalPlan> {
+    let LogicalPlan::Aggregate(aggr) = aggr_plan else {
+        return Err(datafusion_common::DataFusionError::Plan(
+            "step_aggr_to_upper_aggr only accepts Aggregate plan".to_string(),
+        ));
+    };
+    if !is_all_aggr_exprs_steppable(&aggr.aggr_expr) {
+        return Err(datafusion_common::DataFusionError::NotImplemented(
+            "Some aggregate expressions are not steppable".to_string(),
+        ));
+    }
+    let mut upper_aggr_expr = vec![];
+    for aggr_expr in &aggr.aggr_expr {
+        let Some(aggr_func) = get_aggr_func(aggr_expr) else {
+            return Err(datafusion_common::DataFusionError::NotImplemented(
+                "Aggregate function not found".to_string(),
+            ));
+        };
+        let col_name = aggr_expr.name_for_alias()?;
+        let input_column =
+            Expr::Column(datafusion_common::Column::new_unqualified(col_name.clone()));
+        let upper_func = match aggr_func.func.name() {
+            "sum" | "min" | "max" | "last_value" | "first_value" => {
+                // aggr_calc(aggr_merge(input_column))) as col_name
+                let mut new_aggr_func = aggr_func.clone();
+                new_aggr_func.args = vec![input_column.clone()];
+                new_aggr_func
+            }
+            "count" => {
+                // sum(input_column) as col_name
+                let mut new_aggr_func = aggr_func.clone();
+                new_aggr_func.func = sum_udaf();
+                new_aggr_func.args = vec![input_column.clone()];
+                new_aggr_func
+            }
+            UDDSKETCH_STATE_NAME => {
+                // udd_merge(bucket_size, error_rate input_column) as col_name
+                let mut new_aggr_func = aggr_func.clone();
+                new_aggr_func.func = Arc::new(UddSketchState::merge_udf_impl());
+                new_aggr_func.args[2] = input_column.clone();
+                new_aggr_func
+            }
+            HLL_NAME => {
+                // hll_merge(input_column) as col_name
+                let mut new_aggr_func = aggr_func.clone();
+                new_aggr_func.func = Arc::new(HllState::merge_udf_impl());
+                new_aggr_func.args = vec![input_column.clone()];
+                new_aggr_func
+            }
+            _ => {
+                return Err(datafusion_common::DataFusionError::NotImplemented(format!(
+                    "Aggregate function {} is not supported for Step aggregation",
+                    aggr_func.func.name()
+                )))
+            }
+        };
+
+        // deal with nested alias case
+        let mut new_aggr_expr = aggr_expr.clone();
+        {
+            let new_aggr_func = get_aggr_func_mut(&mut new_aggr_expr).unwrap();
+            *new_aggr_func = upper_func;
+        }
+
+        // make the column name the same, so parent can recognize it
+        upper_aggr_expr.push(new_aggr_expr.alias(col_name));
+    }
+    let mut new_aggr = aggr.clone();
+    new_aggr.aggr_expr = upper_aggr_expr;
+    // group by expr also need alias to avoid duplicated computing
+
+    let mut new_group_expr = new_aggr.group_expr.clone();
+    for expr in &mut new_group_expr {
+        if let Expr::Column(_) = expr {
+            // already a column, no need to change
+            continue;
+        }
+        let col_name = expr.name_for_alias()?;
+        let input_column =
+            Expr::Column(datafusion_common::Column::new_unqualified(col_name.clone()));
+        *expr = input_column.alias(col_name);
+    }
+    new_aggr.group_expr = new_group_expr;
+    // return the new logical plan
+    Ok(LogicalPlan::Aggregate(new_aggr))
+}
+
+/// Check if the given aggregate expression is steppable.
+/// As in if it can be split into multiple steps:
+/// i.e. on datanode first call `state(input)` then
+/// on frontend call `calc(merge(state))` to get the final result.
+///
+pub fn is_all_aggr_exprs_steppable(aggr_exprs: &[Expr]) -> bool {
+    let step_action = HashSet::from([
+        "sum",
+        "count",
+        "min",
+        "max",
+        "first_value",
+        "last_value",
+        UDDSKETCH_STATE_NAME,
+        HLL_NAME,
+    ]);
+    aggr_exprs.iter().all(|expr| {
+        if let Some(aggr_func) = get_aggr_func(expr) {
+            if aggr_func.distinct {
+                // Distinct aggregate functions are not steppable(yet).
+                return false;
+            }
+            step_action.contains(aggr_func.func.name())
+        } else {
+            false
+        }
+    })
+}
+
+pub fn get_aggr_func(expr: &Expr) -> Option<&datafusion_expr::expr::AggregateFunction> {
+    let mut expr_ref = expr;
+    while let Expr::Alias(alias) = expr_ref {
+        expr_ref = &alias.expr;
+    }
+    if let Expr::AggregateFunction(aggr_func) = expr_ref {
+        Some(aggr_func)
+    } else {
+        None
+    }
+}
+
+pub fn get_aggr_func_mut(expr: &mut Expr) -> Option<&mut datafusion_expr::expr::AggregateFunction> {
+    let mut expr_ref = expr;
+    while let Expr::Alias(alias) = expr_ref {
+        expr_ref = &mut alias.expr;
+    }
+    if let Expr::AggregateFunction(aggr_func) = expr_ref {
+        Some(aggr_func)
+    } else {
+        None
+    }
+}
+
 #[allow(dead_code)]
 pub enum Commutativity {
     Commutative,
     PartialCommutative,
     ConditionalCommutative(Option<Transformer>),
-    TransformedCommutative(Option<Transformer>),
+    TransformedCommutative {
+        transformer: Option<Transformer>,
+        /// whether the transformer changes the child to parent
+        expand_on_parent: bool,
+    },
     NonCommutative,
     Unimplemented,
     /// For unrelated plans like DDL
@@ -55,7 +203,18 @@ impl Categorizer {
             LogicalPlan::Filter(filter) => Self::check_expr(&filter.predicate),
             LogicalPlan::Window(_) => Commutativity::Unimplemented,
             LogicalPlan::Aggregate(aggr) => {
-                if !Self::check_partition(&aggr.group_expr, &partition_cols) {
+                let is_all_steppable = is_all_aggr_exprs_steppable(&aggr.aggr_expr);
+                let is_partition = Self::check_partition(&aggr.group_expr, &partition_cols);
+                if !is_partition && is_all_steppable {
+                    debug!("Plan is steppable: {plan}");
+                    return Commutativity::TransformedCommutative {
+                        transformer: Some(Arc::new(|plan: &LogicalPlan| {
+                            step_aggr_to_upper_aggr(plan).ok()
+                        })),
+                        expand_on_parent: true,
+                    };
+                }
+                if !is_partition {
                     return Commutativity::NonCommutative;
                 }
                 for expr in &aggr.aggr_expr {

tests/cases/distributed/explain/step_aggr.result

@@ -0,0 +1,409 @@
+CREATE TABLE integers(
+    host STRING,
+    i BIGINT,
+    ts TIMESTAMP TIME INDEX
+) PARTITION ON COLUMNS (host) (
+    host < '550-A',
+    host >= '550-A'
+    AND host < '550-W',
+    host >= '550-W'
+);
+
+Affected Rows: 0
+
+INSERT INTO integers (host, i, ts) VALUES
+    ('550-A', 1, '2023-01-01 00:00:00'),
+    ('550-A', 2, '2023-01-01 01:00:00'),
+    ('550-W', 3, '2023-01-01 02:00:00'),
+    ('550-W', 4, '2023-01-01 03:00:00');
+
+Affected Rows: 4
+
+-- count
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers;
+
++---------------+-------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]                         |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[count(integers.i)]]          |
+|               |   TableScan: integers]                                                                                |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[count(integers.i)]                                           |
+|               |   CoalescePartitionsExec                                                                              |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[count(integers.i)]                                     |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]                |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[count(integers.i)]] |
+|               |   TableScan: integers]                                                                                  |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[count(integers.i)]                        |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[count(integers.i)]                           |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-----------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                      |
++---------------+-----------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: count(integers.i)                                                                             |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]                |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[count(integers.i)]] |
+|               |   TableScan: integers]                                                                                    |
+| physical_plan | ProjectionExec: expr=[count(integers.i)@1 as count(integers.i)]                                           |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[count(integers.i)]                        |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[count(integers.i)]                           |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                           |
++---------------+-----------------------------------------------------------------------------------------------------------+
+
+-- sum
+EXPLAIN SELECT
+    sum(i)
+FROM
+    integers;
+
++---------------+-------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[sum(sum(integers.i)) AS sum(integers.i)]]                             |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[sum(integers.i)]]            |
+|               |   TableScan: integers]                                                                                |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[sum(integers.i)]                                             |
+|               |   CoalescePartitionsExec                                                                              |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[sum(integers.i)]                                       |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    sum(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[sum(sum(integers.i)) AS sum(integers.i)]]                    |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[sum(integers.i)]]   |
+|               |   TableScan: integers]                                                                                  |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[sum(integers.i)]                          |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[sum(integers.i)]                             |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    sum(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-----------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                      |
++---------------+-----------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: sum(integers.i)                                                                               |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[sum(sum(integers.i)) AS sum(integers.i)]]                    |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[sum(integers.i)]]   |
+|               |   TableScan: integers]                                                                                    |
+| physical_plan | ProjectionExec: expr=[sum(integers.i)@1 as sum(integers.i)]                                               |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[sum(integers.i)]                          |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[sum(integers.i)]                             |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                           |
++---------------+-----------------------------------------------------------------------------------------------------------+
+
+-- min
+EXPLAIN SELECT
+    min(i)
+FROM
+    integers;
+
++---------------+-------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[min(min(integers.i)) AS min(integers.i)]]                             |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[min(integers.i)]]            |
+|               |   TableScan: integers]                                                                                |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[min(integers.i)]                                             |
+|               |   CoalescePartitionsExec                                                                              |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[min(integers.i)]                                       |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    min(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[min(min(integers.i)) AS min(integers.i)]]                    |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[min(integers.i)]]   |
+|               |   TableScan: integers]                                                                                  |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[min(integers.i)]                          |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[min(integers.i)]                             |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    min(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-----------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                      |
++---------------+-----------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: min(integers.i)                                                                               |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[min(min(integers.i)) AS min(integers.i)]]                    |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[min(integers.i)]]   |
+|               |   TableScan: integers]                                                                                    |
+| physical_plan | ProjectionExec: expr=[min(integers.i)@1 as min(integers.i)]                                               |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[min(integers.i)]                          |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[min(integers.i)]                             |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                           |
++---------------+-----------------------------------------------------------------------------------------------------------+
+
+-- max
+EXPLAIN SELECT
+    max(i)
+FROM
+    integers;
+
++---------------+-------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[max(max(integers.i)) AS max(integers.i)]]                             |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[max(integers.i)]]            |
+|               |   TableScan: integers]                                                                                |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[max(integers.i)]                                             |
+|               |   CoalescePartitionsExec                                                                              |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[max(integers.i)]                                       |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    max(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[max(max(integers.i)) AS max(integers.i)]]                    |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[max(integers.i)]]   |
+|               |   TableScan: integers]                                                                                  |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[max(integers.i)]                          |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[max(integers.i)]                             |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    max(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-----------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                      |
++---------------+-----------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: max(integers.i)                                                                               |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[max(max(integers.i)) AS max(integers.i)]]                    |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[max(integers.i)]]   |
+|               |   TableScan: integers]                                                                                    |
+| physical_plan | ProjectionExec: expr=[max(integers.i)@1 as max(integers.i)]                                               |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[max(integers.i)]                          |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[max(integers.i)]                             |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ] |
+|               |                                                                                                           |
++---------------+-----------------------------------------------------------------------------------------------------------+
+
+-- uddsketch_state
+EXPLAIN SELECT
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers;
+
++---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                                                                                       |
++---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[uddsketch_merge(Int64(128), Float64(0.01), uddsketch_state(Int64(128),Float64(0.01),integers.i)) AS uddsketch_state(Int64(128),Float64(0.01),integers.i)]] |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[uddsketch_state(Int64(128), Float64(0.01), CAST(integers.i AS Float64))]]                                         |
+|               |   TableScan: integers]                                                                                                                                                                     |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                             |
+|               |   CoalescePartitionsExec                                                                                                                                                                   |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                       |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]                                                                                      |
+|               |                                                                                                                                                                                            |
++---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[uddsketch_merge(Int64(128), Float64(0.01), uddsketch_state(Int64(128),Float64(0.01),integers.i)) AS uddsketch_state(Int64(128),Float64(0.01),integers.i)]] |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[uddsketch_state(Int64(128), Float64(0.01), CAST(integers.i AS Float64))]]                                         |
+|               |   TableScan: integers]                                                                                                                                                                                |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                   |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                                                                                                                         |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                                                                                                               |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                      |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]                                                                                               |
+|               |                                                                                                                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: uddsketch_state(Int64(128),Float64(0.01),integers.i)                                                                                                                                        |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[uddsketch_merge(Int64(128), Float64(0.01), uddsketch_state(Int64(128),Float64(0.01),integers.i)) AS uddsketch_state(Int64(128),Float64(0.01),integers.i)]] |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[uddsketch_state(Int64(128), Float64(0.01), CAST(integers.i AS Float64))]]                                         |
+|               |   TableScan: integers]                                                                                                                                                                                  |
+| physical_plan | ProjectionExec: expr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)@1 as uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                   |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                   |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                                                                                                                         |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                                                                                                               |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[uddsketch_state(Int64(128),Float64(0.01),integers.i)]                                                                                      |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]                                                                                               |
+|               |                                                                                                                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+
+-- hll
+EXPLAIN SELECT
+    hll(i)
+FROM
+    integers;
+
++---------------+----------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                     |
++---------------+----------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[hll_merge(hll(integers.i)) AS hll(integers.i)]]                          |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[hll(CAST(integers.i AS Utf8))]] |
+|               |   TableScan: integers]                                                                                   |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[hll(integers.i)]                                                |
+|               |   CoalescePartitionsExec                                                                                 |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[hll(integers.i)]                                          |
+|               |       MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]    |
+|               |                                                                                                          |
++---------------+----------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    hll(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                |
++---------------+---------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[hll_merge(hll(integers.i)) AS hll(integers.i)]]                          |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[hll(CAST(integers.i AS Utf8))]] |
+|               |   TableScan: integers]                                                                                              |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[hll(integers.i)]                                      |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                                       |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                             |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[hll(integers.i)]                                         |
+|               |         MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]             |
+|               |                                                                                                                     |
++---------------+---------------------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    hll(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+-----------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                  |
++---------------+-----------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Projection: hll(integers.i)                                                                                           |
+|               |   Aggregate: groupBy=[[integers.ts]], aggr=[[hll_merge(hll(integers.i)) AS hll(integers.i)]]                          |
+|               |     MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[hll(CAST(integers.i AS Utf8))]] |
+|               |   TableScan: integers]                                                                                                |
+| physical_plan | ProjectionExec: expr=[hll(integers.i)@1 as hll(integers.i)]                                                           |
+|               |   AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[hll(integers.i)]                                      |
+|               |     CoalesceBatchesExec: target_batch_size=8192                                                                       |
+|               |       RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                             |
+|               |         AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[hll(integers.i)]                                         |
+|               |           MergeScanExec: peers=[4402341478400(1025, 0), 4402341478401(1025, 1), 4402341478402(1025, 2), ]             |
+|               |                                                                                                                       |
++---------------+-----------------------------------------------------------------------------------------------------------------------+
+
+DROP TABLE integers;
+
+Affected Rows: 0
+

tests/cases/distributed/explain/step_aggr.sql

@@ -0,0 +1,144 @@
+CREATE TABLE integers(
+    host STRING,
+    i BIGINT,
+    ts TIMESTAMP TIME INDEX
+) PARTITION ON COLUMNS (host) (
+    host < '550-A',
+    host >= '550-A'
+    AND host < '550-W',
+    host >= '550-W'
+);
+
+INSERT INTO integers (host, i, ts) VALUES
+    ('550-A', 1, '2023-01-01 00:00:00'),
+    ('550-A', 2, '2023-01-01 01:00:00'),
+    ('550-W', 3, '2023-01-01 02:00:00'),
+    ('550-W', 4, '2023-01-01 03:00:00');
+
+-- count
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+-- sum
+EXPLAIN SELECT
+    sum(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    sum(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    sum(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+-- min
+EXPLAIN SELECT
+    min(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    min(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    min(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+-- max
+EXPLAIN SELECT
+    max(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    max(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    max(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+-- uddsketch_state
+EXPLAIN SELECT
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    uddsketch_state(128, 0.01, i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+-- hll
+EXPLAIN SELECT
+    hll(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    hll(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    hll(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+DROP TABLE integers;

tests/cases/distributed/explain/step_aggr_basic.result

@@ -0,0 +1,80 @@
+CREATE TABLE integers(
+    host STRING,
+    i BIGINT,
+    ts TIMESTAMP TIME INDEX
+) PARTITION ON COLUMNS (host) (
+    host < '550-A',
+    host >= '550-A'
+    AND host < '550-W',
+    host >= '550-W'
+);
+
+Affected Rows: 0
+
+-- count
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers;
+
++---------------+-------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                  |
++---------------+-------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]                         |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[]], aggr=[[count(integers.i)]]          |
+|               |   TableScan: integers]                                                                                |
+| physical_plan | AggregateExec: mode=Final, gby=[], aggr=[count(integers.i)]                                           |
+|               |   CoalescePartitionsExec                                                                              |
+|               |     AggregateExec: mode=Partial, gby=[], aggr=[count(integers.i)]                                     |
+|               |       MergeScanExec: peers=[4398046511104(1024, 0), 4398046511105(1024, 1), 4398046511106(1024, 2), ] |
+|               |                                                                                                       |
++---------------+-------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    ts,
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
++---------------+---------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                    |
++---------------+---------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[integers.ts]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]                |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[integers.ts]], aggr=[[count(integers.i)]] |
+|               |   TableScan: integers]                                                                                  |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[ts@0 as ts], aggr=[count(integers.i)]                        |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                           |
+|               |     RepartitionExec: partitioning=Hash([ts@0], 20), input_partitions=20                                 |
+|               |       AggregateExec: mode=Partial, gby=[ts@0 as ts], aggr=[count(integers.i)]                           |
+|               |         MergeScanExec: peers=[4398046511104(1024, 0), 4398046511105(1024, 1), 4398046511106(1024, 2), ] |
+|               |                                                                                                         |
++---------------+---------------------------------------------------------------------------------------------------------+
+
+EXPLAIN SELECT
+    date_bin('1 hour'::INTERVAL, ts),
+    count(i)
+FROM
+    integers
+GROUP BY
+    date_bin('1 hour'::INTERVAL, ts);
+
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| plan_type     | plan                                                                                                                                                              |
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| logical_plan  | Aggregate: groupBy=[[date_bin(Utf8("1 hour"),integers.ts) AS date_bin(Utf8("1 hour"),integers.ts)]], aggr=[[sum(count(integers.i)) AS count(integers.i)]]         |
+|               |   MergeScan [is_placeholder=false, input=Aggregate: groupBy=[[date_bin(CAST(Utf8("1 hour") AS Interval(MonthDayNano)), integers.ts)]], aggr=[[count(integers.i)]] |
+|               |   TableScan: integers]                                                                                                                                            |
+| physical_plan | AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("1 hour"),integers.ts)@0 as date_bin(Utf8("1 hour"),integers.ts)], aggr=[count(integers.i)]              |
+|               |   CoalesceBatchesExec: target_batch_size=8192                                                                                                                     |
+|               |     RepartitionExec: partitioning=Hash([date_bin(Utf8("1 hour"),integers.ts)@0], 20), input_partitions=20                                                         |
+|               |       AggregateExec: mode=Partial, gby=[date_bin(Utf8("1 hour"),integers.ts)@0 as date_bin(Utf8("1 hour"),integers.ts)], aggr=[count(integers.i)]                 |
+|               |         MergeScanExec: peers=[4398046511104(1024, 0), 4398046511105(1024, 1), 4398046511106(1024, 2), ]                                                           |
+|               |                                                                                                                                                                   |
++---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+
+DROP TABLE integers;
+
+Affected Rows: 0
+

tests/cases/distributed/explain/step_aggr_basic.sql

@@ -0,0 +1,34 @@
+CREATE TABLE integers(
+    host STRING,
+    i BIGINT,
+    ts TIMESTAMP TIME INDEX
+) PARTITION ON COLUMNS (host) (
+    host < '550-A',
+    host >= '550-A'
+    AND host < '550-W',
+    host >= '550-W'
+);
+
+-- count
+EXPLAIN SELECT
+    count(i)
+FROM
+    integers;
+
+EXPLAIN SELECT
+    ts,
+    count(i)
+FROM
+    integers
+GROUP BY
+    ts;
+
+EXPLAIN SELECT
+    date_bin('1 hour'::INTERVAL, ts),
+    count(i)
+FROM
+    integers
+GROUP BY
+    date_bin('1 hour'::INTERVAL, ts);
+
+DROP TABLE integers;