refactor: unify function registry (Part 2)

Signed-off-by: Zhenchi <zhongzc_arch@outlook.com>

Cargo.lock

@@ -2336,6 +2336,7 @@ dependencies = [
  "num-traits",
  "once_cell",
  "paste",
+ "promql",
  "s2",
  "serde",
  "serde_json",

config/config.md

@@ -232,7 +232,6 @@
 | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
-| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for frontend side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |
@@ -405,7 +404,6 @@
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
-| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for datanode side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |

config/datanode.example.toml

@@ -44,13 +44,6 @@ runtime_size = 8
 max_recv_message_size = "512MB"
 ## The maximum send message size for gRPC server.
 max_send_message_size = "512MB"
-## Compression mode for datanode side Arrow IPC service. Available options:
-## - `none`: disable all compression
-## - `transport`: only enable gRPC transport compression (zstd)
-## - `arrow_ipc`: only enable Arrow IPC compression (lz4)
-## - `all`: enable all compression.
-## Default to `none`
-flight_compression = "arrow_ipc"
 
 ## gRPC server TLS options, see `mysql.tls` section.
 [grpc.tls]

config/frontend.example.toml

@@ -54,13 +54,6 @@ bind_addr = "127.0.0.1:4001"
 server_addr = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
-## Compression mode for frontend side Arrow IPC service. Available options:
-## - `none`: disable all compression
-## - `transport`: only enable gRPC transport compression (zstd)
-## - `arrow_ipc`: only enable Arrow IPC compression (lz4)
-## - `all`: enable all compression.
-## Default to `none`
-flight_compression = "arrow_ipc"
 
 ## gRPC server TLS options, see `mysql.tls` section.
 [grpc.tls]

src/client/src/client.rs

@@ -162,23 +162,12 @@ impl Client {
             .as_bytes() as usize
     }
 
-    pub fn make_flight_client(
-        &self,
-        send_compression: bool,
-        accept_compression: bool,
-    ) -> Result<FlightClient> {
+    pub fn make_flight_client(&self) -> Result<FlightClient> {
         let (addr, channel) = self.find_channel()?;
 
-        let mut client = FlightServiceClient::new(channel)
+        let client = FlightServiceClient::new(channel)
             .max_decoding_message_size(self.max_grpc_recv_message_size())
             .max_encoding_message_size(self.max_grpc_send_message_size());
-        // todo(hl): support compression methods.
-        if send_compression {
-            client = client.send_compressed(CompressionEncoding::Zstd);
-        }
-        if accept_compression {
-            client = client.accept_compressed(CompressionEncoding::Zstd);
-        }
 
         Ok(FlightClient { addr, client })
     }

src/client/src/client_manager.rs

@@ -49,16 +49,7 @@ impl NodeManager for NodeClients {
     async fn datanode(&self, datanode: &Peer) -> DatanodeRef {
         let client = self.get_client(datanode).await;
 
-        let ChannelConfig {
-            send_compression,
-            accept_compression,
-            ..
-        } = self.channel_manager.config();
-        Arc::new(RegionRequester::new(
-            client,
-            *send_compression,
-            *accept_compression,
-        ))
+        Arc::new(RegionRequester::new(client))
     }
 
     async fn flownode(&self, flownode: &Peer) -> FlownodeRef {

src/client/src/database.rs

@@ -287,7 +287,7 @@ impl Database {
         let mut request = tonic::Request::new(request);
         Self::put_hints(request.metadata_mut(), hints)?;
 
-        let mut client = self.client.make_flight_client(false, false)?;
+        let mut client = self.client.make_flight_client()?;
 
         let response = client.mut_inner().do_get(request).await.or_else(|e| {
             let tonic_code = e.code();
@@ -409,7 +409,7 @@ impl Database {
             MetadataValue::from_str(db_to_put).context(InvalidTonicMetadataValueSnafu)?,
         );
 
-        let mut client = self.client.make_flight_client(false, false)?;
+        let mut client = self.client.make_flight_client()?;
         let response = client.mut_inner().do_put(request).await?;
         let response = response
             .into_inner()

src/client/src/region.rs

@@ -46,8 +46,6 @@ use crate::{metrics, Client, Error};
 #[derive(Debug)]
 pub struct RegionRequester {
     client: Client,
-    send_compression: bool,
-    accept_compression: bool,
 }
 
 #[async_trait]
@@ -91,18 +89,12 @@ impl Datanode for RegionRequester {
 }
 
 impl RegionRequester {
-    pub fn new(client: Client, send_compression: bool, accept_compression: bool) -> Self {
-        Self {
-            client,
-            send_compression,
-            accept_compression,
-        }
+    pub fn new(client: Client) -> Self {
+        Self { client }
     }
 
     pub async fn do_get_inner(&self, ticket: Ticket) -> Result<SendableRecordBatchStream> {
-        let mut flight_client = self
-            .client
-            .make_flight_client(self.send_compression, self.accept_compression)?;
+        let mut flight_client = self.client.make_flight_client()?;
         let response = flight_client
             .mut_inner()
             .do_get(ticket)

src/cmd/src/frontend.rs

@@ -364,16 +364,12 @@ impl StartCommand {
 
         // frontend to datanode need not timeout.
         // Some queries are expected to take long time.
-        let mut channel_config = ChannelConfig {
+        let channel_config = ChannelConfig {
             timeout: None,
             tcp_nodelay: opts.datanode.client.tcp_nodelay,
             connect_timeout: Some(opts.datanode.client.connect_timeout),
             ..Default::default()
         };
-        if opts.grpc.flight_compression.transport_compression() {
-            channel_config.accept_compression = true;
-            channel_config.send_compression = true;
-        }
         let client = NodeClients::new(channel_config);
 
         let instance = FrontendBuilder::new(

src/common/function/Cargo.toml

@@ -47,6 +47,7 @@ num = "0.4"
 num-traits = "0.2"
 once_cell.workspace = true
 paste.workspace = true
+promql.workspace = true
 s2 = { version = "0.0.12", optional = true }
 serde.workspace = true
 serde_json.workspace = true

src/common/function/src/function_registry.rs

@@ -39,7 +39,7 @@ use crate::system::SystemFunction;
 
 #[derive(Default)]
 pub struct FunctionRegistry {
-    functions: RwLock<HashMap<String, ScalarFunctionFactory>>,
+    scalar_functions: RwLock<HashMap<String, ScalarFunctionFactory>>,
     async_functions: RwLock<HashMap<String, AsyncFunctionRef>>,
     aggregate_functions: RwLock<HashMap<String, AggregateUDF>>,
 }
@@ -48,7 +48,7 @@ impl FunctionRegistry {
     pub fn register(&self, func: impl Into<ScalarFunctionFactory>) {
         let func = func.into();
         let _ = self
-            .functions
+            .scalar_functions
             .write()
             .unwrap()
             .insert(func.name().to_string(), func);
@@ -87,13 +87,17 @@ impl FunctionRegistry {
             .collect()
     }
 
-    #[cfg(test)]
-    pub fn get_function(&self, name: &str) -> Option<ScalarFunctionFactory> {
-        self.functions.read().unwrap().get(name).cloned()
+    pub fn get_scalar_function(&self, name: &str) -> Option<ScalarFunctionFactory> {
+        self.scalar_functions.read().unwrap().get(name).cloned()
     }
 
     pub fn scalar_functions(&self) -> Vec<ScalarFunctionFactory> {
-        self.functions.read().unwrap().values().cloned().collect()
+        self.scalar_functions
+            .read()
+            .unwrap()
+            .values()
+            .cloned()
+            .collect()
     }
 
     pub fn aggregate_functions(&self) -> Vec<AggregateUDF> {
@@ -144,6 +148,11 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
     // Approximate functions
     ApproximateFunction::register(&function_registry);
 
+    // PromQL aggregate functions
+    for aggr in promql::functions::aggr_funcs() {
+        function_registry.register_aggr(aggr);
+    }
+
     Arc::new(function_registry)
 });
 
@@ -156,10 +165,10 @@ mod tests {
     fn test_function_registry() {
         let registry = FunctionRegistry::default();
 
-        assert!(registry.get_function("test_and").is_none());
+        assert!(registry.get_scalar_function("test_and").is_none());
         assert!(registry.scalar_functions().is_empty());
         registry.register_scalar(TestAndFunction);
-        let _ = registry.get_function("test_and").unwrap();
+        let _ = registry.get_scalar_function("test_and").unwrap();
         assert_eq!(1, registry.scalar_functions().len());
     }
 }

src/common/grpc/src/channel_manager.rs

@@ -296,8 +296,6 @@ pub struct ChannelConfig {
     pub max_recv_message_size: ReadableSize,
     // Max gRPC sending(encoding) message size
     pub max_send_message_size: ReadableSize,
-    pub send_compression: bool,
-    pub accept_compression: bool,
 }
 
 impl Default for ChannelConfig {
@@ -318,8 +316,6 @@ impl Default for ChannelConfig {
             client_tls: None,
             max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
             max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
-            send_compression: false,
-            accept_compression: false,
         }
     }
 }
@@ -570,8 +566,6 @@ mod tests {
                 client_tls: None,
                 max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
                 max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
-                send_compression: false,
-                accept_compression: false,
             },
             default_cfg
         );
@@ -616,8 +610,6 @@ mod tests {
                 }),
                 max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
                 max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
-                send_compression: false,
-                accept_compression: false,
             },
             cfg
         );

src/common/grpc/src/flight.rs

@@ -64,7 +64,6 @@ impl Default for FlightEncoder {
 }
 
 impl FlightEncoder {
-    /// Creates new [FlightEncoder] with compression disabled.
     pub fn with_compression_disabled() -> Self {
         let write_options = writer::IpcWriteOptions::default()
             .try_with_compression(None)

src/datanode/src/datanode.rs

@@ -372,7 +372,6 @@ impl DatanodeBuilder {
             opts.max_concurrent_queries,
             //TODO: revaluate the hardcoded timeout on the next version of datanode concurrency limiter.
             Duration::from_millis(100),
-            opts.grpc.flight_compression,
         );
 
         let object_store_manager = Self::build_object_store_manager(&opts.storage).await?;

src/datanode/src/region_server.rs

@@ -50,7 +50,6 @@ use query::QueryEngineRef;
 use servers::error::{self as servers_error, ExecuteGrpcRequestSnafu, Result as ServerResult};
 use servers::grpc::flight::{FlightCraft, FlightRecordBatchStream, TonicStream};
 use servers::grpc::region_server::RegionServerHandler;
-use servers::grpc::FlightCompression;
 use session::context::{QueryContextBuilder, QueryContextRef};
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::metric_engine_consts::{
@@ -81,7 +80,6 @@ use crate::event_listener::RegionServerEventListenerRef;
 #[derive(Clone)]
 pub struct RegionServer {
     inner: Arc<RegionServerInner>,
-    flight_compression: FlightCompression,
 }
 
 pub struct RegionStat {
@@ -95,7 +93,6 @@ impl RegionServer {
         query_engine: QueryEngineRef,
         runtime: Runtime,
         event_listener: RegionServerEventListenerRef,
-        flight_compression: FlightCompression,
     ) -> Self {
         Self::with_table_provider(
             query_engine,
@@ -104,7 +101,6 @@ impl RegionServer {
             Arc::new(DummyTableProviderFactory),
             0,
             Duration::from_millis(0),
-            flight_compression,
         )
     }
 
@@ -115,7 +111,6 @@ impl RegionServer {
         table_provider_factory: TableProviderFactoryRef,
         max_concurrent_queries: usize,
         concurrent_query_limiter_timeout: Duration,
-        flight_compression: FlightCompression,
     ) -> Self {
         Self {
             inner: Arc::new(RegionServerInner::new(
@@ -128,7 +123,6 @@ impl RegionServer {
                     concurrent_query_limiter_timeout,
                 ),
             )),
-            flight_compression,
         }
     }
 
@@ -542,11 +536,7 @@ impl FlightCraft for RegionServer {
             .trace(tracing_context.attach(info_span!("RegionServer::handle_read")))
             .await?;
 
-        let stream = Box::pin(FlightRecordBatchStream::new(
-            result,
-            tracing_context,
-            self.flight_compression,
-        ));
+        let stream = Box::pin(FlightRecordBatchStream::new(result, tracing_context));
         Ok(Response::new(stream))
     }
 }

src/datanode/src/tests.rs

@@ -28,7 +28,6 @@ use query::dataframe::DataFrame;
 use query::planner::LogicalPlanner;
 use query::query_engine::{DescribeResult, QueryEngineState};
 use query::{QueryEngine, QueryEngineContext};
-use servers::grpc::FlightCompression;
 use session::context::QueryContextRef;
 use store_api::metadata::RegionMetadataRef;
 use store_api::region_engine::{
@@ -98,7 +97,6 @@ pub fn mock_region_server() -> RegionServer {
         Arc::new(MockQueryEngine),
         Runtime::builder().build().unwrap(),
         Box::new(NoopRegionServerEventListener),
-        FlightCompression::default(),
     )
 }
 

src/flow/src/adapter.rs

@@ -316,7 +316,7 @@ impl StreamingEngine {
             );
 
             METRIC_FLOW_ROWS
-                .with_label_values(&["out-streaming"])
+                .with_label_values(&["out"])
                 .inc_by(total_rows as u64);
 
             let now = self.tick_manager.tick();

src/flow/src/adapter/flownode_impl.rs

@@ -46,7 +46,7 @@ use crate::error::{
     IllegalCheckTaskStateSnafu, InsertIntoFlowSnafu, InternalSnafu, JoinTaskSnafu, ListFlowsSnafu,
     NoAvailableFrontendSnafu, SyncCheckTaskSnafu, UnexpectedSnafu,
 };
-use crate::metrics::{METRIC_FLOW_ROWS, METRIC_FLOW_TASK_COUNT};
+use crate::metrics::METRIC_FLOW_TASK_COUNT;
 use crate::repr::{self, DiffRow};
 use crate::{Error, FlowId};
 
@@ -689,9 +689,6 @@ impl FlowEngine for FlowDualEngine {
         let mut to_stream_engine = Vec::with_capacity(request.requests.len());
         let mut to_batch_engine = request.requests;
 
-        let mut batching_row_cnt = 0;
-        let mut streaming_row_cnt = 0;
-
         {
             // not locking this, or recover flows will be starved when also handling flow inserts
             let src_table2flow = self.src_table2flow.read().await;
@@ -701,11 +698,9 @@ impl FlowEngine for FlowDualEngine {
                 let is_in_stream = src_table2flow.in_stream(table_id);
                 let is_in_batch = src_table2flow.in_batch(table_id);
                 if is_in_stream {
-                    streaming_row_cnt += req.rows.as_ref().map(|rs| rs.rows.len()).unwrap_or(0);
                     to_stream_engine.push(req.clone());
                 }
                 if is_in_batch {
-                    batching_row_cnt += req.rows.as_ref().map(|rs| rs.rows.len()).unwrap_or(0);
                     return true;
                 }
                 if !is_in_batch && !is_in_stream {
@@ -718,14 +713,6 @@ impl FlowEngine for FlowDualEngine {
             // can't use drop due to https://github.com/rust-lang/rust/pull/128846
         }
 
-        METRIC_FLOW_ROWS
-            .with_label_values(&["in-streaming"])
-            .inc_by(streaming_row_cnt as u64);
-
-        METRIC_FLOW_ROWS
-            .with_label_values(&["in-batching"])
-            .inc_by(batching_row_cnt as u64);
-
         let streaming_engine = self.streaming_engine.clone();
         let stream_handler: JoinHandle<Result<(), Error>> =
             common_runtime::spawn_global(async move {

src/flow/src/batching_mode/engine.rs

@@ -42,7 +42,6 @@ use crate::error::{
     ExternalSnafu, FlowAlreadyExistSnafu, FlowNotFoundSnafu, TableNotFoundMetaSnafu,
     UnexpectedSnafu, UnsupportedSnafu,
 };
-use crate::metrics::METRIC_FLOW_ROWS;
 use crate::{CreateFlowArgs, Error, FlowId, TableName};
 
 /// Batching mode Engine, responsible for driving all the batching mode tasks
@@ -156,10 +155,6 @@ impl BatchingEngine {
                         let Some(expr) = &task.config.time_window_expr else {
                             continue;
                         };
-                        let row_cnt: usize = entry.iter().map(|rows| rows.rows.len()).sum();
-                        METRIC_FLOW_ROWS
-                            .with_label_values(&[&format!("{}-batching-in", task.config.flow_id)])
-                            .inc_by(row_cnt as u64);
                         let involved_time_windows = expr.handle_rows(entry.clone()).await?;
                         let mut state = task.state.write().unwrap();
                         state

src/flow/src/batching_mode/state.rs

@@ -32,7 +32,6 @@ 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_WAIT_TIME,
 };
 use crate::{Error, FlowId};
 
@@ -91,28 +90,20 @@ impl TaskState {
         time_window_size: &Option<Duration>,
         max_timeout: Option<Duration>,
     ) -> Instant {
-        // = last query duration, capped by [max(min_run_interval, time_window_size), max_timeout], note at most `max_timeout`
-        let lower = MIN_REFRESH_DURATION.max(time_window_size.unwrap_or_default());
-        let next_duration = self.last_query_duration.max(lower);
-        let next_duration = if let Some(max_timeout) = max_timeout {
-            next_duration.min(max_timeout)
-        } else {
-            next_duration
-        };
+        let last_duration = max_timeout
+            .unwrap_or(self.last_query_duration)
+            .min(self.last_query_duration)
+            .max(MIN_REFRESH_DURATION);
 
-        METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME
-            .with_label_values(&[
-                flow_id.to_string().as_str(),
-                time_window_size
-                    .unwrap_or_default()
-                    .as_secs_f64()
-                    .to_string()
-                    .as_str(),
-            ])
-            .observe(next_duration.as_secs_f64());
+        let next_duration = time_window_size
+            .map(|t| {
+                let half = t / 2;
+                half.max(last_duration)
+            })
+            .unwrap_or(last_duration);
 
         // if have dirty time window, execute immediately to clean dirty time window
-        /*if self.dirty_time_windows.windows.is_empty() {
+        if self.dirty_time_windows.windows.is_empty() {
             self.last_update_time + next_duration
         } else {
             debug!(
@@ -122,9 +113,7 @@ impl TaskState {
                 self.dirty_time_windows.windows
             );
             Instant::now()
-        }*/
-
-        self.last_update_time + next_duration
+        }
     }
 }
 
@@ -217,60 +206,47 @@ impl DirtyTimeWindows {
 
         // get the first `window_cnt` time windows
         let max_time_range = window_size * window_cnt as i32;
+        let nth = {
+            let mut cur_time_range = chrono::Duration::zero();
+            let mut nth_key = None;
+            for (idx, (start, end)) in self.windows.iter().enumerate() {
+                // if time range is too long, stop
+                if cur_time_range > max_time_range {
+                    nth_key = Some(*start);
+                    break;
+                }
 
-        let mut to_be_query = BTreeMap::new();
-        let mut new_windows = self.windows.clone();
-        let mut cur_time_range = chrono::Duration::zero();
-        for (idx, (start, end)) in self.windows.iter().enumerate() {
-            let first_end = start
-                .add_duration(window_size.to_std().unwrap())
-                .context(TimeSnafu)?;
-            let end = end.unwrap_or(first_end);
+                // if we have enough time windows, stop
+                if idx >= window_cnt {
+                    nth_key = Some(*start);
+                    break;
+                }
 
-            // if time range is too long, stop
-            if cur_time_range >= max_time_range {
-                break;
+                if let Some(end) = end {
+                    if let Some(x) = end.sub(start) {
+                        cur_time_range += x;
+                    }
+                }
             }
 
-            // if we have enough time windows, stop
-            if idx >= window_cnt {
-                break;
-            }
+            nth_key
+        };
+        let first_nth = {
+            if let Some(nth) = nth {
+                let mut after = self.windows.split_off(&nth);
+                std::mem::swap(&mut self.windows, &mut after);
 
-            let Some(x) = end.sub(start) else {
-                continue;
-            };
-            if cur_time_range + x <= max_time_range {
-                to_be_query.insert(*start, Some(end));
-                new_windows.remove(start);
-                cur_time_range += x;
+                after
             } else {
-                // too large a window, split it
-                // split at window_size * times
-                let surplus = max_time_range - cur_time_range;
-                let times = surplus.num_seconds() / window_size.num_seconds();
-
-                let split_offset = window_size * times as i32;
-                let split_at = start
-                    .add_duration(split_offset.to_std().unwrap())
-                    .context(TimeSnafu)?;
-                to_be_query.insert(*start, Some(split_at));
-
-                // remove the original window
-                new_windows.remove(start);
-                new_windows.insert(split_at, Some(end));
-                cur_time_range += split_offset;
-                break;
+                std::mem::take(&mut self.windows)
             }
-        }
-
-        self.windows = new_windows;
+        };
 
         METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT
             .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(to_be_query.len() as f64);
+            .observe(first_nth.len() as f64);
 
-        let full_time_range = to_be_query
+        let full_time_range = first_nth
             .iter()
             .fold(chrono::Duration::zero(), |acc, (start, end)| {
                 if let Some(end) = end {
@@ -285,7 +261,7 @@ impl DirtyTimeWindows {
             .observe(full_time_range);
 
         let mut expr_lst = vec![];
-        for (start, end) in to_be_query.into_iter() {
+        for (start, end) in first_nth.into_iter() {
             // align using time window exprs
             let (start, end) = if let Some(ctx) = task_ctx {
                 let Some(time_window_expr) = &ctx.config.time_window_expr else {
@@ -519,64 +495,6 @@ mod test {
                     "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:21' AS TIMESTAMP)))",
                 )
             ),
-            // split range
-            (
-                Vec::from_iter((0..20).map(|i|Timestamp::new_second(i*3)).chain(std::iter::once(
-                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
-                ))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        60
-                    )),
-                ),
-                (
-                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
-                    Some(Timestamp::new_second(
-                        60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1) + 3
-                    )),
-                )]),
-                Some(
-                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
-                )
-            ),
-            // split 2 min into 1 min
-            (
-                Vec::from_iter((0..40).map(|i|Timestamp::new_second(i*3))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        40 * 3
-                    )),
-                )]),
-                Some(
-                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
-                )
-            ),
-            // split 3s + 1min into 3s + 57s
-            (
-                Vec::from_iter(std::iter::once(Timestamp::new_second(0)).chain((0..40).map(|i|Timestamp::new_second(20+i*3)))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        3
-                    )),
-                ),(
-                    Timestamp::new_second(20),
-                    Some(Timestamp::new_second(
-                        140
-                    )),
-                )]),
-                Some(
-                    "(((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:03' AS TIMESTAMP))) OR ((ts >= CAST('1970-01-01 00:00:20' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:17' AS TIMESTAMP))))",
-                )
-            ),
             // expired
             (
                 vec![
@@ -593,8 +511,6 @@ mod test {
                 None
             ),
         ];
-        // let len = testcases.len();
-        // let testcases = testcases[(len - 2)..(len - 1)].to_vec();
         for (lower_bounds, (window_size, expire_lower_bound), expected, expected_filter_expr) in
             testcases
         {

src/flow/src/batching_mode/task.rs

@@ -61,9 +61,7 @@ use crate::error::{
     SubstraitEncodeLogicalPlanSnafu, UnexpectedSnafu,
 };
 use crate::metrics::{
-    METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT, METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME,
-    METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY, METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT,
-    METRIC_FLOW_ROWS,
+    METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME, METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY,
 };
 use crate::{Error, FlowId};
 
@@ -373,9 +371,6 @@ impl BatchingTask {
                 "Flow {flow_id} executed, affected_rows: {affected_rows:?}, elapsed: {:?}",
                 elapsed
             );
-            METRIC_FLOW_ROWS
-                .with_label_values(&[format!("{}-out-batching", flow_id).as_str()])
-                .inc_by(*affected_rows as _);
         } else if let Err(err) = &res {
             warn!(
                 "Failed to execute Flow {flow_id} on frontend {:?}, result: {err:?}, elapsed: {:?} with query: {}",
@@ -415,7 +410,6 @@ impl BatchingTask {
         engine: QueryEngineRef,
         frontend_client: Arc<FrontendClient>,
     ) {
-        let flow_id_str = self.config.flow_id.to_string();
         loop {
             // first check if shutdown signal is received
             // if so, break the loop
@@ -433,9 +427,6 @@ impl BatchingTask {
                     Err(TryRecvError::Empty) => (),
                 }
             }
-            METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT
-                .with_label_values(&[&flow_id_str])
-                .inc();
 
             let new_query = match self.gen_insert_plan(&engine).await {
                 Ok(new_query) => new_query,
@@ -482,9 +473,6 @@ impl BatchingTask {
                 }
                 // TODO(discord9): this error should have better place to go, but for now just print error, also more context is needed
                 Err(err) => {
-                    METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT
-                        .with_label_values(&[&flow_id_str])
-                        .inc();
                     match new_query {
                         Some(query) => {
                             common_telemetry::error!(err; "Failed to execute query for flow={} with query: {query}", self.config.flow_id)

src/flow/src/metrics.rs

@@ -35,13 +35,6 @@ lazy_static! {
         vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,]
     )
     .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME: HistogramVec = register_histogram_vec!(
-        "greptime_flow_batching_engine_wait_time_secs",
-        "flow batching engine wait time between query(seconds)",
-        &["flow_id", "time_window_size"],
-        vec![0.0, 5., 10., 20., 40., 80., 160., 320., 640.,]
-    )
-    .unwrap();
     pub static ref METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY: HistogramVec = register_histogram_vec!(
         "greptime_flow_batching_engine_slow_query_secs",
         "flow batching engine slow query(seconds)",
@@ -65,20 +58,6 @@ lazy_static! {
             vec![60., 4. * 60., 16. * 60., 64. * 60., 256. * 60.]
         )
         .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_START_QUERY_CNT: IntCounterVec =
-        register_int_counter_vec!(
-            "greptime_flow_batching_start_query_count",
-            "flow batching engine started query count",
-            &["flow_id"],
-        )
-        .unwrap();
-    pub static ref METRIC_FLOW_BATCHING_ENGINE_ERROR_CNT: IntCounterVec =
-        register_int_counter_vec!(
-            "greptime_flow_batching_error_count",
-            "flow batching engine error count per flow id",
-            &["flow_id"],
-        )
-        .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/frontend/src/server.rs

@@ -154,7 +154,6 @@ where
             ServerGrpcQueryHandlerAdapter::arc(self.instance.clone()),
             user_provider.clone(),
             runtime,
-            opts.grpc.flight_compression,
         );
 
         let grpc_server = builder

src/promql/src/functions.rs

@@ -34,6 +34,7 @@ pub use changes::Changes;
 use datafusion::arrow::array::{ArrayRef, Float64Array, TimestampMillisecondArray};
 use datafusion::error::DataFusionError;
 use datafusion::physical_plan::ColumnarValue;
+use datafusion_expr::{AggregateUDF, ScalarUDF};
 pub use deriv::Deriv;
 pub use extrapolate_rate::{Delta, Increase, Rate};
 pub use holt_winters::HoltWinters;
@@ -44,6 +45,39 @@ pub use quantile_aggr::{quantile_udaf, QUANTILE_NAME};
 pub use resets::Resets;
 pub use round::Round;
 
+/// Range functions for PromQL.
+pub fn range_funcs() -> Vec<ScalarUDF> {
+    vec![
+        IDelta::<false>::scalar_udf(),
+        IDelta::<true>::scalar_udf(),
+        Rate::scalar_udf(),
+        Increase::scalar_udf(),
+        Delta::scalar_udf(),
+        Resets::scalar_udf(),
+        Changes::scalar_udf(),
+        Deriv::scalar_udf(),
+        Round::scalar_udf(),
+        AvgOverTime::scalar_udf(),
+        MinOverTime::scalar_udf(),
+        MaxOverTime::scalar_udf(),
+        SumOverTime::scalar_udf(),
+        CountOverTime::scalar_udf(),
+        LastOverTime::scalar_udf(),
+        AbsentOverTime::scalar_udf(),
+        PresentOverTime::scalar_udf(),
+        StddevOverTime::scalar_udf(),
+        StdvarOverTime::scalar_udf(),
+        QuantileOverTime::scalar_udf(),
+        PredictLinear::scalar_udf(),
+        HoltWinters::scalar_udf(),
+    ]
+}
+
+/// Aggregate functions for PromQL.
+pub fn aggr_funcs() -> Vec<AggregateUDF> {
+    vec![quantile_udaf()]
+}
+
 /// Extracts an array from a `ColumnarValue`.
 ///
 /// If the `ColumnarValue` is a scalar, it converts it to an array of size 1.

src/promql/src/functions/quantile_aggr.rs

@@ -40,8 +40,8 @@ pub struct QuantileAccumulator {
 
 /// Create a quantile `AggregateUDF` for PromQL quantile operator,
 /// which calculates φ-quantile (0 ≤ φ ≤ 1) over dimensions
-pub fn quantile_udaf() -> Arc<AggregateUDF> {
-    Arc::new(create_udaf(
+pub fn quantile_udaf() -> AggregateUDF {
+    create_udaf(
         QUANTILE_NAME,
         // Input type: (φ, values)
         vec![DataType::Float64, DataType::Float64],
@@ -63,7 +63,7 @@ pub fn quantile_udaf() -> Arc<AggregateUDF> {
             )]
             .into(),
         )]),
-    ))
+    )
 }
 
 impl QuantileAccumulator {

src/query/src/promql/planner.rs

@@ -1948,7 +1948,7 @@ impl PromPlanner {
             token::T_QUANTILE => {
                 let q = Self::get_param_value_as_f64(op, param)?;
                 non_col_args.push(lit(q));
-                quantile_udaf()
+                Arc::new(quantile_udaf())
             }
             token::T_AVG => avg_udaf(),
             token::T_COUNT_VALUES | token::T_COUNT => count_udaf(),

src/query/src/query_engine/default_serializer.rs

@@ -28,11 +28,6 @@ use datafusion::execution::{FunctionRegistry, SessionStateBuilder};
 use datafusion::logical_expr::LogicalPlan;
 use datafusion_expr::UserDefinedLogicalNode;
 use greptime_proto::substrait_extension::MergeScan as PbMergeScan;
-use promql::functions::{
-    quantile_udaf, AbsentOverTime, AvgOverTime, Changes, CountOverTime, Delta, Deriv, IDelta,
-    Increase, LastOverTime, MaxOverTime, MinOverTime, PresentOverTime, Rate, Resets, Round,
-    StddevOverTime, StdvarOverTime, SumOverTime,
-};
 use prost::Message;
 use session::context::QueryContextRef;
 use snafu::ResultExt;
@@ -117,12 +112,15 @@ impl SubstraitPlanDecoder for DefaultPlanDecoder {
         let mut session_state = SessionStateBuilder::new_from_existing(self.session_state.clone())
             .with_catalog_list(catalog_list)
             .build();
+
         // Substrait decoder will look up the UDFs in SessionState, so we need to register them
         // Note: the query context must be passed to set the timezone
         // We MUST register the UDFs after we build the session state, otherwise the UDFs will be lost
         // if they have the same name as the default UDFs or their alias.
         // e.g. The default UDF `to_char()` has an alias `date_format()`, if we register a UDF with the name `date_format()`
         // before we build the session state, the UDF will be lost.
+
+        // Scalar functions
         for func in FUNCTION_REGISTRY.scalar_functions() {
             let udf = func.provide(FunctionContext {
                 query_ctx: self.query_ctx.clone(),
@@ -133,6 +131,15 @@ impl SubstraitPlanDecoder for DefaultPlanDecoder {
                 .context(RegisterUdfSnafu { name: func.name() })?;
         }
 
+        // PromQL range functions
+        for func in promql::functions::range_funcs() {
+            let name = func.name().to_string();
+            session_state
+                .register_udf(Arc::new(func))
+                .context(RegisterUdfSnafu { name })?;
+        }
+
+        // Aggregate functions
         for func in FUNCTION_REGISTRY.aggregate_functions() {
             let name = func.name().to_string();
             session_state
@@ -140,29 +147,6 @@ impl SubstraitPlanDecoder for DefaultPlanDecoder {
                 .context(RegisterUdfSnafu { name })?;
         }
 
-        let _ = session_state.register_udaf(quantile_udaf());
-
-        let _ = session_state.register_udf(Arc::new(IDelta::<false>::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(IDelta::<true>::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Rate::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Increase::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Delta::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Resets::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Changes::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Deriv::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(Round::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(AvgOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(MinOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(MaxOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(SumOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(CountOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(LastOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(AbsentOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(PresentOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(StddevOverTime::scalar_udf()));
-        let _ = session_state.register_udf(Arc::new(StdvarOverTime::scalar_udf()));
-        // TODO(ruihang): add quantile_over_time, predict_linear, holt_winters, round
-
         let logical_plan = DFLogicalSubstraitConvertor
             .decode(message, session_state)
             .await

src/servers/src/grpc.rs

@@ -66,8 +66,6 @@ pub struct GrpcOptions {
     pub max_recv_message_size: ReadableSize,
     /// Max gRPC sending(encoding) message size
     pub max_send_message_size: ReadableSize,
-    /// Compression mode in Arrow Flight service.
-    pub flight_compression: FlightCompression,
     pub runtime_size: usize,
     #[serde(default = "Default::default")]
     pub tls: TlsOption,
@@ -116,7 +114,6 @@ impl Default for GrpcOptions {
             server_addr: String::new(),
             max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
             max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
-            flight_compression: FlightCompression::ArrowIpc,
             runtime_size: 8,
             tls: TlsOption::default(),
         }
@@ -135,30 +132,6 @@ impl GrpcOptions {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Default)]
-#[serde(rename_all = "snake_case")]
-pub enum FlightCompression {
-    /// Disable all compression in Arrow Flight service.
-    #[default]
-    None,
-    /// Enable only transport layer compression (zstd).
-    Transport,
-    /// Enable only payload compression (lz4)
-    ArrowIpc,
-    /// Enable all compression.
-    All,
-}
-
-impl FlightCompression {
-    pub fn transport_compression(&self) -> bool {
-        self == &FlightCompression::Transport || self == &FlightCompression::All
-    }
-
-    pub fn arrow_compression(&self) -> bool {
-        self == &FlightCompression::ArrowIpc || self == &FlightCompression::All
-    }
-}
-
 pub struct GrpcServer {
     // states
     shutdown_tx: Mutex<Option<Sender<()>>>,

src/servers/src/grpc/flight.rs

@@ -45,7 +45,7 @@ use tonic::{Request, Response, Status, Streaming};
 use crate::error::{InvalidParameterSnafu, ParseJsonSnafu, Result, ToJsonSnafu};
 pub use crate::grpc::flight::stream::FlightRecordBatchStream;
 use crate::grpc::greptime_handler::{get_request_type, GreptimeRequestHandler};
-use crate::grpc::{FlightCompression, TonicResult};
+use crate::grpc::TonicResult;
 use crate::http::header::constants::GREPTIME_DB_HEADER_NAME;
 use crate::http::AUTHORIZATION_HEADER;
 use crate::{error, hint_headers};
@@ -195,14 +195,9 @@ impl FlightCraft for GreptimeRequestHandler {
             protocol = "grpc",
             request_type = get_request_type(&request)
         );
-        let flight_compression = self.flight_compression;
         async {
             let output = self.handle_request(request, hints).await?;
-            let stream = to_flight_data_stream(
-                output,
-                TracingContext::from_current_span(),
-                flight_compression,
-            );
+            let stream = to_flight_data_stream(output, TracingContext::from_current_span());
             Ok(Response::new(stream))
         }
         .trace(span)
@@ -370,16 +365,14 @@ impl Stream for PutRecordBatchRequestStream {
 fn to_flight_data_stream(
     output: Output,
     tracing_context: TracingContext,
-    flight_compression: FlightCompression,
 ) -> TonicStream<FlightData> {
     match output.data {
         OutputData::Stream(stream) => {
-            let stream = FlightRecordBatchStream::new(stream, tracing_context, flight_compression);
+            let stream = FlightRecordBatchStream::new(stream, tracing_context);
             Box::pin(stream) as _
         }
         OutputData::RecordBatches(x) => {
-            let stream =
-                FlightRecordBatchStream::new(x.as_stream(), tracing_context, flight_compression);
+            let stream = FlightRecordBatchStream::new(x.as_stream(), tracing_context);
             Box::pin(stream) as _
         }
         OutputData::AffectedRows(rows) => {

src/servers/src/grpc/flight/stream.rs

@@ -30,7 +30,6 @@ use tokio::task::JoinHandle;
 
 use crate::error;
 use crate::grpc::flight::TonicResult;
-use crate::grpc::FlightCompression;
 
 #[pin_project(PinnedDrop)]
 pub struct FlightRecordBatchStream {
@@ -42,27 +41,18 @@ pub struct FlightRecordBatchStream {
 }
 
 impl FlightRecordBatchStream {
-    pub fn new(
-        recordbatches: SendableRecordBatchStream,
-        tracing_context: TracingContext,
-        compression: FlightCompression,
-    ) -> Self {
+    pub fn new(recordbatches: SendableRecordBatchStream, tracing_context: TracingContext) -> Self {
         let (tx, rx) = mpsc::channel::<TonicResult<FlightMessage>>(1);
         let join_handle = common_runtime::spawn_global(async move {
             Self::flight_data_stream(recordbatches, tx)
                 .trace(tracing_context.attach(info_span!("flight_data_stream")))
                 .await
         });
-        let encoder = if compression.arrow_compression() {
-            FlightEncoder::default()
-        } else {
-            FlightEncoder::with_compression_disabled()
-        };
         Self {
             rx,
             join_handle,
             done: false,
-            encoder,
+            encoder: FlightEncoder::with_compression_disabled(),
         }
     }
 
@@ -171,11 +161,7 @@ mod test {
         let recordbatches = RecordBatches::try_new(schema.clone(), vec![recordbatch.clone()])
             .unwrap()
             .as_stream();
-        let mut stream = FlightRecordBatchStream::new(
-            recordbatches,
-            TracingContext::default(),
-            FlightCompression::default(),
-        );
+        let mut stream = FlightRecordBatchStream::new(recordbatches, TracingContext::default());
 
         let mut raw_data = Vec::with_capacity(2);
         raw_data.push(stream.next().await.unwrap().unwrap());

src/servers/src/grpc/greptime_handler.rs

@@ -49,7 +49,7 @@ use crate::error::{
     JoinTaskSnafu, NotFoundAuthHeaderSnafu, Result, UnknownHintSnafu,
 };
 use crate::grpc::flight::{PutRecordBatchRequest, PutRecordBatchRequestStream};
-use crate::grpc::{FlightCompression, TonicResult};
+use crate::grpc::TonicResult;
 use crate::metrics;
 use crate::metrics::{METRIC_AUTH_FAILURE, METRIC_SERVER_GRPC_DB_REQUEST_TIMER};
 use crate::query_handler::grpc::ServerGrpcQueryHandlerRef;
@@ -59,7 +59,6 @@ pub struct GreptimeRequestHandler {
     handler: ServerGrpcQueryHandlerRef,
     user_provider: Option<UserProviderRef>,
     runtime: Option<Runtime>,
-    pub(crate) flight_compression: FlightCompression,
 }
 
 impl GreptimeRequestHandler {
@@ -67,13 +66,11 @@ impl GreptimeRequestHandler {
         handler: ServerGrpcQueryHandlerRef,
         user_provider: Option<UserProviderRef>,
         runtime: Option<Runtime>,
-        flight_compression: FlightCompression,
     ) -> Self {
         Self {
             handler,
             user_provider,
             runtime,
-            flight_compression,
         }
     }
 

tests-integration/src/grpc/flight.rs

@@ -34,7 +34,7 @@ mod test {
     use itertools::Itertools;
     use servers::grpc::builder::GrpcServerBuilder;
     use servers::grpc::greptime_handler::GreptimeRequestHandler;
-    use servers::grpc::{FlightCompression, GrpcServerConfig};
+    use servers::grpc::GrpcServerConfig;
     use servers::query_handler::grpc::ServerGrpcQueryHandlerAdapter;
     use servers::server::Server;
 
@@ -94,7 +94,6 @@ mod test {
             )
             .ok(),
             Some(runtime.clone()),
-            FlightCompression::default(),
         );
         let mut grpc_server = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime)
             .flight_handler(Arc::new(greptime_request_handler))
@@ -140,7 +139,8 @@ mod test {
         let schema = record_batches[0].schema.arrow_schema().clone();
 
         let stream = futures::stream::once(async move {
-            let mut schema_data = FlightEncoder::default().encode(FlightMessage::Schema(schema));
+            let mut schema_data =
+                FlightEncoder::with_compression_disabled().encode(FlightMessage::Schema(schema));
             let metadata = DoPutMetadata::new(0);
             schema_data.app_metadata = serde_json::to_vec(&metadata).unwrap().into();
             // first message in "DoPut" stream should carry table name in flight descriptor
@@ -155,7 +155,7 @@ mod test {
             tokio_stream::iter(record_batches)
                 .enumerate()
                 .map(|(i, x)| {
-                    let mut encoder = FlightEncoder::default();
+                    let mut encoder = FlightEncoder::with_compression_disabled();
                     let message = FlightMessage::RecordBatch(x.into_df_record_batch());
                     let mut data = encoder.encode(message);
                     let metadata = DoPutMetadata::new((i + 1) as i64);

tests-integration/src/test_util.rs

@@ -42,7 +42,7 @@ use object_store::test_util::TempFolder;
 use object_store::ObjectStore;
 use servers::grpc::builder::GrpcServerBuilder;
 use servers::grpc::greptime_handler::GreptimeRequestHandler;
-use servers::grpc::{FlightCompression, GrpcOptions, GrpcServer, GrpcServerConfig};
+use servers::grpc::{GrpcOptions, GrpcServer, GrpcServerConfig};
 use servers::http::{HttpOptions, HttpServerBuilder, PromValidationMode};
 use servers::metrics_handler::MetricsHandler;
 use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
@@ -585,7 +585,6 @@ pub async fn setup_grpc_server_with(
         ServerGrpcQueryHandlerAdapter::arc(fe_instance_ref.clone()),
         user_provider.clone(),
         Some(runtime.clone()),
-        FlightCompression::default(),
     );
 
     let flight_handler = Arc::new(greptime_request_handler.clone());

tests-integration/tests/http.rs

@@ -1025,7 +1025,6 @@ bind_addr = "127.0.0.1:4001"
 server_addr = "127.0.0.1:4001"
 max_recv_message_size = "512MiB"
 max_send_message_size = "512MiB"
-flight_compression = "arrow_ipc"
 runtime_size = 8
 
 [grpc.tls]