From 93f598ad71687049d597e39df1ccf8c2e06a41e9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 01:16:16 +0800 Subject: [PATCH 1/5] feat(table): support data evolution compaction --- crates/paimon/src/spec/core_options.rs | 10 +- crates/paimon/src/table/compact.rs | 144 ++- .../src/table/compact/data_evolution.rs | 700 +++++++++++ .../paimon/src/table/data_evolution_writer.rs | 14 +- crates/paimon/src/table/mod.rs | 2 +- crates/paimon/src/table/table_commit.rs | 91 +- .../src/table/table_commit/parity_tests.rs | 85 ++ .../table_commit/row_id_conflict_tests.rs | 26 +- .../table/table_commit/row_tracking_tests.rs | 2 +- crates/paimon/tests/append_compact_test.rs | 4 - .../data_evolution_compact_layout_test.rs | 164 +++ .../tests/data_evolution_compact_test.rs | 1085 +++++++++++++++++ docs/src/compaction.md | 65 +- 13 files changed, 2332 insertions(+), 60 deletions(-) create mode 100644 crates/paimon/src/table/compact/data_evolution.rs create mode 100644 crates/paimon/tests/data_evolution_compact_layout_test.rs create mode 100644 crates/paimon/tests/data_evolution_compact_test.rs diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index 052fc6e90..f85edca0b 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -1639,17 +1639,21 @@ impl<'a> CoreOptions<'a> { .unwrap_or(DEFAULT_TARGET_FILE_SIZE) } - /// Minimum append input files for an underfilled compaction task. Default is 5. + /// Minimum input files for a compaction task. Default is 5. Data evolution + /// permits one file for a forced rewrite; ordinary append requires two. pub fn compaction_min_file_num(&self) -> crate::Result { + let minimum = if self.data_evolution_enabled() { 1 } else { 2 }; self.options .get(COMPACTION_MIN_FILE_NUM_OPTION) .map_or(Ok(5), |raw| { raw.parse::() .ok() - .filter(|count| *count >= 2) + .filter(|count| *count >= minimum) .map(|count| count as usize) .ok_or_else(|| crate::Error::ConfigInvalid { - message: format!("compaction.min.file-num must be at least 2, got '{raw}'"), + message: format!( + "compaction.min.file-num must be at least {minimum}, got '{raw}'" + ), }) }) } diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index abfec835b..76fae4793 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -15,9 +15,10 @@ // specific language governing permissions and limitations // under the License. -//! Snapshot-bound append compaction and full-bucket primary-key compaction. +//! Snapshot-bound append, data-evolution and primary-key compaction. mod append; +pub(super) mod data_evolution; mod primary_key; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -42,9 +43,8 @@ use crate::{Error, Result}; /// Creates planners, executors and a committer with one shared configuration. /// Output size and append task selection use the captured table options. -/// Supports ordinary append tables and fixed or dynamic bucket PK tables, -/// without row tracking, deletion vectors, data evolution, query authorization -/// or dedicated blob/vector storage. +/// Supports ordinary append, data-evolution normal files, and fixed or dynamic +/// bucket PK tables. Data evolution preserves row IDs and logical deletions. pub struct CompactBuilder<'a> { table: &'a Table, partition_filter: Option, @@ -92,11 +92,15 @@ impl<'a> CompactBuilder<'a> { message: "Compaction requires a positive target-file-size".into(), }); } - let append = if self.table.schema().primary_keys().is_empty() { - Some(append::AppendCompactPlanner::new(&options)?) - } else { - None - }; + let append = + if self.table.schema().primary_keys().is_empty() && !options.data_evolution_enabled() { + Some(append::AppendCompactPlanner::new(&options)?) + } else { + None + }; + if options.data_evolution_enabled() { + options.compaction_min_file_num()?; + } Ok(CompactPlanner { table: self.table.clone(), append, @@ -122,7 +126,7 @@ impl<'a> CompactBuilder<'a> { } } -/// Selects append small files or whole PK buckets from each captured snapshot. +/// Selects append files, contiguous data-evolution ranges or whole PK buckets. pub struct CompactPlanner { table: Table, append: Option, @@ -168,6 +172,10 @@ pub struct CompactTask { } impl CompactTask { + fn needs_captured_inputs(&self) -> bool { + self.is_primary_key() || self.is_data_evolution() + } + pub fn snapshot_id(&self) -> i64 { self.context.snapshot.id() } @@ -193,7 +201,8 @@ impl CompactPlanner { if let Some(filter) = &self.partition_filter { reader.with_filter(filter.clone()); } - let source = reader.new_scan().plan_all().await?; + let scan = reader.new_scan().with_scan_all_files(); + let source = scan.plan_all().await?; let Some(snapshot_id) = source.snapshot_id() else { return Ok(CompactPlan { snapshot_id: None, @@ -216,21 +225,55 @@ impl CompactPlanner { }; let mut groups = BTreeMap::<_, Vec>::new(); let primary_key = !self.table.schema().primary_keys().is_empty(); - for split in source.splits() { - let group = groups - .entry(( - split.partition().to_serialized_bytes(), - split.bucket(), - split.total_buckets(), - )) - .or_default(); - for file in split.data_files() { - validate_file(file, primary_key)?; - group.push(file.clone()); + let data_evolution = self.table.schema().core_options().data_evolution_enabled(); + if data_evolution { + // Query planning may discard shadowed column providers. A rewrite + // must remove every normal file in each captured row-ID group. + for entry in scan.plan_manifest_entries(&context.snapshot).await? { + if !super::source::is_data_evolution_normal_file(entry.file()) { + continue; + } + data_evolution::validate_file(entry.file())?; + groups + .entry(( + entry.partition().to_vec(), + entry.bucket(), + entry.total_buckets(), + )) + .or_default() + .push(entry.file().clone()); + } + } else { + for split in source.splits() { + let group = groups + .entry(( + split.partition().to_serialized_bytes(), + split.bucket(), + split.total_buckets(), + )) + .or_default(); + for file in split.data_files() { + validate_file(file, primary_key)?; + group.push(file.clone()); + } } } let mut tasks = Vec::new(); for ((partition, bucket, total_buckets), files) in groups { + if data_evolution { + for files in data_evolution::plan(files, &self.table.schema().core_options())? { + data_evolution::validate_storage_layout(&self.table, &files).await?; + tasks.push(CompactTask { + context: context.clone(), + task_id: tasks.len(), + partition: partition.clone(), + bucket, + total_buckets, + files, + }); + } + continue; + } if primary_key { // L0 requires upgrading; multiple levels require merging. Files // at one positive level already form a non-overlapping sorted @@ -346,6 +389,20 @@ impl CompactExecutor { .with_bucket_path(bucket_path) .with_data_files(task.files.clone()) .build()?; + if task.is_data_evolution() { + let files = data_evolution::execute( + &table, + task, + &split, + partition_path, + self.resources.clone(), + ) + .await?; + return Ok(CompactResult { + task: task.clone(), + files, + }); + } if task.is_primary_key() { let files = primary_key::execute(&table, task, &split, self.resources.clone()).await?; return Ok(CompactResult { @@ -411,6 +468,10 @@ impl CompactExecutor { } impl CompactTask { + fn is_data_evolution(&self) -> bool { + self.context.schema.core_options().data_evolution_enabled() + } + fn is_primary_key(&self) -> bool { !self.context.schema.primary_keys().is_empty() } @@ -429,6 +490,9 @@ impl CompactTask { } fn validate_inputs(&self, active: &[ManifestEntry]) -> Result<()> { + if self.is_data_evolution() { + data_evolution::validate_inputs(&self.partition, self.bucket, &self.files, active)?; + } let bucket_files = active .iter() .filter(|entry| { @@ -467,7 +531,11 @@ impl CompactTask { )); } ensure_snapshot_supported(table, &self.context.snapshot)?; - let minimum_files = if self.is_primary_key() { 1 } else { 2 }; + let minimum_files = if self.is_primary_key() || self.is_data_evolution() { + 1 + } else { + 2 + }; if table.schema().core_options().target_file_size() <= 0 || self.files.len() < minimum_files || self.bucket < 0 @@ -476,11 +544,18 @@ impl CompactTask { } let mut names = HashSet::new(); for file in &self.files { - validate_file(file, self.is_primary_key())?; + if self.is_data_evolution() { + data_evolution::validate_file(file)?; + } else { + validate_file(file, self.is_primary_key())?; + } if !names.insert(&file.file_name) { return Err(invalid("Duplicate compaction input file")); } } + if self.is_data_evolution() { + data_evolution::row_range(&self.files)?; + } Ok(()) } } @@ -537,7 +612,11 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul .sum::(); let mut output_rows = 0u64; for file in &result.files { - validate_file(file, task.is_primary_key())?; + if task.is_data_evolution() { + data_evolution::validate_file(file)?; + } else { + validate_file(file, task.is_primary_key())?; + } if file.schema_id != task.context.schema.id() || file.file_source != Some(1) || file.level != task.output_level() @@ -556,7 +635,9 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul 2, )); } - if !task.is_primary_key() && input_rows != output_rows { + if task.is_data_evolution() { + data_evolution::validate_outputs(task, &result.files)?; + } else if !task.is_primary_key() && input_rows != output_rows { return Err(invalid("Compaction must preserve the input row count")); } let options = task.context.schema.core_options(); @@ -596,7 +677,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul snapshot_id: first.task.snapshot_id(), read_snapshot: first .task - .is_primary_key() + .needs_captured_inputs() .then(|| Box::new(first.task.context.snapshot.clone())), commit_suffix: hex::encode(digest.finalize()), metrics, @@ -608,6 +689,10 @@ pub(super) fn ensure_supported(table: &Table) -> Result<()> { table.ensure_not_branch_reference_for_write()?; let options = table.schema().core_options(); let properties = table.schema().options(); + if options.data_evolution_enabled() { + data_evolution::ensure_supported(table)?; + return options.validate_data_file_path_directory(); + } if table.schema().primary_keys().is_empty() && options.bucket() > 0 && !properties @@ -671,6 +756,7 @@ pub(super) fn ensure_supported(table: &Table) -> Result<()> { fn ensure_snapshot_supported(table: &Table, snapshot: &Snapshot) -> Result<()> { if snapshot.index_manifest().is_some() + && !table.schema().core_options().data_evolution_enabled() && (table.schema().primary_keys().is_empty() || table.schema().core_options().bucket() != -1) { @@ -694,6 +780,10 @@ pub(super) async fn validate_snapshot_indexes( }; let path = table.snapshot_manager().manifest_path(name); let entries = IndexManifest::read(table.file_io(), &path).await?; + if table.schema().core_options().data_evolution_enabled() { + data_evolution::validate_indexes(&entries)?; + return Ok(Some(name.to_string())); + } if entries.iter().any(|entry| { entry.index_file.index_type != "HASH" || entry.index_file.deletion_vectors_ranges.is_some() diff --git a/crates/paimon/src/table/compact/data_evolution.rs b/crates/paimon/src/table/compact/data_evolution.rs new file mode 100644 index 000000000..c8dd6ce53 --- /dev/null +++ b/crates/paimon/src/table/compact/data_evolution.rs @@ -0,0 +1,700 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Normal-file compaction, following Java DataEvolutionCompactCoordinator and +//! DataEvolutionNormalCompactTask. Physical slots and logical deletions survive. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Arc; + +use futures::TryStreamExt; +use roaring::RoaringBitmap; + +use super::{invalid, CompactTask}; +use crate::resource::ResourceContext; +use crate::spec::{ + bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, FileKind, + IndexManifest, IndexManifestEntry, ManifestEntry, PartitionComputer, Snapshot, +}; +use crate::table::data_file_index_writer::FileIndexOptions; +use crate::table::data_file_writer::DataFileWriter; +use crate::table::index_file_path::committed_index_file_path; +use crate::table::source::{data_evolution_anchor_file, is_data_evolution_normal_file}; +use crate::table::stats_filter::group_by_overlapping_row_id; +use crate::table::{DataEvolutionDeleteWriter, DataSplit, Table}; +use crate::{Error, Result}; + +pub(super) fn ensure_supported(table: &Table) -> Result<()> { + let options = table.schema().core_options(); + if table.is_format_table() + || !options.row_tracking_enabled() + || !table.schema().primary_keys().is_empty() + || options.bucket() != -1 + || options.query_auth_enabled() + || options.rowkind_field().is_some() + || options.clustering_incremental_enabled() + || table + .schema() + .options() + .contains_key("record-level.expire-time") + { + return Err(Error::Unsupported { + message: "Data-evolution compaction requires an unaware append table with row tracking, without query authorization, rowkind fields, incremental clustering or record expiration".into(), + }); + } + for key in [ + "data-evolution.compaction.rewrite-row-ids", + "data-evolution.compaction.split-large-files", + ] { + if table + .schema() + .options() + .get(key) + .is_some_and(|v| v.eq_ignore_ascii_case("true")) + { + return Err(Error::Unsupported { + message: format!("Data-evolution compaction does not support {key}=true"), + }); + } + } + Ok(()) +} + +pub(super) fn validate_file(file: &DataFileMeta) -> Result<()> { + if !is_data_evolution_normal_file(file) + || file.file_size < 0 + || file.row_count <= 0 + || file.level != 0 + || file + .first_row_id + .is_none_or(|id| id < 0 || id.checked_add(file.row_count).is_none()) + || file.min_sequence_number <= 0 + || file.max_sequence_number < file.min_sequence_number + || file.delete_row_count.is_some_and(|count| count != 0) + || file.write_cols.as_ref().is_some_and(|cols| { + cols.iter() + .any(|name| crate::spec::is_row_tracking_column(name)) + }) + { + return Err(invalid("Data-evolution compaction requires normal files with valid row IDs, row counts and sequence bounds")); + } + Ok(()) +} + +/// Check alignment within each overlapping group and continuity between groups. +pub(super) fn row_range(files: &[DataFileMeta]) -> Result<(i64, i64)> { + let groups = group_by_overlapping_row_id(files.to_vec()); + let mut range: Option<(i64, i64)> = None; + for group in groups { + for file in &group { + validate_file(file)?; + } + let current = group[0] + .row_id_range() + .ok_or_else(|| invalid("Missing data-evolution row range"))?; + if group + .iter() + .any(|file| file.row_id_range() != Some(current)) + { + return Err(invalid( + "Data-evolution normal files must have aligned row-ID ranges", + )); + } + match range { + Some((first, last)) if last.checked_add(1) == Some(current.0) => { + range = Some((first, current.1)) + } + Some(_) => { + return Err(invalid( + "Data-evolution compaction cannot cross a row-ID gap", + )) + } + None => range = Some(current), + } + } + range.ok_or_else(|| invalid("Empty data-evolution compaction range")) +} + +fn output_range(files: &[DataFileMeta], options: &CoreOptions<'_>) -> Result<(i64, i64)> { + let (first, last) = row_range(files)?; + // The current Rust DV writer addresses local positions with u32. Merging + // otherwise deletable groups must not make later DELETEs unrepresentable. + if options.deletion_vectors_enabled() && last - first > i64::from(u32::MAX) { + return Err(Error::Unsupported { + message: "Data-evolution compaction with deletion vectors cannot produce more than 4294967296 physical row slots in one file; 64-bit deletion-vector writing is not supported".into(), + }); + } + Ok((first, last)) +} + +/// Java packs whole aligned groups by sum(max(file_size, open_file_cost)). +/// A heavy group is isolated; adjacent groups fill bins until weight > target. +pub(super) fn plan( + files: Vec, + options: &CoreOptions<'_>, +) -> Result>> { + let min_files = options.compaction_min_file_num()?; + let target = i128::from(options.target_file_size()); + let mut tasks = Vec::new(); + let mut pending = Vec::new(); + let mut weight = 0i128; + let mut previous_end: Option = None; + let drain = |pending: &mut Vec, tasks: &mut Vec>| { + let files = std::mem::take(pending); + if files.len() >= min_files { + tasks.push(files); + } + }; + for group in group_by_overlapping_row_id(files) { + let (first, last) = row_range(&group)?; + let group_weight = group + .iter() + .map(|file| i128::from(file.file_size.max(options.source_split_open_file_cost()))) + .sum::(); + if previous_end.is_some_and(|end| end.checked_add(1) != Some(first)) + || group_weight > target + { + drain(&mut pending, &mut tasks); + weight = 0; + } + pending.extend(group); + weight += group_weight; + if weight > target { + drain(&mut pending, &mut tasks); + weight = 0; + } + previous_end = Some(last); + } + drain(&mut pending, &mut tasks); + for task in &tasks { + output_range(task, options)?; + } + Ok(tasks) +} + +pub(super) fn validate_inputs( + partition: &[u8], + bucket: i32, + files: &[DataFileMeta], + active: &[ManifestEntry], +) -> Result<()> { + let (first, last) = row_range(files)?; + let names = files + .iter() + .map(|file| file.file_name.as_str()) + .collect::>(); + if active.iter().any(|entry| { + entry.partition() == partition + && entry.bucket() == bucket + && is_data_evolution_normal_file(entry.file()) + && entry + .file() + .row_id_range() + .is_some_and(|(start, end)| start <= last && end >= first) + && !names.contains(entry.file().file_name.as_str()) + }) { + return Err(invalid("Data-evolution compaction inputs changed or omit files from the captured row-ID groups")); + } + Ok(()) +} + +pub(in crate::table) fn validate_replacement_inputs( + delta: &[ManifestEntry], + active: &[ManifestEntry], +) -> Result<()> { + // Keep disjoint submitted task ranges separate: committing a subset must + // not turn its gaps into selected rows or require untouched groups. + let mut buckets = BTreeMap::<_, Vec>::new(); + for entry in delta + .iter() + .filter(|entry| *entry.kind() == FileKind::Delete) + { + buckets + .entry((entry.partition().to_vec(), entry.bucket())) + .or_default() + .push(entry.file().clone()); + } + for ((partition, bucket), files) in buckets { + for group in group_by_overlapping_row_id(files) { + validate_inputs(&partition, bucket, &group, active)?; + } + } + Ok(()) +} + +fn normal_fields(schema: &crate::spec::TableSchema) -> Vec { + let options = schema.core_options(); + let inline = options.blob_inline_fields(); + schema + .fields() + .iter() + .filter(|field| { + !(field.data_type().is_blob_file_field() && !inline.contains(field.name()) + || options.vector_file_format().is_some() + && matches!(field.data_type(), DataType::Vector(_))) + }) + .cloned() + .collect() +} + +async fn column_sequences( + table: &Table, + inputs: &[DataFileMeta], + fields: &[DataField], + max_sequence: i64, +) -> Result>> { + let mut sequences = HashMap::::new(); + for file in inputs { + let physical = input_fields(table, file).await?; + for (position, field) in physical.iter().enumerate() { + let sequence = file + .column_max_sequence_numbers + .as_ref() + .filter(|values| values.len() == physical.len()) + .and_then(|values| values.get(position)) + .copied() + .filter(|value| *value > 0 && *value <= file.max_sequence_number) + .unwrap_or(file.max_sequence_number); + sequences + .entry(field.id()) + .and_modify(|current| *current = (*current).max(sequence)) + .or_insert(sequence); + } + } + let result = fields + .iter() + .map(|field| sequences.get(&field.id()).copied().unwrap_or(max_sequence)) + .collect::>(); + Ok(result + .iter() + .any(|value| *value != max_sequence) + .then_some(result)) +} + +async fn input_fields(table: &Table, file: &DataFileMeta) -> Result> { + let schema = table.schema_manager().schema(file.schema_id).await?; + match &file.write_cols { + Some(columns) => { + crate::table::data_evolution_fields::project_by_paths(schema.fields(), columns) + } + None => Ok(schema.data_file_fields(None)), + } +} + +/// A layout option can start writing VECTOR fields to dedicated files while +/// older normal files still contain inline values. Dropping those providers +/// would turn readable values into NULL; migration requires a separate rewrite. +pub(in crate::table) async fn validate_storage_layout( + table: &Table, + files: &[DataFileMeta], +) -> Result<()> { + let normal_ids = normal_fields(table.schema()) + .into_iter() + .map(|field| field.id()) + .collect::>(); + let dedicated = table + .schema() + .fields() + .iter() + .filter(|field| !normal_ids.contains(&field.id())) + .map(|field| (field.id(), field.name())) + .collect::>(); + if dedicated.is_empty() { + return Ok(()); + } + for file in files { + for field in input_fields(table, file).await? { + if let Some(name) = dedicated.get(&field.id()) { + return Err(Error::Unsupported { + message: format!("Data-evolution compaction cannot discard historical inline field '{name}' from '{}'; migrate its values to dedicated storage before compacting", file.file_name), + }); + } + } + } + Ok(()) +} + +pub(super) async fn execute( + table: &Table, + task: &CompactTask, + split: &DataSplit, + partition_path: String, + resources: Option, +) -> Result> { + validate_storage_layout(table, task.files()).await?; + let (first, last) = output_range(task.files(), &table.schema().core_options())?; + let fields = normal_fields(table.schema()); + let min_sequence = task + .files + .iter() + .map(|file| file.min_sequence_number) + .min() + .unwrap(); + let max_sequence = task + .files + .iter() + .map(|file| file.max_sequence_number) + .max() + .unwrap(); + let sequences = column_sequences(table, &task.files, &fields, max_sequence).await?; + let options = table.schema().core_options(); + // Inline BLOB descriptors and views are normal columns. Preserve their + // references instead of fetching payloads or resolving upstream values. + let read_table = table.copy_with_options(HashMap::from([ + ("blob-as-descriptor".into(), "true".into()), + ("blob-view.resolve.enabled".into(), "false".into()), + ])); + let mut builder = read_table.new_read_builder(); + builder.with_read_type(fields.clone()); + let mut read = builder.new_read()?.with_ordered(true); + if let Some(resources) = &resources { + read = read.with_resources(resources.clone()); + } + // The executor's split has no deletion files: retain even deleted physical + // slots. Commit preparation reattaches the latest logical deletions. + let mut batches = read.to_arrow(std::slice::from_ref(split))?; + let index = FileIndexOptions::parse(table.schema().options(), &fields)?; + let mut writer = DataFileWriter::new( + table.file_io().clone(), + table.location().to_string(), + partition_path, + task.bucket, + table.schema().id(), + i64::MAX, + options.file_compression().to_string(), + options.file_compression_zstd_level(), + options.write_buffer_size()? as i64, + options.file_format(), + fields.clone(), + table.schema().options().clone(), + Some(1), + Some(first), + Some( + fields + .iter() + .map(|field| field.name().to_string()) + .collect(), + ), + )? + .with_file_index(index.map(Arc::new)) + .with_resources(resources); + let written = async { + let mut rows = 0i64; + while let Some(batch) = batches.try_next().await? { + rows += batch.num_rows() as i64; + writer.write(&batch).await?; + } + if rows != last - first + 1 { + return Err(invalid("Data-evolution compaction input row count changed")); + } + let mut files = writer.prepare_commit().await?; + for file in &mut files { + file.min_sequence_number = min_sequence; + file.max_sequence_number = max_sequence; + file.column_max_sequence_numbers = sequences.clone(); + } + validate_outputs(task, &files)?; + Ok(files) + } + .await; + if written.is_err() { + writer.abort().await; + } + written +} + +pub(super) fn validate_outputs(task: &CompactTask, files: &[DataFileMeta]) -> Result<()> { + let expected = output_range(&task.files, &task.context.schema.core_options())?; + let min_sequence = task + .files + .iter() + .map(|file| file.min_sequence_number) + .min() + .unwrap(); + let max_sequence = task + .files + .iter() + .map(|file| file.max_sequence_number) + .max() + .unwrap(); + if files.len() != 1 + || files[0].row_id_range() != Some(expected) + || files[0].min_sequence_number != min_sequence + || files[0].max_sequence_number != max_sequence + || files[0].write_cols + != Some( + normal_fields(&task.context.schema) + .iter() + .map(|field| field.name().to_string()) + .collect(), + ) + { + return Err(invalid( + "Data-evolution compaction must preserve its row-ID range, columns and sequence bounds", + )); + } + Ok(()) +} + +pub(super) fn validate_indexes(entries: &[IndexManifestEntry]) -> Result<()> { + if entries.iter().any(|entry| { + entry.index_file.index_type != "DELETION_VECTORS" + && entry.index_file.global_index_meta.is_none() + }) { + return Err(Error::Unsupported { + message: "Data-evolution compaction supports deletion vectors and global indexes" + .into(), + }); + } + Ok(()) +} + +/// Rebase DVs on the latest snapshot on every CAS attempt. A concurrent DELETE +/// is carried forward, including deletions in untouched groups of this bucket. +pub(in crate::table) async fn rewrite_deletion_vectors( + table: &Table, + snapshot: &Snapshot, + delta: &[ManifestEntry], + created_files: &mut Vec, +) -> Result>> { + let Some(name) = snapshot.index_manifest() else { + return Ok(None); + }; + let mut indexes = IndexManifest::read( + table.file_io(), + &table.snapshot_manager().manifest_path(name), + ) + .await?; + let mut buckets = BTreeMap::<_, (Vec, Vec)>::new(); + for entry in delta { + let files = buckets + .entry((entry.partition().to_vec(), entry.bucket())) + .or_default(); + match entry.kind() { + FileKind::Delete => files.0.push(entry.file().clone()), + FileKind::Add => files.1.push(entry.file().clone()), + } + } + let writer = DataEvolutionDeleteWriter::new(table)?; + let mut changed = false; + for ((partition, bucket), (before, after)) in buckets { + let groups = group_by_overlapping_row_id(before); + let anchors = groups + .iter() + .map(|group| Ok(data_evolution_anchor_file(group)?.file_name.clone())) + .collect::>>()?; + // Like Java AppendDeleteFileMaintainer, replace only index files + // touched by selected anchors. Untouched files remain independent. + let (mut bitmaps, removed) = writer + .read_existing_bucket_deletion_vectors(&partition, bucket, snapshot, Some(&anchors)) + .await?; + let mut moved = false; + for group in groups { + let anchor = data_evolution_anchor_file(&group)?; + let Some(bitmap) = bitmaps.shift_remove(&anchor.file_name) else { + continue; + }; + moved = true; + let old_first = anchor.first_row_id.unwrap(); + for position in bitmap { + if i64::from(position) >= anchor.row_count { + return Err(invalid( + "Deletion vector position is outside its compact input", + )); + } + let absolute = old_first + i64::from(position); + let output = after + .iter() + .find(|file| { + file.row_id_range() + .is_some_and(|(start, end)| start <= absolute && absolute <= end) + }) + .ok_or_else(|| invalid("Compacted files do not cover a deleted row ID"))?; + let relative = u32::try_from(absolute - output.first_row_id.unwrap()) + .map_err(|_| invalid("Compacted deletion vector position exceeds u32::MAX"))?; + bitmaps + .entry(output.file_name.clone()) + .or_insert_with(RoaringBitmap::new) + .insert(relative); + } + } + if !moved { + continue; + } + let file = writer + .write_deletion_vector_index_file(&partition, bucket, bitmaps) + .await?; + let computer = PartitionComputer::new( + table.schema().partition_keys(), + table.schema().fields(), + table.schema().core_options().partition_default_name(), + table.schema().core_options().legacy_partition_name(), + )?; + let bucket_dir = bucket_path( + &table.data_file_location(), + Some(&computer), + &BinaryRow::from_serialized_bytes(&partition)?, + bucket, + )?; + created_files.push(committed_index_file_path( + table.location(), + &bucket_dir, + table.schema().core_options().index_file_in_data_file_dir(), + &file, + )); + indexes.retain(|entry| { + !removed.iter().any(|old| { + entry.partition == partition && entry.bucket == bucket && entry.index_file == *old + }) + }); + indexes.push(IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition, + bucket, + index_file: file, + }); + changed = true; + } + Ok(changed.then_some(indexes)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::stats::BinaryTableStats; + + fn file(name: &str, first: i64, rows: i64, bytes: i64) -> DataFileMeta { + DataFileMeta { + file_name: name.into(), + file_size: bytes, + row_count: rows, + min_key: vec![], + max_key: vec![], + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 1, + max_sequence_number: 1, + schema_id: 0, + level: 0, + extra_files: vec![], + creation_time: None, + delete_row_count: Some(0), + embedded_index: None, + file_source: Some(0), + value_stats_cols: None, + external_path: None, + first_row_id: Some(first), + write_cols: None, + column_max_sequence_numbers: None, + } + } + + fn planned( + files: Vec, + target: i64, + open_cost: i64, + min: usize, + ) -> Vec> { + let options = HashMap::from([ + ("target-file-size".into(), format!("{target}b")), + ( + "source.split.open-file-cost".into(), + format!("{open_cost}b"), + ), + ("compaction.min.file-num".into(), min.to_string()), + ]); + plan(files, &CoreOptions::new(&options)) + .unwrap() + .into_iter() + .map(|group| group.into_iter().map(|file| file.file_name).collect()) + .collect() + } + + #[test] + fn java_contiguous_bin_boundary_and_open_cost() { + let files = vec![ + file("a", 0, 100, 100), + file("b", 100, 100, 100), + file("c", 200, 100, 100), + ]; + assert_eq!(planned(files.clone(), 199, 1, 2), vec![vec!["a", "b"]]); + assert_eq!(planned(files.clone(), 200, 1, 2), vec![vec!["a", "b", "c"]]); + assert_eq!( + planned(files, 199, 200, 2), + Vec::>::new(), + "each heavy singleton is isolated" + ); + } + + #[test] + fn java_gap_and_large_file_boundaries() { + let files = vec![ + file("a", 0, 10, 10), + file("b", 10, 10, 10), + file("large", 20, 10, 1000), + file("c", 100, 10, 10), + file("d", 110, 10, 10), + ]; + assert_eq!( + planned(files, 100, 1, 2), + vec![vec!["a", "b"], vec!["c", "d"]] + ); + } + + #[test] + fn versions_stay_together_even_when_the_group_exceeds_target() { + let files = vec![ + file("tail", 0, 10, 1), + file("base", 10, 10, 80), + file("partial", 10, 10, 80), + file("last", 20, 10, 1), + ]; + assert_eq!(planned(files, 100, 1, 2), vec![vec!["base", "partial"]]); + } + + #[test] + fn rejects_gaps_misaligned_versions_and_overflow() { + assert!(row_range(&[file("a", 0, 10, 1), file("b", 20, 10, 1)]).is_err()); + assert!(row_range(&[file("a", 0, 10, 1), file("b", 5, 10, 1)]).is_err()); + assert!(row_range(&[file("a", i64::MAX, 1, 1)]).is_err()); + } + + #[test] + fn dv_output_range_keeps_every_row_addressable_after_compaction() { + let mut options = HashMap::from([ + ("data-evolution.enabled".into(), "true".into()), + ("deletion-vectors.enabled".into(), "true".into()), + ("compaction.min.file-num".into(), "2".into()), + ]); + let max_rows = i64::from(u32::MAX) + 1; + let too_wide = vec![file("a", 0, max_rows, 1), file("b", max_rows, 1, 1)]; + let error = plan(too_wide.clone(), &CoreOptions::new(&options)).unwrap_err(); + assert!(matches!(error, Error::Unsupported { .. })); + assert!(error.to_string().contains("64-bit deletion-vector writing")); + let at_limit = vec![file("a", 0, max_rows - 1, 1), file("b", max_rows - 1, 1, 1)]; + assert_eq!( + plan(at_limit, &CoreOptions::new(&options)).unwrap().len(), + 1 + ); + options.insert("deletion-vectors.enabled".into(), "false".into()); + assert_eq!( + plan(too_wide, &CoreOptions::new(&options)).unwrap().len(), + 1 + ); + } +} diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index 470f7aa02..e14e17816 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -747,7 +747,7 @@ impl DataEvolutionDeleteWriter { snapshot: &Snapshot, ) -> Result> { let (mut bitmaps, deleted_index_files) = self - .read_existing_bucket_deletion_vectors(&partition, bucket, snapshot) + .read_existing_bucket_deletion_vectors(&partition, bucket, snapshot, None) .await?; let mut changed = false; @@ -820,11 +820,12 @@ impl DataEvolutionDeleteWriter { }) } - async fn read_existing_bucket_deletion_vectors( + pub(in crate::table) async fn read_existing_bucket_deletion_vectors( &self, partition: &[u8], bucket: i32, snapshot: &Snapshot, + selected_anchors: Option<&HashSet>, ) -> Result<(IndexMap, Vec)> { let Some(index_manifest_name) = snapshot.index_manifest() else { return Ok((IndexMap::new(), Vec::new())); @@ -845,6 +846,13 @@ impl DataEvolutionDeleteWriter { || entry.bucket != bucket || entry.partition != partition || entry.index_file.index_type != DELETION_VECTORS_INDEX_TYPE + || selected_anchors.is_some_and(|anchors| { + entry + .index_file + .deletion_vectors_ranges + .as_ref() + .is_none_or(|ranges| !ranges.keys().any(|name| anchors.contains(name))) + }) { continue; } @@ -873,7 +881,7 @@ impl DataEvolutionDeleteWriter { Ok((bitmaps, deleted_index_files)) } - async fn write_deletion_vector_index_file( + pub(in crate::table) async fn write_deletion_vector_index_file( &self, partition: &[u8], bucket: i32, diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 8e64944e4..a56f6e022 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -707,7 +707,7 @@ impl Table { WriteBuilder::new(self) } - /// Plan and execute append or fixed/dynamic bucket PK compaction. + /// Plan and execute append, data-evolution or fixed/dynamic bucket PK compaction. pub fn new_compact_builder(&self) -> CompactBuilder<'_> { CompactBuilder::new(self) } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 3592b191d..3a2245438 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -1360,6 +1360,29 @@ impl TableCommit { ) -> Result<(Snapshot, Vec)> { let new_snapshot_id = latest_snapshot.as_ref().map(|s| s.id() + 1).unwrap_or(1); + if self.data_evolution_enabled + && resolved.kind == CommitKind::COMPACT + && self + .table + .schema() + .core_options() + .deletion_vectors_enabled() + { + if let Some(snapshot) = latest_snapshot { + if let Some(indexes) = super::compact::data_evolution::rewrite_deletion_vectors( + &self.table, + snapshot, + &resolved.entries, + created_files, + ) + .await? + { + resolved.index_manifest_changed = true; + resolved.new_index_manifest_entries = Some(indexes); + } + } + } + // Row tracking let mut next_row_id: Option = None; if self.row_tracking_enabled { @@ -1897,8 +1920,8 @@ impl TableCommit { source: None, }); } - // Concurrent writes may replace the HASH index after planning. - // Reuse the latest manifest on every retry, never the captured one. + // Concurrent writes may replace HASH indexes, deletion vectors + // or global indexes. Rebase on the latest manifest every retry. let index_manifest_name = match latest_snapshot { Some(snapshot) => { super::compact::validate_snapshot_indexes(&self.table, snapshot).await? @@ -1930,9 +1953,31 @@ impl TableCommit { .iter() .map(|e| (e.partition(), e.bucket())) .collect::>(); + let deleted_ids = deletes + .iter() + .map(|entry| commit_file_identifier(entry)) + .collect::>(); + if self.data_evolution_enabled { + super::compact::data_evolution::validate_replacement_inputs( + entries, + &captured_files, + )?; + super::compact::data_evolution::validate_storage_layout( + &self.table, + &deletes + .iter() + .map(|entry| entry.file().clone()) + .collect::>(), + ) + .await?; + } let captured_files = captured_files .iter() - .filter(|e| buckets.contains(&(e.partition(), e.bucket()))) + .filter(|e| { + buckets.contains(&(e.partition(), e.bucket())) + && (!self.data_evolution_enabled + || deleted_ids.contains(&commit_file_identifier(e))) + }) .map(|entry| (commit_file_identifier(entry), entry)) .collect::>(); if captured_files.len() != deletes.len() @@ -1946,8 +1991,11 @@ impl TableCommit { }) { return Err(crate::Error::DataInvalid { - message: "PK compaction must include the complete captured bucket" - .into(), + message: if self.data_evolution_enabled { + "Data-evolution compaction inputs must match the captured row-ID groups" + } else { + "PK compaction must include the complete captured bucket" + }.into(), source: None, }); } @@ -1985,6 +2033,12 @@ impl TableCommit { }); } } + if self.data_evolution_enabled { + super::compact::data_evolution::validate_replacement_inputs( + entries, + base_data_files.as_deref().unwrap_or_default(), + )?; + } self.check_compact_lsm_conflicts( base_data_files.as_deref().unwrap_or_default(), entries, @@ -2818,7 +2872,7 @@ impl TableCommit { } let next_row_id = latest_snapshot.and_then(Snapshot::next_row_id); - self.check_row_id_existence(base_entries, delta_entries, next_row_id)?; + self.check_row_id_existence(base_entries, delta_entries, next_row_id, commit_kind)?; self.check_row_id_range_conflicts(commit_kind, check_from_snapshot, &merged_entries)?; self.check_row_id_from_snapshot(latest_snapshot, delta_entries, check_from_snapshot) @@ -3053,6 +3107,7 @@ impl TableCommit { base_entries: &[ManifestEntry], delta_entries: &[ManifestEntry], next_row_id: Option, + commit_kind: &CommitKind, ) -> Result<()> { // File identity deliberately excludes row IDs. Check DELETE metadata // before identifier-based merging can cancel a reassigned current ADD. @@ -3120,14 +3175,30 @@ impl TableCommit { for entry in files_to_check { let first_row_id = entry.file().first_row_id.unwrap(); - if is_dedicated_storage_file(entry.file()) { + if is_dedicated_storage_file(entry.file()) || commit_kind == &CommitKind::COMPACT { if let Some((start, end)) = entry.file().row_id_range() { let covered_by_existing = existing_ranges .get(&(entry.partition().to_vec(), entry.bucket())) .is_some_and(|ranges| { - ranges.iter().any(|&(base_start, base_end)| { - base_start <= start && end <= base_end - }) + if commit_kind == &CommitKind::COMPACT { + let mut ranges = ranges.clone(); + ranges.sort_unstable(); + let mut cursor = start; + for (base_start, base_end) in ranges { + if base_start > cursor { + break; + } + if base_end >= end { + return true; + } + cursor = cursor.max(base_end.saturating_add(1)); + } + false + } else { + ranges.iter().any(|&(base_start, base_end)| { + base_start <= start && end <= base_end + }) + } }); if covered_by_existing { continue; diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs b/crates/paimon/src/table/table_commit/parity_tests.rs index bd2724255..00a7b3c6a 100644 --- a/crates/paimon/src/table/table_commit/parity_tests.rs +++ b/crates/paimon/src/table/table_commit/parity_tests.rs @@ -357,6 +357,91 @@ async fn publication_errors_retry_without_duplicate_snapshots() { } } +#[tokio::test] +async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_published_vectors() { + use arrow_array::{Int32Array, RecordBatch}; + use futures::TryStreamExt; + + struct RacingCompactCommit { + table: Table, + calls: std::sync::atomic::AtomicUsize, + publish_first: bool, + } + + #[async_trait::async_trait] + impl SnapshotCommit for RacingCompactCommit { + async fn commit( + &self, + _: Option<&str>, + snapshot: &Snapshot, + _: &str, + _: &[PartitionStatistics], + ) -> Result { + if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 { + if self.publish_first { + assert!(self.table.snapshot_manager().commit_snapshot(snapshot).await?); + return Err(crate::Error::IoUnsupported { message: "lost compact response".into() }); + } + // Commit another DELETE after compact prepared its DV file, + // then force its CAS to lose. Retry must remap the newer DV. + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&self.table)?; + delete.add_row_ids([2])?; + self.table.new_write_builder().new_commit().commit(delete.prepare_commit().await?).await?; + return Ok(false); + } + self.table.snapshot_manager().commit_snapshot(snapshot).await + } + } + + for publish_first in [false, true] { + let io = test_file_io(); + let path = format!("memory:/de-compact-cas-{publish_first}"); + setup_dirs(&io, &path).await; + let schema = crate::spec::Schema::builder() + .column("id", DataType::Int(crate::spec::IntType::new())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("deletion-vectors.enabled", "true") + .option("compaction.min.file-num", "2") + .build().unwrap(); + let table = Table::new(io.clone(), Identifier::new("default", "de_compact_cas"), path.clone(), TableSchema::new(0, &schema), None); + io.mkdirs(&format!("{path}/schema")).await.unwrap(); + io.new_output(&format!("{path}/schema/schema-0")).unwrap().write(serde_json::to_vec(table.schema()).unwrap().into()).await.unwrap(); + for ids in [vec![0, 1], vec![2, 3]] { + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + write.write_arrow_batch(&RecordBatch::try_from_iter([("id", Arc::new(Int32Array::from(ids)) as arrow_array::ArrayRef)]).unwrap()).await.unwrap(); + builder.new_commit().commit(write.prepare_commit().await.unwrap()).await.unwrap(); + } + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&table).unwrap(); + delete.add_row_ids([1]).unwrap(); + table.new_write_builder().new_commit().commit(delete.prepare_commit().await.unwrap()).await.unwrap(); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let result = builder.new_executor().unwrap().execute(&plan.tasks()[0]).await.unwrap(); + let publisher = Arc::new(RacingCompactCommit { + table: table.clone(), calls: std::sync::atomic::AtomicUsize::new(0), publish_first, + }); + let mut commit = builder.new_commit().unwrap(); + commit.snapshot_commit = publisher.clone(); + commit.commit_min_retry_wait_ms = 0; + commit.commit_max_retry_wait_ms = 0; + let metrics = commit.commit_compact(vec![result.clone()]).await.unwrap(); + assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), if publish_first { 1 } else { 2 }); + let after = table.snapshot_manager().get_latest_snapshot().await.unwrap().unwrap(); + assert_eq!(after.id(), if publish_first { 4 } else { 5 }); + assert_eq!(after.commit_kind(), &CommitKind::COMPACT); + assert_eq!(commit.commit_compact(vec![result]).await.unwrap(), metrics); + assert_eq!(table.snapshot_manager().get_latest_snapshot_id().await.unwrap(), Some(after.id())); + let read = table.new_read_builder(); + let scan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read.new_read().unwrap().to_arrow(scan.splits()).unwrap().try_collect().await.unwrap(); + let mut ids = batches.iter().flat_map(|batch| batch.column(0).as_any().downcast_ref::().unwrap().values().iter().copied()).collect::>(); + ids.sort_unstable(); + assert_eq!(ids, if publish_first { vec![0, 2, 3] } else { vec![0, 3] }); + } +} + #[tokio::test] async fn mixed_append_cannot_bypass_stale_dv_check() { let io = test_file_io(); diff --git a/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs index 99e90a4f1..83daebc8a 100644 --- a/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs +++ b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs @@ -24,7 +24,7 @@ fn row_id_delete_must_match_the_current_file_layout() { file.first_row_id = Some(0); let stale_delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, file, 2); let error = commit - .check_row_id_existence(&[current], &[stale_delete], Some(20)) + .check_row_id_existence(&[current], &[stale_delete], Some(20), &CommitKind::APPEND) .expect_err("a DELETE must not cancel the current file after row-ID reassignment"); assert!(error.to_string().contains("Row ID existence conflict")); assert!(matches!( @@ -47,7 +47,7 @@ fn row_id_existence_preserves_missing_range_details() { file.first_row_id = Some(7); let entry = ManifestEntry::new(FileKind::Add, vec![], 3, -1, file, 2); let error = commit - .check_row_id_existence(&[], &[entry], Some(20)) + .check_row_id_existence(&[], &[entry], Some(20), &CommitKind::APPEND) .unwrap_err(); assert!(matches!(error, crate::Error::RowIdExistenceConflict { ref file_name, first_row_id: Some(7), row_count: 4, bucket: 3, @@ -55,6 +55,22 @@ fn row_id_existence_preserves_missing_range_details() { } if file_name == "opaque '(4): file.parquet")); } +#[test] +fn compact_row_id_existence_requires_contiguous_normal_coverage() { + let commit = setup_data_evolution_commit(&test_file_io(), "memory:/compact_range_existence"); + let entry = |name: &str, first: i64, count: i64| { + let mut file = test_data_file(name, count); + file.first_row_id = Some(first); + ManifestEntry::new(FileKind::Add, vec![], 0, -1, file, 2) + }; + let output = entry("compact.parquet", 0, 10); + let contiguous = [entry("a.parquet", 0, 5), entry("b.parquet", 5, 5)]; + commit.check_row_id_existence(&contiguous, std::slice::from_ref(&output), Some(10), &CommitKind::COMPACT).unwrap(); + assert!(commit.check_row_id_existence(&contiguous, std::slice::from_ref(&output), Some(10), &CommitKind::APPEND).is_err(), "DML must still match an exact source range"); + let gap = [entry("a.parquet", 0, 4), entry("b.parquet", 5, 5), entry("stale.blob", 0, 10)]; + assert!(commit.check_row_id_existence(&gap, &[output], Some(10), &CommitKind::COMPACT).is_err(), "neither a gap nor a dedicated file can supply normal row slots"); +} + #[test] fn row_id_delete_metadata_check_does_not_require_next_row_id() { let commit = setup_data_evolution_commit(&test_file_io(), "memory:/delete_without_counter"); @@ -66,7 +82,7 @@ fn row_id_delete_metadata_check_does_not_require_next_row_id() { old.first_row_id = Some(0); let old = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, old, 2); assert!( - matches!(commit.check_row_id_existence(&[current], &[old], None), + matches!(commit.check_row_id_existence(&[current], &[old], None, &CommitKind::APPEND), Err(crate::Error::RowIdExistenceConflict { current_first_row_id, current_row_count: Some(count), .. }) if current_first_row_id == current_id && count == current_count) ); @@ -75,9 +91,9 @@ fn row_id_delete_metadata_check_does_not_require_next_row_id() { let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1, file.clone(), 2); let delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, file, 2); commit - .check_row_id_existence(&[current], &[delete], None) + .check_row_id_existence(&[current], &[delete], None, &CommitKind::APPEND) .unwrap(); - commit.check_row_id_existence(&[], &[], None).unwrap(); + commit.check_row_id_existence(&[], &[], None, &CommitKind::APPEND).unwrap(); } fn row_id_check_message(name: &str, column: &str) -> CommitMessage { diff --git a/crates/paimon/src/table/table_commit/row_tracking_tests.rs b/crates/paimon/src/table/table_commit/row_tracking_tests.rs index 16f89ccb2..eff3e6123 100644 --- a/crates/paimon/src/table/table_commit/row_tracking_tests.rs +++ b/crates/paimon/src/table/table_commit/row_tracking_tests.rs @@ -393,7 +393,7 @@ fn row_tracking_dedicated_file_cannot_supply_normal_range_existence() { for name in ["update.blob", "update.parquet"] { let delta = vec![entry(name, 0, 10)]; assert!(commit - .check_row_id_existence(&base, &delta, Some(10)) + .check_row_id_existence(&base, &delta, Some(10), &CommitKind::APPEND) .is_err()); } // The merged-layout check must also catch an older dedicated file that diff --git a/crates/paimon/tests/append_compact_test.rs b/crates/paimon/tests/append_compact_test.rs index a67da1847..adbab7301 100644 --- a/crates/paimon/tests/append_compact_test.rs +++ b/crates/paimon/tests/append_compact_test.rs @@ -468,10 +468,6 @@ async fn compact_keeps_partitions_and_buckets_separate_and_honors_partition_filt async fn compact_refuses_unsupported_tables_invalid_options_and_foreign_tasks() { for options in [ vec![("row-tracking.enabled", "true")], - vec![ - ("data-evolution.enabled", "true"), - ("row-tracking.enabled", "true"), - ], vec![("deletion-vectors.enabled", "true")], vec![("bucket", "1")], vec![ diff --git a/crates/paimon/tests/data_evolution_compact_layout_test.rs b/crates/paimon/tests/data_evolution_compact_layout_test.rs new file mode 100644 index 000000000..5158025dc --- /dev/null +++ b/crates/paimon/tests/data_evolution_compact_layout_test.rs @@ -0,0 +1,164 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod common; + +use std::sync::Arc; + +use arrow_array::{ + types::Float32Type, Array, ArrayRef, FixedSizeListArray, Float32Array, Int32Array, RecordBatch, +}; +use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs, write_batch}; +use futures::TryStreamExt; +use paimon::spec::{DataType, FloatType, IntType, Schema, SchemaChange, TableSchema, VectorType}; +use paimon::table::Table; + +async fn vector_rows(table: &Table) -> Vec<(i32, Option>)> { + let read = table.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut rows = Vec::new(); + for batch in batches { + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let vectors = batch + .column_by_name("embedding") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + let value = (!vectors.is_null(row)).then(|| { + let vector = vectors.value(row); + vector + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }); + rows.push((ids.value(row), value)); + } + } + rows.sort_by_key(|row| row.0); + rows +} + +#[tokio::test] +async fn compact_rejects_historical_inline_vectors_after_enabling_dedicated_format() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column( + "embedding", + DataType::Vector( + VectorType::try_new(true, 2, DataType::Float(FloatType::new())).unwrap(), + ), + ) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("compaction.min.file-num", "2") + .build() + .unwrap(); + let location = format!("memory:/de_compact_vector_layout/{}", uuid::Uuid::new_v4()); + let (io, original) = memory_table(&location, TableSchema::new(0, &schema)); + setup_dirs(&io, &location).await; + persist_table_schema(&io, &location, original.schema()).await; + for id in [1, 2] { + write_batch( + &original, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef), + ( + "embedding", + Arc::new( + FixedSizeListArray::from_iter_primitive::( + vec![Some(vec![Some(id as f32), Some(10.0)])], + 2, + ), + ) as ArrayRef, + ), + ]) + .unwrap(), + ) + .await; + } + let expected = vec![(1, Some(vec![1.0, 10.0])), (2, Some(vec![2.0, 10.0]))]; + assert_eq!(vector_rows(&original).await, expected); + let original_plan = original + .new_compact_builder() + .new_planner() + .unwrap() + .plan() + .await + .unwrap(); + let schema = original + .schema() + .apply_changes(vec![SchemaChange::SetOption { + key: "vector.file.format".into(), + value: "parquet".into(), + }]) + .unwrap(); + persist_table_schema(&io, &location, &schema).await; + let evolved = Table::new(io, original.identifier().clone(), location, schema, None); + assert_eq!( + vector_rows(&evolved).await, + expected, + "the supported reader must still see inline vectors before compaction" + ); + let builder = evolved.new_compact_builder(); + let error = builder.new_planner().unwrap().plan().await.unwrap_err(); + assert!(matches!(error, paimon::Error::Unsupported { .. })); + assert!(error + .to_string() + .contains("historical inline field 'embedding'")); + // Workers independently enforce the layout guard when handed a serialized + // task, even if its captured table schema matches the current options. + let mut task = serde_json::to_value(&original_plan.tasks()[0]).unwrap(); + task["context"]["schema"] = serde_json::to_value(evolved.schema()).unwrap(); + let task: paimon::table::CompactTask = serde_json::from_value(task).unwrap(); + let error = builder + .new_executor() + .unwrap() + .execute(&task) + .await + .unwrap_err(); + assert!(matches!(error, paimon::Error::Unsupported { .. })); + assert_eq!( + evolved + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + Some(2) + ); + assert_eq!( + vector_rows(&evolved).await, + expected, + "rejected compaction must preserve every historical inline vector" + ); +} diff --git a/crates/paimon/tests/data_evolution_compact_test.rs b/crates/paimon/tests/data_evolution_compact_test.rs new file mode 100644 index 000000000..b1dbcaf35 --- /dev/null +++ b/crates/paimon/tests/data_evolution_compact_test.rs @@ -0,0 +1,1085 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod common; + +use std::sync::Arc; + +use arrow_array::{Array, ArrayRef, Int32Array, Int64Array, RecordBatch}; +use common::incremental_helpers::{memory_table, persist_table_schema, setup_dirs, write_batch}; +use futures::TryStreamExt; +use paimon::spec::{CommitKind, DataType, IntType, Schema, TableSchema}; +use paimon::table::{CompactPlan, CompactResult, DataEvolutionDeleteWriter, Table}; + +async fn table(options: &[(&str, &str)]) -> Table { + let mut schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("compaction.min.file-num", "2"); + for (key, value) in options { + schema = schema.option(*key, *value); + } + create_table(schema.build().unwrap()).await +} + +async fn create_table(schema: Schema) -> Table { + let location = format!("memory:/de_compact/{}", uuid::Uuid::new_v4()); + let (io, table) = memory_table(&location, TableSchema::new(0, &schema)); + setup_dirs(&io, &location).await; + persist_table_schema(&io, &location, table.schema()).await; + table +} + +async fn append(table: &Table, start: i32) { + write_batch( + table, + &RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(vec![start, start + 1])) as ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![Some(10), None])) as ArrayRef, + ), + ]) + .unwrap(), + ) + .await; +} + +async fn update(table: &Table, row_id: i64, value: Option) { + let builder = table.new_write_builder(); + let mut writer = builder + .new_data_evolution_writer(vec!["value".into()]) + .unwrap(); + writer + .add_matched_batch( + RecordBatch::try_from_iter([ + ( + "_ROW_ID", + Arc::new(Int64Array::from(vec![row_id])) as ArrayRef, + ), + ("value", Arc::new(Int32Array::from(vec![value])) as ArrayRef), + ]) + .unwrap(), + ) + .unwrap(); + builder + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); +} + +async fn delete(table: &Table, ids: &[i64]) { + let mut writer = DataEvolutionDeleteWriter::new(table).unwrap(); + writer.add_row_ids(ids.iter().copied()).unwrap(); + table + .new_write_builder() + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); +} + +async fn rows(table: &Table) -> Vec<(i64, i32, Option)> { + let mut builder = table.new_read_builder(); + builder + .with_projection(&["_ROW_ID", "id", "value"]) + .unwrap(); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut result = vec![]; + for batch in batches { + let row_ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let ids = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + result.extend((0..batch.num_rows()).map(|i| { + ( + row_ids.value(i), + ids.value(i), + (!values.is_null(i)).then(|| values.value(i)), + ) + })); + } + result.sort_unstable(); + result +} + +async fn execute(table: &Table) -> (CompactPlan, Vec) { + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let plan: CompactPlan = serde_json::from_slice(&serde_json::to_vec(&plan).unwrap()).unwrap(); + let executor = builder.new_executor().unwrap(); + let mut results = vec![]; + for task in plan.tasks() { + results.push(executor.execute(task).await.unwrap()); + } + ( + plan, + serde_json::from_slice(&serde_json::to_vec(&results).unwrap()).unwrap(), + ) +} + +#[tokio::test] +async fn compact_merges_contiguous_ranges_and_latest_columns_without_reassigning_rows() { + for format in ["parquet", "orc", "avro", "row"] { + let table = table(&[ + ("file.format", format), + ("file.compression", "none"), + ("target-file-row-num", "2"), + ]) + .await; + append(&table, 0).await; + append(&table, 2).await; + update(&table, 0, Some(99)).await; + let expected = rows(&table).await; + let before = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let (plan, results) = execute(&table).await; + assert_eq!(plan.tasks().len(), 1, "{format}"); + assert_eq!(plan.tasks()[0].files().len(), 3); + assert_eq!( + results[0].files().len(), + 1, + "write-time row limit must not split a compact range" + ); + let output = &results[0].files()[0]; + assert_eq!(output.first_row_id, Some(0)); + assert_eq!(output.row_count, 4); + assert_eq!( + (output.min_sequence_number, output.max_sequence_number), + (1, 3) + ); + let committer = table.new_compact_builder().new_commit().unwrap(); + let metrics = committer.commit_compact(results.clone()).await.unwrap(); + assert_eq!(metrics.rows_written, 4); + assert_eq!(metrics.files_removed, 3); + assert_eq!(rows(&table).await, expected, "{format}"); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&before))).await, + expected + ); + let after = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + assert_eq!(after.next_row_id(), before.next_row_id()); + assert_eq!(after.commit_kind(), &CommitKind::COMPACT); + assert_eq!(committer.commit_compact(results).await.unwrap(), metrics); + assert!(execute(&table).await.0.tasks().is_empty()); + update(&table, 3, Some(123)).await; + append(&table, 4).await; + assert_eq!( + rows(&table).await, + vec![ + (0, 0, Some(99)), + (1, 1, None), + (2, 2, Some(10)), + (3, 3, Some(123)), + (4, 4, Some(10)), + (5, 5, None) + ] + ); + } +} + +#[tokio::test] +async fn compact_rejects_concurrent_column_updates_but_accepts_disjoint_appends() { + let table = table(&[]).await; + append(&table, 0).await; + update(&table, 0, Some(99)).await; + let (_, stale) = execute(&table).await; + update(&table, 1, Some(55)).await; + let committer = table.new_compact_builder().new_commit().unwrap(); + assert!(committer.commit_compact(stale).await.is_err()); + assert_eq!(rows(&table).await, vec![(0, 0, Some(99)), (1, 1, Some(55))]); + let (_, fresh) = execute(&table).await; + append(&table, 2).await; + committer.commit_compact(fresh).await.unwrap(); + assert_eq!( + rows(&table).await, + vec![ + (0, 0, Some(99)), + (1, 1, Some(55)), + (2, 2, Some(10)), + (3, 3, None) + ] + ); +} + +#[tokio::test] +async fn compact_moves_latest_deletion_vectors_and_keeps_physical_slots() { + for in_data_dir in ["true", "false"] { + let table = table(&[ + ("deletion-vectors.enabled", "true"), + ("index-file-in-data-file-dir", in_data_dir), + ]) + .await; + append(&table, 0).await; + append(&table, 2).await; + update(&table, 0, Some(99)).await; + delete(&table, &[1]).await; + let before = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let (_, results) = execute(&table).await; + assert_eq!( + results[0].files()[0].row_count, + 4, + "deleted positions must be written too" + ); + delete(&table, &[2]).await; + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let indexes = paimon::spec::IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(snapshot.index_manifest().unwrap()), + ) + .await + .unwrap(); + assert_eq!(indexes.len(), 1); + let name = &indexes[0].index_file.file_name; + let bucket_path = format!("{}/bucket-0/{name}", table.location()); + let index_path = format!("{}/index/{name}", table.location()); + assert_eq!( + table.file_io().exists(&bucket_path).await.unwrap(), + in_data_dir == "true" + ); + assert_eq!( + table.file_io().exists(&index_path).await.unwrap(), + in_data_dir == "false" + ); + assert_eq!(rows(&table).await, vec![(0, 0, Some(99)), (3, 3, None)]); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&before))).await, + vec![(0, 0, Some(99)), (2, 2, Some(10)), (3, 3, None)] + ); + update(&table, 3, Some(77)).await; + delete(&table, &[0]).await; + assert_eq!(rows(&table).await, vec![(3, 3, Some(77))]); + } +} + +#[tokio::test] +async fn compact_retains_dedicated_blob_and_vector_files() { + use arrow_array::{types::Float32Type, FixedSizeListArray, LargeBinaryArray}; + use paimon::spec::{BlobType, FloatType, VectorType}; + + let table = create_table( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("payload", DataType::Blob(BlobType::new())) + .column( + "embedding", + DataType::Vector( + VectorType::try_new(true, 2, DataType::Float(FloatType::new())).unwrap(), + ), + ) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("compaction.min.file-num", "2") + .option("vector.file.format", "parquet") + .build() + .unwrap(), + ) + .await; + for id in [0, 1] { + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef), + ( + "value", + Arc::new(Int32Array::from(vec![id * 10])) as ArrayRef, + ), + ( + "payload", + Arc::new(LargeBinaryArray::from(vec![Some(b"retained".as_slice())])) + as ArrayRef, + ), + ( + "embedding", + Arc::new( + FixedSizeListArray::from_iter_primitive::( + vec![Some(vec![Some(id as f32), Some(1.0)])], + 2, + ), + ) as ArrayRef, + ), + ]) + .unwrap(), + ) + .await; + } + let mut read = table.new_read_builder(); + read.with_projection(&["id", "payload", "embedding"]) + .unwrap(); + let original = read.new_scan().plan().await.unwrap(); + let blobs = original + .splits() + .iter() + .flat_map(|split| split.data_files()) + .filter(|file| file.file_name.ends_with(".blob") || file.file_name.contains(".vector.")) + .cloned() + .collect::>(); + assert_eq!(blobs.len(), 4); + let (_, results) = execute(&table).await; + assert_eq!( + results[0].files()[0].write_cols, + Some(vec!["id".into(), "value".into()]) + ); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let latest = read.new_scan().plan().await.unwrap(); + for blob in blobs { + assert!(latest + .splits() + .iter() + .flat_map(|split| split.data_files()) + .any(|file| file == &blob)); + } + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(latest.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 2 + ); + for batch in batches { + let payload = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(payload + .iter() + .all(|value| value == Some(b"retained".as_slice()))); + let vectors = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + let ids = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + let vector = vectors.value(row); + let values = vector + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!(values.values().as_ref(), &[ids.value(row) as f32, 1.0]); + } + } +} + +#[tokio::test] +async fn compact_keeps_global_indexes_and_column_freshness() { + use paimon::spec::{Datum, IndexManifest, PredicateBuilder}; + + let table = table(&[("global-index.column-update-action", "ignore")]).await; + append(&table, 0).await; + append(&table, 2).await; + table + .new_global_index_build_builder() + .with_index_type("bitmap") + .with_index_column("id") + .execute() + .await + .unwrap(); + update(&table, 0, Some(99)).await; + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let indexes = IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(snapshot.index_manifest().unwrap()), + ) + .await + .unwrap(); + let (_, results) = execute(&table).await; + assert_eq!( + results[0].files()[0].column_max_sequence_numbers, + Some(vec![2, 4]) + ); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let latest = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + assert_eq!( + IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(latest.index_manifest().unwrap()) + ) + .await + .unwrap(), + indexes + ); + let mut builder = table.new_read_builder(); + builder.with_filter( + PredicateBuilder::new(table.schema().fields()) + .equal("id", Datum::Int(0)) + .unwrap(), + ); + let plan = builder.new_scan().plan().await.unwrap(); + let batches: Vec = builder + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 1 + ); + assert_eq!( + batches[0] + .column_by_name("value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 99 + ); +} + +#[tokio::test] +async fn compact_preserves_inline_blob_descriptors_without_fetching_payloads() { + use arrow_array::LargeBinaryArray; + use paimon::spec::{BlobDescriptor, BlobType}; + use std::collections::HashMap; + + let table = create_table( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("payload", DataType::Blob(BlobType::new())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("blob-descriptor-field", "payload") + .option("compaction.min.file-num", "2") + .build() + .unwrap(), + ) + .await; + let descriptor = BlobDescriptor::new("memory:/absent/upstream".into(), 3, 20).serialize(); + for id in [0, 1] { + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef), + ( + "payload", + Arc::new(LargeBinaryArray::from(vec![Some(descriptor.as_slice())])) as ArrayRef, + ), + ]) + .unwrap(), + ) + .await; + } + let (_, results) = execute(&table).await; + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let descriptors = table.copy_with_options(HashMap::from([( + "blob-as-descriptor".into(), + "true".into(), + )])); + let read = descriptors.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!( + batches.iter().map(|batch| batch.num_rows()).sum::(), + 2 + ); + for batch in batches { + let payload = batch + .column_by_name("payload") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert!(payload + .iter() + .all(|value| value == Some(descriptor.as_slice()))); + } +} + +#[tokio::test] +async fn compact_rejects_incomplete_serialized_groups_and_changed_output_ranges() { + let table = table(&[]).await; + append(&table, 0).await; + update(&table, 0, Some(99)).await; + update(&table, 1, Some(88)).await; + let (plan, results) = execute(&table).await; + assert_eq!( + plan.tasks()[0].files().len(), + 3, + "shadowed versions must still be removed" + ); + let mut modified = serde_json::to_value(&plan).unwrap(); + modified["tasks"][0]["files"] + .as_array_mut() + .unwrap() + .remove(0); + let modified: CompactPlan = serde_json::from_value(modified).unwrap(); + assert!(table + .new_compact_builder() + .new_executor() + .unwrap() + .execute(&modified.tasks()[0]) + .await + .is_err()); + let mut modified = serde_json::to_value(&results).unwrap(); + modified[0]["files"][0]["_FIRST_ROW_ID"] = 1.into(); + let modified: Vec = serde_json::from_value(modified).unwrap(); + assert!(table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(modified) + .await + .is_err()); + assert_eq!(rows(&table).await, vec![(0, 0, Some(99)), (1, 1, Some(88))]); +} + +#[tokio::test] +async fn compact_partition_filter_and_subsets_keep_unselected_files_and_deletions() { + use paimon::spec::{Datum, PredicateBuilder}; + + let table = create_table( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .partition_keys(["id"]) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("deletion-vectors.enabled", "true") + .option("compaction.min.file-num", "2") + .build() + .unwrap(), + ) + .await; + for id in [10, 20] { + for value in [1, 2] { + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![id])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![value])) as ArrayRef), + ]) + .unwrap(), + ) + .await; + } + } + delete(&table, &[1, 3]).await; + let expected = rows(&table).await; + let builder = table + .new_compact_builder() + .with_partition_filter( + PredicateBuilder::new(table.schema().fields()) + .equal("id", Datum::Int(10)) + .unwrap(), + ) + .unwrap(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks().len(), 1); + let result = builder + .new_executor() + .unwrap() + .execute(&plan.tasks()[0]) + .await + .unwrap(); + builder + .new_commit() + .unwrap() + .commit_compact(vec![result]) + .await + .unwrap(); + assert_eq!(rows(&table).await, expected); + let (remaining, results) = execute(&table).await; + assert_eq!(remaining.tasks().len(), 1); + assert_eq!(remaining.tasks()[0].files()[0].first_row_id, Some(2)); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert_eq!(rows(&table).await, expected); + + // Multiple tasks sharing one DV index must also be independently committable. + let table = self::table(&[ + ("target-file-size", "1b"), + ("deletion-vectors.enabled", "true"), + ]) + .await; + append(&table, 0).await; + append(&table, 2).await; + update(&table, 0, Some(7)).await; + update(&table, 2, Some(8)).await; + delete(&table, &[1, 3]).await; + let (plan, mut results) = execute(&table).await; + assert_eq!(plan.tasks().len(), 2); + let expected = rows(&table).await; + let commit = table.new_compact_builder().new_commit().unwrap(); + let first = results.remove(0); + commit.commit_compact(vec![first.clone()]).await.unwrap(); + assert_eq!(rows(&table).await, expected); + commit.commit_compact(results).await.unwrap(); + commit.commit_compact(vec![first]).await.unwrap(); + assert_eq!(rows(&table).await, expected); +} + +#[tokio::test] +async fn compact_preserves_field_ids_through_rename_drop_readd_and_type_promotion() { + use paimon::spec::{BigIntType, SchemaChange}; + + let original = table(&[]).await; + append(&original, 0).await; + update(&original, 0, Some(99)).await; + let schema = original + .schema() + .apply_changes(vec![ + SchemaChange::rename_column("value".into(), "renamed".into()), + SchemaChange::update_column_type("renamed".into(), DataType::BigInt(BigIntType::new())), + SchemaChange::add_column("value".into(), DataType::Int(IntType::new())), + ]) + .unwrap(); + persist_table_schema(original.file_io(), original.location(), &schema).await; + let table = original.copy_with_resolved_schema(schema, "main").unwrap(); + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![2])) as ArrayRef), + ("renamed", Arc::new(Int64Array::from(vec![123])) as ArrayRef), + ("value", Arc::new(Int32Array::from(vec![77])) as ArrayRef), + ]) + .unwrap(), + ) + .await; + let (_, results) = execute(&table).await; + assert_eq!(results.len(), 1); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let read = table.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .with_ordered(true) + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let merged = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap(); + assert_eq!( + merged + .column_by_name("renamed") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(), + &Int64Array::from(vec![Some(99), None, Some(123)]) + ); + assert_eq!( + merged + .column_by_name("value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(), + &Int32Array::from(vec![None, None, Some(77)]) + ); + assert_eq!( + rows(&original.copy_with_pinned_snapshot(Some( + &original.snapshot_manager().get_snapshot(2).await.unwrap() + ))) + .await, + vec![(0, 0, Some(99)), (1, 1, None)] + ); +} + +#[tokio::test] +async fn compact_merges_nested_updates_without_replacing_unwritten_children() { + use arrow_array::{StringArray, StructArray}; + use paimon::spec::{DataField, RowType, VarCharType}; + + let table = create_table( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column( + "profile", + DataType::Row(RowType::new(vec![ + DataField::new(0, "age".into(), DataType::Int(IntType::new())), + DataField::new( + 1, + "name".into(), + DataType::VarChar(VarCharType::string_type()), + ), + ])), + ) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("data-evolution.nested-field.enabled", "true") + .option("compaction.min.file-num", "2") + .build() + .unwrap(), + ) + .await; + let profile = StructArray::from(vec![ + ( + Arc::new(arrow_schema::Field::new( + "age", + arrow_schema::DataType::Int32, + true, + )), + Arc::new(Int32Array::from(vec![Some(10), None])) as ArrayRef, + ), + ( + Arc::new(arrow_schema::Field::new( + "name", + arrow_schema::DataType::Utf8, + true, + )), + Arc::new(StringArray::from(vec![Some("alice"), Some("bob")])) as ArrayRef, + ), + ]); + write_batch( + &table, + &RecordBatch::try_from_iter([ + ("id", Arc::new(Int32Array::from(vec![0, 1])) as ArrayRef), + ("profile", Arc::new(profile) as ArrayRef), + ]) + .unwrap(), + ) + .await; + let builder = table.new_write_builder(); + let mut writer = builder + .new_data_evolution_writer(vec!["profile.age".into()]) + .unwrap(); + writer + .add_matched_batch( + RecordBatch::try_from_iter([ + ("_ROW_ID", Arc::new(Int64Array::from(vec![1])) as ArrayRef), + ( + "profile.age", + Arc::new(Int32Array::from(vec![Some(42)])) as ArrayRef, + ), + ]) + .unwrap(), + ) + .unwrap(); + builder + .new_commit() + .commit(writer.prepare_commit().await.unwrap()) + .await + .unwrap(); + let (_, results) = execute(&table).await; + builder.new_commit().commit_compact(results).await.unwrap(); + let read = table.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let merged = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap(); + let profile = merged + .column_by_name("profile") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + assert_eq!( + profile + .column_by_name("age") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(), + &Int32Array::from(vec![10, 42]) + ); + assert_eq!( + profile + .column_by_name("name") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(), + &StringArray::from(vec!["alice", "bob"]) + ); +} + +#[tokio::test] +async fn java_minimum_one_allows_single_file_schema_materialization() { + use paimon::spec::SchemaChange; + + let original = table(&[("compaction.min.file-num", "1")]).await; + append(&original, 0).await; + let schema = original + .schema() + .apply_changes(vec![SchemaChange::add_column( + "added".into(), + DataType::Int(IntType::new()), + )]) + .unwrap(); + persist_table_schema(original.file_io(), original.location(), &schema).await; + let table = original.copy_with_resolved_schema(schema, "main").unwrap(); + let (plan, results) = execute(&table).await; + assert_eq!(plan.tasks().len(), 1); + assert_eq!(plan.tasks()[0].files().len(), 1); + assert_eq!(results[0].files()[0].schema_id, table.schema().id()); + assert_eq!(results[0].files()[0].first_row_id, Some(0)); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert_eq!(rows(&table).await, vec![(0, 0, Some(10)), (1, 1, None)]); + let read = table.new_read_builder(); + let plan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert!(batches + .iter() + .all(|batch| batch.column_by_name("added").unwrap().null_count() == batch.num_rows())); +} + +#[tokio::test] +async fn compact_subset_preserves_unrelated_deletion_vector_indexes_without_opening_them() { + use paimon::spec::IndexManifest; + use paimon::table::CommitMessage; + + let table = table(&[ + ("target-file-size", "1b"), + ("deletion-vectors.enabled", "true"), + ]) + .await; + append(&table, 0).await; + append(&table, 2).await; + update(&table, 0, Some(99)).await; + delete(&table, &[1]).await; + let scan = table + .new_read_builder() + .new_scan() + .plan_all() + .await + .unwrap(); + let other = scan + .splits() + .iter() + .flat_map(|split| split.data_files()) + .find(|file| file.first_row_id == Some(2)) + .unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let indexes = IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(snapshot.index_manifest().unwrap()), + ) + .await + .unwrap(); + assert_eq!(indexes.len(), 1); + let original = &indexes[0]; + let original_bytes = table + .file_io() + .new_input(&format!( + "{}/index/{}", + table.location(), + original.index_file.file_name + )) + .unwrap() + .read() + .await + .unwrap(); + let mut unrelated = original.index_file.clone(); + unrelated.file_name = format!("independent-dv-{}", uuid::Uuid::new_v4()); + let ranges = unrelated.deletion_vectors_ranges.as_mut().unwrap(); + let meta = ranges.values().next().unwrap().clone(); + ranges.clear(); + ranges.insert(other.file_name.clone(), meta); + let unrelated_path = format!("{}/index/{}", table.location(), unrelated.file_name); + table + .file_io() + .new_output(&unrelated_path) + .unwrap() + .write(original_bytes.clone()) + .await + .unwrap(); + let mut message = CommitMessage::new(original.partition.clone(), original.bucket, vec![]); + message.new_index_files.push(unrelated.clone()); + table + .new_write_builder() + .new_commit() + .commit(vec![message]) + .await + .unwrap(); + + let (plan, results) = execute(&table).await; + assert_eq!( + plan.tasks().len(), + 1, + "only the updated A group is selected" + ); + table.file_io().delete_file(&unrelated_path).await.unwrap(); + table + .new_compact_builder() + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let latest = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let indexes = IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(latest.index_manifest().unwrap()), + ) + .await + .unwrap(); + assert!( + indexes.iter().any(|entry| entry.index_file == unrelated), + "Java leaves wholly untouched DV index entries verbatim" + ); + table + .file_io() + .new_output(&unrelated_path) + .unwrap() + .write(original_bytes) + .await + .unwrap(); + assert_eq!(rows(&table).await, vec![(0, 0, Some(99)), (2, 2, Some(10))]); +} diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 712621700..e82e6d049 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -19,8 +19,9 @@ under the License. # Table Compaction -Rust supports small-file compaction for ordinary append tables and full-bucket -compaction for primary-key tables with fixed or dynamic buckets. `CompactBuilder` creates +Rust supports small-file compaction for ordinary append tables, normal-file +compaction for Data Evolution tables, and full-bucket compaction for primary-key +tables with fixed or dynamic buckets. `CompactBuilder` creates a `CompactPlanner`, a `CompactExecutor`, and a `TableCommit`. Planning and execution do not publish snapshots; `commit_compact` atomically publishes the selected results in a `COMPACT` snapshot. @@ -40,7 +41,7 @@ let metrics = committer.commit_compact(results).await?; println!("Compacted {} files into {}", metrics.files_removed, metrics.files_added); ``` -Compaction uses the table's `target-file-size` option for its output size target. +Ordinary append and PK compaction use `target-file-size` as the output size target. For unaware append tables (`bucket=-1`), task selection follows the batch packing rules in Java's [`AppendCompactCoordinator`](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/append/AppendCompactCoordinator.java): @@ -180,15 +181,67 @@ Cross-partition dynamic buckets are also supported: subsequent writers rebuild their key-location index from the compacted data as usual. Compaction does not rebalance buckets or reclaim HASH index capacity. +## Data Evolution compaction + +The same API compacts normal structured files in unaware append tables with +`data-evolution.enabled=true` and `row-tracking.enabled=true`, following Java's +`DataEvolutionCompactCoordinator` and `DataEvolutionNormalCompactTask`. + +Within each partition, planning keeps every version of an aligned row-ID group +together, including shadowed column files. Adjacent groups are packed using +`sum(max(file_size, source.split.open-file-cost))` as the weight. A bin is drained +when its weight strictly exceeds `target-file-size`; a group that alone exceeds +the target is isolated. A row-ID gap also drains the bin. A drained bin produces +a task only when it contains at least `compaction.min.file-num` physical files. +The ordinary append small-file ratio does not apply. +Data Evolution permits `compaction.min.file-num=1` to force a single-file +rewrite, for example to materialize the current schema. + +Each task streams the latest normal-column values into one file covering the +same contiguous row-ID range. Like Java's default normal compaction, write-time +size and row limits do not roll this output; `target-file-size` guides planning. +The minimum and maximum input sequence numbers and per-column maximum sequence +numbers are retained. Blob and Vector files remain attached to their stable row +IDs and are not rewritten by these tasks. + +Deleted physical slots are retained. When deletion vectors are enabled, commit +preparation moves their positions from the previous normal anchor files to the +new files. This uses the latest snapshot on every commit attempt, preserving +concurrent deletions and vectors belonging to untouched groups. Global indexes +remain valid because row IDs and values are preserved. Historical snapshots +continue to reference their original data and deletion vectors. + +Concurrent normal-column writes in a selected row-ID range cause commit to fail; +plan again from the latest snapshot. Disjoint appends can commit alongside the +rewrite. Subsets, serialization and idempotent retries use the same compaction +API as append and PK tables. `rows_written` counts physical slots, including +deleted rows; it counts overlapping column versions only once. + +Changing storage options does not migrate values in existing files. If selected +normal files still provide a field that the current schema stores in dedicated +files, compaction is rejected before writing or publishing output. Migrate those +historical inline values first; otherwise omitting the field would lose data. +With deletion vectors enabled, one output file is limited to 4,294,967,296 +physical row slots so every local position remains deletable by the current +Rust writer. Wider rewrites are rejected before execution; 64-bit deletion-vector +writing is not yet supported. + +Dedicated Blob/Vector compaction, splitting large normal files, and materializing +deletions into reassigned row IDs are not provided. Setting +`data-evolution.compaction.split-large-files=true` or +`data-evolution.compaction.rewrite-row-ids=true` is rejected. + ## Current scope Compaction-generated changelogs (`full-compaction` / `lookup`) are unsupported. -Row tracking, Data Evolution, deletion vectors, query authorization, rowkind -fields, postpone buckets and dedicated blob/vector storage are unsupported. +Outside Data Evolution mode, row tracking, deletion vectors and dedicated +blob/vector storage are unsupported. Query authorization, rowkind fields and +postpone buckets are unsupported in all modes. Ordered bucketed append, append incremental clustering, primary-key clustering override (`pk-clustering-override`) and record-level expiration (`record-level.expire-time`) are also unsupported. -Snapshot index manifests may contain only dynamic-bucket HASH indexes. +Snapshot index manifests may contain dynamic-bucket HASH indexes, or deletion +vectors and global indexes in Data Evolution mode. Embedded and sidecar file indexes are regenerated according to the captured table options. Language bindings and background compaction scheduling are not yet provided. From 64427adc391fb66e880a7697b1661a41513f1ced Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 08:43:05 +0800 Subject: [PATCH 2/5] refactor(table): separate compaction planners by table mode --- crates/paimon/src/table/compact.rs | 138 +++++--------- crates/paimon/src/table/compact/append.rs | 27 +++ .../src/table/compact/data_evolution.rs | 171 ++++++++++++------ .../paimon/src/table/compact/primary_key.rs | 39 +++- docs/src/compaction.md | 4 + 5 files changed, 231 insertions(+), 148 deletions(-) diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 76fae4793..46f553064 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -33,7 +33,7 @@ use super::data_file_writer::DataFileWriter; use super::partition_filter::validate_partition_filter; use super::snapshot_manifest::read_snapshot_entries; use super::write_builder::{ensure_table_write_authorized, validate_commit_user}; -use super::{DataSplit, Table, TableCommit}; +use super::{DataSplit, Plan, Table, TableCommit}; use crate::resource::ResourceContext; use crate::spec::{ bucket_path_under, BinaryRow, ChangelogProducer, DataFileMeta, FileKind, IndexManifest, @@ -92,18 +92,18 @@ impl<'a> CompactBuilder<'a> { message: "Compaction requires a positive target-file-size".into(), }); } - let append = - if self.table.schema().primary_keys().is_empty() && !options.data_evolution_enabled() { - Some(append::AppendCompactPlanner::new(&options)?) - } else { - None - }; - if options.data_evolution_enabled() { - options.compaction_min_file_num()?; - } + let kind = if options.data_evolution_enabled() { + CompactPlannerKind::DataEvolution(data_evolution::DataEvolutionCompactPlanner::new( + &options, + )?) + } else if self.table.schema().primary_keys().is_empty() { + CompactPlannerKind::Append(append::AppendCompactPlanner::new(&options)?) + } else { + CompactPlannerKind::PrimaryKey(primary_key::PrimaryKeyCompactPlanner) + }; Ok(CompactPlanner { table: self.table.clone(), - append, + kind, partition_filter: self.partition_filter.clone(), }) } @@ -126,13 +126,38 @@ impl<'a> CompactBuilder<'a> { } } -/// Selects append files, contiguous data-evolution ranges or whole PK buckets. +/// Captures a shared snapshot boundary and delegates to the table-mode planner. pub struct CompactPlanner { table: Table, - append: Option, + kind: CompactPlannerKind, partition_filter: Option, } +enum CompactPlannerKind { + Append(append::AppendCompactPlanner), + PrimaryKey(primary_key::PrimaryKeyCompactPlanner), + DataEvolution(data_evolution::DataEvolutionCompactPlanner), +} + +type BucketFiles = BTreeMap<(Vec, i32, i32), Vec>; + +/// Append and PK planners share the split-to-bucket metadata representation. +/// Eligibility and task selection remain with each table-mode planner. +fn split_files_by_bucket(source: &Plan) -> BucketFiles { + let mut groups = BucketFiles::new(); + for split in source.splits() { + groups + .entry(( + split.partition().to_serialized_bytes(), + split.bucket(), + split.total_buckets(), + )) + .or_default() + .extend_from_slice(split.data_files()); + } + groups +} + #[derive(Debug, Clone, Serialize, Deserialize)] struct CompactContext { plan_id: String, @@ -223,88 +248,13 @@ impl CompactPlanner { schema: self.table.schema().clone(), snapshot, }; - let mut groups = BTreeMap::<_, Vec>::new(); - let primary_key = !self.table.schema().primary_keys().is_empty(); - let data_evolution = self.table.schema().core_options().data_evolution_enabled(); - if data_evolution { - // Query planning may discard shadowed column providers. A rewrite - // must remove every normal file in each captured row-ID group. - for entry in scan.plan_manifest_entries(&context.snapshot).await? { - if !super::source::is_data_evolution_normal_file(entry.file()) { - continue; - } - data_evolution::validate_file(entry.file())?; - groups - .entry(( - entry.partition().to_vec(), - entry.bucket(), - entry.total_buckets(), - )) - .or_default() - .push(entry.file().clone()); + let tasks = match &self.kind { + CompactPlannerKind::Append(planner) => planner.plan_tasks(&source, &context)?, + CompactPlannerKind::PrimaryKey(planner) => planner.plan_tasks(&source, &context)?, + CompactPlannerKind::DataEvolution(planner) => { + planner.plan_tasks(&self.table, &scan, &context).await? } - } else { - for split in source.splits() { - let group = groups - .entry(( - split.partition().to_serialized_bytes(), - split.bucket(), - split.total_buckets(), - )) - .or_default(); - for file in split.data_files() { - validate_file(file, primary_key)?; - group.push(file.clone()); - } - } - } - let mut tasks = Vec::new(); - for ((partition, bucket, total_buckets), files) in groups { - if data_evolution { - for files in data_evolution::plan(files, &self.table.schema().core_options())? { - data_evolution::validate_storage_layout(&self.table, &files).await?; - tasks.push(CompactTask { - context: context.clone(), - task_id: tasks.len(), - partition: partition.clone(), - bucket, - total_buckets, - files, - }); - } - continue; - } - if primary_key { - // L0 requires upgrading; multiple levels require merging. Files - // at one positive level already form a non-overlapping sorted - // run, including when the previous compaction rolled files. - let first_level = files.first().map(|file| file.level); - if files - .iter() - .any(|file| file.level == 0 || Some(file.level) != first_level) - { - tasks.push(CompactTask { - context: context.clone(), - task_id: tasks.len(), - partition, - bucket, - total_buckets, - files, - }); - } - continue; - } - for files in self.append.as_ref().unwrap().plan(files) { - tasks.push(CompactTask { - context: context.clone(), - task_id: tasks.len(), - partition: partition.clone(), - bucket, - total_buckets, - files, - }); - } - } + }; Ok(CompactPlan { snapshot_id: Some(snapshot_id), tasks, diff --git a/crates/paimon/src/table/compact/append.rs b/crates/paimon/src/table/compact/append.rs index 8e4f1e63f..58fb92f36 100644 --- a/crates/paimon/src/table/compact/append.rs +++ b/crates/paimon/src/table/compact/append.rs @@ -16,9 +16,12 @@ // under the License. use crate::spec::{CoreOptions, DataFileMeta}; +use crate::table::Plan; use crate::Result; use std::collections::VecDeque; +use super::{split_files_by_bucket, validate_file, CompactContext, CompactTask}; + /// Unaware batch packing and bucketed best-effort windows from Java. pub(super) struct AppendCompactPlanner { bucketed: bool, @@ -39,6 +42,30 @@ impl AppendCompactPlanner { }) } + pub(super) fn plan_tasks( + &self, + source: &Plan, + context: &CompactContext, + ) -> Result> { + let mut tasks = Vec::new(); + for ((partition, bucket, total_buckets), files) in split_files_by_bucket(source) { + for file in &files { + validate_file(file, false)?; + } + for files in self.plan(files) { + tasks.push(CompactTask { + context: context.clone(), + task_id: tasks.len(), + partition: partition.clone(), + bucket, + total_buckets, + files, + }); + } + } + Ok(tasks) + } + pub(super) fn plan(&self, mut files: Vec) -> Vec> { if self.bucketed { return self.plan_bucketed(files); diff --git a/crates/paimon/src/table/compact/data_evolution.rs b/crates/paimon/src/table/compact/data_evolution.rs index c8dd6ce53..4e4b6f56c 100644 --- a/crates/paimon/src/table/compact/data_evolution.rs +++ b/crates/paimon/src/table/compact/data_evolution.rs @@ -24,7 +24,7 @@ use std::sync::Arc; use futures::TryStreamExt; use roaring::RoaringBitmap; -use super::{invalid, CompactTask}; +use super::{invalid, BucketFiles, CompactContext, CompactTask}; use crate::resource::ResourceContext; use crate::spec::{ bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, FileKind, @@ -35,7 +35,7 @@ use crate::table::data_file_writer::DataFileWriter; use crate::table::index_file_path::committed_index_file_path; use crate::table::source::{data_evolution_anchor_file, is_data_evolution_normal_file}; use crate::table::stats_filter::group_by_overlapping_row_id; -use crate::table::{DataEvolutionDeleteWriter, DataSplit, Table}; +use crate::table::{DataEvolutionDeleteWriter, DataSplit, Table, TableScan}; use crate::{Error, Result}; pub(super) fn ensure_supported(table: &Table) -> Result<()> { @@ -129,11 +129,11 @@ pub(super) fn row_range(files: &[DataFileMeta]) -> Result<(i64, i64)> { range.ok_or_else(|| invalid("Empty data-evolution compaction range")) } -fn output_range(files: &[DataFileMeta], options: &CoreOptions<'_>) -> Result<(i64, i64)> { +fn output_range(files: &[DataFileMeta], deletion_vectors_enabled: bool) -> Result<(i64, i64)> { let (first, last) = row_range(files)?; // The current Rust DV writer addresses local positions with u32. Merging // otherwise deletable groups must not make later DELETEs unrepresentable. - if options.deletion_vectors_enabled() && last - first > i64::from(u32::MAX) { + if deletion_vectors_enabled && last - first > i64::from(u32::MAX) { return Err(Error::Unsupported { message: "Data-evolution compaction with deletion vectors cannot produce more than 4294967296 physical row slots in one file; 64-bit deletion-vector writing is not supported".into(), }); @@ -141,49 +141,103 @@ fn output_range(files: &[DataFileMeta], options: &CoreOptions<'_>) -> Result<(i6 Ok((first, last)) } -/// Java packs whole aligned groups by sum(max(file_size, open_file_cost)). -/// A heavy group is isolated; adjacent groups fill bins until weight > target. -pub(super) fn plan( - files: Vec, - options: &CoreOptions<'_>, -) -> Result>> { - let min_files = options.compaction_min_file_num()?; - let target = i128::from(options.target_file_size()); - let mut tasks = Vec::new(); - let mut pending = Vec::new(); - let mut weight = 0i128; - let mut previous_end: Option = None; - let drain = |pending: &mut Vec, tasks: &mut Vec>| { - let files = std::mem::take(pending); - if files.len() >= min_files { - tasks.push(files); - } - }; - for group in group_by_overlapping_row_id(files) { - let (first, last) = row_range(&group)?; - let group_weight = group - .iter() - .map(|file| i128::from(file.file_size.max(options.source_split_open_file_cost()))) - .sum::(); - if previous_end.is_some_and(|end| end.checked_add(1) != Some(first)) - || group_weight > target - { - drain(&mut pending, &mut tasks); - weight = 0; +/// Select complete normal-file groups and pack adjacent row-ID ranges. +pub(super) struct DataEvolutionCompactPlanner { + min_file_num: usize, + target_file_size: i128, + open_file_cost: i64, + deletion_vectors_enabled: bool, +} + +impl DataEvolutionCompactPlanner { + pub(super) fn new(options: &CoreOptions<'_>) -> Result { + Ok(Self { + min_file_num: options.compaction_min_file_num()?, + target_file_size: i128::from(options.target_file_size()), + open_file_cost: options.source_split_open_file_cost(), + deletion_vectors_enabled: options.deletion_vectors_enabled(), + }) + } + + pub(super) async fn plan_tasks( + &self, + table: &Table, + scan: &TableScan<'_>, + context: &CompactContext, + ) -> Result> { + let mut groups = BucketFiles::new(); + // Read captured manifests rather than query providers so shadowed + // column versions are also included in each normal-file rewrite. + for entry in scan.plan_manifest_entries(&context.snapshot).await? { + if !is_data_evolution_normal_file(entry.file()) { + continue; + } + validate_file(entry.file())?; + groups + .entry(( + entry.partition().to_vec(), + entry.bucket(), + entry.total_buckets(), + )) + .or_default() + .push(entry.file().clone()); } - pending.extend(group); - weight += group_weight; - if weight > target { - drain(&mut pending, &mut tasks); - weight = 0; + let mut tasks = Vec::new(); + for ((partition, bucket, total_buckets), files) in groups { + for files in self.plan(files)? { + validate_storage_layout(table, &files).await?; + tasks.push(CompactTask { + context: context.clone(), + task_id: tasks.len(), + partition: partition.clone(), + bucket, + total_buckets, + files, + }); + } } - previous_end = Some(last); + Ok(tasks) } - drain(&mut pending, &mut tasks); - for task in &tasks { - output_range(task, options)?; + + /// Java packs whole groups by sum(max(file_size, open_file_cost)). A heavy + /// group is isolated; adjacent groups fill bins until weight > target. + fn plan(&self, files: Vec) -> Result>> { + let mut tasks = Vec::new(); + let mut pending = Vec::new(); + let mut weight = 0i128; + let mut previous_end: Option = None; + let drain = |pending: &mut Vec, tasks: &mut Vec>| { + let files = std::mem::take(pending); + if files.len() >= self.min_file_num { + tasks.push(files); + } + }; + for group in group_by_overlapping_row_id(files) { + let (first, last) = row_range(&group)?; + let group_weight = group + .iter() + .map(|file| i128::from(file.file_size.max(self.open_file_cost))) + .sum::(); + if previous_end.is_some_and(|end| end.checked_add(1) != Some(first)) + || group_weight > self.target_file_size + { + drain(&mut pending, &mut tasks); + weight = 0; + } + pending.extend(group); + weight += group_weight; + if weight > self.target_file_size { + drain(&mut pending, &mut tasks); + weight = 0; + } + previous_end = Some(last); + } + drain(&mut pending, &mut tasks); + for task in &tasks { + output_range(task, self.deletion_vectors_enabled)?; + } + Ok(tasks) } - Ok(tasks) } pub(super) fn validate_inputs( @@ -336,7 +390,10 @@ pub(super) async fn execute( resources: Option, ) -> Result> { validate_storage_layout(table, task.files()).await?; - let (first, last) = output_range(task.files(), &table.schema().core_options())?; + let (first, last) = output_range( + task.files(), + table.schema().core_options().deletion_vectors_enabled(), + )?; let fields = normal_fields(table.schema()); let min_sequence = task .files @@ -418,7 +475,13 @@ pub(super) async fn execute( } pub(super) fn validate_outputs(task: &CompactTask, files: &[DataFileMeta]) -> Result<()> { - let expected = output_range(&task.files, &task.context.schema.core_options())?; + let expected = output_range( + &task.files, + task.context + .schema + .core_options() + .deletion_vectors_enabled(), + )?; let min_sequence = task .files .iter() @@ -618,7 +681,9 @@ mod tests { ), ("compaction.min.file-num".into(), min.to_string()), ]); - plan(files, &CoreOptions::new(&options)) + DataEvolutionCompactPlanner::new(&CoreOptions::new(&options)) + .unwrap() + .plan(files) .unwrap() .into_iter() .map(|group| group.into_iter().map(|file| file.file_name).collect()) @@ -683,17 +748,19 @@ mod tests { ]); let max_rows = i64::from(u32::MAX) + 1; let too_wide = vec![file("a", 0, max_rows, 1), file("b", max_rows, 1, 1)]; - let error = plan(too_wide.clone(), &CoreOptions::new(&options)).unwrap_err(); + let planner = DataEvolutionCompactPlanner::new(&CoreOptions::new(&options)).unwrap(); + let error = planner.plan(too_wide.clone()).unwrap_err(); assert!(matches!(error, Error::Unsupported { .. })); assert!(error.to_string().contains("64-bit deletion-vector writing")); let at_limit = vec![file("a", 0, max_rows - 1, 1), file("b", max_rows - 1, 1, 1)]; - assert_eq!( - plan(at_limit, &CoreOptions::new(&options)).unwrap().len(), - 1 - ); + assert_eq!(planner.plan(at_limit).unwrap().len(), 1); options.insert("deletion-vectors.enabled".into(), "false".into()); assert_eq!( - plan(too_wide, &CoreOptions::new(&options)).unwrap().len(), + DataEvolutionCompactPlanner::new(&CoreOptions::new(&options)) + .unwrap() + .plan(too_wide) + .unwrap() + .len(), 1 ); } diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index e228838a9..454c751fd 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -20,7 +20,7 @@ use std::sync::Arc; use arrow_schema::SchemaRef; use futures::TryStreamExt; -use super::CompactTask; +use super::{split_files_by_bucket, validate_file, CompactContext, CompactTask}; use crate::resource::ResourceContext; use crate::spec::{ BigIntType, DataField, DataFileMeta, DataType, MergeEngine, RowKind, TinyIntType, @@ -37,9 +37,44 @@ use crate::table::sort_merge::{ use crate::table::table_read::{ configured_mosaic_prefetch, configured_parquet_read_budget, MAX_MERGE_INPUT_STREAMS, }; -use crate::table::{DataSplit, Table}; +use crate::table::{DataSplit, Plan, Table}; use crate::Result; +/// Select complete captured buckets that need an L0 upgrade or level merge. +pub(super) struct PrimaryKeyCompactPlanner; + +impl PrimaryKeyCompactPlanner { + pub(super) fn plan_tasks( + &self, + source: &Plan, + context: &CompactContext, + ) -> Result> { + let mut tasks = Vec::new(); + for ((partition, bucket, total_buckets), files) in split_files_by_bucket(source) { + for file in &files { + validate_file(file, true)?; + } + // One positive level is already a non-overlapping sorted run, + // including when the previous compaction rolled several files. + let first_level = files.first().map(|file| file.level); + if files + .iter() + .any(|file| file.level == 0 || Some(file.level) != first_level) + { + tasks.push(CompactTask { + context: context.clone(), + task_id: tasks.len(), + partition, + bucket, + total_buckets, + files, + }); + } + } + Ok(tasks) + } +} + /// Keep the same winner as a deduplicate query, including DELETE/UPDATE_BEFORE. /// A later write may carry a lower user sequence, so even full compaction /// must retain a winning retract rather than resurrect an older value. diff --git a/docs/src/compaction.md b/docs/src/compaction.md index e82e6d049..56f4b4722 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -26,6 +26,10 @@ a `CompactPlanner`, a `CompactExecutor`, and a `TableCommit`. Planning and execution do not publish snapshots; `commit_compact` atomically publishes the selected results in a `COMPACT` snapshot. +The builder selects a dedicated append, primary-key or Data Evolution planner +for the table mode. They share snapshot capture and the public plan/task types; +each planner owns its file eligibility, grouping and task-selection rules. + ```rust let builder = table.new_compact_builder(); let planner = builder.new_planner()?; From 33c9ba122273dcdd225f834946ff680957b35e55 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 08:53:11 +0800 Subject: [PATCH 3/5] test(table): cover deletes racing with data evolution compaction --- .../src/table/table_commit/parity_tests.rs | 183 ++++++++++++++++++ 1 file changed, 183 insertions(+) diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs b/crates/paimon/src/table/table_commit/parity_tests.rs index 00a7b3c6a..32a17bdbf 100644 --- a/crates/paimon/src/table/table_commit/parity_tests.rs +++ b/crates/paimon/src/table/table_commit/parity_tests.rs @@ -442,6 +442,189 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe } } +#[tokio::test] +async fn stale_delete_after_compact_cas_loss_requires_repreparation() { + use arrow_array::{Int32Array, RecordBatch}; + use futures::TryStreamExt; + + struct CompactBeforeDeleteCommit { + table: Table, + calls: std::sync::atomic::AtomicUsize, + } + + #[async_trait::async_trait] + impl SnapshotCommit for CompactBeforeDeleteCommit { + async fn commit( + &self, + _: Option<&str>, + snapshot: &Snapshot, + _: &str, + _: &[PartitionStatistics], + ) -> Result { + if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 { + // DELETE has finished preparing its snapshot. Let COMPACT win + // publication, so DELETE must validate its old DV again. + let builder = self.table.new_compact_builder(); + let plan = builder.new_planner()?.plan().await?; + assert_eq!(plan.tasks().len(), 1); + let result = builder.new_executor()?.execute(&plan.tasks()[0]).await?; + builder.new_commit()?.commit_compact(vec![result]).await?; + return Ok(false); + } + self.table + .snapshot_manager() + .commit_snapshot(snapshot) + .await + } + } + + async fn visible_ids(table: &Table) -> Vec { + let read = table.new_read_builder(); + let scan = read.new_scan().plan().await.unwrap(); + let batches: Vec = read + .new_read() + .unwrap() + .to_arrow(scan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut ids = batches + .iter() + .flat_map(|batch| { + batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .values() + .iter() + .copied() + }) + .collect::>(); + ids.sort_unstable(); + ids + } + + for existing_dv in [false, true] { + let io = test_file_io(); + let path = format!("memory:/de-delete-after-compact-{existing_dv}"); + setup_dirs(&io, &path).await; + let schema = crate::spec::Schema::builder() + .column("id", DataType::Int(crate::spec::IntType::new())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("deletion-vectors.enabled", "true") + .option("compaction.min.file-num", "2") + .build() + .unwrap(); + let table = Table::new( + io.clone(), + Identifier::new("default", "de_delete_after_compact"), + path.clone(), + TableSchema::new(0, &schema), + None, + ); + io.mkdirs(&format!("{path}/schema")).await.unwrap(); + io.new_output(&format!("{path}/schema/schema-0")) + .unwrap() + .write(serde_json::to_vec(table.schema()).unwrap().into()) + .await + .unwrap(); + for ids in [vec![0, 1], vec![2, 3]] { + let builder = table.new_write_builder(); + let mut write = builder.new_write().unwrap(); + let batch = RecordBatch::try_from_iter([( + "id", + Arc::new(Int32Array::from(ids)) as arrow_array::ArrayRef, + )]) + .unwrap(); + write.write_arrow_batch(&batch).await.unwrap(); + builder + .new_commit() + .commit(write.prepare_commit().await.unwrap()) + .await + .unwrap(); + } + if existing_dv { + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&table).unwrap(); + delete.add_row_ids([1]).unwrap(); + table + .new_write_builder() + .new_commit() + .commit(delete.prepare_commit().await.unwrap()) + .await + .unwrap(); + } + + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&table).unwrap(); + delete.add_row_ids([2]).unwrap(); + let messages = delete.prepare_commit().await.unwrap(); + let publisher = Arc::new(CompactBeforeDeleteCommit { + table: table.clone(), + calls: std::sync::atomic::AtomicUsize::new(0), + }); + let mut commit = table.new_write_builder().new_commit(); + commit.snapshot_commit = publisher.clone(); + commit.commit_min_retry_wait_ms = 0; + commit.commit_max_retry_wait_ms = 0; + let error = commit.commit(messages.clone()).await.unwrap_err(); + let expected_error = if existing_dv { + "Cannot replace missing deletion vector index" + } else { + "Deletion vector references missing data file" + }; + assert!(error.to_string().contains(expected_error), "{error}"); + assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), 1); + let compact_snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + assert_eq!(compact_snapshot.commit_kind(), &CommitKind::COMPACT); + assert_eq!( + visible_ids(&table).await, + if existing_dv { + vec![0, 2, 3] + } else { + vec![0, 1, 2, 3] + } + ); + + // Replaying the same stale messages cannot repair their old anchors. + let error = commit.commit(messages).await.unwrap_err(); + assert!(error.to_string().contains(expected_error), "{error}"); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + Some(compact_snapshot.id()) + ); + + // Reprepare from current files instead. Stable row IDs still identify + // the requested row, and the already committed deletion is retained. + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&table).unwrap(); + delete.add_row_ids([2]).unwrap(); + table + .new_write_builder() + .new_commit() + .commit(delete.prepare_commit().await.unwrap()) + .await + .unwrap(); + assert_eq!( + visible_ids(&table).await, + if existing_dv { + vec![0, 3] + } else { + vec![0, 1, 3] + } + ); + } +} + #[tokio::test] async fn mixed_append_cannot_bypass_stale_dv_check() { let io = test_file_io(); From 3bbc05c0ab3be61407a2c31b73125d6e17d77cb3 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 08:53:17 +0800 Subject: [PATCH 4/5] refactor(table): cache deletion vector commit option --- crates/paimon/src/table/table_commit.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 3a2245438..6769098b5 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -207,6 +207,7 @@ pub struct TableCommit { manifest_sidecar_enabled: bool, row_tracking_enabled: bool, data_evolution_enabled: bool, + deletion_vectors_enabled: bool, partition_default_name: String, snapshot_properties: Option>, commit_callbacks: Vec>, @@ -233,6 +234,7 @@ impl TableCommit { let manifest_sidecar_enabled = core_options.manifest_sidecar_enabled(); let row_tracking_enabled = core_options.row_tracking_enabled(); let data_evolution_enabled = core_options.data_evolution_enabled(); + let deletion_vectors_enabled = core_options.deletion_vectors_enabled(); let partition_default_name = core_options.partition_default_name().to_string(); Self { table, @@ -250,6 +252,7 @@ impl TableCommit { manifest_sidecar_enabled, row_tracking_enabled, data_evolution_enabled, + deletion_vectors_enabled, partition_default_name, snapshot_properties: None, commit_callbacks: Vec::new(), @@ -1362,11 +1365,7 @@ impl TableCommit { if self.data_evolution_enabled && resolved.kind == CommitKind::COMPACT - && self - .table - .schema() - .core_options() - .deletion_vectors_enabled() + && self.deletion_vectors_enabled { if let Some(snapshot) = latest_snapshot { if let Some(indexes) = super::compact::data_evolution::rewrite_deletion_vectors( From ed85070e91ffdcab7f9c240efd4b5ce6967001b1 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 09:01:10 +0800 Subject: [PATCH 5/5] test(table): strengthen data evolution compaction recovery coverage --- .../src/table/table_commit/parity_tests.rs | 27 ++++++++++++++----- .../tests/data_evolution_compact_test.rs | 21 +++++++++++---- docs/src/compaction.md | 4 +++ 3 files changed, 40 insertions(+), 12 deletions(-) diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs b/crates/paimon/src/table/table_commit/parity_tests.rs index 32a17bdbf..33e090a5e 100644 --- a/crates/paimon/src/table/table_commit/parity_tests.rs +++ b/crates/paimon/src/table/table_commit/parity_tests.rs @@ -366,6 +366,7 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe table: Table, calls: std::sync::atomic::AtomicUsize, publish_first: bool, + delete_after_publish: bool, } #[async_trait::async_trait] @@ -380,6 +381,13 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 { if self.publish_first { assert!(self.table.snapshot_manager().commit_snapshot(snapshot).await?); + if self.delete_after_publish { + // The retry must find its successful COMPACT even when + // a newer DELETE has already replaced the migrated DV. + let mut delete = crate::table::DataEvolutionDeleteWriter::new(&self.table)?; + delete.add_row_ids([2])?; + self.table.new_write_builder().new_commit().commit(delete.prepare_commit().await?).await?; + } return Err(crate::Error::IoUnsupported { message: "lost compact response".into() }); } // Commit another DELETE after compact prepared its DV file, @@ -393,9 +401,9 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe } } - for publish_first in [false, true] { + for (publish_first, delete_after_publish) in [(false, false), (true, false), (true, true)] { let io = test_file_io(); - let path = format!("memory:/de-compact-cas-{publish_first}"); + let path = format!("memory:/de-compact-cas-{publish_first}-{delete_after_publish}"); setup_dirs(&io, &path).await; let schema = crate::spec::Schema::builder() .column("id", DataType::Int(crate::spec::IntType::new())) @@ -420,7 +428,7 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe let plan = builder.new_planner().unwrap().plan().await.unwrap(); let result = builder.new_executor().unwrap().execute(&plan.tasks()[0]).await.unwrap(); let publisher = Arc::new(RacingCompactCommit { - table: table.clone(), calls: std::sync::atomic::AtomicUsize::new(0), publish_first, + table: table.clone(), calls: std::sync::atomic::AtomicUsize::new(0), publish_first, delete_after_publish, }); let mut commit = builder.new_commit().unwrap(); commit.snapshot_commit = publisher.clone(); @@ -429,16 +437,21 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe let metrics = commit.commit_compact(vec![result.clone()]).await.unwrap(); assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), if publish_first { 1 } else { 2 }); let after = table.snapshot_manager().get_latest_snapshot().await.unwrap().unwrap(); - assert_eq!(after.id(), if publish_first { 4 } else { 5 }); - assert_eq!(after.commit_kind(), &CommitKind::COMPACT); - assert_eq!(commit.commit_compact(vec![result]).await.unwrap(), metrics); + assert_eq!(after.id(), if publish_first && !delete_after_publish { 4 } else { 5 }); + assert_eq!(after.commit_kind(), if delete_after_publish { &CommitKind::OVERWRITE } else { &CommitKind::COMPACT }); + assert_eq!(commit.commit_compact(vec![result.clone()]).await.unwrap(), metrics); + // Recovery after coordinator restart must not need the captured source + // snapshot once the COMPACT identity is already present in history. + io.delete_file(&table.snapshot_manager().snapshot_path(plan.snapshot_id().unwrap())).await.unwrap(); + let recovered = table.new_compact_builder().with_commit_user(commit.commit_user.clone()).unwrap().new_commit().unwrap(); + assert_eq!(recovered.commit_compact(vec![result]).await.unwrap(), metrics); assert_eq!(table.snapshot_manager().get_latest_snapshot_id().await.unwrap(), Some(after.id())); let read = table.new_read_builder(); let scan = read.new_scan().plan().await.unwrap(); let batches: Vec = read.new_read().unwrap().to_arrow(scan.splits()).unwrap().try_collect().await.unwrap(); let mut ids = batches.iter().flat_map(|batch| batch.column(0).as_any().downcast_ref::().unwrap().values().iter().copied()).collect::>(); ids.sort_unstable(); - assert_eq!(ids, if publish_first { vec![0, 2, 3] } else { vec![0, 3] }); + assert_eq!(ids, if publish_first && !delete_after_publish { vec![0, 2, 3] } else { vec![0, 3] }); } } diff --git a/crates/paimon/tests/data_evolution_compact_test.rs b/crates/paimon/tests/data_evolution_compact_test.rs index b1dbcaf35..f8369cca2 100644 --- a/crates/paimon/tests/data_evolution_compact_test.rs +++ b/crates/paimon/tests/data_evolution_compact_test.rs @@ -719,19 +719,30 @@ async fn compact_partition_filter_and_subsets_keep_unselected_files_and_deletion .await; append(&table, 0).await; append(&table, 2).await; + append(&table, 4).await; update(&table, 0, Some(7)).await; update(&table, 2, Some(8)).await; - delete(&table, &[1, 3]).await; + update(&table, 4, Some(9)).await; + delete(&table, &[1, 3, 5]).await; let (plan, mut results) = execute(&table).await; - assert_eq!(plan.tasks().len(), 2); + assert_eq!(plan.tasks().len(), 3); let expected = rows(&table).await; let commit = table.new_compact_builder().new_commit().unwrap(); + let last = results.remove(2); let first = results.remove(0); - commit.commit_compact(vec![first.clone()]).await.unwrap(); + // A subset may leave a gap occupied by an unsubmitted task. Its shared + // index must retain that middle group's original anchor and deletions. + commit + .commit_compact(vec![last.clone(), first.clone()]) + .await + .unwrap(); assert_eq!(rows(&table).await, expected); + delete(&table, &[2]).await; + let after_delete = rows(&table).await; + assert_eq!(after_delete, vec![(0, 0, Some(7)), (4, 4, Some(9))]); commit.commit_compact(results).await.unwrap(); - commit.commit_compact(vec![first]).await.unwrap(); - assert_eq!(rows(&table).await, expected); + commit.commit_compact(vec![first, last]).await.unwrap(); + assert_eq!(rows(&table).await, after_delete); } #[tokio::test] diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 56f4b4722..9e76a5a31 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -215,6 +215,10 @@ concurrent deletions and vectors belonging to untouched groups. Global indexes remain valid because row IDs and values are preserved. Historical snapshots continue to reference their original data and deletion vectors. +If compaction publishes before a prepared DELETE, that DELETE fails when it +references replaced anchor files or DV indexes. Prepare the DELETE again from +the latest snapshot; replaying the same stale messages does not rebase it. + Concurrent normal-column writes in a selected row-ID range cause commit to fail; plan again from the latest snapshot. Disjoint appends can commit alongside the rewrite. Subsets, serialization and idempotent retries use the same compaction