fix(query): fix lazy topn in srf functions (#17945)

* fix(query): fix lazy topn in srf functions

* update

Author: sundy-li

src/query/service/tests/it/pipelines/filter/random_filter_expr.rs

@@ -152,6 +152,7 @@ fn convert_predicate_tree_to_scalar_expr(node: PredicateNode, data_type: &DataTy
                 data_type: Box::new(data_type.clone()),
                 visibility: Visibility::Visible,
                 virtual_expr: None,
+                is_srf: false,
             };
             let scalar_expr = ScalarExpr::BoundColumnRef(BoundColumnRef { span: None, column });
             ScalarExpr::FunctionCall(FunctionCall {

src/query/service/tests/it/sql/planner/optimizer/test_utils.rs

@@ -123,6 +123,7 @@ pub fn create_table_bound_column_ref(
         table_index,
         visibility: Visibility::Visible,
         virtual_expr: None,
+        is_srf: false,
     };
     ScalarExpr::BoundColumnRef(BoundColumnRef { column, span: None })
 }

src/query/sql/src/planner/binder/bind_context.rs

@@ -515,6 +515,16 @@ impl BindContext {
             .collect()
     }
 
+    pub fn has_srf_recursive(&self) -> bool {
+        self.columns.iter().any(|x| x.is_srf)
+            || !self.srf_info.srfs.is_empty()
+            || self
+                .parent
+                .as_ref()
+                .map(|p| p.has_srf_recursive())
+                .unwrap_or(false)
+    }
+
     /// Return data scheme.
     pub fn output_schema(&self) -> DataSchemaRef {
         let fields = self

src/query/sql/src/planner/binder/bind_table_reference/bind_table_function.rs

@@ -285,6 +285,7 @@ impl Binder {
                         Box::new(data_type),
                         Visibility::Visible,
                     )
+                    .is_srf(true)
                     .build();
                     bind_context.add_column_binding(column_binding);
 

src/query/sql/src/planner/binder/column_binding.rs

@@ -39,6 +39,8 @@ pub struct ColumnBinding {
     // Opitonal virtual expr, used by virtual computed column and variant virtual column,
     // `virtual_expr` will be parsed and bind to a `ScalarExpr`.
     pub virtual_expr: Option<String>,
+    // A flag indicates the column binding from srf or not
+    pub is_srf: bool,
 }
 
 const DUMMY_INDEX: usize = usize::MAX;
@@ -77,6 +79,7 @@ impl ColumnBinding {
             data_type,
             visibility: Visibility::Visible,
             virtual_expr: None,
+            is_srf: false,
         }
     }
 
@@ -110,6 +113,7 @@ pub struct ColumnBindingBuilder {
     pub visibility: Visibility,
 
     pub virtual_expr: Option<String>,
+    pub is_srf: bool,
 }
 
 impl ColumnBindingBuilder {
@@ -129,6 +133,7 @@ impl ColumnBindingBuilder {
             data_type,
             visibility,
             virtual_expr: None,
+            is_srf: false,
         }
     }
 
@@ -157,6 +162,11 @@ impl ColumnBindingBuilder {
         self
     }
 
+    pub fn is_srf(mut self, is_srf: bool) -> ColumnBindingBuilder {
+        self.is_srf = is_srf;
+        self
+    }
+
     pub fn build(self) -> ColumnBinding {
         ColumnBinding {
             database_name: self.database_name,
@@ -168,6 +178,7 @@ impl ColumnBindingBuilder {
             data_type: self.data_type,
             visibility: self.visibility,
             virtual_expr: self.virtual_expr,
+            is_srf: self.is_srf,
         }
     }
 }

src/query/sql/src/planner/binder/select.rs

@@ -512,7 +512,7 @@ impl Binder {
             || stmt.qualify.is_some()
             || !bind_context.aggregate_info.group_items.is_empty()
             || !bind_context.aggregate_info.aggregate_functions.is_empty()
-            || !bind_context.srf_info.srfs.is_empty()
+            || bind_context.has_srf_recursive()
         {
             return Ok(());
         }
@@ -579,7 +579,6 @@ impl Binder {
         }
 
         let cols = metadata.columns();
-
         let virtual_cols = cols
             .iter()
             .filter(|col| matches!(col, ColumnEntry::VirtualColumn(_)))

src/query/sql/src/planner/optimizer/optimizers/operator/aggregate/normalize_aggregate.rs

@@ -151,6 +151,7 @@ impl RuleNormalizeAggregateOptimizer {
                                 data_type: work_c.return_type.clone(),
                                 visibility: Visibility::Visible,
                                 column_name: work_c.display_name.clone(),
+                                is_srf: false,
                             },
                         }),
                     })

src/query/storages/parquet/src/parquet_reader/reader/row_group_reader.rs

@@ -80,7 +80,7 @@ impl RowGroupReader {
             op,
             &part.meta,
             page_locations.as_deref(),
-            read_settings.with_enable_cache(true),
+            *read_settings,
         );
         let mut selection = part
             .selectors

src/query/storages/parquet/src/source.rs

@@ -274,11 +274,14 @@ impl Processor for ParquetSource {
             State::Init => {
                 if let Some(part) = self.ctx.get_partition() {
                     match ParquetPart::from_part(&part)? {
+                        // From Copy Table from Stage, we don't enable cache
                         ParquetPart::RowGroup(part) => {
                             if let Some(reader) = self
                                 .row_group_reader
                                 .create_read_policy(
-                                    &ReadSettings::from_ctx(&self.ctx)?,
+                                    &ReadSettings::from_ctx(&self.ctx)?.with_enable_cache(
+                                        !matches!(self.source_type, ParquetSourceType::StageTable),
+                                    ),
                                     part,
                                     &mut self.topk_sorter,
                                     self.transformer.clone(),
@@ -343,16 +346,16 @@ impl ParquetSource {
             read_metadata_async_cached(path, &op, Some(part.compressed_size), &part.dedup_key)
                 .await?;
 
-        if matches!(self.source_type, ParquetSourceType::StageTable) {
+        let from_stage_table = matches!(self.source_type, ParquetSourceType::StageTable);
+        if from_stage_table {
             check_parquet_schema(
                 self.row_group_reader.schema_desc(),
                 meta.file_metadata().schema_descr(),
                 "first_file",
                 part.file.as_str(),
             )?;
         }
-        self.transformer
-            .match_by_field_name(matches!(self.source_type, ParquetSourceType::StageTable));
+        self.transformer.match_by_field_name(from_stage_table);
         // The schema of the table in iceberg may be inconsistent with the schema in parquet
         let reader = if self.row_group_reader.schema_desc().root_schema()
             != meta.file_metadata().schema_descr().root_schema()
@@ -402,7 +405,7 @@ impl ParquetSource {
 
             let reader = reader
                 .create_read_policy(
-                    &ReadSettings::from_ctx(&self.ctx)?,
+                    &ReadSettings::from_ctx(&self.ctx)?.with_enable_cache(!from_stage_table),
                     &part,
                     &mut self.topk_sorter,
                     self.transformer.clone(),

tests/sqllogictests/suites/query/lateral.test

@@ -63,6 +63,14 @@ select * from t, lateral(select unnest([a, a+1, a+2]) as b) t1 order by a, b
 3 4
 3 5
 
+statement ok
+create table test_mm(payload variant, create_date Date);
+
+query T
+select test_mm.payload payload from test_mm,
+lateral flatten(payload:key) walks
+where create_date > date_add(hour, -6, current_timestamp())
+and walks.value:source_person_id::varchar = 'xvcvz'  limit 1;
 
 statement ok
 CREATE TABLE user_activities(user_id int, activities variant)