fix(fuse/mutation): avoid panic when filtering by _block_name

  Mutation planning was passing planner metadata column indexes (including internal
  columns like _block_name) as Fuse table schema field indexes, which could
  cause TableSchema::project() to index out of bounds during DELETE.

  Convert mutation read columns to base column_positions (schema FieldIndex),
  carry internal columns separately, and generate them from block_meta_index at
  execution time for filter evaluation.

Author: dantengsky

src/query/service/src/interpreters/interpreter_mutation.rs

@@ -28,6 +28,7 @@ use databend_common_expression::SendableDataBlockStream;
 use databend_common_expression::types::UInt64Type;
 use databend_common_pipeline::core::ProcessorPtr;
 use databend_common_pipeline::sinks::EmptySink;
+use databend_common_sql::ColumnEntry;
 use databend_common_sql::binder::MutationStrategy;
 use databend_common_sql::binder::MutationType;
 use databend_common_sql::executor::physical_plans::MutationKind;
@@ -278,6 +279,21 @@ async fn mutation_source_partitions(
         (None, vec![])
     };
 
+    let mut read_column_positions = Vec::new();
+    if !filter_used_columns.is_empty() {
+        let metadata = mutation.metadata.read();
+        for column_index in &filter_used_columns {
+            if let ColumnEntry::BaseTableColumn(base) = metadata.column(*column_index) {
+                if let Some(pos) = base.column_position {
+                    // column_position is 1-based
+                    read_column_positions.push(pos - 1);
+                }
+            }
+        }
+        read_column_positions.sort_unstable();
+        read_column_positions.dedup();
+    }
+
     let (is_lazy, is_delete) = if mutation.mutation_type == MutationType::Delete {
         let cluster = ctx.get_cluster();
         let is_lazy = fuse_table.is_column_oriented()
@@ -292,7 +308,7 @@ async fn mutation_source_partitions(
         .mutation_read_partitions(
             ctx,
             table_snapshot,
-            filter_used_columns,
+            read_column_positions,
             filters,
             is_lazy,
             is_delete,

src/query/service/src/physical_plans/physical_mutation_source.rs

@@ -68,6 +68,8 @@ pub struct MutationSource {
     pub output_schema: DataSchemaRef,
     pub input_type: MutationType,
     pub read_partition_columns: ColumnSet,
+    pub read_column_positions: Vec<usize>,
+    pub internal_columns: Vec<databend_common_catalog::plan::InternalColumn>,
     pub truncate_table: bool,
 
     pub partitions: Partitions,
@@ -110,6 +112,8 @@ impl IPhysicalPlan for MutationSource {
             output_schema: self.output_schema.clone(),
             input_type: self.input_type.clone(),
             read_partition_columns: self.read_partition_columns.clone(),
+            read_column_positions: self.read_column_positions.clone(),
+            internal_columns: self.internal_columns.clone(),
             truncate_table: self.truncate_table,
             partitions: self.partitions.clone(),
             statistics: self.statistics.clone(),
@@ -143,8 +147,7 @@ impl IPhysicalPlan for MutationSource {
             );
         }
 
-        let read_partition_columns: Vec<usize> =
-            self.read_partition_columns.clone().into_iter().collect();
+        let read_partition_columns = self.read_column_positions.clone();
 
         let is_lazy = self.partitions.partitions_type() == PartInfoType::LazyLevel && is_delete;
         if is_lazy {
@@ -191,13 +194,14 @@ impl IPhysicalPlan for MutationSource {
         } else {
             MutationAction::Update
         };
-        let col_indices = self.read_partition_columns.clone().into_iter().collect();
+        let col_indices = self.read_column_positions.clone();
         let update_mutation_with_filter =
             self.input_type == MutationType::Update && filter.is_some();
         table.add_mutation_source(
             builder.ctx.clone(),
             filter,
             col_indices,
+            self.internal_columns.clone(),
             &mut builder.main_pipeline,
             mutation_action,
         )?;
@@ -262,6 +266,25 @@ impl PhysicalPlanBuilder {
         }
         let output_schema = DataSchemaRefExt::create(fields);
 
+        let mut read_column_positions = Vec::new();
+        let mut internal_columns = Vec::new();
+        for column_index in mutation_source.read_partition_columns.iter() {
+            match metadata.column(*column_index) {
+                databend_common_sql::ColumnEntry::BaseTableColumn(base) => {
+                    if let Some(pos) = base.column_position {
+                        // column_position is 1-based
+                        read_column_positions.push(pos - 1);
+                    }
+                }
+                databend_common_sql::ColumnEntry::InternalColumn(internal) => {
+                    internal_columns.push(internal.internal_column.clone());
+                }
+                _ => {}
+            }
+        }
+        read_column_positions.sort_unstable();
+        read_column_positions.dedup();
+
         let truncate_table =
             mutation_source.mutation_type == MutationType::Delete && filters.is_none();
         Ok(PhysicalPlan::new(MutationSource {
@@ -271,6 +294,8 @@ impl PhysicalPlanBuilder {
             filters,
             input_type: mutation_source.mutation_type.clone(),
             read_partition_columns: mutation_source.read_partition_columns.clone(),
+            read_column_positions,
+            internal_columns,
             truncate_table,
             meta: PhysicalPlanMeta::new("MutationSource"),
             partitions: mutation_info.partitions.clone(),

src/query/service/tests/it/storages/fuse/operations/mutation/deletion.rs

@@ -13,8 +13,12 @@
 //  limitations under the License.
 
 use databend_common_base::base::tokio;
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::ScalarRef;
+use databend_common_expression::types::NumberScalar;
 use databend_query::test_kits::*;
+use futures::TryStreamExt;
 
 #[tokio::test(flavor = "multi_thread")]
 async fn test_deletion_mutator_multiple_empty_segments() -> Result<()> {
@@ -56,3 +60,63 @@ async fn test_deletion_mutator_multiple_empty_segments() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_delete_by_block_name() -> Result<()> {
+    let fixture = TestFixture::setup().await?;
+    let db = fixture.default_db_name();
+    let tbl = fixture.default_table_name();
+
+    fixture.create_default_database().await?;
+    fixture.create_default_table().await?;
+
+    fixture
+        .execute_command(&format!("insert into {db}.{tbl} values(1, (2, 3))"))
+        .await?;
+
+    let blocks = fixture
+        .execute_query(&format!("select _block_name from {db}.{tbl} limit 1"))
+        .await?
+        .try_collect::<Vec<_>>()
+        .await?;
+    let block_name = match blocks.get(0).and_then(|b| b.get_by_offset(0).index(0)) {
+        Some(ScalarRef::String(s)) => s.to_string(),
+        Some(other) => {
+            return Err(ErrorCode::Internal(format!(
+                "unexpected _block_name scalar: {other:?}"
+            )));
+        }
+        None => {
+            return Err(ErrorCode::Internal(
+                "failed to fetch _block_name".to_string(),
+            ));
+        }
+    };
+
+    fixture
+        .execute_command(&format!(
+            "delete from {db}.{tbl} where _block_name = '{block_name}'"
+        ))
+        .await?;
+
+    let count_blocks = fixture
+        .execute_query(&format!("select count() from {db}.{tbl}"))
+        .await?
+        .try_collect::<Vec<_>>()
+        .await?;
+    let count = match count_blocks
+        .get(0)
+        .and_then(|b| b.get_by_offset(0).index(0))
+    {
+        Some(ScalarRef::Number(NumberScalar::UInt64(v))) => v,
+        Some(other) => {
+            return Err(ErrorCode::Internal(format!(
+                "unexpected count() scalar: {other:?}"
+            )));
+        }
+        None => return Err(ErrorCode::Internal("failed to fetch count()".to_string())),
+    };
+    assert_eq!(count, 0);
+
+    Ok(())
+}

src/query/storages/fuse/src/operations/mutation/processors/mutation_source.rs

@@ -17,6 +17,8 @@ use std::ops::Not;
 use std::sync::Arc;
 
 use databend_common_base::base::ProgressValues;
+use databend_common_catalog::plan::InternalColumn;
+use databend_common_catalog::plan::InternalColumnMeta;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::build_origin_block_row_num;
 use databend_common_catalog::plan::gen_mutation_stream_meta;
@@ -82,6 +84,7 @@ pub struct MutationSource {
     filter: Arc<Option<Expr>>,
     block_reader: Arc<BlockReader>,
     remain_reader: Arc<Option<BlockReader>>,
+    internal_columns: Vec<InternalColumn>,
     operators: Vec<BlockOperator>,
     storage_format: FuseStorageFormat,
     action: MutationAction,
@@ -99,6 +102,7 @@ impl MutationSource {
         filter: Arc<Option<Expr>>,
         block_reader: Arc<BlockReader>,
         remain_reader: Arc<Option<BlockReader>>,
+        internal_columns: Vec<InternalColumn>,
         operators: Vec<BlockOperator>,
         storage_format: FuseStorageFormat,
     ) -> Result<ProcessorPtr> {
@@ -109,6 +113,7 @@ impl MutationSource {
             filter,
             block_reader,
             remain_reader,
+            internal_columns,
             operators,
             storage_format,
             action,
@@ -186,6 +191,11 @@ impl Processor for MutationSource {
                 if let Some(filter) = self.filter.as_ref() {
                     assert_eq!(filter.data_type(), &DataType::Boolean);
 
+                    let base_cols_len = data_block.num_columns();
+                    if !self.internal_columns.is_empty() {
+                        data_block = self.add_internal_columns(data_block, fuse_part)?;
+                    }
+
                     let func_ctx = self.ctx.get_function_context()?;
                     let evaluator = Evaluator::new(&data_block, &func_ctx, &BUILTIN_FUNCTIONS);
 
@@ -195,6 +205,12 @@ impl Processor for MutationSource {
                         .try_downcast::<BooleanType>()
                         .unwrap();
 
+                    if data_block.num_columns() != base_cols_len {
+                        let projections: std::collections::BTreeSet<usize> =
+                            (0..base_cols_len).collect();
+                        data_block = data_block.project(&projections);
+                    }
+
                     let affect_rows = match &predicates {
                         Value::Scalar(v) => {
                             if *v {
@@ -417,6 +433,39 @@ impl Processor for MutationSource {
 }
 
 impl MutationSource {
+    fn add_internal_columns(
+        &self,
+        mut data_block: DataBlock,
+        part: &FuseBlockPartInfo,
+    ) -> Result<DataBlock> {
+        let Some(block_meta_index) = part.block_meta_index.as_ref() else {
+            return Err(ErrorCode::Internal(
+                "block_meta_index is missing for internal columns".to_string(),
+            ));
+        };
+
+        let num_rows = data_block.num_rows();
+        let internal_column_meta = InternalColumnMeta {
+            segment_idx: block_meta_index.segment_idx,
+            block_id: block_meta_index.block_id,
+            block_location: block_meta_index.block_location.clone(),
+            segment_location: block_meta_index.segment_location.clone(),
+            snapshot_location: block_meta_index.snapshot_location.clone(),
+            offsets: None,
+            base_block_ids: None,
+            inner: None,
+            matched_rows: block_meta_index.matched_rows.clone(),
+            matched_scores: block_meta_index.matched_scores.clone(),
+            vector_scores: block_meta_index.vector_scores.clone(),
+        };
+
+        for internal_column in &self.internal_columns {
+            let entry = internal_column.generate_column_values(&internal_column_meta, num_rows);
+            data_block.add_entry(entry);
+        }
+        Ok(data_block)
+    }
+
     fn update_mutation_status(&self, num_rows: usize) {
         let (update_rows, deleted_rows) = if self.action == MutationAction::Update {
             (num_rows as u64, 0)

src/query/storages/fuse/src/operations/mutation_source.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use databend_common_catalog::plan::Filters;
+use databend_common_catalog::plan::InternalColumn;
 use databend_common_catalog::plan::PartInfoPtr;
 use databend_common_catalog::plan::PartStatistics;
 use databend_common_catalog::plan::Partitions;
@@ -53,6 +54,7 @@ impl FuseTable {
         ctx: Arc<dyn TableContext>,
         filter: Option<RemoteExpr<String>>,
         col_indices: Vec<usize>,
+        internal_columns: Vec<InternalColumn>,
         pipeline: &mut Pipeline,
         mutation_action: MutationAction,
     ) -> Result<()> {
@@ -65,10 +67,25 @@ impl FuseTable {
             };
         let projection = Projection::Columns(col_indices.clone());
         let update_stream_columns = self.change_tracking_enabled();
-        let block_reader =
-            self.create_block_reader(ctx.clone(), projection, false, update_stream_columns, false)?;
+        let query_internal_columns = !internal_columns.is_empty();
+        let block_reader = self.create_block_reader(
+            ctx.clone(),
+            projection,
+            query_internal_columns,
+            update_stream_columns,
+            false,
+        )?;
 
-        let schema = block_reader.schema().as_ref().clone();
+        let mut schema = block_reader.schema().as_ref().clone();
+        if query_internal_columns {
+            for internal_column in &internal_columns {
+                schema.add_internal_field(
+                    internal_column.column_name(),
+                    internal_column.table_data_type(),
+                    internal_column.column_id(),
+                );
+            }
+        }
         let filter_expr = Arc::new(filter.map(|v| {
             v.as_expr(&BUILTIN_FUNCTIONS)
                 .project_column_ref(|name| schema.index_of(name))
@@ -120,6 +137,7 @@ impl FuseTable {
                     filter_expr.clone(),
                     block_reader.clone(),
                     remain_reader.clone(),
+                    internal_columns.clone(),
                     ops.clone(),
                     self.storage_format,
                 )