From c0af3dd780595f8c2fd6339c039ccdf14d930b03 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 13:44:35 +0800 Subject: [PATCH 1/4] feat(table): support incremental deduplicate compaction --- crates/paimon/src/table/compact.rs | 41 +- crates/paimon/src/table/compact/append.rs | 1 + .../src/table/compact/data_evolution.rs | 1 + .../paimon/src/table/compact/primary_key.rs | 20 +- .../src/table/compact/primary_key/planner.rs | 408 ++++++++++++++++++ crates/paimon/src/table/table_commit.rs | 11 +- .../tests/pk_compact/incremental_planning.rs | 397 +++++++++++++++++ crates/paimon/tests/pk_compact_test.rs | 11 +- docs/src/compaction.md | 34 +- 9 files changed, 908 insertions(+), 16 deletions(-) create mode 100644 crates/paimon/src/table/compact/primary_key/planner.rs create mode 100644 crates/paimon/tests/pk_compact/incremental_planning.rs diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index e1c611688..3ebfb4cad 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -208,6 +208,8 @@ pub struct CompactTask { bucket: i32, total_buckets: i32, files: Vec, + #[serde(default)] + primary_key: Option, } impl CompactTask { @@ -468,6 +470,9 @@ impl CompactTask { } fn output_level(&self) -> i32 { + if let Some(selection) = &self.primary_key { + return selection.output_level; + } if self.is_primary_key() { self.files .iter() @@ -480,7 +485,7 @@ impl CompactTask { } } - fn validate_inputs(&self, active: &[ManifestEntry]) -> Result<()> { + pub(super) fn validate_inputs(&self, active: &[ManifestEntry]) -> Result<()> { if self.is_data_evolution() { data_evolution::validate_inputs(&self.partition, self.bucket, &self.files, active)?; } @@ -493,7 +498,21 @@ impl CompactTask { }) .map(|entry| (entry.file().file_name.as_str(), entry)) .collect::>(); - if self.is_primary_key() && bucket_files.len() != self.files.len() { + if self.is_primary_key() && primary_key::planner::enabled(&self.context.schema)? { + let selected = primary_key::planner::select( + bucket_files + .values() + .map(|entry| entry.file().clone()) + .collect(), + self.context.schema.options(), + )? + .ok_or_else(|| invalid("No incremental PK compaction is eligible"))?; + if selected.0 != self.files || Some(&selected.1) != self.primary_key.as_ref() { + return Err(invalid( + "PK compaction inputs do not match incremental run selection", + )); + } + } else if self.is_primary_key() && bucket_files.len() != self.files.len() { return Err(invalid( "PK compaction must include the complete captured bucket", )); @@ -522,6 +541,18 @@ impl CompactTask { )); } ensure_snapshot_supported(table, &self.context.snapshot)?; + let incremental = + self.is_primary_key() && primary_key::planner::enabled(&self.context.schema)?; + // Legacy results may be recovered as already committed before their + // captured inputs are validated. Uncommitted tasks still need the + // exact incremental selection in validate_inputs. + if self + .primary_key + .as_ref() + .is_some_and(|task| !incremental || task.output_level <= 0) + { + return Err(invalid("Invalid incremental PK compaction selection")); + } let minimum_files = if self.is_primary_key() || self.is_data_evolution() { 1 } else { @@ -560,6 +591,7 @@ pub(super) struct PreparedCompact { pub read_snapshot: Option>, pub commit_suffix: String, pub metrics: CompactMetrics, + pub pk_tasks: Vec, } pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Result { @@ -573,9 +605,13 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul let mut entries = Vec::new(); let mut changelog_entries = Vec::new(); let mut metrics = CompactMetrics::default(); + let mut pk_tasks = Vec::new(); for result in results { let task = &result.task; task.validate_target(table)?; + if task.is_primary_key() { + pk_tasks.push(task.clone()); + } if task.context.plan_id != first.task.context.plan_id || !task .context @@ -719,6 +755,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul .then(|| Box::new(first.task.context.snapshot.clone())), commit_suffix: hex::encode(digest.finalize()), metrics, + pk_tasks, }) } diff --git a/crates/paimon/src/table/compact/append.rs b/crates/paimon/src/table/compact/append.rs index 58fb92f36..b83c7b32c 100644 --- a/crates/paimon/src/table/compact/append.rs +++ b/crates/paimon/src/table/compact/append.rs @@ -60,6 +60,7 @@ impl AppendCompactPlanner { bucket, total_buckets, files, + primary_key: None, }); } } diff --git a/crates/paimon/src/table/compact/data_evolution.rs b/crates/paimon/src/table/compact/data_evolution.rs index 4e4b6f56c..2f16a2a36 100644 --- a/crates/paimon/src/table/compact/data_evolution.rs +++ b/crates/paimon/src/table/compact/data_evolution.rs @@ -193,6 +193,7 @@ impl DataEvolutionCompactPlanner { bucket, total_buckets, files, + primary_key: None, }); } } diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index 156c79858..dc8c35172 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -16,6 +16,7 @@ // under the License. pub(super) mod lookup; +pub(in crate::table) mod planner; use std::sync::Arc; @@ -43,7 +44,7 @@ use crate::table::table_read::{ use crate::table::{DataSplit, Plan, Table}; use crate::Result; -/// Select files using the existing primary-key bucket compaction policy. +/// Use incremental sorted runs when merging subsets does not need history lookup. pub(super) struct PrimaryKeyCompactPlanner; impl PrimaryKeyCompactPlanner { @@ -57,6 +58,22 @@ impl PrimaryKeyCompactPlanner { for file in &files { validate_file(file, true)?; } + if planner::enabled(&context.schema)? { + if let Some((files, primary_key)) = + planner::select(files, context.schema.options())? + { + tasks.push(CompactTask { + context: context.clone(), + task_id: tasks.len(), + partition, + bucket, + total_buckets, + files, + primary_key: Some(primary_key), + }); + } + continue; + } // 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); @@ -71,6 +88,7 @@ impl PrimaryKeyCompactPlanner { bucket, total_buckets, files, + primary_key: None, }); } } diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs new file mode 100644 index 000000000..ff5b130d0 --- /dev/null +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -0,0 +1,408 @@ +// 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. + +//! Java UniversalCompaction run selection. Explicit compact drains L0 using +//! forcePickL0; positive-level maintenance uses size/run triggers, never a full +//! bucket override. + +use crate::spec::{ChangelogProducer, DataFileMeta, MergeEngine, TableSchema}; +use crate::{Error, Result}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(in crate::table) struct PrimaryKeyTask { + pub output_level: i32, +} + +pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { + let options = schema.core_options(); + Ok(options.merge_engine()? == MergeEngine::Deduplicate + && !options.deletion_vectors_enabled() + && matches!( + options.try_changelog_producer()?, + ChangelogProducer::None | ChangelogProducer::Input + )) +} + +fn number( + options: &std::collections::HashMap, + key: &str, + default: i32, + minimum: i32, +) -> Result { + let value = options + .get(key) + .map(|value| value.parse::()) + .transpose() + .map_err(|_| Error::ConfigInvalid { + message: format!("Invalid {key}"), + })? + .unwrap_or(default); + if value < minimum { + return Err(Error::ConfigInvalid { + message: format!("{key} must be at least {minimum}"), + }); + } + Ok(value) +} + +pub(in crate::table::compact) fn select( + files: Vec, + properties: &std::collections::HashMap, +) -> Result, PrimaryKeyTask)>> { + let trigger = number(properties, "num-sorted-run.compaction-trigger", 5, 1)?; + let configured_levels = number(properties, "num-levels", trigger.saturating_add(1), 2)?; + let levels = configured_levels.max( + files + .iter() + .map(|file| file.level.saturating_add(1)) + .max() + .unwrap_or(2), + ); + let ratio = number(properties, "compaction.size-ratio", 1, 0)?; + let amplification = number( + properties, + "compaction.max-size-amplification-percent", + 200, + 0, + )?; + let mut l0 = files + .iter() + .filter(|file| file.level == 0) + .cloned() + .collect::>(); + l0.sort_by(|a, b| { + b.max_sequence_number + .cmp(&a.max_sequence_number) + .then_with(|| a.min_sequence_number.cmp(&b.min_sequence_number)) + .then_with(|| { + a.creation_time + .unwrap_or_default() + .cmp(&b.creation_time.unwrap_or_default()) + }) + .then_with(|| a.file_name.cmp(&b.file_name)) + }); + let mut positive = BTreeMap::>::new(); + for file in files.into_iter().filter(|file| file.level > 0) { + positive.entry(file.level).or_default().push(file); + } + let l0_count = l0.len(); + let mut runs = l0 + .into_iter() + .map(|file| (0, vec![file])) + .collect::>(); + for (level, mut files) in positive { + files.sort_by(|a, b| a.file_name.cmp(&b.file_name)); + runs.push((level, files)); + } + if runs.is_empty() { + return Ok(None); + } + let sizes = runs + .iter() + .map(|(_, files)| { + files + .iter() + .map(|file| file.file_size as u128) + .sum::() + }) + .collect::>(); + let mut count; + if l0_count > 0 { + // Java prepare-commit forces L0 progress for lookup; the explicit Rust + // compact API requests the same bounded incremental progress here. + count = l0_count; + } else { + if runs.len() < trigger as usize { + return Ok(None); + } + let additional = sizes[..sizes.len() - 1].iter().sum::(); + if additional * 100 > amplification as u128 * sizes[sizes.len() - 1] { + count = runs.len(); + } else { + count = 1; + } + } + let mut size = sizes[..count].iter().sum::(); + while count < runs.len() { + let next = &runs[count]; + if size * (100 + ratio as u128) < sizes[count] * 100 + && (count <= 1 && l0_count == 0 || next.0 > 1) + { + break; + } + size += sizes[count]; + count += 1; + } + if l0_count == 0 && count == 1 { + if runs.len() <= trigger as usize { + return Ok(None); + } + count = runs.len() - trigger as usize + 1; + size = sizes[..count].iter().sum(); + while count < runs.len() + && (size * (100 + ratio as u128) >= sizes[count] * 100 || runs[count].0 <= 1) + { + size += sizes[count]; + count += 1; + } + } + let output_level = if count == runs.len() { + levels - 1 + } else { + (runs[count].0 - 1).max(1) + }; + let selected = runs[..count] + .iter() + .flat_map(|(_, files)| files.clone()) + .collect::>(); + if selected.len() == 1 && selected[0].level == output_level { + return Ok(None); + } + Ok(Some((selected, PrimaryKeyTask { output_level }))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::stats::BinaryTableStats; + use std::collections::HashMap; + fn file(name: &str, level: i32, size: i64) -> DataFileMeta { + DataFileMeta { + file_name: name.into(), + file_size: size, + row_count: 1, + min_key: vec![], + max_key: vec![], + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 1, + schema_id: 0, + level, + 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: None, + write_cols: None, + column_max_sequence_numbers: None, + } + } + #[test] + fn java_force_l0_stops_before_a_large_higher_level_and_outputs_below_it() { + let (selected, task) = select( + vec![file("new", 0, 1), file("anchor", 5, 1000)], + &HashMap::new(), + ) + .unwrap() + .unwrap(); + assert_eq!( + selected + .iter() + .map(|file| file.file_name.as_str()) + .collect::>(), + vec!["new"] + ); + assert_eq!(task.output_level, 4); + let (selected, task) = select( + vec![ + file("new", 0, 1), + file("l1", 1, 100), + file("anchor", 5, 10000), + ], + &HashMap::new(), + ) + .unwrap() + .unwrap(); + assert_eq!( + selected.len(), + 2, + "Java always consumes L1 rather than outputting L0" + ); + assert_eq!(task.output_level, 4); + } + #[test] + fn positive_run_count_and_size_amplification_trigger_incremental_maintenance() { + let options = HashMap::from([ + ("num-sorted-run.compaction-trigger".into(), "2".into()), + ("num-levels".into(), "6".into()), + ]); + assert!( + select(vec![file("small", 3, 1), file("anchor", 5, 1000)], &options) + .unwrap() + .is_none() + ); + let (selected, task) = select( + vec![ + file("small", 3, 1), + file("medium", 4, 10), + file("anchor", 5, 1000), + ], + &options, + ) + .unwrap() + .unwrap(); + assert_eq!(selected.len(), 2); + assert_eq!(task.output_level, 4); + let (selected, task) = select( + vec![ + file("large", 1, 1000), + file("other", 3, 1000), + file("old", 5, 100), + ], + &options, + ) + .unwrap() + .unwrap(); + assert_eq!(selected.len(), 3); + assert_eq!(task.output_level, 5); + } + + #[test] + fn selection_is_stable_and_keeps_each_positive_run_together() { + let mut older = file("older-l0", 0, 1); + older.max_sequence_number = 10; + let mut newer = file("newer-l0", 0, 1); + newer.max_sequence_number = 20; + let files = vec![ + file("anchor", 5, 100000), + file("b", 1, 100), + older, + file("a", 1, 100), + newer, + ]; + let expected = select(files.clone(), &HashMap::new()).unwrap().unwrap(); + assert_eq!( + expected + .0 + .iter() + .map(|file| file.file_name.as_str()) + .collect::>(), + vec!["newer-l0", "older-l0", "a", "b"] + ); + assert_eq!(expected.1.output_level, 4); + let reversed = files.into_iter().rev().collect(); + assert_eq!( + select(reversed, &HashMap::new()).unwrap().unwrap(), + expected + ); + let mut ties = vec![]; + for (name, minimum, time) in [("a-later", 2, 2), ("z-earlier", 2, 1), ("z-minimum", 1, 2)] { + let mut row = file(name, 0, 1); + row.min_sequence_number = minimum; + row.max_sequence_number = 20; + row.creation_time = chrono::DateTime::from_timestamp_millis(time); + ties.push(row); + } + let selected = select(ties, &HashMap::new()).unwrap().unwrap(); + assert_eq!( + selected + .0 + .iter() + .map(|file| file.file_name.as_str()) + .collect::>(), + vec!["z-minimum", "z-earlier", "a-later"] + ); + } + + #[test] + fn size_ratio_and_amplification_use_java_threshold_boundaries() { + let options = HashMap::from([("num-sorted-run.compaction-trigger".into(), "3".into())]); + for (middle, all) in [(199, false), (200, true)] { + let selected = select( + vec![ + file("small", 2, 1), + file("middle", 3, middle), + file("old", 5, 100), + ], + &options, + ) + .unwrap(); + if all { + assert_eq!(selected.unwrap().0.len(), 3); + } else { + assert!(selected.is_none()); + } + } + for (first, merge) in [(199, false), (200, true)] { + let selected = select( + vec![ + file("first", 2, first), + file("second", 3, 202), + file("old", 5, 100000), + ], + &options, + ) + .unwrap(); + if merge { + let (files, task) = selected.unwrap(); + assert_eq!(files.len(), 2); + assert_eq!(task.output_level, 4); + } else { + assert!(selected.is_none()); + } + } + // Arithmetic must not overflow when byte sizes exceed i64 percentages. + let (files, task) = select( + vec![file("first", 0, i64::MAX), file("old", 5, i64::MAX)], + &HashMap::new(), + ) + .unwrap() + .unwrap(); + assert_eq!(files.len(), 2); + assert_eq!(task.output_level, 5); + } + + #[test] + fn singletons_empty_inputs_and_existing_levels_do_not_force_a_full_rewrite() { + assert!(select(vec![], &HashMap::new()).unwrap().is_none()); + assert!(select(vec![file("old", 5, 100)], &HashMap::new()) + .unwrap() + .is_none()); + let options = HashMap::from([("num-levels".into(), "3".into())]); + let (_, task) = select(vec![file("new", 0, 1)], &options).unwrap().unwrap(); + assert_eq!(task.output_level, 2); + let (_, task) = select(vec![file("new", 0, 100), file("old", 8, 1)], &options) + .unwrap() + .unwrap(); + assert_eq!(task.output_level, 8, "existing higher levels remain valid"); + } + + #[test] + fn invalid_run_policy_options_are_reported_before_selection() { + for (name, bad, minimum) in [ + ("num-sorted-run.compaction-trigger", "0", "1"), + ("num-levels", "1", "2"), + ("compaction.size-ratio", "-1", "0"), + ("compaction.max-size-amplification-percent", "-1", "0"), + ] { + for value in [bad, "invalid", "2147483648"] { + let options = HashMap::from([(name.into(), value.into())]); + let error = select(vec![file("new", 0, 1)], &options).unwrap_err(); + assert!(matches!(error, Error::ConfigInvalid { .. })); + assert!(error.to_string().contains(name)); + } + let options = HashMap::from([(name.into(), minimum.into())]); + assert!(select(vec![file("new", 0, 1)], &options).unwrap().is_some()); + } + } +} diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 9d6c41874..cfebd90d3 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -443,6 +443,7 @@ impl TableCommit { CommitEntriesPlan::Compact { entries: prepared.entries, changelog_entries: prepared.changelog_entries, + pk_tasks: prepared.pk_tasks, schema_id: prepared.schema_id, read_snapshot: prepared.read_snapshot, planned_snapshot_id: prepared.snapshot_id, @@ -1911,6 +1912,7 @@ impl TableCommit { schema_id, read_snapshot, planned_snapshot_id, + pk_tasks, } => { if latest_snapshot .as_ref() @@ -1953,6 +1955,9 @@ impl TableCommit { None, ) .await?; + for task in pk_tasks { + task.validate_inputs(&captured_files)?; + } let deletes = entries .iter() .filter(|e| *e.kind() == FileKind::Delete) @@ -1983,8 +1988,7 @@ impl TableCommit { .iter() .filter(|e| { buckets.contains(&(e.partition(), e.bucket())) - && (!self.data_evolution_enabled - || deleted_ids.contains(&commit_file_identifier(e))) + && deleted_ids.contains(&commit_file_identifier(e)) }) .map(|entry| (commit_file_identifier(entry), entry)) .collect::>(); @@ -2002,7 +2006,7 @@ impl TableCommit { 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" + "PK compaction must match its captured incremental inputs" }.into(), source: None, }); @@ -3929,6 +3933,7 @@ enum CommitEntriesPlan { schema_id: i64, read_snapshot: Option>, planned_snapshot_id: i64, + pk_tasks: Vec, }, /// Caller-provided entries. May contain `FileKind::Delete` entries from CoW /// rewrites, in which case `resolve_commit` auto-promotes to `CommitKind::OVERWRITE`. diff --git a/crates/paimon/tests/pk_compact/incremental_planning.rs b/crates/paimon/tests/pk_compact/incremental_planning.rs new file mode 100644 index 000000000..9d3ab1c7a --- /dev/null +++ b/crates/paimon/tests/pk_compact/incremental_planning.rs @@ -0,0 +1,397 @@ +// 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. + +use super::*; + +async fn compact_once(table: &Table) -> Vec { + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(table, &plan).await; + builder + .new_commit() + .unwrap() + .commit_compact(results.clone()) + .await + .unwrap(); + results +} + +async fn active_files(table: &Table) -> Vec { + let scan = table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .unwrap(); + scan.splits() + .iter() + .flat_map(|split| split.data_files()) + .cloned() + .collect() +} + +#[tokio::test] +async fn incremental_deduplicate_keeps_large_history_and_preserves_rows_and_snapshots() { + for producer in ["none", "input"] { + let table = table(&[("bucket", "1"), ("changelog-producer", producer)]).await; + let original = (0..1000).map(|key| (key, key)).collect::>(); + write_batch( + &table, + &make_batch((0..1000).collect(), (0..1000).collect()), + ) + .await; + let first = compact_once(&table).await; + let anchor = first[0].files()[0].clone(); + assert_eq!(anchor.level, 5); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let historical = table.copy_with_pinned_snapshot(Some(&snapshot)); + for key in [42, 43] { + write_batch( + &table, + &make_batch_with_kinds( + vec![key, key + 100], + vec![10000 + key, key + 100], + vec![0, 3], + ), + ) + .await; + let before = rows(&table).await; + let before_sequences = sequences(&table).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks().len(), 1); + assert!(plan.tasks()[0] + .files() + .iter() + .all(|file| file.file_name != anchor.file_name)); + let results = execute(&table, &plan).await; + assert!(results[0].files().iter().all(|file| file.level == 4)); + builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert!(active_files(&table).await.contains(&anchor)); + assert_eq!(rows(&table).await, before); + assert_eq!(sequences(&table).await, before_sequences); + assert_eq!(rows(&historical).await, original); + assert!(builder + .new_planner() + .unwrap() + .plan() + .await + .unwrap() + .tasks() + .is_empty()); + } + let actual = rows(&table).await; + assert_eq!(actual.len(), 998); + assert!(actual.contains(&(42, 10042)) && actual.contains(&(43, 10043))); + assert!(!actual.iter().any(|(key, _)| [142, 143].contains(key))); + } +} + +fn sequenced_batch(ids: Vec, values: Vec, seq: i32, kind: i8) -> RecordBatch { + let count = ids.len(); + RecordBatch::try_from_iter([ + ( + "id", + std::sync::Arc::new(Int32Array::from(ids)) as arrow_array::ArrayRef, + ), + ( + "value", + std::sync::Arc::new(Int32Array::from(values)) as arrow_array::ArrayRef, + ), + ( + "seq", + std::sync::Arc::new(Int32Array::from(vec![seq; count])) as arrow_array::ArrayRef, + ), + ( + "_VALUE_KIND", + std::sync::Arc::new(arrow_array::Int8Array::from(vec![kind; count])) + as arrow_array::ArrayRef, + ), + ]) + .unwrap() +} + +#[tokio::test] +async fn incremental_deduplicate_keeps_user_sequence_winners_and_tombstones() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("seq", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("sequence.field", "seq") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + write_batch( + &table, + &sequenced_batch((0..1000).collect(), (0..1000).collect(), 100, 0), + ) + .await; + let anchor = compact_once(&table).await[0].files()[0].clone(); + for (sequence, kind, expected) in [ + (50, 0, Some(42)), + (150, 0, Some(150)), + (200, 3, None), + (180, 0, None), + (300, 0, Some(300)), + ] { + write_batch( + &table, + &sequenced_batch(vec![42], vec![sequence], sequence, kind), + ) + .await; + let before = rows(&table).await; + let results = compact_once(&table).await; + assert!(!results.is_empty()); + assert!(results + .iter() + .flat_map(|result| result.files()) + .all(|file| file.level == 4)); + assert_eq!(rows(&table).await, before); + assert_eq!( + before + .iter() + .find(|(key, _)| *key == 42) + .map(|(_, value)| *value), + expected + ); + assert!(active_files(&table).await.contains(&anchor)); + } +} + +#[tokio::test] +async fn incremental_selection_is_validated_for_execution_and_commit() { + let table = table(&[("bucket", "1")]).await; + write_batch( + &table, + &make_batch((0..1000).collect(), (0..1000).collect()), + ) + .await; + let anchor = compact_once(&table).await[0].files()[0].clone(); + write_batch(&table, &make_batch(vec![42], vec![4200])).await; + write_batch(&table, &make_batch(vec![43], vec![4300])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let captured = plan.snapshot_id(); + let results = execute(&table, &plan).await; + for mutation in [ + "omit-input", + "include-history", + "output-level", + "missing-selection", + ] { + let mutate = |task: &mut serde_json::Value| match mutation { + "omit-input" => { + task["files"].as_array_mut().unwrap().pop(); + } + "include-history" => task["files"] + .as_array_mut() + .unwrap() + .push(serde_json::to_value(&anchor).unwrap()), + "output-level" => task["primary_key"]["output_level"] = 5.into(), + "missing-selection" => { + task.as_object_mut().unwrap().remove("primary_key"); + } + _ => unreachable!(), + }; + let mut encoded = serde_json::to_value(&plan.tasks()[0]).unwrap(); + mutate(&mut encoded); + let task: CompactTask = serde_json::from_value(encoded).unwrap(); + assert!( + builder + .new_executor() + .unwrap() + .execute(&task) + .await + .is_err(), + "execute {mutation}" + ); + let mut encoded = serde_json::to_value(&results).unwrap(); + mutate(&mut encoded[0]["task"]); + if mutation == "output-level" { + for file in encoded[0]["files"].as_array_mut().unwrap() { + file["level"] = 5.into(); + } + } + let changed: Vec = serde_json::from_value(encoded).unwrap(); + assert!( + builder + .new_commit() + .unwrap() + .commit_compact(changed) + .await + .is_err(), + "commit {mutation}" + ); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + captured + ); + } + builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert_eq!( + rows(&table).await.iter().find(|(key, _)| *key == 42), + Some(&(42, 4200)) + ); + assert!(active_files(&table).await.contains(&anchor)); +} + +#[tokio::test] +async fn incremental_compaction_preserves_concurrent_l0_appends_and_retry_identity() { + let table = table(&[("bucket", "1")]).await; + write_batch( + &table, + &make_batch((0..1000).collect(), (0..1000).collect()), + ) + .await; + let anchor = compact_once(&table).await[0].files()[0].clone(); + write_batch(&table, &make_batch(vec![42], vec![4200])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(&table, &plan).await; + write_batch(&table, &make_batch(vec![42], vec![4201])).await; + let committer = builder.new_commit().unwrap(); + committer.commit_compact(results.clone()).await.unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + committer.commit_compact(results).await.unwrap(); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + snapshot + ); + assert_eq!( + rows(&table).await.iter().find(|(key, _)| *key == 42), + Some(&(42, 4201)) + ); + assert!(active_files(&table).await.contains(&anchor)); + compact_once(&table).await; + assert_eq!( + rows(&table).await.iter().find(|(key, _)| *key == 42), + Some(&(42, 4201)) + ); +} + +#[tokio::test] +async fn order_sensitive_aggregation_keeps_its_history_in_the_merge() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("seq", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("sequence.field", "seq") + .option("merge-engine", "aggregation") + .option("fields.value.aggregate-function", "first_value") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + write_batch( + &table, + &sequenced_batch((0..1000).collect(), vec![20; 1000], 20, 0), + ) + .await; + let anchor = compact_once(&table).await[0].files()[0].clone(); + write_batch(&table, &sequenced_batch(vec![42], vec![10], 10, 0)).await; + write_batch(&table, &sequenced_batch(vec![42], vec![30], 30, 0)).await; + let before = rows(&table).await; + assert_eq!(before.iter().find(|(key, _)| *key == 42), Some(&(42, 10))); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().contains(&anchor)); + let results = execute(&table, &plan).await; + builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert_eq!(rows(&table).await, before); +} + +#[tokio::test] +async fn legacy_completed_results_recover_but_uncommitted_tasks_require_replanning() { + let table = table(&[("bucket", "1")]).await; + write_batch(&table, &make_batch(vec![1], vec![10])).await; + compact_once(&table).await; + write_batch( + &table, + &make_batch((0..1000).collect(), (0..1000).collect()), + ) + .await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks()[0].files().len(), 2); + let results = execute(&table, &plan).await; + assert!(results[0].files().iter().all(|file| file.level == 5)); + // Older serialized tasks have no selection metadata. Their full-bucket + // outputs are recoverable once committed, but must never execute anew. + let mut legacy = serde_json::to_value(&results).unwrap(); + legacy[0]["task"] + .as_object_mut() + .unwrap() + .remove("primary_key"); + let legacy: Vec = serde_json::from_value(legacy).unwrap(); + let committer = builder.new_commit().unwrap(); + assert!(committer.commit_compact(legacy.clone()).await.is_err()); + committer.commit_compact(results).await.unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + committer.commit_compact(legacy).await.unwrap(); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + snapshot + ); + assert_eq!( + rows(&table).await, + (0..1000).map(|key| (key, key)).collect::>() + ); +} diff --git a/crates/paimon/tests/pk_compact_test.rs b/crates/paimon/tests/pk_compact_test.rs index a79323a9f..4021bf428 100644 --- a/crates/paimon/tests/pk_compact_test.rs +++ b/crates/paimon/tests/pk_compact_test.rs @@ -23,6 +23,9 @@ mod engine_tests; #[path = "pk_compact/lookup_changelog.rs"] mod lookup_changelog_tests; +#[path = "pk_compact/incremental_planning.rs"] +mod incremental_planning_tests; + #[path = "pk_compact/dynamic_bucket.rs"] mod dynamic_tests; @@ -151,7 +154,7 @@ async fn pk_compaction_rejects_concurrent_overlapping_positive_level_files() { .unwrap(); let mut messages = writer.prepare_commit().await.unwrap(); for file in &mut messages[0].new_files { - file.level = 1; + file.level = results[0].files()[0].level; } write.new_commit().commit(messages).await.unwrap(); let result = builder.new_commit().unwrap().commit_compact(results).await; @@ -283,7 +286,7 @@ async fn pk_compaction_merges_versions_and_preserves_sequences_deletes_and_histo let results = execute(&table, &plan).await; assert_eq!(results[0].files().len(), 1, "{format}"); let file = &results[0].files()[0]; - assert_eq!(file.level, 1); + assert_eq!(file.level, 5); assert_eq!(file.delete_row_count, Some(1)); assert_eq!((file.min_sequence_number, file.max_sequence_number), (2, 5)); assert_eq!(file.file_source, Some(1)); @@ -506,7 +509,7 @@ async fn pk_compaction_rolls_ordered_files_and_rebuilds_indexes_with_level_stats let plan = builder.new_planner().unwrap().plan().await.unwrap(); let results = execute(&table, &plan).await; assert_eq!(results[0].files().len(), 3); - assert!(results[0].files().iter().all(|file| file.level == 1 + assert!(results[0].files().iter().all(|file| file.level == 5 && file.row_count <= 2 && (file.embedded_index.is_some() || !file.extra_files.is_empty()) && !file.value_stats.null_counts().is_empty())); @@ -582,7 +585,7 @@ async fn pk_compaction_refuses_partial_buckets_and_unsupported_layouts() { .await .unwrap_err() .to_string() - .contains("complete captured bucket")); + .contains("incremental run selection")); let results = execute(&table, &plan).await; let mut encoded = serde_json::to_value(&results).unwrap(); encoded[0]["task"]["files"].as_array_mut().unwrap().pop(); diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 2f8121f2d..dfed66bf1 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -115,12 +115,34 @@ snapshot expiration and storage reclamation are separate maintenance work. The same API compacts primary-key tables with `deduplicate`, `first-row`, `partial-update` or `aggregation` merge engines. Changelog producers `none`, `input` and `lookup` are supported; `first-row` supports `none` or `lookup`. -The planner includes every active file of each selected partition/bucket, -including large files. Append size and count selection options do not split PK -buckets. A single level-0 file is also eligible. A bucket whose files already form one sorted -run at a single positive level needs no rewrite, including when the previous -compaction rolled multiple output files. Output is key ordered at the highest input level, or level 1 -when all inputs are level 0. +For `deduplicate` with `none` or `input`, the planner chooses incremental sorted +runs using Java's universal-compaction size and count policy. Each L0 file is a +run; all files at one positive level form a single run. Explicit compaction +includes every L0 run and absorbs adjacent runs according to their sizes. A +large higher-level run can remain in place while smaller runs are rewritten. +Without L0, positive-level maintenance requires the configured run-count and +size triggers. A single positive run needs no rewrite, even when it spans +several files. + +The policy uses `num-sorted-run.compaction-trigger` (default `5`), `num-levels` +(default trigger plus one), `compaction.size-ratio` (default `1` percent), and +`compaction.max-size-amplification-percent` (default `200`). When selected runs +reach the oldest run, output goes to the highest configured or existing level; +otherwise it goes immediately below the next unselected run. Explicit full +compaction is not exposed. Outputs retain winning retracts, so older values +in untouched files do not reappear after a delete or out-of-order update. + +`lookup`, `first-row`, `partial-update` and `aggregation` retain their current +selection behavior: every active file in a selected bucket participates, and +output uses the highest input level or level 1 for all-L0 inputs. Their state +and changelog generation must see history before incremental selection can be +enabled. Append size and count options do not control PK selection. + +Tasks capture their selected runs and output level. Executors and committers +recompute the selection against the captured snapshot and reject altered or +incomplete inputs. Older serialized deduplicate tasks without this selection +must be replanned before execution or publication; already committed results +remain recoverable through the existing idempotent retry path. Compaction uses each engine's existing merge function and persists its merge state, including original sequence metadata and synthesized DELETE records: From a930b6f2eb7f741d4f367b3e04e6af68e9255d23 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 15:03:35 +0800 Subject: [PATCH 2/4] fix(table): align incremental compaction with Java strategy order --- crates/paimon/src/table/compact.rs | 8 +- .../src/table/compact/primary_key/planner.rs | 136 +++++++++++++----- crates/paimon/src/table/table_commit.rs | 23 ++- docs/src/compaction.md | 5 +- 4 files changed, 130 insertions(+), 42 deletions(-) diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 3ebfb4cad..41f05bda1 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -489,8 +489,14 @@ impl CompactTask { if self.is_data_evolution() { data_evolution::validate_inputs(&self.partition, self.bucket, &self.files, active)?; } + self.validate_bucket_inputs(active.iter()) + } + + pub(super) fn validate_bucket_inputs<'a>( + &self, + active: impl Iterator, + ) -> Result<()> { let bucket_files = active - .iter() .filter(|entry| { entry.partition() == self.partition && entry.bucket() == self.bucket diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index ff5b130d0..0e9b1200e 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -15,9 +15,9 @@ // specific language governing permissions and limitations // under the License. -//! Java UniversalCompaction run selection. Explicit compact drains L0 using -//! forcePickL0; positive-level maintenance uses size/run triggers, never a full -//! bucket override. +//! Java UniversalCompaction run selection. Apply normal size/run triggers +//! before forcing L0 progress, as ForceUpLevel0Compaction does, without a +//! full-bucket override. use crate::spec::{ChangelogProducer, DataFileMeta, MergeEngine, TableSchema}; use crate::{Error, Result}; @@ -122,46 +122,30 @@ pub(in crate::table::compact) fn select( .sum::() }) .collect::>(); - let mut count; - if l0_count > 0 { - // Java prepare-commit forces L0 progress for lookup; the explicit Rust - // compact API requests the same bounded incremental progress here. - count = l0_count; - } else { - if runs.len() < trigger as usize { - return Ok(None); - } + let mut count = None; + if runs.len() >= trigger as usize { let additional = sizes[..sizes.len() - 1].iter().sum::(); if additional * 100 > amplification as u128 * sizes[sizes.len() - 1] { - count = runs.len(); + count = Some(runs.len()); } else { - count = 1; + count = pick_for_size_ratio(&runs, &sizes, ratio, 1, false); } - } - let mut size = sizes[..count].iter().sum::(); - while count < runs.len() { - let next = &runs[count]; - if size * (100 + ratio as u128) < sizes[count] * 100 - && (count <= 1 && l0_count == 0 || next.0 > 1) - { - break; + if count.is_none() && runs.len() > trigger as usize { + count = pick_for_size_ratio( + &runs, + &sizes, + ratio, + runs.len() - trigger as usize + 1, + false, + ); } - size += sizes[count]; - count += 1; } - if l0_count == 0 && count == 1 { - if runs.len() <= trigger as usize { - return Ok(None); - } - count = runs.len() - trigger as usize + 1; - size = sizes[..count].iter().sum(); - while count < runs.len() - && (size * (100 + ratio as u128) >= sizes[count] * 100 || runs[count].0 <= 1) - { - size += sizes[count]; - count += 1; - } + if count.is_none() && l0_count > 0 { + count = pick_for_size_ratio(&runs, &sizes, ratio, l0_count, true); } + let Some(count) = count else { + return Ok(None); + }; let output_level = if count == runs.len() { levels - 1 } else { @@ -177,6 +161,30 @@ pub(in crate::table::compact) fn select( Ok(Some((selected, PrimaryKeyTask { output_level }))) } +fn pick_for_size_ratio( + runs: &[(i32, Vec)], + sizes: &[u128], + ratio: i32, + mut count: usize, + force: bool, +) -> Option { + let mut size = sizes[..count].iter().sum::(); + let mut triggered = force || count > 1; + while count < runs.len() { + // Match Java's double arithmetic, including threshold rounding for + // large runs, rather than comparing exact integer cross-products. + if size as f64 * (100.0 + f64::from(ratio)) / 100.0 < sizes[count] as f64 + && (!triggered || runs[count].0 > 1) + { + break; + } + size += sizes[count]; + count += 1; + triggered = true; + } + triggered.then_some(count) +} + #[cfg(test)] mod tests { use super::*; @@ -405,4 +413,60 @@ mod tests { assert!(select(vec![file("new", 0, 1)], &options).unwrap().is_some()); } } + + #[test] + fn normal_maintenance_takes_precedence_over_forcing_l0() { + let (files, task) = select( + vec![ + file("new", 0, 1), + file("l2", 2, 1000), + file("l3", 3, 1000), + file("l4", 4, 1000), + file("old", 5, 100), + ], + &HashMap::new(), + ) + .unwrap() + .unwrap(); + assert_eq!( + files.len(), + 5, + "space amplification must run even when L0 exists" + ); + assert_eq!(task.output_level, 5); + + let options = HashMap::from([("num-sorted-run.compaction-trigger".into(), "2".into())]); + let (files, task) = select( + vec![ + file("new", 0, 1), + file("small", 3, 100), + file("old", 5, 100000), + ], + &options, + ) + .unwrap() + .unwrap(); + assert_eq!( + files.len(), + 2, + "run-count maintenance must precede forced L0 progress" + ); + assert_eq!(task.output_level, 4); + } + + #[test] + fn large_run_size_ratio_matches_java_double_threshold() { + let options = HashMap::from([("num-sorted-run.compaction-trigger".into(), "2".into())]); + let (files, task) = select( + vec![ + file("newer", 2, 100_000_000_000_099), + file("older", 5, 101_000_000_000_100), + ], + &options, + ) + .unwrap() + .expect("Java rounds the 1% threshold up to the older run's size"); + assert_eq!(files.len(), 2); + assert_eq!(task.output_level, 5); + } } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index cfebd90d3..e83abab81 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -1955,9 +1955,6 @@ impl TableCommit { None, ) .await?; - for task in pk_tasks { - task.validate_inputs(&captured_files)?; - } let deletes = entries .iter() .filter(|e| *e.kind() == FileKind::Delete) @@ -1966,6 +1963,26 @@ impl TableCommit { .iter() .map(|e| (e.partition(), e.bucket())) .collect::>(); + if !pk_tasks.is_empty() { + // Index once for the submitted buckets instead of + // rescanning the whole snapshot for every PK task. + let mut by_bucket = HashMap::<_, Vec<&ManifestEntry>>::new(); + for entry in &captured_files { + let key = (entry.partition(), entry.bucket()); + if buckets.contains(&key) { + by_bucket.entry(key).or_default().push(entry); + } + } + for task in pk_tasks { + task.validate_bucket_inputs( + by_bucket + .get(&(task.partition(), task.bucket())) + .into_iter() + .flatten() + .copied(), + )?; + } + } let deleted_ids = deletes .iter() .map(|entry| commit_file_identifier(entry)) diff --git a/docs/src/compaction.md b/docs/src/compaction.md index dfed66bf1..889e27a11 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -117,8 +117,9 @@ The same API compacts primary-key tables with `deduplicate`, `none`, `input` and `lookup` are supported; `first-row` supports `none` or `lookup`. For `deduplicate` with `none` or `input`, the planner chooses incremental sorted runs using Java's universal-compaction size and count policy. Each L0 file is a -run; all files at one positive level form a single run. Explicit compaction -includes every L0 run and absorbs adjacent runs according to their sizes. A +run; all files at one positive level form a single run. Normal size and run-count +triggers take precedence. If they select no work, explicit compaction forces +progress on every L0 run and absorbs adjacent runs according to their sizes. A large higher-level run can remain in place while smaller runs are rewritten. Without L0, positive-level maintenance requires the configured run-count and size triggers. A single positive run needs no rewrite, even when it spans From ab80a8f01414db17bcac7af491ddddeac95df217 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 15:26:21 +0800 Subject: [PATCH 3/4] refactor(table): flatten compaction task output level --- crates/paimon/src/table/compact.rs | 13 ++--- crates/paimon/src/table/compact/append.rs | 2 +- .../src/table/compact/data_evolution.rs | 2 +- .../paimon/src/table/compact/primary_key.rs | 6 +- .../src/table/compact/primary_key/planner.rs | 56 +++++++++---------- .../tests/pk_compact/incremental_planning.rs | 11 ++-- 6 files changed, 43 insertions(+), 47 deletions(-) diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 41f05bda1..74cb729b8 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -209,7 +209,7 @@ pub struct CompactTask { total_buckets: i32, files: Vec, #[serde(default)] - primary_key: Option, + output_level: Option, } impl CompactTask { @@ -470,8 +470,8 @@ impl CompactTask { } fn output_level(&self) -> i32 { - if let Some(selection) = &self.primary_key { - return selection.output_level; + if let Some(output_level) = self.output_level { + return output_level; } if self.is_primary_key() { self.files @@ -513,7 +513,7 @@ impl CompactTask { self.context.schema.options(), )? .ok_or_else(|| invalid("No incremental PK compaction is eligible"))?; - if selected.0 != self.files || Some(&selected.1) != self.primary_key.as_ref() { + if selected.0 != self.files || Some(selected.1) != self.output_level { return Err(invalid( "PK compaction inputs do not match incremental run selection", )); @@ -553,9 +553,8 @@ impl CompactTask { // captured inputs are validated. Uncommitted tasks still need the // exact incremental selection in validate_inputs. if self - .primary_key - .as_ref() - .is_some_and(|task| !incremental || task.output_level <= 0) + .output_level + .is_some_and(|output_level| !incremental || output_level <= 0) { return Err(invalid("Invalid incremental PK compaction selection")); } diff --git a/crates/paimon/src/table/compact/append.rs b/crates/paimon/src/table/compact/append.rs index b83c7b32c..375b9dad6 100644 --- a/crates/paimon/src/table/compact/append.rs +++ b/crates/paimon/src/table/compact/append.rs @@ -60,7 +60,7 @@ impl AppendCompactPlanner { bucket, total_buckets, files, - primary_key: None, + output_level: None, }); } } diff --git a/crates/paimon/src/table/compact/data_evolution.rs b/crates/paimon/src/table/compact/data_evolution.rs index 2f16a2a36..07ea02f6b 100644 --- a/crates/paimon/src/table/compact/data_evolution.rs +++ b/crates/paimon/src/table/compact/data_evolution.rs @@ -193,7 +193,7 @@ impl DataEvolutionCompactPlanner { bucket, total_buckets, files, - primary_key: None, + output_level: None, }); } } diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index dc8c35172..5f7ae1119 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -59,7 +59,7 @@ impl PrimaryKeyCompactPlanner { validate_file(file, true)?; } if planner::enabled(&context.schema)? { - if let Some((files, primary_key)) = + if let Some((files, output_level)) = planner::select(files, context.schema.options())? { tasks.push(CompactTask { @@ -69,7 +69,7 @@ impl PrimaryKeyCompactPlanner { bucket, total_buckets, files, - primary_key: Some(primary_key), + output_level: Some(output_level), }); } continue; @@ -88,7 +88,7 @@ impl PrimaryKeyCompactPlanner { bucket, total_buckets, files, - primary_key: None, + output_level: None, }); } } diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index 0e9b1200e..f519949d2 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -21,14 +21,8 @@ use crate::spec::{ChangelogProducer, DataFileMeta, MergeEngine, TableSchema}; use crate::{Error, Result}; -use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub(in crate::table) struct PrimaryKeyTask { - pub output_level: i32, -} - pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { let options = schema.core_options(); Ok(options.merge_engine()? == MergeEngine::Deduplicate @@ -64,7 +58,7 @@ fn number( pub(in crate::table::compact) fn select( files: Vec, properties: &std::collections::HashMap, -) -> Result, PrimaryKeyTask)>> { +) -> Result, i32)>> { let trigger = number(properties, "num-sorted-run.compaction-trigger", 5, 1)?; let configured_levels = number(properties, "num-levels", trigger.saturating_add(1), 2)?; let levels = configured_levels.max( @@ -158,7 +152,7 @@ pub(in crate::table::compact) fn select( if selected.len() == 1 && selected[0].level == output_level { return Ok(None); } - Ok(Some((selected, PrimaryKeyTask { output_level }))) + Ok(Some((selected, output_level))) } fn pick_for_size_ratio( @@ -217,7 +211,7 @@ mod tests { } #[test] fn java_force_l0_stops_before_a_large_higher_level_and_outputs_below_it() { - let (selected, task) = select( + let (selected, output_level) = select( vec![file("new", 0, 1), file("anchor", 5, 1000)], &HashMap::new(), ) @@ -230,8 +224,8 @@ mod tests { .collect::>(), vec!["new"] ); - assert_eq!(task.output_level, 4); - let (selected, task) = select( + assert_eq!(output_level, 4); + let (selected, output_level) = select( vec![ file("new", 0, 1), file("l1", 1, 100), @@ -246,7 +240,7 @@ mod tests { 2, "Java always consumes L1 rather than outputting L0" ); - assert_eq!(task.output_level, 4); + assert_eq!(output_level, 4); } #[test] fn positive_run_count_and_size_amplification_trigger_incremental_maintenance() { @@ -259,7 +253,7 @@ mod tests { .unwrap() .is_none() ); - let (selected, task) = select( + let (selected, output_level) = select( vec![ file("small", 3, 1), file("medium", 4, 10), @@ -270,8 +264,8 @@ mod tests { .unwrap() .unwrap(); assert_eq!(selected.len(), 2); - assert_eq!(task.output_level, 4); - let (selected, task) = select( + assert_eq!(output_level, 4); + let (selected, output_level) = select( vec![ file("large", 1, 1000), file("other", 3, 1000), @@ -282,7 +276,7 @@ mod tests { .unwrap() .unwrap(); assert_eq!(selected.len(), 3); - assert_eq!(task.output_level, 5); + assert_eq!(output_level, 5); } #[test] @@ -307,7 +301,7 @@ mod tests { .collect::>(), vec!["newer-l0", "older-l0", "a", "b"] ); - assert_eq!(expected.1.output_level, 4); + assert_eq!(expected.1, 4); let reversed = files.into_iter().rev().collect(); assert_eq!( select(reversed, &HashMap::new()).unwrap().unwrap(), @@ -362,22 +356,22 @@ mod tests { ) .unwrap(); if merge { - let (files, task) = selected.unwrap(); + let (files, output_level) = selected.unwrap(); assert_eq!(files.len(), 2); - assert_eq!(task.output_level, 4); + assert_eq!(output_level, 4); } else { assert!(selected.is_none()); } } // Arithmetic must not overflow when byte sizes exceed i64 percentages. - let (files, task) = select( + let (files, output_level) = select( vec![file("first", 0, i64::MAX), file("old", 5, i64::MAX)], &HashMap::new(), ) .unwrap() .unwrap(); assert_eq!(files.len(), 2); - assert_eq!(task.output_level, 5); + assert_eq!(output_level, 5); } #[test] @@ -387,12 +381,12 @@ mod tests { .unwrap() .is_none()); let options = HashMap::from([("num-levels".into(), "3".into())]); - let (_, task) = select(vec![file("new", 0, 1)], &options).unwrap().unwrap(); - assert_eq!(task.output_level, 2); - let (_, task) = select(vec![file("new", 0, 100), file("old", 8, 1)], &options) + let (_, output_level) = select(vec![file("new", 0, 1)], &options).unwrap().unwrap(); + assert_eq!(output_level, 2); + let (_, output_level) = select(vec![file("new", 0, 100), file("old", 8, 1)], &options) .unwrap() .unwrap(); - assert_eq!(task.output_level, 8, "existing higher levels remain valid"); + assert_eq!(output_level, 8, "existing higher levels remain valid"); } #[test] @@ -416,7 +410,7 @@ mod tests { #[test] fn normal_maintenance_takes_precedence_over_forcing_l0() { - let (files, task) = select( + let (files, output_level) = select( vec![ file("new", 0, 1), file("l2", 2, 1000), @@ -433,10 +427,10 @@ mod tests { 5, "space amplification must run even when L0 exists" ); - assert_eq!(task.output_level, 5); + assert_eq!(output_level, 5); let options = HashMap::from([("num-sorted-run.compaction-trigger".into(), "2".into())]); - let (files, task) = select( + let (files, output_level) = select( vec![ file("new", 0, 1), file("small", 3, 100), @@ -451,13 +445,13 @@ mod tests { 2, "run-count maintenance must precede forced L0 progress" ); - assert_eq!(task.output_level, 4); + assert_eq!(output_level, 4); } #[test] fn large_run_size_ratio_matches_java_double_threshold() { let options = HashMap::from([("num-sorted-run.compaction-trigger".into(), "2".into())]); - let (files, task) = select( + let (files, output_level) = select( vec![ file("newer", 2, 100_000_000_000_099), file("older", 5, 101_000_000_000_100), @@ -467,6 +461,6 @@ mod tests { .unwrap() .expect("Java rounds the 1% threshold up to the older run's size"); assert_eq!(files.len(), 2); - assert_eq!(task.output_level, 5); + assert_eq!(output_level, 5); } } diff --git a/crates/paimon/tests/pk_compact/incremental_planning.rs b/crates/paimon/tests/pk_compact/incremental_planning.rs index 9d3ab1c7a..b24b1fc28 100644 --- a/crates/paimon/tests/pk_compact/incremental_planning.rs +++ b/crates/paimon/tests/pk_compact/incremental_planning.rs @@ -200,6 +200,9 @@ async fn incremental_selection_is_validated_for_execution_and_commit() { let plan = builder.new_planner().unwrap().plan().await.unwrap(); let captured = plan.snapshot_id(); let results = execute(&table, &plan).await; + let task_json = serde_json::to_value(&plan.tasks()[0]).unwrap(); + assert_eq!(task_json["output_level"], 4); + assert!(task_json.get("primary_key").is_none()); for mutation in [ "omit-input", "include-history", @@ -214,13 +217,13 @@ async fn incremental_selection_is_validated_for_execution_and_commit() { .as_array_mut() .unwrap() .push(serde_json::to_value(&anchor).unwrap()), - "output-level" => task["primary_key"]["output_level"] = 5.into(), + "output-level" => task["output_level"] = 5.into(), "missing-selection" => { - task.as_object_mut().unwrap().remove("primary_key"); + task.as_object_mut().unwrap().remove("output_level"); } _ => unreachable!(), }; - let mut encoded = serde_json::to_value(&plan.tasks()[0]).unwrap(); + let mut encoded = task_json.clone(); mutate(&mut encoded); let task: CompactTask = serde_json::from_value(encoded).unwrap(); assert!( @@ -371,7 +374,7 @@ async fn legacy_completed_results_recover_but_uncommitted_tasks_require_replanni legacy[0]["task"] .as_object_mut() .unwrap() - .remove("primary_key"); + .remove("output_level"); let legacy: Vec = serde_json::from_value(legacy).unwrap(); let committer = builder.new_commit().unwrap(); assert!(committer.commit_compact(legacy.clone()).await.is_err()); From 2f72825967dff30807c902cf55c92a1d5603ed2b Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 17:23:50 +0800 Subject: [PATCH 4/4] refactor(table): validate compaction from captured file changes --- crates/paimon/src/table/compact.rs | 98 ++++++++++-- crates/paimon/src/table/table_commit.rs | 28 +--- .../tests/pk_compact/incremental_planning.rs | 144 +++++++++++------- docs/src/compaction.md | 12 +- 4 files changed, 188 insertions(+), 94 deletions(-) diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 74cb729b8..d06ebd302 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -485,18 +485,12 @@ impl CompactTask { } } - pub(super) fn validate_inputs(&self, active: &[ManifestEntry]) -> Result<()> { + fn validate_inputs(&self, active: &[ManifestEntry]) -> Result<()> { if self.is_data_evolution() { data_evolution::validate_inputs(&self.partition, self.bucket, &self.files, active)?; } - self.validate_bucket_inputs(active.iter()) - } - - pub(super) fn validate_bucket_inputs<'a>( - &self, - active: impl Iterator, - ) -> Result<()> { let bucket_files = active + .iter() .filter(|entry| { entry.partition() == self.partition && entry.bucket() == self.bucket @@ -587,6 +581,79 @@ impl CompactTask { } } +/// Validate published PK file changes without depending on executor task metadata. +pub(super) fn validate_primary_key_changes( + schema: &TableSchema, + changes: &[ManifestEntry], + captured: &[ManifestEntry], +) -> Result<()> { + fn group(entry: &ManifestEntry) -> (&[u8], i32, i32) { + (entry.partition(), entry.bucket(), entry.total_buckets()) + } + let mut inputs = HashMap::<_, Vec<&ManifestEntry>>::new(); + for entry in changes + .iter() + .filter(|entry| *entry.kind() == FileKind::Delete) + { + inputs.entry(group(entry)).or_default().push(entry); + } + let mut active = HashMap::<_, Vec<&DataFileMeta>>::new(); + for entry in captured { + let key = group(entry); + if inputs.contains_key(&key) { + active.entry(key).or_default().push(entry.file()); + } + } + let incremental = primary_key::planner::enabled(schema)?; + let mut output_levels = HashMap::new(); + for (key, removed) in inputs { + let files = active + .remove(&key) + .unwrap_or_default() + .into_iter() + .cloned() + .collect::>(); + let (selected, output_level) = if incremental { + primary_key::planner::select(files, schema.options())? + .ok_or_else(|| invalid("No incremental PK compaction is eligible"))? + } else { + let level = files + .iter() + .map(|file| file.level) + .max() + .unwrap_or(0) + .max(1); + (files, level) + }; + let expected = selected + .iter() + .map(|file| (file.file_name.as_str(), file)) + .collect::>(); + if expected.len() != removed.len() + || removed.iter().any(|entry| { + expected + .get(entry.file().file_name.as_str()) + .is_none_or(|file| **file != *entry.file()) + }) + { + return Err(invalid( + "PK compaction inputs do not match captured run selection", + )); + } + output_levels.insert(key, output_level); + } + if changes + .iter() + .filter(|entry| *entry.kind() == FileKind::Add) + .any(|entry| output_levels.get(&group(entry)) != Some(&entry.file().level)) + { + return Err(invalid( + "PK compaction output level does not match captured run selection", + )); + } + Ok(()) +} + /// Validated inputs for the existing atomic snapshot publication machinery. pub(super) struct PreparedCompact { pub entries: Vec, @@ -596,7 +663,6 @@ pub(super) struct PreparedCompact { pub read_snapshot: Option>, pub commit_suffix: String, pub metrics: CompactMetrics, - pub pk_tasks: Vec, } pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Result { @@ -610,13 +676,9 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul let mut entries = Vec::new(); let mut changelog_entries = Vec::new(); let mut metrics = CompactMetrics::default(); - let mut pk_tasks = Vec::new(); for result in results { let task = &result.task; task.validate_target(table)?; - if task.is_primary_key() { - pk_tasks.push(task.clone()); - } if task.context.plan_id != first.task.context.plan_id || !task .context @@ -644,6 +706,13 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul .iter() .map(|file| file.row_count as u64) .sum::(); + // Missing PK task metadata must not block valid prepared file changes; + // their output level is checked against the captured selection at commit. + let output_level = if task.is_primary_key() { + task.output_level + } else { + Some(0) + }; let mut output_rows = 0u64; for file in &result.files { if task.is_data_evolution() { @@ -653,7 +722,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul } if file.schema_id != task.context.schema.id() || file.file_source != Some(1) - || file.level != task.output_level() + || output_level.is_some_and(|level| file.level != level) || !outputs.insert(file.file_name.clone()) || task.files.iter().any(|old| old.file_name == file.file_name) { @@ -760,7 +829,6 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul .then(|| Box::new(first.task.context.snapshot.clone())), commit_suffix: hex::encode(digest.finalize()), metrics, - pk_tasks, }) } diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index e83abab81..007fb1b74 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -443,7 +443,6 @@ impl TableCommit { CommitEntriesPlan::Compact { entries: prepared.entries, changelog_entries: prepared.changelog_entries, - pk_tasks: prepared.pk_tasks, schema_id: prepared.schema_id, read_snapshot: prepared.read_snapshot, planned_snapshot_id: prepared.snapshot_id, @@ -1912,7 +1911,6 @@ impl TableCommit { schema_id, read_snapshot, planned_snapshot_id, - pk_tasks, } => { if latest_snapshot .as_ref() @@ -1963,25 +1961,12 @@ impl TableCommit { .iter() .map(|e| (e.partition(), e.bucket())) .collect::>(); - if !pk_tasks.is_empty() { - // Index once for the submitted buckets instead of - // rescanning the whole snapshot for every PK task. - let mut by_bucket = HashMap::<_, Vec<&ManifestEntry>>::new(); - for entry in &captured_files { - let key = (entry.partition(), entry.bucket()); - if buckets.contains(&key) { - by_bucket.entry(key).or_default().push(entry); - } - } - for task in pk_tasks { - task.validate_bucket_inputs( - by_bucket - .get(&(task.partition(), task.bucket())) - .into_iter() - .flatten() - .copied(), - )?; - } + if !self.table.schema().primary_keys().is_empty() { + super::compact::validate_primary_key_changes( + self.table.schema(), + entries, + &captured_files, + )?; } let deleted_ids = deletes .iter() @@ -3950,7 +3935,6 @@ enum CommitEntriesPlan { schema_id: i64, read_snapshot: Option>, planned_snapshot_id: i64, - pk_tasks: Vec, }, /// Caller-provided entries. May contain `FileKind::Delete` entries from CoW /// rewrites, in which case `resolve_commit` auto-promotes to `CommitKind::OVERWRITE`. diff --git a/crates/paimon/tests/pk_compact/incremental_planning.rs b/crates/paimon/tests/pk_compact/incremental_planning.rs index b24b1fc28..af2be93ba 100644 --- a/crates/paimon/tests/pk_compact/incremental_planning.rs +++ b/crates/paimon/tests/pk_compact/incremental_planning.rs @@ -207,7 +207,7 @@ async fn incremental_selection_is_validated_for_execution_and_commit() { "omit-input", "include-history", "output-level", - "missing-selection", + "legacy-output-level", ] { let mutate = |task: &mut serde_json::Value| match mutation { "omit-input" => { @@ -218,7 +218,7 @@ async fn incremental_selection_is_validated_for_execution_and_commit() { .unwrap() .push(serde_json::to_value(&anchor).unwrap()), "output-level" => task["output_level"] = 5.into(), - "missing-selection" => { + "legacy-output-level" => { task.as_object_mut().unwrap().remove("output_level"); } _ => unreachable!(), @@ -239,19 +239,28 @@ async fn incremental_selection_is_validated_for_execution_and_commit() { mutate(&mut encoded[0]["task"]); if mutation == "output-level" { for file in encoded[0]["files"].as_array_mut().unwrap() { - file["level"] = 5.into(); + file["_LEVEL"] = 5.into(); + } + } + if mutation == "legacy-output-level" { + for file in encoded[0]["files"].as_array_mut().unwrap() { + file["_LEVEL"] = 1.into(); } } let changed: Vec = serde_json::from_value(encoded).unwrap(); - assert!( - builder - .new_commit() - .unwrap() - .commit_compact(changed) - .await - .is_err(), - "commit {mutation}" - ); + if matches!(mutation, "output-level" | "legacy-output-level") { + let expected = if mutation == "output-level" { 5 } else { 1 }; + assert!(changed[0].files().iter().all(|file| file.level == expected)); + } + let error = builder + .new_commit() + .unwrap() + .commit_compact(changed) + .await + .expect_err(&format!("commit {mutation}")); + if mutation == "legacy-output-level" { + assert!(error.to_string().contains("output level"), "{error}"); + } assert_eq!( table .snapshot_manager() @@ -354,47 +363,78 @@ async fn order_sensitive_aggregation_keeps_its_history_in_the_merge() { } #[tokio::test] -async fn legacy_completed_results_recover_but_uncommitted_tasks_require_replanning() { - let table = table(&[("bucket", "1")]).await; - write_batch(&table, &make_batch(vec![1], vec![10])).await; - compact_once(&table).await; - write_batch( - &table, - &make_batch((0..1000).collect(), (0..1000).collect()), - ) - .await; - let builder = table.new_compact_builder(); - let plan = builder.new_planner().unwrap().plan().await.unwrap(); - assert_eq!(plan.tasks()[0].files().len(), 2); - let results = execute(&table, &plan).await; - assert!(results[0].files().iter().all(|file| file.level == 5)); - // Older serialized tasks have no selection metadata. Their full-bucket - // outputs are recoverable once committed, but must never execute anew. - let mut legacy = serde_json::to_value(&results).unwrap(); - legacy[0]["task"] - .as_object_mut() - .unwrap() - .remove("output_level"); - let legacy: Vec = serde_json::from_value(legacy).unwrap(); - let committer = builder.new_commit().unwrap(); - assert!(committer.commit_compact(legacy.clone()).await.is_err()); - committer.commit_compact(results).await.unwrap(); - let snapshot = table - .snapshot_manager() - .get_latest_snapshot_id() - .await - .unwrap(); - committer.commit_compact(legacy).await.unwrap(); - assert_eq!( - table +async fn legacy_results_validate_file_changes_and_recover_idempotently() { + for keep_history in [false, true] { + let table = table(&[("bucket", "1")]).await; + let initial = if keep_history { + make_batch((0..1000).collect(), (0..1000).collect()) + } else { + make_batch(vec![1], vec![10]) + }; + write_batch(&table, &initial).await; + compact_once(&table).await; + let update = if keep_history { + make_batch(vec![42], vec![4200]) + } else { + make_batch((0..1000).collect(), (0..1000).collect()) + }; + write_batch(&table, &update).await; + let mut expected = (0..1000).map(|key| (key, key)).collect::>(); + if keep_history { + expected[42].1 = 4200; + } + assert_eq!(rows(&table).await, expected); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!( + plan.tasks()[0].files().len(), + if keep_history { 1 } else { 2 } + ); + let results = execute(&table, &plan).await; + assert!(results[0] + .files() + .iter() + .all(|file| file.level == if keep_history { 4 } else { 5 })); + // Older tasks cannot execute without their output level. Prepared results + // can still be published when their file changes match the captured policy. + let mut legacy = serde_json::to_value(&results).unwrap(); + legacy[0]["task"] + .as_object_mut() + .unwrap() + .remove("output_level"); + let legacy: Vec = serde_json::from_value(legacy).unwrap(); + let task = + serde_json::from_value(serde_json::to_value(&legacy).unwrap()[0]["task"].clone()) + .unwrap(); + assert!(builder + .new_executor() + .unwrap() + .execute(&task) + .await + .is_err()); + let committer = builder.new_commit().unwrap(); + committer.commit_compact(legacy.clone()).await.unwrap(); + let published = table .snapshot_manager() .get_latest_snapshot_id() .await - .unwrap(), - snapshot - ); - assert_eq!( - rows(&table).await, - (0..1000).map(|key| (key, key)).collect::>() - ); + .unwrap(); + committer.commit_compact(results).await.unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + assert_eq!(snapshot, published); + committer.commit_compact(legacy).await.unwrap(); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + snapshot + ); + assert_eq!(rows(&table).await, expected); + } } diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 889e27a11..d2ac5ab9b 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -139,11 +139,13 @@ output uses the highest input level or level 1 for all-L0 inputs. Their state and changelog generation must see history before incremental selection can be enabled. Append size and count options do not control PK selection. -Tasks capture their selected runs and output level. Executors and committers -recompute the selection against the captured snapshot and reject altered or -incomplete inputs. Older serialized deduplicate tasks without this selection -must be replanned before execution or publication; already committed results -remain recoverable through the existing idempotent retry path. +Tasks capture their selected runs and output level. Executors validate the task +against the captured snapshot. Committers independently recompute the selection +from that snapshot and validate the files being removed and the output levels. +Older serialized deduplicate tasks without an output level must be replanned +before execution. Prepared results can still be published if their file changes +match the captured selection; already committed results remain recoverable +through the existing idempotent retry path. Compaction uses each engine's existing merge function and persists its merge state, including original sequence metadata and synthesized DELETE records: