This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new f671fe4b feat(table): support incremental deduplicate compaction
(#1253)
f671fe4b is described below
commit f671fe4b9392df5af7fd6ec2c691dfda7d4555d8
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 29 17:35:41 2026 +0800
feat(table): support incremental deduplicate compaction (#1253)
---
crates/paimon/src/table/compact.rs | 114 ++++-
crates/paimon/src/table/compact/append.rs | 1 +
crates/paimon/src/table/compact/data_evolution.rs | 1 +
crates/paimon/src/table/compact/primary_key.rs | 20 +-
.../src/table/compact/primary_key/planner.rs | 466 +++++++++++++++++++++
crates/paimon/src/table/table_commit.rs | 12 +-
.../tests/pk_compact/incremental_planning.rs | 440 +++++++++++++++++++
crates/paimon/tests/pk_compact_test.rs | 11 +-
docs/src/compaction.md | 37 +-
9 files changed, 1086 insertions(+), 16 deletions(-)
diff --git a/crates/paimon/src/table/compact.rs
b/crates/paimon/src/table/compact.rs
index e1c61168..d06ebd30 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<DataFileMeta>,
+ #[serde(default)]
+ output_level: Option<i32>,
}
impl CompactTask {
@@ -468,6 +470,9 @@ impl CompactTask {
}
fn output_level(&self) -> i32 {
+ if let Some(output_level) = self.output_level {
+ return output_level;
+ }
if self.is_primary_key() {
self.files
.iter()
@@ -493,7 +498,21 @@ impl CompactTask {
})
.map(|entry| (entry.file().file_name.as_str(), entry))
.collect::<HashMap<_, _>>();
- 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.output_level {
+ 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,17 @@ 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
+ .output_level
+ .is_some_and(|output_level| !incremental || 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 {
@@ -551,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::<Vec<_>>();
+ 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::<HashMap<_, _>>();
+ 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<ManifestEntry>,
@@ -603,6 +706,13 @@ pub(super) fn prepare_results(table: &Table, results:
&[CompactResult]) -> Resul
.iter()
.map(|file| file.row_count as u64)
.sum::<u64>();
+ // 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() {
@@ -612,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)
{
diff --git a/crates/paimon/src/table/compact/append.rs
b/crates/paimon/src/table/compact/append.rs
index 58fb92f3..375b9dad 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,
+ output_level: None,
});
}
}
diff --git a/crates/paimon/src/table/compact/data_evolution.rs
b/crates/paimon/src/table/compact/data_evolution.rs
index 4e4b6f56..07ea02f6 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,
+ output_level: None,
});
}
}
diff --git a/crates/paimon/src/table/compact/primary_key.rs
b/crates/paimon/src/table/compact/primary_key.rs
index 156c7985..5f7ae111 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, output_level)) =
+ planner::select(files, context.schema.options())?
+ {
+ tasks.push(CompactTask {
+ context: context.clone(),
+ task_id: tasks.len(),
+ partition,
+ bucket,
+ total_buckets,
+ files,
+ output_level: Some(output_level),
+ });
+ }
+ 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,
+ 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
new file mode 100644
index 00000000..f519949d
--- /dev/null
+++ b/crates/paimon/src/table/compact/primary_key/planner.rs
@@ -0,0 +1,466 @@
+// 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. 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};
+use std::collections::BTreeMap;
+
+pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result<bool>
{
+ 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<String, String>,
+ key: &str,
+ default: i32,
+ minimum: i32,
+) -> Result<i32> {
+ let value = options
+ .get(key)
+ .map(|value| value.parse::<i32>())
+ .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<DataFileMeta>,
+ properties: &std::collections::HashMap<String, String>,
+) -> Result<Option<(Vec<DataFileMeta>, 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(
+ 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::<Vec<_>>();
+ 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::<i32, Vec<DataFileMeta>>::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::<Vec<_>>();
+ 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::<u128>()
+ })
+ .collect::<Vec<_>>();
+ let mut count = None;
+ if runs.len() >= trigger as usize {
+ let additional = sizes[..sizes.len() - 1].iter().sum::<u128>();
+ if additional * 100 > amplification as u128 * sizes[sizes.len() - 1] {
+ count = Some(runs.len());
+ } else {
+ count = pick_for_size_ratio(&runs, &sizes, ratio, 1, false);
+ }
+ if count.is_none() && runs.len() > trigger as usize {
+ count = pick_for_size_ratio(
+ &runs,
+ &sizes,
+ ratio,
+ runs.len() - trigger as usize + 1,
+ false,
+ );
+ }
+ }
+ 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 {
+ (runs[count].0 - 1).max(1)
+ };
+ let selected = runs[..count]
+ .iter()
+ .flat_map(|(_, files)| files.clone())
+ .collect::<Vec<_>>();
+ if selected.len() == 1 && selected[0].level == output_level {
+ return Ok(None);
+ }
+ Ok(Some((selected, output_level)))
+}
+
+fn pick_for_size_ratio(
+ runs: &[(i32, Vec<DataFileMeta>)],
+ sizes: &[u128],
+ ratio: i32,
+ mut count: usize,
+ force: bool,
+) -> Option<usize> {
+ let mut size = sizes[..count].iter().sum::<u128>();
+ 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::*;
+ 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, output_level) = 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<_>>(),
+ vec!["new"]
+ );
+ assert_eq!(output_level, 4);
+ let (selected, output_level) = 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!(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, output_level) = select(
+ vec![
+ file("small", 3, 1),
+ file("medium", 4, 10),
+ file("anchor", 5, 1000),
+ ],
+ &options,
+ )
+ .unwrap()
+ .unwrap();
+ assert_eq!(selected.len(), 2);
+ assert_eq!(output_level, 4);
+ let (selected, output_level) = select(
+ vec![
+ file("large", 1, 1000),
+ file("other", 3, 1000),
+ file("old", 5, 100),
+ ],
+ &options,
+ )
+ .unwrap()
+ .unwrap();
+ assert_eq!(selected.len(), 3);
+ assert_eq!(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<_>>(),
+ vec!["newer-l0", "older-l0", "a", "b"]
+ );
+ assert_eq!(expected.1, 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<_>>(),
+ 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, output_level) = selected.unwrap();
+ assert_eq!(files.len(), 2);
+ assert_eq!(output_level, 4);
+ } else {
+ assert!(selected.is_none());
+ }
+ }
+ // Arithmetic must not overflow when byte sizes exceed i64 percentages.
+ 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!(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 (_, 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!(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());
+ }
+ }
+
+ #[test]
+ fn normal_maintenance_takes_precedence_over_forcing_l0() {
+ let (files, output_level) = 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!(output_level, 5);
+
+ let options =
HashMap::from([("num-sorted-run.compaction-trigger".into(), "2".into())]);
+ let (files, output_level) = 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!(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, output_level) = 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!(output_level, 5);
+ }
+}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 9d6c4187..007fb1b7 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -1961,6 +1961,13 @@ impl TableCommit {
.iter()
.map(|e| (e.partition(), e.bucket()))
.collect::<HashSet<_>>();
+ 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()
.map(|entry| commit_file_identifier(entry))
@@ -1983,8 +1990,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::<HashMap<_, _>>();
@@ -2002,7 +2008,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,
});
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 00000000..af2be93b
--- /dev/null
+++ b/crates/paimon/tests/pk_compact/incremental_planning.rs
@@ -0,0 +1,440 @@
+// 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<CompactResult> {
+ 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<paimon::spec::DataFileMeta> {
+ 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::<Vec<_>>();
+ 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<i32>, values: Vec<i32>, 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;
+ 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",
+ "output-level",
+ "legacy-output-level",
+ ] {
+ 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["output_level"] = 5.into(),
+ "legacy-output-level" => {
+ task.as_object_mut().unwrap().remove("output_level");
+ }
+ _ => unreachable!(),
+ };
+ let mut encoded = task_json.clone();
+ 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();
+ }
+ }
+ if mutation == "legacy-output-level" {
+ for file in encoded[0]["files"].as_array_mut().unwrap() {
+ file["_LEVEL"] = 1.into();
+ }
+ }
+ let changed: Vec<CompactResult> =
serde_json::from_value(encoded).unwrap();
+ 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()
+ .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_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::<Vec<_>>();
+ 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<CompactResult> =
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();
+ 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/crates/paimon/tests/pk_compact_test.rs
b/crates/paimon/tests/pk_compact_test.rs
index a79323a9..4021bf42 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 2f8121f2..d2ac5ab9 100644
--- a/docs/src/compaction.md
+++ b/docs/src/compaction.md
@@ -115,12 +115,37 @@ 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. 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
+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 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:
