feat: support storing different columns in separate files

src/query/AGENTS.md

@@ -6,3 +6,17 @@
 - Read [`sql/README.md`](sql/README.md) for planner structure and optimizer test-support entry points, [`functions/README.md`](functions/README.md) for function registration and tests, and [`ast/README.md`](ast/README.md) for parser-specific workflow.
 - Do not infer ownership from directory names alone when a nearby README already defines the boundary.
 - Query-side behavior changes that remain in the branch should usually add coverage close to the affected crate and add SQL regression tests under `tests/suites/` or `tests/sqllogictests/` when user-visible SQL behavior changes.
+
+## Code Review Rules
+
+### Fuse Partial UPDATE support boundary
+
+- Review Fuse Partial UPDATE and column-group code only within its documented supported domain.
+- Do not report findings whose reproduction requires combining Partial UPDATE with column-oriented
+  segments, partitioned tables, stored or virtual computed columns, Inverted/Ngram/Vector/Spatial
+  indexes, Fuse virtual columns, or `REPLACE INTO`. These combinations are deliberately
+  unsupported; do not require runtime guards, fallbacks, group-aware reads, or derived-metadata
+  maintenance unless a change explicitly expands the feature scope.
+- Continue to report issues reproducible with supported Partial UPDATE behavior, including ordinary
+  Bloom, change tracking, HLL, compression, cluster statistics, metadata compatibility, and
+  GC/Vacuum.

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -135,7 +135,13 @@ pub async fn do_vacuum2(
         .await?;
     let mut gc_root_blocks = HashSet::new();
     for segment in segments {
-        gc_root_blocks.extend(segment?.block_metas()?.iter().map(|b| b.location.0.clone()));
+        for block in segment?.block_metas()? {
+            gc_root_blocks.extend(
+                block
+                    .data_file_locations()
+                    .map(|location| location.0.clone()),
+            );
+        }
     }
     ctx.set_status_info(&format!(
         "Read segments for table {}, elapsed: {:?}, total protected blocks: {}",

src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs

@@ -16,10 +16,13 @@ use std::path::Path;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_storages_fuse::FuseTable;
+use databend_common_storages_fuse::io::TableMetaLocationGenerator;
 use databend_enterprise_query::test_kits::context::EESetup;
 use databend_query::sessions::QueryContext;
 use databend_query::sessions::TableContextTableAccess;
 use databend_query::test_kits::TestFixture;
+use databend_query::test_kits::latest_default_block_meta;
 use databend_storages_common_io::dedup_file_locations;
 
 // TODO investigate this
@@ -118,6 +121,104 @@ async fn test_vacuum2_all() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_vacuum2_preserves_active_partial_update_files() -> anyhow::Result<()> {
+    let fixture = TestFixture::setup_with_custom(EESetup::new()).await?;
+    fixture
+        .default_session()
+        .get_settings()
+        .set_data_retention_time_in_days(0)?;
+    let db = fixture.default_db_name();
+    let table_name = fixture.default_table_name();
+
+    fixture.create_default_database().await?;
+    fixture
+        .execute_command(&format!(
+            "create table {db}.{table_name} (id int, a int, b int) engine=fuse \
+             bloom_index_columns='id,a,b' enable_partial_update=true"
+        ))
+        .await?;
+    fixture
+        .execute_command(&format!(
+            "insert into {db}.{table_name} values (1, 10, 20), (2, 30, 40)"
+        ))
+        .await?;
+    fixture
+        .execute_command("set enable_partial_update = 1")
+        .await?;
+
+    fixture
+        .execute_command(&format!(
+            "update {db}.{table_name} set a = a + 1 where id = 1"
+        ))
+        .await?;
+    let first_update = latest_default_block_meta(&fixture).await?;
+    let obsolete_group = first_update.location.0.clone();
+    let obsolete_bloom = first_update
+        .column_groups
+        .iter()
+        .find(|group| group.location.0 == obsolete_group)
+        .and_then(|group| {
+            group.bloom.as_ref().map(|bloom| {
+                TableMetaLocationGenerator::gen_bloom_index_location_with_version(
+                    &group.location.0,
+                    bloom.format_version,
+                )
+            })
+        })
+        .expect("updated group should have an ordinary Bloom file");
+
+    fixture
+        .execute_command(&format!(
+            "update {db}.{table_name} set a = a + 1 where id = 1"
+        ))
+        .await?;
+    fixture
+        .execute_command(&format!(
+            "update {db}.{table_name} set b = b + 1 where id = 1"
+        ))
+        .await?;
+
+    let current = latest_default_block_meta(&fixture).await?;
+    assert!(
+        current
+            .column_groups
+            .iter()
+            .all(|group| group.location.0 != obsolete_group)
+    );
+
+    fixture
+        .execute_command(&format!("call system$fuse_vacuum2('{db}', '{table_name}')"))
+        .await?;
+
+    let table = fixture.latest_default_table().await?;
+    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+    let operator = fuse_table.get_operator();
+    for group in &current.column_groups {
+        operator.stat(&group.location.0).await?;
+        if let Some(bloom) = &group.bloom {
+            let bloom_location = TableMetaLocationGenerator::gen_bloom_index_location_with_version(
+                &group.location.0,
+                bloom.format_version,
+            );
+            operator.stat(&bloom_location).await?;
+        }
+    }
+    assert!(operator.stat(&obsolete_group).await.is_err());
+    assert!(operator.stat(&obsolete_bloom).await.is_err());
+
+    let rows = fixture
+        .execute_query(&format!(
+            "select count(*) from {db}.{table_name} \
+             where (id = 1 and a = 12 and b = 21) \
+                or (id = 2 and a = 30 and b = 40)"
+        ))
+        .await?;
+    assert_eq!(databend_query::test_kits::query_count(rows).await?, 2);
+
+    Ok(())
+}
+
 /// Verifies that dedup_file_locations correctly removes duplicates and reports samples.
 #[test]
 fn test_dedup_file_locations() {

src/query/service/src/interpreters/common/table_option_validation.rs

@@ -37,6 +37,7 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_DATA_RETENTION_PERIOD_IN_HOURS;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_PARQUET_DICTIONARY;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_VIRTUAL_COLUMN;
 use databend_common_storages_fuse::FUSE_OPT_KEY_FILE_SIZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_RECLUSTER_DEPTH;
@@ -90,6 +91,7 @@ pub static CREATE_FUSE_OPTIONS: LazyLock<HashSet<&'static str>> = LazyLock::new(
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP);
     r.insert(FUSE_OPT_KEY_ENABLE_AUTO_VACUUM);
     r.insert(FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE);
+    r.insert(FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE);
     r.insert(FUSE_OPT_KEY_ENABLE_VIRTUAL_COLUMN);
     r.insert(FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD);
 
@@ -173,6 +175,7 @@ pub static UNSET_TABLE_OPTIONS_WHITE_LIST: LazyLock<HashSet<&'static str>> = Laz
     r.insert(FUSE_OPT_KEY_DATA_RETENTION_NUM_SNAPSHOTS_TO_KEEP);
     r.insert(FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD);
     r.insert(FUSE_OPT_KEY_ENABLE_VIRTUAL_COLUMN);
+    r.insert(FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE);
     r.insert(OPT_KEY_ENABLE_COPY_DEDUP_FULL_PATH);
     r.insert(FUSE_OPT_KEY_DATA_PAGE_ROWS);
     r.insert(FUSE_OPT_KEY_DATA_PAGE_BYTES);

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

@@ -246,6 +246,7 @@ pub async fn build_mutation_info(
         partitions,
         statistics,
         table_meta_timestamps,
+        partial_update: false,
     })
 }
 

src/query/service/src/interpreters/interpreter_table_create.rs

@@ -47,6 +47,7 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_AGGRESSIVE_RECLUSTER;
 use databend_common_storages_fuse::FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE;
 use databend_common_storages_fuse::FuseSegmentFormat;
 use databend_common_storages_fuse::FuseStorageFormat;
 use databend_common_storages_fuse::io::MetaReaders;
@@ -500,6 +501,7 @@ impl CreateTableInterpreter {
         is_valid_option_of_type::<u32>(&table_meta.options, FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)?;
         // check enable auto analyze.
         is_valid_option_of_type::<u32>(&table_meta.options, FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE)?;
+        is_valid_option_of_type::<bool>(&table_meta.options, FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE)?;
         is_valid_option_of_type::<u32>(&table_meta.options, FUSE_OPT_KEY_AGGRESSIVE_RECLUSTER)?;
         is_valid_option_of_type::<u64>(
             &table_meta.options,

src/query/service/src/interpreters/interpreter_table_set_options.rs

@@ -30,6 +30,7 @@ use databend_common_storages_fuse::FUSE_OPT_KEY_AGGRESSIVE_RECLUSTER;
 use databend_common_storages_fuse::FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_ANALYZE;
 use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_AUTO_VACUUM;
+use databend_common_storages_fuse::FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE;
 use databend_common_storages_fuse::FuseSegmentFormat;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::io::SegmentsIO;
@@ -167,6 +168,10 @@ impl Interpreter for SetOptionsInterpreter {
         // Same as settings of FUSE_OPT_KEY_ENABLE_AUTO_VACUUM, expect value type is unsigned integer
         is_valid_option_of_type::<u32>(&self.plan.set_options, FUSE_OPT_KEY_ENABLE_AUTO_VACUUM)?;
         is_valid_option_of_type::<u32>(&self.plan.set_options, FUSE_OPT_KEY_AGGRESSIVE_RECLUSTER)?;
+        is_valid_option_of_type::<bool>(
+            &self.plan.set_options,
+            FUSE_OPT_KEY_ENABLE_PARTIAL_UPDATE,
+        )?;
         is_valid_option_of_type::<u64>(
             &self.plan.set_options,
             FUSE_OPT_KEY_AUTO_COMPACTION_IMPERFECT_BLOCKS_THRESHOLD,

src/query/service/src/physical_plans/physical_column_mutation.rs

@@ -20,6 +20,7 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataSchema;
 use databend_common_expression::DataSchemaRef;
+use databend_common_expression::FieldIndex;
 use databend_common_expression::RemoteExpr;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_common_meta_app::schema::TableInfo;
@@ -29,6 +30,7 @@ use databend_common_sql::evaluator::BlockOperator;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_storages_fuse::operations::TransformSerializeBlock;
+use databend_common_storages_fuse::statistics::ClusterStatsGenerator;
 use databend_storages_common_table_meta::meta::TableMetaTimestamps;
 
 use crate::physical_plans::format::ColumnMutationFormatter;
@@ -51,6 +53,8 @@ pub struct ColumnMutation {
     pub has_filter_column: bool,
     pub table_meta_timestamps: TableMetaTimestamps,
     pub udf_col_num: usize,
+    /// Field indices written by a partial update.
+    pub partial_update_fields: Option<Vec<FieldIndex>>,
 }
 
 #[typetag::serde]
@@ -98,6 +102,7 @@ impl IPhysicalPlan for ColumnMutation {
             has_filter_column: self.has_filter_column,
             table_meta_timestamps: self.table_meta_timestamps,
             udf_col_num: self.udf_col_num,
+            partial_update_fields: self.partial_update_fields.clone(),
         })
     }
 
@@ -166,12 +171,26 @@ impl IPhysicalPlan for ColumnMutation {
             // Keep only table fields in their schema order. Mutation input may
             // carry derived UDF argument/result columns that must not be
             // serialized back into the table.
-            let mut projection = field_id_to_schema_index.iter().collect::<Vec<_>>();
-            projection.sort_by_key(|(field_id, _)| *field_id);
-            let projection = projection
-                .into_iter()
-                .map(|(_, schema_index)| *schema_index)
-                .collect::<Vec<_>>();
+            let projection = if let Some(updated_fields) = &self.partial_update_fields {
+                updated_fields
+                    .iter()
+                    .map(|field_id| {
+                        field_id_to_schema_index
+                            .get(field_id)
+                            .copied()
+                            .ok_or_else(|| {
+                                ErrorCode::Internal("updated field is not in mutation output")
+                            })
+                    })
+                    .collect::<Result<Vec<_>>>()?
+            } else {
+                let mut projection = field_id_to_schema_index.iter().collect::<Vec<_>>();
+                projection.sort_by_key(|(field_id, _)| *field_id);
+                projection
+                    .into_iter()
+                    .map(|(_, schema_index)| *schema_index)
+                    .collect::<Vec<_>>()
+            };
             block_operators.push(BlockOperator::Project { projection });
 
             builder.main_pipeline.add_transformer(|| {
@@ -188,8 +207,12 @@ impl IPhysicalPlan for ColumnMutation {
             .build_table_by_table_info(&self.table_info, None)?;
         let table = FuseTable::try_from_table(table.as_ref())?;
 
+        let write_column_group = self.partial_update_fields.is_some();
+
         let block_thresholds = table.get_block_thresholds();
-        let cluster_stats_gen = if matches!(self.mutation_kind, MutationKind::Delete) {
+        let cluster_stats_gen = if write_column_group {
+            ClusterStatsGenerator::default()
+        } else if matches!(self.mutation_kind, MutationKind::Delete) {
             let input_schema = DataSchema::from(table.schema_with_stream()).into();
             table.get_cluster_stats_gen(builder.ctx.clone(), 0, block_thresholds, input_schema)?
         } else {
@@ -202,15 +225,27 @@ impl IPhysicalPlan for ColumnMutation {
         };
 
         builder.main_pipeline.add_transform(|input, output| {
-            let proc = TransformSerializeBlock::try_create(
-                builder.ctx.clone(),
-                input,
-                output,
-                table,
-                cluster_stats_gen.clone(),
-                self.mutation_kind,
-                self.table_meta_timestamps,
-            )?;
+            let proc = if write_column_group {
+                TransformSerializeBlock::try_create_for_update(
+                    builder.ctx.clone(),
+                    input,
+                    output,
+                    table,
+                    cluster_stats_gen.clone(),
+                    self.partial_update_fields.clone().unwrap(),
+                    self.table_meta_timestamps,
+                )?
+            } else {
+                TransformSerializeBlock::try_create(
+                    builder.ctx.clone(),
+                    input,
+                    output,
+                    table,
+                    cluster_stats_gen.clone(),
+                    self.mutation_kind,
+                    self.table_meta_timestamps,
+                )?
+            };
             proc.into_processor()
         })
     }

src/query/service/src/physical_plans/physical_commit_sink.rs

@@ -156,7 +156,7 @@ impl IPhysicalPlan for CommitSink {
                 } else {
                     builder
                         .main_pipeline
-                        .add_async_accumulating_transformer(|| {
+                        .try_add_async_accumulating_transformer(|| {
                             let base_segments = if matches!(
                                 kind,
                                 MutationKind::Compact
@@ -194,7 +194,7 @@ impl IPhysicalPlan for CommitSink {
                                 *kind,
                                 self.table_meta_timestamps,
                             )
-                        });
+                        })?;
                 }
 
                 let snapshot_gen = MutationGenerator::new(self.snapshot.clone(), *kind);

src/query/service/src/physical_plans/physical_compact_source.rs

@@ -206,7 +206,7 @@ impl IPhysicalPlan for CompactSource {
             builder.main_pipeline.try_resize(1)?;
             builder
                 .main_pipeline
-                .add_async_accumulating_transformer(|| {
+                .try_add_async_accumulating_transformer(|| {
                     TableMutationAggregator::create(
                         table,
                         builder.ctx.clone(),
@@ -217,7 +217,7 @@ impl IPhysicalPlan for CompactSource {
                         MutationKind::Compact,
                         self.table_meta_timestamps,
                     )
-                });
+                })?;
         }
         Ok(())
     }