From 4900d80a316df3bc55cca58f7f541b2c0bd3fbc0 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 30 Oct 2026 08:44:55 +0800 Subject: [PATCH 1/4] feat(table): support incremental first-row compaction --- crates/paimon/src/table/compact.rs | 27 ++- .../paimon/src/table/compact/primary_key.rs | 31 ++- .../table/compact/primary_key/lookup_store.rs | 7 +- .../src/table/compact/primary_key/planner.rs | 18 +- .../tests/pk_compact/incremental_lookup.rs | 200 ++++++++++++++++++ .../tests/pk_compact/lookup_changelog.rs | 54 +++-- docs/src/compaction.md | 19 +- 7 files changed, 318 insertions(+), 38 deletions(-) diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index d3c56c9db..918e7a0ac 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -115,20 +115,26 @@ impl<'a> CompactBuilder<'a> { ensure_supported(self.table)?; let options = self.table.schema().core_options(); let lookup = options.try_changelog_producer()? == ChangelogProducer::Lookup + || (!self.table.schema().primary_keys().is_empty() + && options.merge_engine()? == MergeEngine::FirstRow) || (options.deletion_vectors_enabled() && !self.table.schema().primary_keys().is_empty()); let local_cache = self.table.file_io().local_cache(); if lookup && local_cache.is_none() { return Err(Error::ConfigInvalid { message: - "Lookup or deletion-vector compaction requires local-cache.enabled=true on the catalog/FileIO" + "Lookup, first-row or deletion-vector compaction requires local-cache.enabled=true on the catalog/FileIO" .into(), }); } Ok(CompactExecutor { table: self.table.clone(), resources: self.resources.clone(), - lookup_cache: if lookup && options.merge_engine()? == MergeEngine::Deduplicate { + lookup_cache: if lookup + && matches!( + options.merge_engine()?, + MergeEngine::Deduplicate | MergeEngine::FirstRow + ) { Some(Arc::new(primary_key::lookup_file::LookupFileCache::new( local_cache.expect("lookup cache validated above"), self.table.schema().options(), @@ -820,13 +826,13 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul } let options = task.context.schema.core_options(); let dv = task.is_primary_key() && options.deletion_vectors_enabled(); - let ignores_retracts = dv + // First-row can leave every accepted key in retained history, even + // with ignore-delete=false, and therefore produce no data files. + let can_omit_keys = dv || (task.is_primary_key() - && options.ignore_delete() - && matches!( - options.merge_engine()?, - MergeEngine::FirstRow | MergeEngine::PartialUpdate - )); + && (options.merge_engine()? == MergeEngine::FirstRow + || (options.ignore_delete() + && options.merge_engine()? == MergeEngine::PartialUpdate))); if dv && result.files.iter().any(|file| { file.delete_row_count != Some(0) || file.row_count > i64::from(u32::MAX) + 1 @@ -853,11 +859,10 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul collected.extend(updates.iter().cloned()); } if task.is_primary_key() - && (output_rows > input_rows - || (input_rows > 0 && output_rows == 0 && !ignores_retracts)) + && (output_rows > input_rows || (input_rows > 0 && output_rows == 0 && !can_omit_keys)) { return Err(invalid( - "PK compaction must retain each key's merged state unless retracts are ignored", + "PK compaction must retain merged keys unless its merge mode can omit them", )); } let lookup = options.try_changelog_producer()? == ChangelogProducer::Lookup; diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index 8320448c9..6b02a08de 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -48,6 +48,10 @@ use crate::table::table_read::{ use crate::table::{DataSplit, Plan, Table}; use crate::Result; +// A lookup hit remains in an unselected file; selected positive rows have only +// FILE_LEVEL_METADATA_KEY and must still be included in the compact output. +const LOOKUP_BASE_METADATA_KEY: &str = "paimon.compact.lookup-base"; + /// Select incremental sorted runs for supported merge modes. pub(super) struct PrimaryKeyCompactPlanner; @@ -182,6 +186,31 @@ impl MergeFunction for CompactReducer { } } +struct CompactFirstRow(Box); + +impl MergeFunction for CompactFirstRow { + fn merge( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> Result { + // Validate all selected records before checking retained history, as + // Java FirstRowMergeFunctionWrapper does when feeding its reducer. + let result = self.0.merge(rows, batches, columns, schema)?; + if rows.iter().any(|row| { + matches!(row.value_kind, 0 | 2) + && matches!(&batches[row.batch_idx], BufferedBatch::Source(batch) + if batch.schema().metadata().contains_key(LOOKUP_BASE_METADATA_KEY)) + }) { + Ok(MergeResult::Omit) + } else { + Ok(result) + } + } +} + fn new_merge_function( config: &KeyValueReadConfig, fields: &[DataField], @@ -220,7 +249,7 @@ fn new_merge_function( }; Ok(if config.merge_engine == MergeEngine::FirstRow { // First-row must still reject retracts unless ignore-delete is enabled. - inner + Box::new(CompactFirstRow(inner)) } else { Box::new(CompactReducer { inner, diff --git a/crates/paimon/src/table/compact/primary_key/lookup_store.rs b/crates/paimon/src/table/compact/primary_key/lookup_store.rs index 85b436e54..298139c75 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup_store.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup_store.rs @@ -18,7 +18,7 @@ use super::lookup_file::LookupFileCache; use crate::io::FileRead; use crate::resource::ResourceContext; -use crate::spec::{BinaryRowBuilder, DataField, DataFileMeta}; +use crate::spec::{BinaryRowBuilder, DataField, DataFileMeta, MergeEngine}; use crate::table::merge_tree_split_generator::KeyComparator; use crate::table::sort_merge::{LookupProvider, FILE_LEVEL_METADATA_KEY}; use crate::table::{DataSplit, Table}; @@ -53,6 +53,7 @@ struct LevelLookup { comparator: Option, resources: Option, schema_identity: Vec, + first_row: bool, deletions: Option, } @@ -191,6 +192,7 @@ pub(in crate::table::compact) async fn provider( comparator, resources, schema_identity, + first_row: table.schema().core_options().merge_engine()? == MergeEngine::FirstRow, deletions: deletions.clone(), }), deletions, @@ -333,6 +335,9 @@ impl LookupProvider for LevelLookup { } let mut metadata = batch.schema().metadata().clone(); metadata.insert(FILE_LEVEL_METADATA_KEY.into(), file.meta.level.to_string()); + if self.first_row { + metadata.insert(super::LOOKUP_BASE_METADATA_KEY.into(), String::new()); + } found = Some( RecordBatch::try_new( Arc::new(batch.schema().as_ref().clone().with_metadata(metadata)), diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index 102583984..72018f356 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -25,11 +25,21 @@ use std::collections::BTreeMap; pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { let options = schema.core_options(); - Ok(options.merge_engine()? == MergeEngine::Deduplicate - && matches!( - options.try_changelog_producer()?, + let producer = options.try_changelog_producer()?; + Ok(match options.merge_engine()? { + MergeEngine::Deduplicate => matches!( + producer, ChangelogProducer::None | ChangelogProducer::Input | ChangelogProducer::Lookup - )) + ), + MergeEngine::FirstRow => { + !options.deletion_vectors_enabled() + && matches!( + producer, + ChangelogProducer::None | ChangelogProducer::Lookup + ) + } + _ => false, + }) } fn number( diff --git a/crates/paimon/tests/pk_compact/incremental_lookup.rs b/crates/paimon/tests/pk_compact/incremental_lookup.rs index abc7e44a5..6a0598412 100644 --- a/crates/paimon/tests/pk_compact/incremental_lookup.rs +++ b/crates/paimon/tests/pk_compact/incremental_lookup.rs @@ -487,3 +487,203 @@ async fn incremental_lookup_prunes_old_schema_history_by_unchanged_keys() { ); } } + +#[tokio::test] +async fn incremental_first_row_preserves_history_and_only_materializes_new_keys() { + for producer in ["none", "lookup"] { + for bucket in ["1", "-1"] { + let (table, probe) = probed_table_with_options(&[ + ("merge-engine", "first-row"), + ("changelog-producer", producer), + ("bucket", bucket), + ("read.batch-size", "7"), + ]) + .await; + let anchor = seed(&table).await; + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let original = (0..1000).map(|key| (2 * key, 2 * key)).collect::>(); + *probe.history.lock().unwrap() = anchor.file_name.clone(); + write_batch( + &table, + &make_batch_with_kinds( + vec![42, 43, 45, 42], + vec![9999, 4300, 4500, 9998], + vec![0, 2, 0, 0], + ), + ) + .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.level == 0), + "large accepted history must remain in place" + ); + let (cold, hot) = cold_and_hot(&table, &plan, &probe, &anchor).await; + for results in [&cold, &hot] { + assert_eq!( + results[0] + .files() + .iter() + .map(|file| file.row_count) + .sum::(), + 2, + "lookup hits must not be copied into a lower level" + ); + assert!(results[0].files().iter().all(|file| file.level == 4)); + } + let commit = builder.new_commit().unwrap(); + commit.commit_compact(hot).await.unwrap(); + let end = latest(&table).await; + commit.commit_compact(cold).await.unwrap(); + assert_eq!(latest(&table).await, end); + let mut expected = original.clone(); + expected.extend([(43, 4300), (45, 4500)]); + expected.sort(); + assert_eq!(rows(&table).await, expected); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&snapshot))).await, + original + ); + let files = table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .unwrap(); + assert!(files + .splits() + .iter() + .flat_map(|split| split.data_files()) + .any(|file| file == &anchor)); + if producer == "lookup" { + let mut expected_events = + vec![("+U".into(), 43, 4300, 1001), ("+I".into(), 45, 4500, 1002)]; + expected_events.sort(); + assert_eq!(events(&table, snapshot.id(), end).await, expected_events); + } + } + } +} + +#[tokio::test] +async fn incremental_first_row_empty_results_support_legacy_metadata_and_concurrent_l0() { + for producer in ["none", "lookup"] { + let (table, _) = probed_table_with_options(&[ + ("merge-engine", "first-row"), + ("changelog-producer", producer), + ]) + .await; + let anchor = seed(&table).await; + write_batch(&table, &make_batch(vec![42, 44], vec![4200, 4400])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let results = execute(&table, &plan).await; + assert!(results[0].files().is_empty()); + assert!(results[0].changelog_files().is_empty()); + write_batch(&table, &make_batch(vec![43], vec![4300])).await; + let mut legacy = serde_json::to_value(&results).unwrap(); + for result in legacy.as_array_mut().unwrap() { + result["task"] + .as_object_mut() + .unwrap() + .remove("output_level"); + } + let commit = builder.new_commit().unwrap(); + let metrics = commit + .commit_compact(serde_json::from_value(legacy).unwrap()) + .await + .unwrap(); + let published = latest(&table).await; + assert_eq!(commit.commit_compact(results).await.unwrap(), metrics); + assert_eq!(latest(&table).await, published); + assert_eq!( + rows(&table).await.len(), + 1000, + "concurrent L0 is not accepted before its own compaction" + ); + let next = table + .new_compact_builder() + .new_planner() + .unwrap() + .plan() + .await + .unwrap(); + assert_eq!(next.tasks().len(), 1); + assert!(next.tasks()[0].files().iter().all(|file| file.level == 0)); + compact(&table).await; + let visible = rows(&table).await; + assert_eq!(visible.len(), 1001); + assert!( + visible.contains(&(42, 42)) + && visible.contains(&(44, 44)) + && visible.contains(&(43, 4300)) + ); + let files = table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .unwrap(); + assert!(files + .splits() + .iter() + .flat_map(|split| split.data_files()) + .any(|file| file == &anchor)); + if producer == "lookup" { + let changes = events(&table, published, latest(&table).await).await; + assert_eq!(changes.len(), 1); + assert_eq!( + (changes[0].0.as_str(), changes[0].1, changes[0].2), + ("+I", 43, 4300) + ); + } + } +} + +#[tokio::test] +async fn incremental_first_row_rewrites_selected_history_and_looks_up_retained_history() { + let (table, _) = probed_table_with_options(&[("merge-engine", "first-row")]).await; + let anchor = seed(&table).await; + write_batch(&table, &make_batch(vec![43], vec![4300])).await; + let accepted = compact(&table).await; + assert_eq!(accepted[0].files()[0].level, 4); + let before = latest(&table).await; + write_batch(&table, &make_batch(vec![43, 45], vec![9999, 4500])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().iter().any(|file| file.level == 4)); + assert!(plan.tasks()[0] + .files() + .iter() + .all(|file| file.file_name != anchor.file_name)); + let result = execute(&table, &plan).await; + assert_eq!( + result[0].metrics().rows_written, + 2, + "selected positive rows must stay in the compact output" + ); + builder + .new_commit() + .unwrap() + .commit_compact(result) + .await + .unwrap(); + let visible = rows(&table).await; + assert_eq!(visible.len(), 1002); + assert!(visible.contains(&(43, 4300)) && visible.contains(&(45, 4500))); + let changes = events(&table, before, latest(&table).await).await; + assert_eq!(changes.len(), 1); + assert_eq!( + (changes[0].0.as_str(), changes[0].1, changes[0].2), + ("+I", 45, 4500) + ); +} diff --git a/crates/paimon/tests/pk_compact/lookup_changelog.rs b/crates/paimon/tests/pk_compact/lookup_changelog.rs index 78666aeea..800ff3750 100644 --- a/crates/paimon/tests/pk_compact/lookup_changelog.rs +++ b/crates/paimon/tests/pk_compact/lookup_changelog.rs @@ -363,7 +363,12 @@ async fn lookup_compaction_first_row_only_emits_new_keys() { #[tokio::test] async fn lookup_first_row_preserves_the_earliest_record_across_positive_levels() { for with_l0 in [false, true] { - let table = lookup_table(&[("merge-engine", "first-row")]).await; + let table = lookup_table(&[ + ("merge-engine", "first-row"), + ("num-sorted-run.compaction-trigger", "2"), + ("num-levels", "6"), + ]) + .await; append_at_level(&table, 10, 3).await; append_at_level(&table, 20, 2).await; let history = table @@ -376,6 +381,16 @@ async fn lookup_first_row_preserves_the_earliest_record_across_positive_levels() write_batch(&table, &make_batch(vec![1], vec![30])).await; } let end = compact(&table).await; + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap() + .commit_kind(), + &CommitKind::COMPACT + ); assert_eq!(rows(&table).await, vec![(1, 10)]); assert!(events(&table, history, end).await.is_empty()); } @@ -864,21 +879,28 @@ async fn lookup_compaction_validates_changelog_flags_before_execution() { #[tokio::test] async fn lookup_compaction_requires_the_catalog_cache() { - let table = table_with_cache( - pk_schema(&[ - ("changelog-producer", "lookup"), - ("local-cache.enabled", "true"), - ]), - &[("local-cache.enabled", "false")], - ) - .await; - assert!(table - .new_compact_builder() - .new_executor() - .err() - .unwrap() - .to_string() - .contains("local-cache.enabled=true")); + for (engine, producer) in [ + ("deduplicate", "lookup"), + ("first-row", "lookup"), + ("first-row", "none"), + ] { + let table = table_with_cache( + pk_schema(&[ + ("merge-engine", engine), + ("changelog-producer", producer), + ("local-cache.enabled", "true"), + ]), + &[("local-cache.enabled", "false")], + ) + .await; + assert!(table + .new_compact_builder() + .new_executor() + .err() + .unwrap() + .to_string() + .contains("local-cache.enabled=true")); + } } #[tokio::test] diff --git a/docs/src/compaction.md b/docs/src/compaction.md index e76928b91..1da5cd41a 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -115,8 +115,9 @@ 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`. -For `deduplicate` with `none`, `input` or `lookup`, the planner chooses incremental -sorted runs using Java's universal-compaction size and count policy. Each L0 file is a +For `deduplicate` with `none`, `input` or `lookup`, and non-DV `first-row` with +`none` or `lookup`, 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 @@ -130,9 +131,17 @@ The policy uses `num-sorted-run.compaction-trigger` (default `5`), `num-levels` `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. Without deletion vectors, outputs retain winning retracts, so older values +compaction is not exposed. Without deletion vectors, `deduplicate` outputs retain winning retracts, so older values in untouched files do not reappear after a delete or out-of-order update. +For `first-row`, both producers require actual catalog/FileIO local-cache. +The executor looks up keys in retained higher-level files. Existing accepted keys +stay in those files and produce neither a new data row nor a changelog event. +Selected positive-level rows still participate in the output. The `lookup` +producer emits only newly accepted keys, preserving their first add's row kind +and sequence. If every input key already exists, compaction can retire its +L0 inputs without creating a data or changelog file. + With deletion vectors, `deduplicate` supports the same incremental selection for `none`, `input` and `lookup`. All three require actual catalog/FileIO local-cache. Lookup Files preserve unfiltered physical row positions; captured deletion vectors @@ -149,7 +158,7 @@ retain their physical deletion metadata, including an empty list when appropriat Each output file is limited to 4,294,967,296 physical rows. Migrating files written without DV is unsupported, including enabling DV through temporary table options. -`first-row`, `partial-update` and `aggregation` retain their current +`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 @@ -158,7 +167,7 @@ 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 +Older serialized incremental 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. From 704973e7cd575317433a08dd645cd0a14fded8f9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 30 Oct 2026 13:57:53 +0800 Subject: [PATCH 2/4] feat(table): compact mergeable primary-key states incrementally --- crates/paimon/src/table/compact.rs | 6 +- .../src/table/compact/primary_key/planner.rs | 95 +++- .../tests/pk_compact/incremental_lookup.rs | 60 ++ .../tests/pk_compact/incremental_stateful.rs | 513 ++++++++++++++++++ .../tests/pk_compact/lookup_changelog.rs | 68 ++- crates/paimon/tests/pk_compact_test.rs | 3 + docs/src/compaction.md | 27 +- 7 files changed, 736 insertions(+), 36 deletions(-) create mode 100644 crates/paimon/tests/pk_compact/incremental_stateful.rs diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 918e7a0ac..fbd047f29 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -130,11 +130,7 @@ impl<'a> CompactBuilder<'a> { Ok(CompactExecutor { table: self.table.clone(), resources: self.resources.clone(), - lookup_cache: if lookup - && matches!( - options.merge_engine()?, - MergeEngine::Deduplicate | MergeEngine::FirstRow - ) { + lookup_cache: if lookup { Some(Arc::new(primary_key::lookup_file::LookupFileCache::new( local_cache.expect("lookup cache validated above"), self.table.schema().options(), diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index 72018f356..3611ee84e 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -19,9 +19,12 @@ //! before forcing L0 progress, as ForceUpLevel0Compaction does, without a //! full-bucket override. -use crate::spec::{ChangelogProducer, DataFileMeta, MergeEngine, TableSchema}; +use crate::spec::{ + AggregationConfig, ChangelogProducer, DataField, DataFileMeta, DataType, MergeEngine, + PartialUpdateConfig, TableSchema, +}; use crate::{Error, Result}; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashSet}; pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { let options = schema.core_options(); @@ -38,10 +41,96 @@ pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { ChangelogProducer::None | ChangelogProducer::Lookup ) } - _ => false, + MergeEngine::PartialUpdate | MergeEngine::Aggregation => { + !options.deletion_vectors_enabled() + && match producer { + ChangelogProducer::Lookup => true, + ChangelogProducer::None | ChangelogProducer::Input => { + plain_state_mergeable(schema)? + } + _ => false, + } + } }) } +// Plain queries combine all positive levels. Only states that can be folded +// again without depending on the original row order may leave history behind. +fn plain_state_mergeable(schema: &TableSchema) -> Result { + let options = schema.core_options(); + let keys = schema.primary_keys(); + match options.merge_engine()? { + MergeEngine::Aggregation => { + let config = AggregationConfig::new(schema.options()); + if config.remove_record_on_delete() { + return Ok(false); + } + let sequence = options.sequence_fields(); + Ok(schema + .fields() + .iter() + .filter(|field| { + !keys.iter().any(|key| key == field.name()) + && !sequence.iter().any(|name| name == &field.name()) + }) + .all(|field| { + mergeable_field( + schema, + field, + config + .agg_function_for_field(field.name()) + .or_else(|| config.default_agg_function()) + .unwrap_or("last_non_null_value"), + ) + })) + } + MergeEngine::PartialUpdate => { + if !options.ignore_delete() || !options.sequence_fields().is_empty() { + return Ok(false); + } + let config = PartialUpdateConfig::new(schema.options()); + let functions = config.validated_aggregate_functions(schema.fields(), keys)?; + let ordering = config + .validated_sequence_groups(schema.fields(), keys)? + .into_iter() + .flat_map(|group| group.sequence_fields) + .collect::>(); + Ok(schema + .fields() + .iter() + .filter(|field| !keys.iter().any(|key| key == field.name())) + .all(|field| { + ordering.contains(field.name()) + || functions + .get(field.name()) + .is_some_and(|function| mergeable_field(schema, field, function)) + })) + } + _ => Ok(false), + } +} + +fn mergeable_field(schema: &TableSchema, field: &DataField, function: &str) -> bool { + match crate::spec::canonical_aggregator_name(function) { + "min" | "max" | "bool_and" | "bool_or" => true, + // Floating sums can round differently when regrouped. Checked integer + // sums can fail in a subset even if the complete merge does not overflow. + "sum" => { + matches!( + field.data_type(), + DataType::TinyInt(_) + | DataType::SmallInt(_) + | DataType::Int(_) + | DataType::BigInt(_) + ) && !schema + .options() + .get(&format!("fields.{}.sum.fail-on-overflow", field.name())) + .is_some_and(|value| value.eq_ignore_ascii_case("true")) + } + _ => false, + } +} + fn number( options: &std::collections::HashMap, key: &str, diff --git a/crates/paimon/tests/pk_compact/incremental_lookup.rs b/crates/paimon/tests/pk_compact/incremental_lookup.rs index 6a0598412..25ea6d4df 100644 --- a/crates/paimon/tests/pk_compact/incremental_lookup.rs +++ b/crates/paimon/tests/pk_compact/incremental_lookup.rs @@ -687,3 +687,63 @@ async fn incremental_first_row_rewrites_selected_history_and_looks_up_retained_h ("+I", 45, 4500) ); } + +#[tokio::test] +async fn stateful_lookup_reuses_cached_history_for_sparse_rows_and_changelog() { + for engine in ["partial-update", "aggregation"] { + for bucket in ["1", "-1"] { + let mut options = vec![ + ("merge-engine", engine), + ("bucket", bucket), + ("read.batch-size", "7"), + ]; + if engine == "aggregation" { + options.push(("fields.value.aggregate-function", "sum")); + } + let (table, probe) = probed_table_with_options(&options).await; + let anchor = seed(&table).await; + let start = latest(&table).await; + *probe.history.lock().unwrap() = anchor.file_name.clone(); + let input = RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(vec![42, 44, 43])) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![Some(4200), None, Some(4300)])) + as arrow_array::ArrayRef, + ), + ]) + .unwrap(); + write_batch(&table, &input).await; + let expected = rows(&table).await; + let new_value = if engine == "aggregation" { 4242 } else { 4200 }; + assert!(expected.contains(&(42, new_value)) && expected.contains(&(44, 44))); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let (cold, hot) = cold_and_hot(&table, &plan, &probe, &anchor).await; + assert_eq!(hot[0].metrics().rows_written, 3); + let commit = builder.new_commit().unwrap(); + commit.commit_compact(hot).await.unwrap(); + let end = latest(&table).await; + commit.commit_compact(cold).await.unwrap(); + assert_eq!(latest(&table).await, end); + assert_eq!(rows(&table).await, expected); + let mut changes = vec![ + ("-U".into(), 42, 42, 21), + ("+U".into(), 42, i64::from(new_value), 1000), + ("-U".into(), 44, 44, 22), + ("+U".into(), 44, 44, 1001), + ("+I".into(), 43, 4300, 1002), + ]; + changes.sort(); + assert_eq!(events(&table, start, end).await, changes); + write_batch(&table, &make_batch(vec![42], vec![1])).await; + let next_expected = rows(&table).await; + compact(&table).await; + assert_eq!(rows(&table).await, next_expected); + } + } +} diff --git a/crates/paimon/tests/pk_compact/incremental_stateful.rs b/crates/paimon/tests/pk_compact/incremental_stateful.rs new file mode 100644 index 000000000..3323bf14e --- /dev/null +++ b/crates/paimon/tests/pk_compact/incremental_stateful.rs @@ -0,0 +1,513 @@ +// 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::lookup_changelog_tests::events; +use super::*; +use std::sync::Arc; + +fn schema(engine: &str, producer: &str, function: &str, checked: bool) -> TableSchema { + let mut 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", if producer == "input" { "-1" } else { "1" }) + .option("merge-engine", engine) + .option("changelog-producer", producer) + .option("fields.value.aggregate-function", function); + if engine == "partial-update" { + schema = schema.option("fields.seq.sequence-group", "value"); + if producer != "lookup" { + schema = schema.option("ignore-delete", "true"); + } + } else { + schema = schema.option("sequence.field", "seq"); + } + if checked { + schema = schema.option("fields.value.sum.fail-on-overflow", "true"); + } + TableSchema::new(0, &schema.build().unwrap()) +} + +fn batch(keys: Vec, value: i32, seq: i32) -> RecordBatch { + let len = keys.len(); + RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(keys)) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![value; len])) as arrow_array::ArrayRef, + ), + ( + "seq", + Arc::new(Int32Array::from(vec![seq; len])) as arrow_array::ArrayRef, + ), + ]) + .unwrap() +} + +async fn compact(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 { + table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .unwrap() + .splits() + .iter() + .flat_map(|split| split.data_files()) + .cloned() + .collect() +} + +#[tokio::test] +async fn stateful_lookup_inserts_history_in_sequence_order_and_does_not_apply_it_twice() { + for engine in ["partial-update", "aggregation"] { + let table = table_with_schema(schema(engine, "lookup", "first_value", false)).await; + write_batch(&table, &batch((0..1000).collect(), 20, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + assert_eq!(anchor.level, 5); + let start = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + write_batch(&table, &batch(vec![42], 10, 10)).await; + write_batch(&table, &batch(vec![42], 30, 30)).await; + let expected = rows(&table).await; + assert_eq!(expected[42], (42, 10)); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let results = execute(&table, &plan).await; + assert_eq!(results[0].metrics().rows_written, 1); + assert_eq!(results[0].files()[0].level, 4); + builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + let end = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + assert_eq!( + events(&table, start, end).await, + vec![("+U".into(), 42, 10), ("-U".into(), 42, 20)] + ); + assert_eq!(rows(&table).await, expected); + assert!(active_files(&table).await.contains(&anchor)); + write_batch(&table, &batch(vec![42], 25, 25)).await; + let before_next = rows(&table).await; + compact(&table).await; + assert_eq!(rows(&table).await, before_next); + } +} + +#[tokio::test] +async fn stateful_plain_mergeable_fields_leave_history_out_of_the_output_state() { + for engine in ["partial-update", "aggregation"] { + for producer in ["none", "input"] { + for (function, expected_value) in [("sum", 60), ("min", 10), ("max", 30)] { + let table = table_with_cache( + schema(engine, producer, function, false), + &[("local-cache.enabled", "false")], + ) + .await; + write_batch(&table, &batch((0..1000).collect(), 20, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + assert_eq!(anchor.level, 5, "{engine}/{producer}/{function}"); + write_batch(&table, &batch(vec![42], 10, 10)).await; + write_batch(&table, &batch(vec![42], 30, 30)).await; + let expected = rows(&table).await; + assert_eq!(expected[42], (42, expected_value)); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let results = execute(&table, &plan).await; + assert_eq!(results[0].metrics().rows_written, 1); + builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap(); + assert!(active_files(&table).await.contains(&anchor)); + assert_eq!( + rows(&table).await, + expected, + "{engine}/{producer}/{function}" + ); + write_batch(&table, &batch(vec![42], 5, 15)).await; + let next_expected = rows(&table).await; + compact(&table).await; + assert_eq!(rows(&table).await, next_expected); + } + } + } +} + +#[tokio::test] +async fn stateful_plain_order_sensitive_and_checked_sum_configs_keep_all_history() { + for engine in ["partial-update", "aggregation"] { + for producer in ["none", "input"] { + for (function, checked, initial, earlier, later, expected_value) in [ + ("first_value", false, 20, 10, 30, 10), + ("sum", true, -i32::MAX, i32::MAX, 1, 1), + ] { + let table = table_with_schema(schema(engine, producer, function, checked)).await; + write_batch(&table, &batch((0..1000).collect(), initial, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + write_batch(&table, &batch(vec![42], earlier, 10)).await; + write_batch(&table, &batch(vec![42], later, 30)).await; + let expected = rows(&table).await; + assert_eq!(expected[42], (42, expected_value)); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + builder + .new_commit() + .unwrap() + .commit_compact(execute(&table, &plan).await) + .await + .unwrap(); + assert_eq!(rows(&table).await[42], expected[42]); + assert!(!active_files(&table).await.contains(&anchor)); + } + } + } +} + +#[tokio::test] +async fn stateful_plain_floating_sum_keeps_history_to_preserve_finite_rounding() { + for producer in ["none", "input"] { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Double(paimon::spec::DoubleType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "aggregation") + .option("changelog-producer", producer) + .option("fields.value.aggregate-function", "sum") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |keys: Vec, value| { + let len = keys.len(); + RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(keys)) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(arrow_array::Float64Array::from(vec![value; len])) + as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + write_batch(&table, &batch((0..1000).collect(), 1e20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + write_batch(&table, &batch(vec![42], -1e20)).await; + write_batch(&table, &batch(vec![42], 3.0)).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + builder + .new_commit() + .unwrap() + .commit_compact(execute(&table, &plan).await) + .await + .unwrap(); + let batches = read(&table, None).await; + let actual = batches + .iter() + .find_map(|batch| { + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column_by_name("value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .find(|&row| ids.value(row) == 42) + .map(|row| values.value(row)) + }) + .unwrap(); + assert_eq!(actual, 3.0); + assert!(!active_files(&table).await.contains(&anchor)); + } +} + +#[tokio::test] +async fn stateful_plain_boolean_fields_are_mergeable_for_both_engines() { + for engine in ["partial-update", "aggregation"] { + for producer in ["none", "input"] { + for (function, initial) in [("bool_and", true), ("bool_or", false)] { + let mut schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Boolean(paimon::spec::BooleanType::new())) + .column("seq", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", engine) + .option("changelog-producer", producer) + .option("fields.value.aggregate-function", function); + if engine == "partial-update" { + schema = schema + .option("fields.seq.sequence-group", "value") + .option("ignore-delete", "true"); + } else { + schema = schema.option("sequence.field", "seq"); + } + let table = table_with_cache( + TableSchema::new(0, &schema.build().unwrap()), + &[("local-cache.enabled", "false")], + ) + .await; + let batch = |keys: Vec, value, seq| { + let len = keys.len(); + RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(keys)) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(arrow_array::BooleanArray::from(vec![value; len])) + as arrow_array::ArrayRef, + ), + ( + "seq", + Arc::new(Int32Array::from(vec![seq; len])) as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + write_batch(&table, &batch((0..1000).collect(), initial, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + assert_eq!(anchor.level, 5); + write_batch(&table, &batch(vec![42], !initial, 10)).await; + write_batch(&table, &batch(vec![42], initial, 30)).await; + let results = compact(&table).await; + assert_eq!(results[0].metrics().rows_written, 1); + assert!(active_files(&table).await.contains(&anchor)); + let mut found = false; + for batch in read(&table, None).await { + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column_by_name("value") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + if let Some(row) = (0..batch.num_rows()).find(|&row| ids.value(row) == 42) { + assert_eq!(values.value(row), !initial); + found = true; + } + } + assert!(found, "updated key must remain visible"); + } + } + } +} + +#[tokio::test] +async fn stateful_plain_sparse_overwrite_keeps_interleaved_column_history() { + for producer in ["none", "input"] { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("other", DataType::Int(IntType::new())) + .column("seq", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "partial-update") + .option("changelog-producer", producer) + .option("ignore-delete", "true") + .option("sequence.field", "seq") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |keys: Vec, value: Option, other: Option, seq| { + let len = keys.len(); + RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(keys)) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![value; len])) as arrow_array::ArrayRef, + ), + ( + "other", + Arc::new(Int32Array::from(vec![other; len])) as arrow_array::ArrayRef, + ), + ( + "seq", + Arc::new(Int32Array::from(vec![seq; len])) as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + write_batch(&table, &batch((0..1000).collect(), Some(20), Some(20), 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + write_batch(&table, &batch(vec![42], Some(10), None, 10)).await; + write_batch(&table, &batch(vec![42], None, Some(30), 30)).await; + assert_eq!(rows(&table).await[42], (42, 20)); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + builder + .new_commit() + .unwrap() + .commit_compact(execute(&table, &plan).await) + .await + .unwrap(); + assert_eq!(rows(&table).await[42], (42, 20)); + assert!(!active_files(&table).await.contains(&anchor)); + let batches = read(&table, None).await; + assert!(batches.iter().any(|batch| { + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let other = batch + .column_by_name("other") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()).any(|row| ids.value(row) == 42 && other.value(row) == 30) + })); + } +} + +#[tokio::test] +async fn stateful_reset_on_delete_preserves_tombstones_and_only_uses_incremental_lookup() { + for producer in ["none", "input", "lookup"] { + 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("merge-engine", "aggregation") + .option("changelog-producer", producer) + .option("fields.value.aggregate-function", "sum") + .option("sequence.field", "seq") + .option("aggregation.remove-record-on-delete", "true") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + write_batch(&table, &batch((0..1000).collect(), 20, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + let start = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + let before_delete = batch(vec![42], 5, 30); + let delete = RecordBatch::try_from_iter( + before_delete + .schema() + .fields() + .iter() + .zip(before_delete.columns()) + .map(|(field, column)| (field.name().as_str(), column.clone())) + .chain(std::iter::once(( + "_VALUE_KIND", + Arc::new(arrow_array::Int8Array::from(vec![3])) as arrow_array::ArrayRef, + ))), + ) + .unwrap(); + write_batch(&table, &delete).await; + assert!(!rows(&table).await.iter().any(|(key, _)| *key == 42)); + let deleted = compact(&table).await; + assert_eq!( + deleted[0].metrics().rows_written, + if producer == "lookup" { 1 } else { 1000 } + ); + assert_eq!( + active_files(&table).await.contains(&anchor), + producer == "lookup" + ); + assert!(!rows(&table).await.iter().any(|(key, _)| *key == 42)); + let end = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + if producer == "lookup" { + assert_eq!( + events(&table, start, end).await, + vec![("-D".into(), 42, 20)] + ); + } + write_batch(&table, &batch(vec![42], 7, 40)).await; + let expected = rows(&table).await; + assert_eq!(expected[42], (42, 12)); + compact(&table).await; + assert_eq!(rows(&table).await, expected); + if producer == "lookup" { + let final_id = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + assert_eq!( + events(&table, end, final_id).await, + vec![("+I".into(), 42, 12)] + ); + } + } +} diff --git a/crates/paimon/tests/pk_compact/lookup_changelog.rs b/crates/paimon/tests/pk_compact/lookup_changelog.rs index 800ff3750..75980d80d 100644 --- a/crates/paimon/tests/pk_compact/lookup_changelog.rs +++ b/crates/paimon/tests/pk_compact/lookup_changelog.rs @@ -883,6 +883,8 @@ async fn lookup_compaction_requires_the_catalog_cache() { ("deduplicate", "lookup"), ("first-row", "lookup"), ("first-row", "none"), + ("partial-update", "lookup"), + ("aggregation", "lookup"), ] { let table = table_with_cache( pk_schema(&[ @@ -905,27 +907,47 @@ async fn lookup_compaction_requires_the_catalog_cache() { #[tokio::test] async fn lookup_compaction_selects_the_lowest_positive_level_as_materialized_history() { - let table = lookup_table(&[ - ("merge-engine", "aggregation"), - ("fields.value.aggregate-function", "sum"), - ]) - .await; - // Simulate Java lookup states: L2 already includes L3. Reducing both would - // produce 30 rather than 20 when the L0 delta is applied. - append_at_level(&table, 10, 3).await; - append_at_level(&table, 15, 2).await; - let history = table - .snapshot_manager() - .get_latest_snapshot_id() - .await - .unwrap() - .unwrap(); - write_batch(&table, &make_batch(vec![1], vec![5])).await; - assert_eq!(rows(&table).await, vec![(1, 20)]); - let end = compact(&table).await; - assert_eq!( - events(&table, history, end).await, - expected(&[("-U", 1, 15), ("+U", 1, 20)]) - ); - assert_eq!(rows(&table).await, vec![(1, 20)]); + for with_l0 in [false, true] { + let table = lookup_table(&[ + ("merge-engine", "aggregation"), + ("fields.value.aggregate-function", "sum"), + ("num-sorted-run.compaction-trigger", "2"), + ("num-levels", "6"), + ]) + .await; + // L2 already includes L3. Reducing both would count the history twice. + append_at_level(&table, 10, 3).await; + append_at_level(&table, 15, 2).await; + let history = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + if with_l0 { + write_batch(&table, &make_batch(vec![1], vec![5])).await; + } + let expected_value = if with_l0 { 20 } else { 15 }; + assert_eq!(rows(&table).await, vec![(1, expected_value)]); + let end = compact(&table).await; + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap() + .commit_kind(), + &CommitKind::COMPACT + ); + assert_eq!( + events(&table, history, end).await, + if with_l0 { + expected(&[("-U", 1, 15), ("+U", 1, 20)]) + } else { + vec![] + } + ); + assert_eq!(rows(&table).await, vec![(1, expected_value)]); + } } diff --git a/crates/paimon/tests/pk_compact_test.rs b/crates/paimon/tests/pk_compact_test.rs index fe45814be..80cdae1d3 100644 --- a/crates/paimon/tests/pk_compact_test.rs +++ b/crates/paimon/tests/pk_compact_test.rs @@ -29,6 +29,9 @@ mod incremental_planning_tests; #[path = "pk_compact/incremental_lookup.rs"] mod incremental_lookup_tests; +#[path = "pk_compact/incremental_stateful.rs"] +mod incremental_stateful_tests; + #[path = "pk_compact/dynamic_bucket.rs"] mod dynamic_tests; diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 1da5cd41a..fd4ae6ebb 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -158,11 +158,28 @@ retain their physical deletion metadata, including an empty list when appropriat Each output file is limited to 4,294,967,296 physical rows. Migrating files written without DV is unsupported, including enabling DV through temporary table options. -`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. +Non-DV `partial-update` and `aggregation` with `lookup` use the same incremental +policy and require actual catalog/FileIO local-cache. The executor inserts the +nearest retained positive-level state into each changed key's merge, in sequence +order. The output contains the materialized state; lookup scans use only the +nearest positive-level state, so retained older levels are not aggregated again. + +With `none` or `input`, these engines use incremental selection only when every +value field can be merged again independently of row order: `min`, `max`, +`bool_and`, `bool_or`, or integer `sum` without checked overflow. Aggregation must +have `aggregation.remove-record-on-delete=false`; its sequence fields keep their +normal metadata role. Partial update must use `ignore-delete=true`, no global +sequence fields, and protect every value field with a sequence group and one of +these aggregate functions. Each group's ordering fields retain its winning +sequence. These modes merge only selected inputs, leaving retained history to +ordinary query merging, and do not require local-cache. + +Other `none`/`input` configurations retain their previous selection: every active +file in the selected bucket participates, and output uses the highest input level +or level 1 for all-L0 inputs. This preserves sparse overwrite and order-sensitive +state such as `first_value`, finite floating-point summation, checked-overflow +behavior and reset-on-delete semantics. Existing retract/state restrictions still +apply. 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 fc67c5ee6c20e5d49db1d23cd220bc83362bf3aa Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 30 Oct 2026 14:27:37 +0800 Subject: [PATCH 3/4] refactor(table): carry lookup levels in KV rows --- .../paimon/src/table/compact/primary_key.rs | 40 +++-- .../src/table/compact/primary_key/lookup.rs | 21 ++- .../table/compact/primary_key/lookup_store.rs | 19 +-- crates/paimon/src/table/kv_file_reader.rs | 28 ++-- crates/paimon/src/table/sort_merge.rs | 157 +++++++++++++++--- 5 files changed, 181 insertions(+), 84 deletions(-) diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index 6b02a08de..35b97f7cb 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -48,10 +48,6 @@ use crate::table::table_read::{ use crate::table::{DataSplit, Plan, Table}; use crate::Result; -// A lookup hit remains in an unselected file; selected positive rows have only -// FILE_LEVEL_METADATA_KEY and must still be included in the compact output. -const LOOKUP_BASE_METADATA_KEY: &str = "paimon.compact.lookup-base"; - /// Select incremental sorted runs for supported merge modes. pub(super) struct PrimaryKeyCompactPlanner; @@ -186,7 +182,10 @@ impl MergeFunction for CompactReducer { } } -struct CompactFirstRow(Box); +struct CompactFirstRow { + inner: Box, + output_level: i32, +} impl MergeFunction for CompactFirstRow { fn merge( @@ -198,22 +197,23 @@ impl MergeFunction for CompactFirstRow { ) -> Result { // Validate all selected records before checking retained history, as // Java FirstRowMergeFunctionWrapper does when feeding its reducer. - let result = self.0.merge(rows, batches, columns, schema)?; - if rows.iter().any(|row| { - matches!(row.value_kind, 0 | 2) - && matches!(&batches[row.batch_idx], BufferedBatch::Source(batch) - if batch.schema().metadata().contains_key(LOOKUP_BASE_METADATA_KEY)) - }) { - Ok(MergeResult::Omit) - } else { - Ok(result) + let result = self.inner.merge(rows, batches, columns, schema)?; + for row in rows { + // Only a lookup hit can be above the output level. Its accepted + // record stays in retained history; selected positive rows must + // still be rewritten into this compaction's output. + if matches!(row.value_kind, 0 | 2) && row.level(batches)? > self.output_level { + return Ok(MergeResult::Omit); + } } + Ok(result) } } fn new_merge_function( config: &KeyValueReadConfig, fields: &[DataField], + output_level: i32, ) -> Result> { if config.merge_engine == MergeEngine::Deduplicate { return Ok(Box::new(CompactDeduplicate)); @@ -249,7 +249,10 @@ fn new_merge_function( }; Ok(if config.merge_engine == MergeEngine::FirstRow { // First-row must still reject retracts unless ignore-delete is enabled. - Box::new(CompactFirstRow(inner)) + Box::new(CompactFirstRow { + inner, + output_level, + }) } else { Box::new(CompactReducer { inner, @@ -343,14 +346,15 @@ pub(super) async fn execute( let schema = table.schema(); let options = schema.core_options(); let lookup_changelog = options.try_changelog_producer()? == ChangelogProducer::Lookup; - let drop_delete = options.deletion_vectors_enabled() && task.output_level() > 0; + let output_level = task.output_level(); + let drop_delete = options.deletion_vectors_enabled() && output_level > 0; let reader = new_reader(table, resources.clone())?.with_lookup(lookup); let mut batches = reader.read_with_merge_function(std::slice::from_ref(split), move |config, fields| { if lookup_changelog { - lookup::new_merge_function(config, fields) + lookup::new_merge_function(config, fields, output_level) } else { - new_merge_function(config, fields) + new_merge_function(config, fields, output_level) } })?; let keys = schema.trimmed_primary_keys(); diff --git a/crates/paimon/src/table/compact/primary_key/lookup.rs b/crates/paimon/src/table/compact/primary_key/lookup.rs index b56c65eab..4be6e90c1 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup.rs @@ -26,9 +26,7 @@ use arrow_schema::{DataType as ArrowType, SchemaRef}; use crate::spec::{DataField, MergeEngine, RowKind, VALUE_KIND_FIELD_NAME}; use crate::table::kv_file_reader::KeyValueReadConfig; -use crate::table::sort_merge::{ - lookup_rows, BufferedBatch, MergeFunction, MergeResult, MergeRow, FILE_LEVEL_METADATA_KEY, -}; +use crate::table::sort_merge::{lookup_rows, BufferedBatch, MergeFunction, MergeResult, MergeRow}; use crate::{Error, Result}; struct ChangelogOptions { @@ -81,10 +79,11 @@ struct LookupChangelogMergeFunction { pub(super) fn new_merge_function( config: &KeyValueReadConfig, fields: &[DataField], + output_level: i32, ) -> Result> { let options = ChangelogOptions::parse(&config.table_options)?; Ok(Box::new(LookupChangelogMergeFunction { - inner: super::new_merge_function(config, fields)?, + inner: super::new_merge_function(config, fields, output_level)?, first_row: config.merge_engine == MergeEngine::FirstRow, value_columns: fields .iter() @@ -317,13 +316,13 @@ impl MergeFunction for LookupChangelogMergeFunction { return Ok((result, None)); } if self.first_row { - let contains_history = rows.iter().any(|row| { - matches!(row.value_kind, 0 | 2) - && matches!(&batches[row.batch_idx], BufferedBatch::Source(batch) - if batch.schema().metadata().get(FILE_LEVEL_METADATA_KEY) - .and_then(|level| level.parse::().ok()) - .is_some_and(|level| level > 0)) - }); + let mut contains_history = false; + for row in rows { + if matches!(row.value_kind, 0 | 2) && row.level(batches)? > 0 { + contains_history = true; + break; + } + } // FirstRowMergeFunctionWrapper emits the first add unchanged, // including UPDATE_AFTER; an existing accepted history emits none. let changelog = if contains_history { diff --git a/crates/paimon/src/table/compact/primary_key/lookup_store.rs b/crates/paimon/src/table/compact/primary_key/lookup_store.rs index 298139c75..c561cdd30 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup_store.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup_store.rs @@ -18,9 +18,9 @@ use super::lookup_file::LookupFileCache; use crate::io::FileRead; use crate::resource::ResourceContext; -use crate::spec::{BinaryRowBuilder, DataField, DataFileMeta, MergeEngine}; +use crate::spec::{BinaryRowBuilder, DataField, DataFileMeta}; use crate::table::merge_tree_split_generator::KeyComparator; -use crate::table::sort_merge::{LookupProvider, FILE_LEVEL_METADATA_KEY}; +use crate::table::sort_merge::{with_level, LookupProvider}; use crate::table::{DataSplit, Table}; use crate::{Error, Result}; use arrow_array::RecordBatch; @@ -53,7 +53,6 @@ struct LevelLookup { comparator: Option, resources: Option, schema_identity: Vec, - first_row: bool, deletions: Option, } @@ -192,7 +191,6 @@ pub(in crate::table::compact) async fn provider( comparator, resources, schema_identity, - first_row: table.schema().core_options().merge_engine()? == MergeEngine::FirstRow, deletions: deletions.clone(), }), deletions, @@ -333,18 +331,7 @@ impl LookupProvider for LevelLookup { .or_default() .insert(position); } - let mut metadata = batch.schema().metadata().clone(); - metadata.insert(FILE_LEVEL_METADATA_KEY.into(), file.meta.level.to_string()); - if self.first_row { - metadata.insert(super::LOOKUP_BASE_METADATA_KEY.into(), String::new()); - } - found = Some( - RecordBatch::try_new( - Arc::new(batch.schema().as_ref().clone().with_metadata(metadata)), - batch.columns().to_vec(), - ) - .map_err(invalid)?, - ); + found = Some(with_level(batch, file.meta.level)?); } } if found.is_some() { diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 2eab24f3a..1970eecdf 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -27,9 +27,9 @@ use super::data_file_reader::{file_index_selection_to_local_ranges, DataFileReader}; use super::sort_merge::{ - AggregateMergeFunction, DeduplicateMergeFunction, FirstRowMergeFunction, LookupMergeFunction, - MergeFunction, PartialUpdateMergeFunction, QueryReducerMergeFunction, SortMergeReaderBuilder, - CHANGELOG_BATCH_METADATA_KEY, FILE_LEVEL_METADATA_KEY, + schema_with_level, with_level, AggregateMergeFunction, DeduplicateMergeFunction, + FirstRowMergeFunction, LookupMergeFunction, MergeFunction, PartialUpdateMergeFunction, + QueryReducerMergeFunction, SortMergeReaderBuilder, CHANGELOG_BATCH_METADATA_KEY, }; use crate::arrow::format::MosaicPrefetchOptions; use crate::arrow::shredding::variant::assemble_variant_extraction_array; @@ -353,10 +353,11 @@ impl KeyValueFileReader { &config.table_fields, &config.primary_keys, ); - let include_file_levels = config - .table_options - .get("changelog-producer") - .is_some_and(|value| value.eq_ignore_ascii_case("lookup")); + let include_file_levels = config.merge_engine == MergeEngine::FirstRow + || config + .table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")); Self { file_io, config, @@ -642,6 +643,11 @@ impl KeyValueFileReader { internal_read_type.extend(user_fields.clone()); let internal_schema = crate::arrow::build_read_arrow_schema(&internal_read_type)?; + let merge_input_schema = if self.include_file_levels { + schema_with_level(internal_schema.as_ref())? + } else { + internal_schema + }; // Output schema: user's read_type order let output_schema = crate::arrow::build_read_arrow_schema(&self.config.read_type)?; @@ -882,11 +888,7 @@ impl KeyValueFileReader { while let Some(batch) = file_stream.next().await { let batch = batch?; if include_file_levels { - let mut metadata = batch.schema().metadata().clone(); - metadata.insert(FILE_LEVEL_METADATA_KEY.into(), file_level.to_string()); - yield RecordBatch::try_new( - Arc::new(batch.schema().as_ref().clone().with_metadata(metadata)), batch.columns().to_vec(), - ).map_err(|error| Error::DataInvalid { message: format!("Invalid lookup input batch: {error}"), source: Some(Box::new(error)) })?; + yield with_level(batch, file_level)?; } else { yield batch; } } } @@ -911,7 +913,7 @@ impl KeyValueFileReader { // still contain duplicate keys. let mut merge_stream = SortMergeReaderBuilder::new( file_streams, - internal_schema.clone(), + merge_input_schema.clone(), key_indices.clone(), seq_index, value_kind_index, diff --git a/crates/paimon/src/table/sort_merge.rs b/crates/paimon/src/table/sort_merge.rs index f9068ef51..89f50bb12 100644 --- a/crates/paimon/src/table/sort_merge.rs +++ b/crates/paimon/src/table/sort_merge.rs @@ -30,10 +30,12 @@ use crate::spec::{AggregationConfig, CoreOptions, DataField, PartialUpdateConfig use crate::table::aggregator::{new_aggregator, FieldAggregator}; use crate::table::ArrowRecordBatchStream; use crate::Error; -use arrow_array::{new_null_array, Array, ArrayRef, Int64Array, Int8Array, RecordBatch}; +use arrow_array::{ + new_null_array, Array, ArrayRef, Int32Array, Int64Array, Int8Array, RecordBatch, +}; use arrow_ord::ord::make_comparator; use arrow_row::{OwnedRow, RowConverter, Rows, SortField}; -use arrow_schema::{SchemaRef, SortOptions}; +use arrow_schema::{Schema, SchemaRef, SortOptions}; use arrow_select::interleave::interleave; use async_stream::try_stream; use futures::StreamExt; @@ -84,7 +86,6 @@ pub(crate) struct MergeRow { pub user_sequence: Option, } -#[cfg(test)] impl MergeRow { fn source_batch<'a>( &self, @@ -108,6 +109,19 @@ impl MergeRow { }), } } + + pub(super) fn level(&self, batches: &[BufferedBatch]) -> crate::Result { + let batch = self.source_batch(batches)?; + let levels = batch + .column_by_name(LEVEL_FIELD_NAME) + .and_then(|column| column.as_any().downcast_ref::()) + .filter(|levels| self.row_idx < levels.len() && levels.is_valid(self.row_idx)) + .ok_or_else(|| Error::DataInvalid { + message: "Merge input must contain a non-null INT _LEVEL for each row".into(), + source: None, + })?; + Ok(levels.value(self.row_idx)) + } } /// Merge result for rows sharing the same primary key. @@ -172,9 +186,36 @@ pub(crate) trait LookupProvider: Send + Sync { async fn lookup(&self, key: &[u8]) -> crate::Result>; } -/// Internal batch metadata used only by lookup compaction. Source levels -/// survive physical reads; changelog batches are routed to a separate writer. -pub(super) const FILE_LEVEL_METADATA_KEY: &str = "paimon.compact.file-level"; +const LEVEL_FIELD_NAME: &str = "_LEVEL"; + +/// Java KeyValue.schemaWithLevel appends the non-null INT SpecialFields.LEVEL. +/// This column carries each input row's file level through merging; it is not +/// part of the data-file schema or the merge output. +pub(super) fn schema_with_level(schema: &Schema) -> crate::Result { + let level = crate::arrow::build_target_arrow_schema(&[DataField::new( + i32::MAX - 3, + LEVEL_FIELD_NAME.to_string(), + crate::spec::DataType::Int(crate::spec::IntType::with_nullable(false)), + )])?; + let mut fields = schema.fields().to_vec(); + fields.push(level.field(0).clone().into()); + Ok(Arc::new(Schema::new_with_metadata( + fields, + schema.metadata().clone(), + ))) +} + +pub(super) fn with_level(batch: RecordBatch, level: i32) -> crate::Result { + let schema = schema_with_level(batch.schema().as_ref())?; + let mut columns = batch.columns().to_vec(); + columns.push(Arc::new(Int32Array::from(vec![level; batch.num_rows()]))); + RecordBatch::try_new(schema, columns).map_err(|error| Error::DataInvalid { + message: format!("Invalid merge input with level: {error}"), + source: Some(Box::new(error)), + }) +} + +/// Route changelog batches to a separate writer. pub(super) const CHANGELOG_BATCH_METADATA_KEY: &str = "paimon.compact.changelog"; /// Lookup states at positive levels are already materialized. The lowest @@ -188,21 +229,7 @@ pub(super) fn lookup_rows( let mut base_level = i32::MAX; let mut levels = Vec::with_capacity(rows.len()); for (index, row) in rows.iter().enumerate() { - let BufferedBatch::Source(batch) = &batches[row.batch_idx] else { - return Err(Error::UnexpectedError { - message: "Lookup input must be a source batch".into(), - source: None, - }); - }; - let level = batch - .schema() - .metadata() - .get(FILE_LEVEL_METADATA_KEY) - .and_then(|level| level.parse::().ok()) - .ok_or_else(|| Error::DataInvalid { - message: "Lookup input is missing its file level".into(), - source: None, - })?; + let level = row.level(batches)?; if level > 0 && (base.is_none() || level < base_level) { base = Some(index); base_level = level; @@ -1656,11 +1683,13 @@ fn sort_merge_stream( } if let Some((lookup, schema)) = &lookup { - let has_base = same_key_rows.iter().any(|row| match &batch_buffer[row.batch_idx] { - BufferedBatch::Source(batch) => batch.schema().metadata().get(FILE_LEVEL_METADATA_KEY) - .and_then(|level| level.parse::().ok()).is_some_and(|level| level > 0), - _ => false, - }); + let mut has_base = false; + for row in &same_key_rows { + if row.level(&batch_buffer)? > 0 { + has_base = true; + break; + } + } if !has_base { if let Some(batch) = lookup.lookup(&winner_key).await? { if batch.num_rows() != 1 || batch.schema().fields() != schema.fields() { @@ -1843,6 +1872,82 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + #[test] + fn level_column_survives_concatenation_and_serialization() { + let input = RecordBatch::try_from_iter([( + "value", + Arc::new(Int64Array::from(vec![10, 20])) as ArrayRef, + )]) + .unwrap(); + let inputs = [ + with_level(input.clone(), 0).unwrap(), + with_level(input, 3).unwrap(), + ]; + let schema = inputs[0].schema(); + let field = schema.fields().last().unwrap(); + assert_eq!(field.name(), "_LEVEL"); + assert_eq!(field.data_type(), &DataType::Int32); + assert!(!field.is_nullable()); + assert_eq!( + field.metadata()["PARQUET:field_id"], + (i32::MAX - 3).to_string() + ); + let batch = arrow_select::concat::concat_batches(&schema, &inputs).unwrap(); + let mut bytes = Vec::new(); + { + let mut writer = arrow_ipc::writer::StreamWriter::try_new(&mut bytes, &schema).unwrap(); + writer.write(&batch).unwrap(); + writer.finish().unwrap(); + } + let restored = arrow_ipc::reader::StreamReader::try_new(bytes.as_slice(), None) + .unwrap() + .next() + .unwrap() + .unwrap(); + assert_eq!(restored.schema(), schema); + let batches = vec![BufferedBatch::Source(restored)]; + let levels = (0..4) + .map(|row_idx| { + MergeRow { + batch_idx: 0, + row_idx, + sequence_number: row_idx as i64, + value_kind: 0, + user_sequence: None, + } + .level(&batches) + .unwrap() + }) + .collect::>(); + assert_eq!(levels, vec![0, 0, 3, 3]); + } + + #[test] + fn lookup_uses_each_rows_level_in_a_mixed_batch() { + let batch = RecordBatch::try_from_iter([( + "_LEVEL", + Arc::new(Int32Array::from(vec![-1, 0, 5, 2])) as ArrayRef, + )]) + .unwrap(); + let batches = vec![BufferedBatch::Source(batch)]; + let rows = (0..4) + .map(|row_idx| MergeRow { + batch_idx: 0, + row_idx, + sequence_number: row_idx as i64, + value_kind: 0, + user_sequence: None, + }) + .collect::>(); + let (base, has_l0, selected) = lookup_rows(&rows, &batches).unwrap(); + assert_eq!(base.unwrap().row_idx, 3); + assert!(has_l0); + assert_eq!( + selected.iter().map(|row| row.row_idx).collect::>(), + vec![0, 1, 3] + ); + } + #[test] fn first_row_merge_keeps_earliest_sequence_and_first_tie() { let rows: Vec<_> = [30, 10, 10, 20] From 295dd8db572e22219e0fa439f84650a5552e320d Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 30 Oct 2026 14:49:16 +0800 Subject: [PATCH 4/4] fix(table): constrain replayable incremental merge states --- .../src/table/compact/primary_key/planner.rs | 29 ++-- .../tests/pk_compact/incremental_stateful.rs | 137 +++++++++++++++++- docs/src/compaction.md | 12 +- 3 files changed, 156 insertions(+), 22 deletions(-) diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index 3611ee84e..c89b53c7c 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -62,17 +62,16 @@ fn plain_state_mergeable(schema: &TableSchema) -> Result { match options.merge_engine()? { MergeEngine::Aggregation => { let config = AggregationConfig::new(schema.options()); - if config.remove_record_on_delete() { + // Global sequence fields use last_value, whose retract clears the + // sequence tuple. A materialized subset can then sort before older + // retained states and change the visible sequence fields. + if config.remove_record_on_delete() || !options.sequence_fields().is_empty() { return Ok(false); } - let sequence = options.sequence_fields(); Ok(schema .fields() .iter() - .filter(|field| { - !keys.iter().any(|key| key == field.name()) - && !sequence.iter().any(|name| name == &field.name()) - }) + .filter(|field| !keys.iter().any(|key| key == field.name())) .all(|field| { mergeable_field( schema, @@ -90,11 +89,21 @@ fn plain_state_mergeable(schema: &TableSchema) -> Result { } let config = PartialUpdateConfig::new(schema.options()); let functions = config.validated_aggregate_functions(schema.fields(), keys)?; - let ordering = config - .validated_sequence_groups(schema.fields(), keys)? - .into_iter() - .flat_map(|group| group.sequence_fields) + let groups = config.validated_sequence_groups(schema.fields(), keys)?; + let ordering = groups + .iter() + .flat_map(|group| group.sequence_fields.iter().cloned()) .collect::>(); + // Another group can overwrite an ordering field with NULL, making + // its aggregate state impossible to replay into retained history. + if groups.iter().any(|group| { + group + .protected_fields + .iter() + .any(|field| ordering.contains(field)) + }) { + return Ok(false); + } Ok(schema .fields() .iter() diff --git a/crates/paimon/tests/pk_compact/incremental_stateful.rs b/crates/paimon/tests/pk_compact/incremental_stateful.rs index 3323bf14e..7faf8db03 100644 --- a/crates/paimon/tests/pk_compact/incremental_stateful.rs +++ b/crates/paimon/tests/pk_compact/incremental_stateful.rs @@ -19,7 +19,13 @@ use super::lookup_changelog_tests::events; use super::*; use std::sync::Arc; -fn schema(engine: &str, producer: &str, function: &str, checked: bool) -> TableSchema { +fn schema( + engine: &str, + producer: &str, + function: &str, + checked: bool, + global_sequence: bool, +) -> TableSchema { let mut schema = Schema::builder() .column("id", DataType::Int(IntType::new())) .column("value", DataType::Int(IntType::new())) @@ -34,8 +40,10 @@ fn schema(engine: &str, producer: &str, function: &str, checked: bool) -> TableS if producer != "lookup" { schema = schema.option("ignore-delete", "true"); } - } else { + } else if global_sequence { schema = schema.option("sequence.field", "seq"); + } else { + schema = schema.option("fields.seq.aggregate-function", "max"); } if checked { schema = schema.option("fields.value.sum.fail-on-overflow", "true"); @@ -90,10 +98,122 @@ async fn active_files(table: &Table) -> Vec { .collect() } +async fn int_rows(table: &Table) -> Vec>> { + let mut rows = Vec::new(); + for batch in read(table, None).await { + let columns = batch + .columns() + .iter() + .map(|column| column.as_any().downcast_ref::().unwrap()) + .collect::>(); + rows.extend((0..batch.num_rows()).map(|row| { + columns + .iter() + .map(|column| (!column.is_null(row)).then(|| column.value(row))) + .collect() + })); + } + rows.sort(); + rows +} + +#[tokio::test] +async fn stateful_plain_global_sequence_retract_keeps_the_null_sequence() { + for producer in ["none", "input", "lookup"] { + for kind in [1, 3] { + let table = + table_with_schema(schema("aggregation", producer, "sum", false, true)).await; + write_batch(&table, &batch((0..1000).collect(), 20, 20)).await; + let anchor = compact(&table).await[0].files()[0].clone(); + write_batch(&table, &batch(vec![42], 30, 30)).await; + let retract = batch(vec![42], 10, 40); + let retract = RecordBatch::try_from_iter( + retract + .schema() + .fields() + .iter() + .zip(retract.columns()) + .map(|(field, column)| (field.name().as_str(), column.clone())) + .chain(std::iter::once(( + "_VALUE_KIND", + Arc::new(arrow_array::Int8Array::from(vec![kind])) as arrow_array::ArrayRef, + ))), + ) + .unwrap(); + write_batch(&table, &retract).await; + let expected = int_rows(&table).await; + assert_eq!(expected[42], vec![Some(42), Some(40), None]); + compact(&table).await; + let actual = int_rows(&table).await; + assert_eq!(actual[42], expected[42], "{producer}/{kind}"); + assert_eq!(actual, expected); + assert_eq!( + active_files(&table).await.contains(&anchor), + producer == "lookup" + ); + } + } +} + +#[tokio::test] +async fn stateful_plain_cross_group_ordering_keeps_all_history() { + for producer in ["none", "input"] { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("s1", DataType::Int(IntType::new())) + .column("s2", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "partial-update") + .option("changelog-producer", producer) + .option("ignore-delete", "true") + .option("fields.s1.sequence-group", "s2") + .option("fields.s2.sequence-group", "value") + .option("fields.value.aggregate-function", "sum") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |keys: Vec, value, s1, s2: Option| { + let len = keys.len(); + RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(keys)) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from(vec![value; len])) as arrow_array::ArrayRef, + ), + ( + "s1", + Arc::new(Int32Array::from(vec![s1; len])) as arrow_array::ArrayRef, + ), + ( + "s2", + Arc::new(Int32Array::from(vec![s2; len])) as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + write_batch(&table, &batch((0..1000).collect(), 20, 0, Some(1))).await; + let anchor = compact(&table).await[0].files()[0].clone(); + write_batch(&table, &batch(vec![42], 1, 1, Some(10))).await; + write_batch(&table, &batch(vec![42], 2, 2, None)).await; + let expected = int_rows(&table).await; + assert_eq!(expected[42], vec![Some(42), Some(21), Some(2), None]); + compact(&table).await; + let actual = int_rows(&table).await; + assert_eq!(actual[42], expected[42], "{producer}"); + assert_eq!(actual, expected); + assert!(!active_files(&table).await.contains(&anchor)); + } +} + #[tokio::test] async fn stateful_lookup_inserts_history_in_sequence_order_and_does_not_apply_it_twice() { for engine in ["partial-update", "aggregation"] { - let table = table_with_schema(schema(engine, "lookup", "first_value", false)).await; + let table = table_with_schema(schema(engine, "lookup", "first_value", false, true)).await; write_batch(&table, &batch((0..1000).collect(), 20, 20)).await; let anchor = compact(&table).await[0].files()[0].clone(); assert_eq!(anchor.level, 5); @@ -144,7 +264,7 @@ async fn stateful_plain_mergeable_fields_leave_history_out_of_the_output_state() for producer in ["none", "input"] { for (function, expected_value) in [("sum", 60), ("min", 10), ("max", 30)] { let table = table_with_cache( - schema(engine, producer, function, false), + schema(engine, producer, function, false, false), &[("local-cache.enabled", "false")], ) .await; @@ -189,7 +309,12 @@ async fn stateful_plain_order_sensitive_and_checked_sum_configs_keep_all_history ("first_value", false, 20, 10, 30, 10), ("sum", true, -i32::MAX, i32::MAX, 1, 1), ] { - let table = table_with_schema(schema(engine, producer, function, checked)).await; + // Exercise checked overflow independently of the global + // sequence fallback; first_value needs interleaved sequences. + let global_sequence = function == "first_value"; + let table = + table_with_schema(schema(engine, producer, function, checked, global_sequence)) + .await; write_batch(&table, &batch((0..1000).collect(), initial, 20)).await; let anchor = compact(&table).await[0].files()[0].clone(); write_batch(&table, &batch(vec![42], earlier, 10)).await; @@ -297,7 +422,7 @@ async fn stateful_plain_boolean_fields_are_mergeable_for_both_engines() { .option("fields.seq.sequence-group", "value") .option("ignore-delete", "true"); } else { - schema = schema.option("sequence.field", "seq"); + schema = schema.option("fields.seq.aggregate-function", "max"); } let table = table_with_cache( TableSchema::new(0, &schema.build().unwrap()), diff --git a/docs/src/compaction.md b/docs/src/compaction.md index fd4ae6ebb..87036989e 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -167,12 +167,12 @@ nearest positive-level state, so retained older levels are not aggregated again. With `none` or `input`, these engines use incremental selection only when every value field can be merged again independently of row order: `min`, `max`, `bool_and`, `bool_or`, or integer `sum` without checked overflow. Aggregation must -have `aggregation.remove-record-on-delete=false`; its sequence fields keep their -normal metadata role. Partial update must use `ignore-delete=true`, no global -sequence fields, and protect every value field with a sequence group and one of -these aggregate functions. Each group's ordering fields retain its winning -sequence. These modes merge only selected inputs, leaving retained history to -ordinary query merging, and do not require local-cache. +have `aggregation.remove-record-on-delete=false` and no global sequence fields. +Partial update must use `ignore-delete=true`, no global sequence fields, and +protect every value field with a sequence group and one of these aggregate +functions. Each group's ordering fields retain its winning sequence and must +not also be protected fields. These modes merge only selected inputs, leaving +retained history to ordinary query merging, and do not require local-cache. Other `none`/`input` configurations retain their previous selection: every active file in the selected bucket participates, and output uses the highest input level