refactor(query): refactor row fetcher for avoid oom (#18440)

zhang2014 · web-flow · commit 082e3b7f4615 · 2025-08-02T23:16:02.000+08:00
* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom

* refactor(query): refactor row fetcher for avoid oom
diff --git a/src/query/catalog/src/table.rs b/src/query/catalog/src/table.rs
@@ -76,6 +76,10 @@ pub trait Table: Sync + Send {
         false
     }
 
+    fn supported_lazy_materialize(&self) -> bool {
+        false
+    }
+
     fn schema(&self) -> Arc<TableSchema> {
         self.get_table_info().schema()
     }
diff --git a/src/query/service/src/pipelines/builders/builder_row_fetch.rs b/src/query/service/src/pipelines/builders/builder_row_fetch.rs
@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
-use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_exception::Result;
 use databend_common_pipeline_core::processors::InputPort;
 use databend_common_pipeline_core::processors::OutputPort;
@@ -24,24 +21,18 @@ use databend_common_pipeline_transforms::processors::create_dummy_item;
 use databend_common_sql::executor::physical_plans::RowFetch;
 use databend_common_sql::executor::PhysicalPlan;
 use databend_common_storages_fuse::operations::row_fetch_processor;
-use databend_common_storages_fuse::TableContext;
-use tokio::sync::Semaphore;
 
 use crate::pipelines::PipelineBuilder;
 impl PipelineBuilder {
     pub(crate) fn build_row_fetch(&mut self, row_fetch: &RowFetch) -> Result<()> {
         self.build_pipeline(&row_fetch.input)?;
-        let max_io_requests = self.ctx.get_settings().get_max_storage_io_requests()? as usize;
-        let row_fetch_runtime = GlobalIORuntime::instance();
-        let row_fetch_semaphore = Arc::new(Semaphore::new(max_io_requests));
+
         let processor = row_fetch_processor(
             self.ctx.clone(),
             row_fetch.row_id_col_offset,
             &row_fetch.source,
             row_fetch.cols_to_fetch.clone(),
             row_fetch.need_wrap_nullable,
-            row_fetch_semaphore,
-            row_fetch_runtime,
         )?;
         if !matches!(&*row_fetch.input, PhysicalPlan::MutationSplit(_)) {
             self.main_pipeline.add_transform(processor)?;
diff --git a/src/query/settings/src/settings_default.rs b/src/query/settings/src/settings_default.rs
@@ -145,11 +145,18 @@ impl DefaultSettings {
                 }),
                 ("max_block_size", DefaultSettingValue {
                     value: UserSettingValue::UInt64(65536),
-                    desc: "Sets the maximum byte size of a single data block that can be read.",
+                    desc: "Sets the maximum rows size of a single data block that can be read.",
                     mode: SettingMode::Both,
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(1..=u64::MAX)),
                 }),
+                ("max_block_bytes", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(50 * 1024 * 1024),
+                    desc: "Sets the maximum byte size of a single data block that can be read.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(1024 * 1024..=u64::MAX)),
+                }),
                 ("sequence_step_size", DefaultSettingValue {
                     value: UserSettingValue::UInt64(65536),
                     desc: "Sets the sequence step size for nextval function.",
diff --git a/src/query/settings/src/settings_getter_setter.rs b/src/query/settings/src/settings_getter_setter.rs
@@ -188,6 +188,10 @@ impl Settings {
         self.try_get_u64("max_block_size")
     }
 
+    pub fn get_max_block_bytes(&self) -> Result<u64> {
+        self.try_get_u64("max_block_bytes")
+    }
+
     // Set max_block_size.
     pub fn set_max_block_size(&self, val: u64) -> Result<()> {
         self.try_set_u64("max_block_size", val)
diff --git a/src/query/sql/src/executor/physical_plans/physical_table_scan.rs b/src/query/sql/src/executor/physical_plans/physical_table_scan.rs
@@ -147,7 +147,16 @@ impl PhysicalPlanBuilder {
 
             let mut prewhere = scan.prewhere.clone();
             let mut used: ColumnSet = required.intersection(&columns).cloned().collect();
-            if scan.is_lazy_table {
+
+            let supported_lazy_materialize = {
+                self.metadata
+                    .read()
+                    .table(scan.table_index)
+                    .table()
+                    .supported_lazy_materialize()
+            };
+
+            if scan.is_lazy_table && supported_lazy_materialize {
                 let lazy_columns = columns.difference(&used).cloned().collect();
                 let mut metadata = self.metadata.write();
                 metadata.set_table_lazy_columns(scan.table_index, lazy_columns);
diff --git a/src/query/sql/src/planner/binder/bind_mutation/mutation_expression.rs b/src/query/sql/src/planner/binder/bind_mutation/mutation_expression.rs
@@ -151,7 +151,14 @@ impl MutationExpression {
                 if *mutation_strategy == MutationStrategy::NotMatchedOnly {
                     update_stream_columns = false;
                 }
-                let is_lazy_table = *mutation_strategy != MutationStrategy::NotMatchedOnly;
+                let is_lazy_table = {
+                    let metadata = binder.metadata.read();
+                    *mutation_strategy != MutationStrategy::NotMatchedOnly
+                        && metadata
+                            .table(target_table_index)
+                            .table()
+                            .supported_lazy_materialize()
+                };
                 target_s_expr =
                     Self::update_target_scan(&target_s_expr, is_lazy_table, update_stream_columns)?;
 
@@ -173,6 +180,12 @@ impl MutationExpression {
                     )
                     .await?;
 
+                if !is_lazy_table {
+                    for column_index in bind_context.column_set().iter() {
+                        required_columns.insert(*column_index);
+                    }
+                }
+
                 Ok(MutationExpressionBindResult {
                     input: join_s_expr,
                     mutation_type,
@@ -276,7 +289,14 @@ impl MutationExpression {
                         target_table_row_id_index: DUMMY_COLUMN_INDEX,
                     })
                 } else {
-                    let is_lazy_table = mutation_type != MutationType::Delete;
+                    let is_lazy_table = {
+                        let metadata = binder.metadata.read();
+                        mutation_type != MutationType::Delete
+                            && metadata
+                                .table(target_table_index)
+                                .table()
+                                .supported_lazy_materialize()
+                    };
                     s_expr =
                         Self::update_target_scan(&s_expr, is_lazy_table, update_stream_columns)?;
 
@@ -321,6 +341,13 @@ impl MutationExpression {
                     let mut rewriter = SubqueryDecorrelatorOptimizer::new(opt_ctx, None);
                     let s_expr = rewriter.optimize_sync(&s_expr)?;
 
+                    // The delete operation only requires the row ID to locate the row to be deleted and does not need to extract any other columns.
+                    if !is_lazy_table && mutation_type != MutationType::Delete {
+                        for column_index in bind_context.column_set().iter() {
+                            required_columns.insert(*column_index);
+                        }
+                    }
+
                     Ok(MutationExpressionBindResult {
                         input: s_expr,
                         mutation_type,
diff --git a/src/query/sql/src/planner/binder/bind_mutation/update.rs b/src/query/sql/src/planner/binder/bind_mutation/update.rs
@@ -313,11 +313,11 @@ impl Binder {
             };
         }
 
-        mutation.required_columns = Box::new(
-            std::iter::once(mutation.row_id_index)
-                .chain(any_columns.into_iter().map(|c| c.new))
-                .collect(),
-        );
+        // update required columns
+        for any_column in any_columns {
+            mutation.required_columns.remove(&any_column.old);
+            mutation.required_columns.insert(any_column.new);
+        }
 
         let aggr_expr =
             self.bind_aggregate(&mut mutation.bind_context, s_expr.unary_child().clone())?;
diff --git a/src/query/sql/src/planner/binder/select.rs b/src/query/sql/src/planner/binder/select.rs
@@ -574,6 +574,10 @@ impl Binder {
             return Ok(());
         }
 
+        if !metadata.table(0).table().supported_lazy_materialize() {
+            return Ok(());
+        }
+
         let cols = metadata.columns();
         let virtual_cols = cols
             .iter()
diff --git a/src/query/storages/fuse/src/fuse_table.rs b/src/query/storages/fuse/src/fuse_table.rs
@@ -769,6 +769,10 @@ impl Table for FuseTable {
         column_id >= VECTOR_SCORE_COLUMN_ID
     }
 
+    fn supported_lazy_materialize(&self) -> bool {
+        !matches!(self.storage_format, FuseStorageFormat::Native)
+    }
+
     fn support_column_projection(&self) -> bool {
         true
     }
diff --git a/src/query/storages/fuse/src/io/read/block/block_reader_merge_io_sync.rs b/src/query/storages/fuse/src/io/read/block/block_reader_merge_io_sync.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::collections::HashSet;
 
 use databend_common_catalog::plan::PartInfoPtr;
@@ -22,6 +23,7 @@ use databend_storages_common_cache::CacheManager;
 use databend_storages_common_cache::TableDataCacheKey;
 use databend_storages_common_io::MergeIOReader;
 use databend_storages_common_io::ReadSettings;
+use databend_storages_common_table_meta::meta::ColumnMeta;
 
 use crate::fuse_part::FuseBlockPartInfo;
 use crate::io::BlockReader;
@@ -35,6 +37,24 @@ impl BlockReader {
         ignore_column_ids: &Option<HashSet<ColumnId>>,
     ) -> Result<BlockReadResult> {
         let part = FuseBlockPartInfo::from_part(part)?;
+        let location = &part.location;
+        let columns_meta = &part.columns_meta;
+
+        self.sync_read_columns_data_by_merge_io_2(
+            settings,
+            location,
+            columns_meta,
+            ignore_column_ids,
+        )
+    }
+
+    pub fn sync_read_columns_data_by_merge_io_2(
+        &self,
+        settings: &ReadSettings,
+        location: &str,
+        columns_meta: &HashMap<ColumnId, ColumnMeta>,
+        ignore_column_ids: &Option<HashSet<ColumnId>>,
+    ) -> Result<BlockReadResult> {
         let column_array_cache = CacheManager::instance().get_table_data_array_cache();
 
         let mut ranges = vec![];
@@ -45,9 +65,10 @@ impl BlockReader {
                     continue;
                 }
             }
-            let block_path = &part.location;
 
-            if let Some(column_meta) = part.columns_meta.get(column_id) {
+            let block_path = location;
+
+            if let Some(column_meta) = columns_meta.get(column_id) {
                 // first, check column array object cache
                 let (offset, len) = column_meta.offset_length();
                 let column_cache_key = TableDataCacheKey::new(block_path, *column_id, offset, len);
@@ -59,12 +80,8 @@ impl BlockReader {
             }
         }
 
-        let merge_io_result = MergeIOReader::sync_merge_io_read(
-            settings,
-            self.operator.clone(),
-            &part.location,
-            &ranges,
-        )?;
+        let merge_io_result =
+            MergeIOReader::sync_merge_io_read(settings, self.operator.clone(), location, &ranges)?;
 
         // for sync read, we disable table data cache
         let cached_column_data = vec![];
diff --git a/src/query/storages/fuse/src/operations/read/fuse_rows_fetcher.rs b/src/query/storages/fuse/src/operations/read/fuse_rows_fetcher.rs
diff --git a/src/query/storages/fuse/src/operations/read/mod.rs b/src/query/storages/fuse/src/operations/read/mod.rs
diff --git a/src/query/storages/fuse/src/operations/read/native_rows_fetcher.rs b/src/query/storages/fuse/src/operations/read/native_rows_fetcher.rs
diff --git a/src/query/storages/fuse/src/operations/read/parquet_rows_fetcher.rs b/src/query/storages/fuse/src/operations/read/parquet_rows_fetcher.rs
diff --git a/tests/sqllogictests/suites/base/06_show/06_0003_show_settings.test b/tests/sqllogictests/suites/base/06_show/06_0003_show_settings.test

Original file line number	Diff line number	Diff line change
`@@ -76,6 +76,10 @@ pub trait Table: Sync + Send {`
`76`	`76`	`false`
`77`	`77`	`}`
`78`	`78`
	`79`	`+ fn supported_lazy_materialize(&self) -> bool {`
	`80`	`+ false`
	`81`	`+ }`
	`82`	`+`
`79`	`83`	`fn schema(&self) -> Arc<TableSchema> {`
`80`	`84`	`self.get_table_info().schema()`
`81`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -769,6 +769,10 @@ impl Table for FuseTable {`
`769`	`769`	`column_id >= VECTOR_SCORE_COLUMN_ID`
`770`	`770`	`}`
`771`	`771`
	`772`	`+ fn supported_lazy_materialize(&self) -> bool {`
	`773`	`+ !matches!(self.storage_format, FuseStorageFormat::Native)`
	`774`	`+ }`
	`775`	`+`
`772`	`776`	`fn support_column_projection(&self) -> bool {`
`773`	`777`	`true`
`774`	`778`	`}`