fix nextest

Signed-off-by: luofucong <luofc@foxmail.com>

Author: MichaelScofield

src/query/src/optimizer/windowed_sort.rs

@@ -115,7 +115,6 @@ impl WindowedSortPhysicalRule {
                             sort_exec.fetch(),
                             scanner_info.partition_ranges.clone(),
                             sort_input,
-                            sort_exec.create_filter(),
                         ))
                     };
 

src/query/src/part_sort.rs

@@ -37,6 +37,7 @@ use datafusion::physical_plan::{
 };
 use datafusion_common::{DataFusionError, internal_err};
 use datafusion_physical_expr::PhysicalSortExpr;
+use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
 use futures::{Stream, StreamExt};
 use itertools::Itertools;
 use parking_lot::RwLock;
@@ -60,7 +61,6 @@ pub struct PartSortExec {
     metrics: ExecutionPlanMetricsSet,
     partition_ranges: Vec<Vec<PartitionRange>>,
     properties: PlanProperties,
-    filter: Arc<RwLock<TopKDynamicFilters>>,
 }
 
 impl PartSortExec {
@@ -69,7 +69,6 @@ impl PartSortExec {
         limit: Option<usize>,
         partition_ranges: Vec<Vec<PartitionRange>>,
         input: Arc<dyn ExecutionPlan>,
-        filter: Arc<RwLock<TopKDynamicFilters>>,
     ) -> Self {
         let metrics = ExecutionPlanMetricsSet::new();
         let properties = input.properties();
@@ -87,7 +86,6 @@ impl PartSortExec {
             metrics,
             partition_ranges,
             properties,
-            filter,
         }
     }
 
@@ -115,7 +113,6 @@ impl PartSortExec {
             input_stream,
             self.partition_ranges[partition].clone(),
             partition,
-            self.filter.clone(),
         )?) as _;
 
         Ok(df_stream)
@@ -172,7 +169,6 @@ impl ExecutionPlan for PartSortExec {
             self.limit,
             self.partition_ranges.clone(),
             new_input.clone(),
-            self.filter.clone(),
         )))
     }
 
@@ -234,7 +230,6 @@ struct PartSortStream {
     metrics: BaselineMetrics,
     context: Arc<TaskContext>,
     root_metrics: ExecutionPlanMetricsSet,
-    filter: Arc<RwLock<TopKDynamicFilters>>,
 }
 
 impl PartSortStream {
@@ -245,9 +240,11 @@ impl PartSortStream {
         input: DfSendableRecordBatchStream,
         partition_ranges: Vec<PartitionRange>,
         partition: usize,
-        filter: Arc<RwLock<TopKDynamicFilters>>,
     ) -> datafusion_common::Result<Self> {
         let buffer = if let Some(limit) = limit {
+            let filter = Arc::new(RwLock::new(TopKDynamicFilters::new(Arc::new(
+                DynamicFilterPhysicalExpr::new(vec![], lit(true)),
+            ))));
             PartSortBuffer::Top(
                 TopK::try_new(
                     partition,
@@ -258,7 +255,7 @@ impl PartSortStream {
                     context.session_config().batch_size(),
                     context.runtime_env(),
                     &sort.metrics,
-                    filter.clone(),
+                    filter,
                 )?,
                 0,
             )
@@ -283,7 +280,6 @@ impl PartSortStream {
             metrics: BaselineMetrics::new(&sort.metrics, partition),
             context,
             root_metrics: sort.metrics.clone(),
-            filter,
         })
     }
 }
@@ -507,6 +503,9 @@ impl PartSortStream {
 
     /// Internal method for sorting `Top` buffer (with limit).
     fn sort_top_buffer(&mut self) -> datafusion_common::Result<DfRecordBatch> {
+        let filter = Arc::new(RwLock::new(TopKDynamicFilters::new(Arc::new(
+            DynamicFilterPhysicalExpr::new(vec![], lit(true)),
+        ))));
         let new_top_buffer = TopK::try_new(
             self.partition,
             self.schema().clone(),
@@ -516,7 +515,7 @@ impl PartSortStream {
             self.context.session_config().batch_size(),
             self.context.runtime_env(),
             &self.root_metrics,
-            self.filter.clone(),
+            filter,
         )?;
         let PartSortBuffer::Top(top_k, _) =
             std::mem::replace(&mut self.buffer, PartSortBuffer::Top(new_top_buffer, 0))
@@ -685,7 +684,6 @@ mod test {
     use arrow::json::ArrayWriter;
     use arrow_schema::{DataType, Field, Schema, SortOptions, TimeUnit};
     use common_time::Timestamp;
-    use datafusion::physical_plan::sorts::sort::SortExec;
     use datafusion_physical_expr::expressions::Column;
     use futures::StreamExt;
     use store_api::region_engine::PartitionRange;
@@ -1044,19 +1042,16 @@ mod test {
                 cols
             })
             .collect_vec();
-        let mock_input = Arc::new(MockInputExec::new(batches, schema.clone()));
+        let mock_input = MockInputExec::new(batches, schema.clone());
 
-        let expr = PhysicalSortExpr {
-            expr: Arc::new(Column::new("ts", 0)),
-            options: opt,
-        };
-        let sort_exec = SortExec::new([expr.clone()].into(), mock_input.clone());
         let exec = PartSortExec::new(
-            expr,
+            PhysicalSortExpr {
+                expr: Arc::new(Column::new("ts", 0)),
+                options: opt,
+            },
             limit,
             vec![ranges.clone()],
-            mock_input,
-            sort_exec.create_filter(),
+            Arc::new(mock_input),
         );
 
         let exec_stream = exec.execute(0, Arc::new(TaskContext::default())).unwrap();

src/servers/src/http/handler.rs

@@ -208,7 +208,7 @@ pub async fn sql_format(
 
     let mut parts: Vec<String> = Vec::with_capacity(stmts.len());
     for stmt in stmts {
-        let mut s = format!("{:#}", stmt);
+        let mut s = format!("{stmt}");
         if !s.trim_end().ends_with(';') {
             s.push(';');
         }

src/sql/src/parsers/create_parser.rs

@@ -2427,8 +2427,7 @@ non TIMESTAMP(6) TIME INDEX,
         let sql = "CREATE VIEW test AS DELETE from demo";
         let result =
             ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
-        assert!(result.is_err());
-        assert_matches!(result, Err(crate::error::Error::Syntax { .. }));
+        assert!(result.is_ok_and(|x| x.len() == 1));
     }
 
     #[test]

tests-integration/tests/http.rs

@@ -662,7 +662,7 @@ pub async fn test_http_sql_slow_query(store_type: StorageType) {
     let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "sql_api").await;
     let client = TestClient::new(app).await;
 
-    let slow_query = "WITH RECURSIVE slow_cte AS (SELECT 1 AS n, md5(CAST(random() AS STRING)) AS hash UNION ALL SELECT n + 1, md5(concat(hash, n)) FROM slow_cte WHERE n < 4500) SELECT COUNT(*) FROM slow_cte";
+    let slow_query = "SELECT count(*) FROM generate_series(1, 1000000000)";
     let encoded_slow_query = encode(slow_query);
 
     let query_params = format!("/v1/sql?sql={encoded_slow_query}");
@@ -1152,12 +1152,12 @@ pub async fn test_prom_http_api(store_type: StorageType) {
     // query `__name__` without match[]
     // create a physical table and a logical table
     let res = client
-        .get("/v1/sql?sql=create table physical_table (`ts` timestamp time index, message string) with ('physical_metric_table' = 'true');")
+        .get("/v1/sql?sql=create table physical_table (`ts` timestamp time index, `message` string) with ('physical_metric_table' = 'true');")
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await);
     let res = client
-        .get("/v1/sql?sql=create table logic_table (`ts` timestamp time index, message string) with ('on_physical_table' = 'physical_table');")
+        .get("/v1/sql?sql=create table logic_table (`ts` timestamp time index, `message` string) with ('on_physical_table' = 'physical_table');")
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await);
@@ -5295,7 +5295,7 @@ pub async fn test_log_query(store_type: StorageType) {
 
     // prepare data with SQL API
     let res = client
-        .get("/v1/sql?sql=create table logs (`ts` timestamp time index, message string);")
+        .get("/v1/sql?sql=create table logs (`ts` timestamp time index, `message` string);")
         .send()
         .await;
     assert_eq!(res.status(), StatusCode::OK, "{:?}", res.text().await);