refactor(query): support lazy read for update from

Author: zhang2014

src/query/sql/src/executor/physical_plans/physical_mutation.rs

@@ -58,6 +58,7 @@ use crate::executor::physical_plans::MutationOrganize;
 use crate::executor::physical_plans::MutationSplit;
 use crate::executor::physical_plans::RowFetch;
 use crate::executor::PhysicalPlanBuilder;
+use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
 use crate::parse_computed_expr;
 use crate::plans::BoundColumnRef;
@@ -283,11 +284,21 @@ impl PhysicalPlanBuilder {
             }));
         }
 
+        let already_enable_lazy_read = {
+            let settings = self.ctx.get_settings();
+            let lazy_read_threshold = settings.get_nondeterministic_update_lazy_read_threshold()?;
+            let rel_expr = RelExpr::with_s_expr(s_expr);
+            let cardinality = rel_expr.derive_cardinality_child(0)?;
+
+            lazy_read_threshold != 0 && lazy_read_threshold >= cardinality.cardinality as u64
+        };
+
         // Construct row fetch plan for lazy columns.
-        if let Some(lazy_columns) = self
-            .metadata
-            .read()
-            .get_table_lazy_columns(target_table_index)
+        if !already_enable_lazy_read
+            && let Some(lazy_columns) = self
+                .metadata
+                .read()
+                .get_table_lazy_columns(target_table_index)
             && !lazy_columns.is_empty()
         {
             plan = PhysicalPlan::RowFetch(build_mutation_row_fetch(

src/query/sql/src/executor/physical_plans/physical_row_fetch.rs

@@ -88,8 +88,8 @@ impl PhysicalPlanBuilder {
             return Err(ErrorCode::Internal("Internal column _row_id is not found"));
         };
 
-        let lazy_columns = metadata
-            .lazy_columns()
+        let lazy_columns = row_fetch
+            .lazy_columns
             .iter()
             .filter(|index| !input_schema.has_field(&index.to_string())) // If the column is already in the input schema, we don't need to fetch it.
             .cloned()
@@ -114,7 +114,8 @@ impl PhysicalPlanBuilder {
             })
             .collect();
 
-        let source = input_plan.try_find_single_data_source();
+        let metadata = self.metadata.read();
+        let source = metadata.get_table_source(&row_fetch.fetch_table_index);
         debug_assert!(source.is_some());
         let source_info = source.cloned().unwrap();
         let table_schema = source_info.source_info.schema();

src/query/sql/src/planner/binder/bind_mutation/update.rs

@@ -355,6 +355,7 @@ impl Binder {
                 need_wrap_nullable: false,
                 row_id_index: row_id.index,
                 lazy_columns: fields_bindings.iter().map(|x| x.index).collect(),
+                fetch_table_index: mutation.target_table_index,
             }));
         }
 

src/query/sql/src/planner/plans/row_fetch.rs

@@ -30,6 +30,7 @@ pub struct RowFetch {
 
     pub lazy_columns: ColumnSet,
     pub row_id_index: IndexType,
+    pub fetch_table_index: IndexType,
 }
 
 impl Operator for RowFetch {