From e4537706f98c237f5446ee87b3aa2fadef21b532 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 11:35:27 +0800 Subject: [PATCH 1/2] feat(table): add lookup changelog production for primary-key compaction --- crates/paimon/src/io/file_io.rs | 4 + crates/paimon/src/table/compact.rs | 102 +++- .../paimon/src/table/compact/primary_key.rs | 131 +++-- .../src/table/compact/primary_key/lookup.rs | 233 +++++++++ crates/paimon/src/table/incremental_scan.rs | 4 +- .../paimon/src/table/kv_data_file_writer.rs | 6 +- crates/paimon/src/table/kv_file_reader.rs | 47 +- crates/paimon/src/table/sort_merge.rs | 104 +++- crates/paimon/src/table/table_commit.rs | 63 ++- .../tests/pk_compact/lookup_changelog.rs | 464 ++++++++++++++++++ crates/paimon/tests/pk_compact_test.rs | 31 +- docs/src/compaction.md | 55 ++- docs/src/sql.md | 5 + 13 files changed, 1179 insertions(+), 70 deletions(-) create mode 100644 crates/paimon/src/table/compact/primary_key/lookup.rs create mode 100644 crates/paimon/tests/pk_compact/lookup_changelog.rs diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs index 79779c3a4..dec7b603a 100644 --- a/crates/paimon/src/io/file_io.rs +++ b/crates/paimon/src/io/file_io.rs @@ -584,6 +584,10 @@ impl FileIO { Ok(self.cache_namespace.to_string()) } + pub(crate) fn local_cache(&self) -> Option> { + self.cache.clone() + } + #[cfg(test)] pub(crate) fn has_local_cache(&self) -> bool { self.cache.is_some() diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 46f553064..e1c611688 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -111,6 +111,20 @@ impl<'a> CompactBuilder<'a> { /// Executor resources are local; input files and output settings come from each task. pub fn new_executor(&self) -> Result { ensure_supported(self.table)?; + if self + .table + .schema() + .core_options() + .try_changelog_producer()? + == ChangelogProducer::Lookup + && self.table.file_io().local_cache().is_none() + { + return Err(Error::ConfigInvalid { + message: + "Lookup compaction requires local-cache.enabled=true on the catalog/FileIO" + .into(), + }); + } Ok(CompactExecutor { table: self.table.clone(), resources: self.resources.clone(), @@ -274,6 +288,8 @@ pub struct CompactExecutor { pub struct CompactResult { task: CompactTask, files: Vec, + #[serde(default)] + changelog_files: Vec, } impl CompactResult { @@ -281,12 +297,27 @@ impl CompactResult { &self.files } + pub fn changelog_files(&self) -> &[DataFileMeta] { + &self.changelog_files + } + pub fn metrics(&self) -> CompactMetrics { CompactMetrics { files_removed: self.task.files.len(), files_added: self.files.len(), rows_written: self.files.iter().map(|file| file.row_count as u64).sum(), bytes_written: self.files.iter().map(|file| file.file_size as u64).sum(), + changelog_files_added: self.changelog_files.len(), + changelog_rows_written: self + .changelog_files + .iter() + .map(|file| file.row_count as u64) + .sum(), + changelog_bytes_written: self + .changelog_files + .iter() + .map(|file| file.file_size as u64) + .sum(), } } } @@ -298,6 +329,12 @@ pub struct CompactMetrics { pub files_added: usize, pub rows_written: u64, pub bytes_written: u64, + #[serde(default)] + pub changelog_files_added: usize, + #[serde(default)] + pub changelog_rows_written: u64, + #[serde(default)] + pub changelog_bytes_written: u64, } impl CompactExecutor { @@ -351,13 +388,16 @@ impl CompactExecutor { return Ok(CompactResult { task: task.clone(), files, + changelog_files: vec![], }); } if task.is_primary_key() { - let files = primary_key::execute(&table, task, &split, self.resources.clone()).await?; + let (files, changelog_files) = + primary_key::execute(&table, task, &split, self.resources.clone()).await?; return Ok(CompactResult { task: task.clone(), files, + changelog_files, }); } let mut read = table.new_read_builder().new_read()?.with_ordered(true); @@ -408,6 +448,7 @@ impl CompactExecutor { Ok(files) => Ok(CompactResult { task: task.clone(), files, + changelog_files: vec![], }), Err(error) => { writer.abort().await; @@ -513,6 +554,7 @@ impl CompactTask { /// Validated inputs for the existing atomic snapshot publication machinery. pub(super) struct PreparedCompact { pub entries: Vec, + pub changelog_entries: Vec, pub schema_id: i64, pub snapshot_id: i64, pub read_snapshot: Option>, @@ -529,6 +571,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul let mut inputs = HashSet::new(); let mut outputs = HashSet::new(); let mut entries = Vec::new(); + let mut changelog_entries = Vec::new(); let mut metrics = CompactMetrics::default(); for result in results { let task = &result.task; @@ -605,11 +648,55 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul "PK compaction must retain each key's merged state unless retracts are ignored", )); } + let lookup = options.try_changelog_producer()? == ChangelogProducer::Lookup; + if !lookup && !result.changelog_files.is_empty() { + return Err(invalid( + "Only lookup compaction may produce changelog files", + )); + } + let mut changelog_rows = 0u64; + for file in &result.changelog_files { + validate_file(file, true)?; + if file.schema_id != task.context.schema.id() + || file.file_source != Some(1) + || file.level != task.output_level() + || file + .delete_row_count + .is_none_or(|count| count > file.row_count) + || !outputs.insert(file.file_name.clone()) + || task.files.iter().any(|old| old.file_name == file.file_name) + { + return Err(invalid("Invalid compaction changelog file")); + } + changelog_rows += file.row_count as u64; + changelog_entries.push(ManifestEntry::new( + FileKind::Add, + task.partition.clone(), + task.bucket, + task.total_buckets, + file.clone(), + 2, + )); + } + let l0_rows = task + .files + .iter() + .filter(|file| file.level == 0) + .map(|file| file.row_count as u64) + .sum::(); + if changelog_rows > l0_rows.saturating_mul(2) { + return Err(invalid( + "Lookup changelogs must be generated by level-0 records", + )); + } let part = result.metrics(); metrics.files_removed += part.files_removed; metrics.files_added += part.files_added; metrics.rows_written += part.rows_written; metrics.bytes_written += part.bytes_written; + metrics.changelog_files_added += part.changelog_files_added; + metrics.changelog_rows_written += part.changelog_rows_written; + metrics.changelog_bytes_written += part.changelog_bytes_written; } ids.sort_unstable(); if ids.windows(2).any(|pair| pair[0] == pair[1]) { @@ -623,6 +710,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul } Ok(PreparedCompact { entries, + changelog_entries, schema_id: first.task.context.schema.id(), snapshot_id: first.task.snapshot_id(), read_snapshot: first @@ -692,15 +780,21 @@ pub(super) fn ensure_supported(table: &Table) -> Result<()> { && (options.bucket() == 0 || !matches!( options.try_changelog_producer()?, - ChangelogProducer::None | ChangelogProducer::Input + ChangelogProducer::None | ChangelogProducer::Input | ChangelogProducer::Lookup ) || (options.merge_engine()? == MergeEngine::FirstRow - && options.try_changelog_producer()? != ChangelogProducer::None)) + && !matches!( + options.try_changelog_producer()?, + ChangelogProducer::None | ChangelogProducer::Lookup + ))) { return Err(Error::Unsupported { - message: "PK compaction requires fixed or dynamic buckets and changelog-producer=none or input; first-row requires changelog-producer=none".into(), + message: "PK compaction requires fixed or dynamic buckets and changelog-producer=none, input or lookup; first-row requires changelog-producer=none or lookup".into(), }); } + if options.try_changelog_producer()? == ChangelogProducer::Lookup { + primary_key::lookup::validate_options(properties)?; + } options.validate_data_file_path_directory() } diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index 454c751fd..f03282640 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +pub(super) mod lookup; + use std::sync::Arc; use arrow_schema::SchemaRef; @@ -23,8 +25,8 @@ use futures::TryStreamExt; use super::{split_files_by_bucket, validate_file, CompactContext, CompactTask}; use crate::resource::ResourceContext; use crate::spec::{ - BigIntType, DataField, DataFileMeta, DataType, MergeEngine, RowKind, TinyIntType, - SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, + BigIntType, ChangelogProducer, DataField, DataFileMeta, DataType, MergeEngine, RowKind, + TinyIntType, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME, }; use crate::table::data_file_index_writer::FileIndexOptions; @@ -33,6 +35,7 @@ use crate::table::kv_data_file_writer::{KeyValueFileOptions, KeyValueRollingWrit use crate::table::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig}; use crate::table::sort_merge::{ compare_sequence_order, BufferedBatch, MergeFunction, MergeResult, MergeRow, + CHANGELOG_BATCH_METADATA_KEY, }; use crate::table::table_read::{ configured_mosaic_prefetch, configured_parquet_read_budget, MAX_MERGE_INPUT_STREAMS, @@ -40,7 +43,7 @@ use crate::table::table_read::{ use crate::table::{DataSplit, Plan, Table}; use crate::Result; -/// Select complete captured buckets that need an L0 upgrade or level merge. +/// Select files using the existing primary-key bucket compaction policy. pub(super) struct PrimaryKeyCompactPlanner; impl PrimaryKeyCompactPlanner { @@ -76,8 +79,8 @@ impl PrimaryKeyCompactPlanner { } /// Keep the same winner as a deduplicate query, including DELETE/UPDATE_BEFORE. -/// A later write may carry a lower user sequence, so even full compaction -/// must retain a winning retract rather than resurrect an older value. +/// A later write may carry a lower user sequence, so compaction must retain +/// a winning retract rather than resurrect an older value. struct CompactDeduplicate; impl MergeFunction for CompactDeduplicate { @@ -159,7 +162,7 @@ fn new_merge_function( if config.merge_engine == MergeEngine::Deduplicate { return Ok(Box::new(CompactDeduplicate)); } - let inner = KeyValueFileReader::new_merge_function(config, fields)?; + let inner = KeyValueFileReader::new_unwrapped_merge_function(config, fields)?; let aggregate_functions = match config.merge_engine { MergeEngine::PartialUpdate => crate::spec::PartialUpdateConfig::new(&config.table_options) .validated_aggregate_functions(&config.table_fields, &config.primary_keys)? @@ -213,16 +216,8 @@ fn new_merge_function( }) } -pub(super) async fn execute( - table: &Table, - task: &CompactTask, - split: &DataSplit, - resources: Option, -) -> Result> { - let schema = table.schema(); - let options = schema.core_options(); - let merge_engine = options.merge_engine()?; - let mut read_type = vec![ +fn read_type(table: &Table) -> Vec { + let mut fields = vec![ DataField::new( SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME.into(), @@ -234,7 +229,21 @@ pub(super) async fn execute( DataType::TinyInt(TinyIntType::new()), ), ]; - read_type.extend_from_slice(schema.fields()); + fields.extend_from_slice(table.schema().fields()); + fields +} + +pub(super) async fn execute( + table: &Table, + task: &CompactTask, + split: &DataSplit, + resources: Option, +) -> Result<(Vec, Vec)> { + let schema = table.schema(); + let options = schema.core_options(); + let merge_engine = options.merge_engine()?; + let lookup_changelog = options.try_changelog_producer()? == ChangelogProducer::Lookup; + let read_type = read_type(table); let budget = configured_parquet_read_budget(table)?; let budget = match &resources { Some(resources) => Arc::new(budget.with_resources(resources.clone()).without_prefetch()), @@ -266,7 +275,13 @@ pub(super) async fn execute( }, ); let mut batches = - reader.read_with_merge_function(std::slice::from_ref(split), new_merge_function)?; + reader.read_with_merge_function(std::slice::from_ref(split), move |config, fields| { + if lookup_changelog { + lookup::new_merge_function(config, fields) + } else { + new_merge_function(config, fields) + } + })?; let keys = schema.trimmed_primary_keys(); let primary_key_indices = keys .iter() @@ -301,39 +316,79 @@ pub(super) async fn execute( schema.options(), )?; let file_format = options.file_format(); + let file_options = KeyValueFileOptions { + file_io: table.file_io(), + paths: &paths, + table_options: schema.options(), + value_fields: schema.fields(), + primary_key_indices: &primary_key_indices, + primary_key_types: &primary_key_types, + schema_id: schema.id(), + file_format: &file_format, + file_prefix: options.data_file_prefix(), + file_compression: options.file_compression(), + file_compression_zstd_level: options.file_compression_zstd_level(), + write_buffer_size: options.write_buffer_size()? as i64, + level: task.output_level(), + file_source: 1, + is_changelog: false, + index_options: index.as_ref(), + managed_blob_enabled: false, + resources: resources.as_ref(), + }; let mut writer = KeyValueRollingWriter::new( + file_options, + options.target_file_size(), + options.target_file_row_num()?, + ); + let changelog_format = options.changelog_file_format(); + let mut changelog_writer = KeyValueRollingWriter::new( KeyValueFileOptions { - file_io: table.file_io(), - paths: &paths, - table_options: schema.options(), - value_fields: schema.fields(), - primary_key_indices: &primary_key_indices, - primary_key_types: &primary_key_types, - schema_id: schema.id(), - file_format: &file_format, - file_prefix: options.data_file_prefix(), - file_compression: options.file_compression(), - file_compression_zstd_level: options.file_compression_zstd_level(), - write_buffer_size: options.write_buffer_size()? as i64, - level: task.output_level(), - file_source: 1, - is_changelog: false, - index_options: index.as_ref(), - managed_blob_enabled: false, - resources: resources.as_ref(), + file_format: &changelog_format, + file_prefix: options.changelog_file_prefix(), + file_compression: options.changelog_file_compression(), + is_changelog: true, + index_options: None, + ..file_options }, options.target_file_size(), options.target_file_row_num()?, ); + let result = async { while let Some(batch) = batches.try_next().await? { - writer.write(&batch).await?; + if batch + .schema() + .metadata() + .contains_key(CHANGELOG_BATCH_METADATA_KEY) + { + let mut metadata = batch.schema().metadata().clone(); + metadata.remove(CHANGELOG_BATCH_METADATA_KEY); + let batch = arrow_array::RecordBatch::try_new( + Arc::new(batch.schema().as_ref().clone().with_metadata(metadata)), + batch.columns().to_vec(), + ) + .map_err(|error| { + super::invalid(&format!("Invalid changelog output batch: {error}")) + })?; + changelog_writer.write(&batch).await?; + } else { + writer.write(&batch).await?; + } } - writer.prepare_commit().await + // Keep both prepared file lists owned by their writers until both have + // closed successfully, so either failure can clean up the whole task. + writer.finish().await?; + changelog_writer.finish().await?; + Ok(( + writer.prepare_commit().await?, + changelog_writer.prepare_commit().await?, + )) } .await; if result.is_err() { writer.abort().await; + changelog_writer.abort().await; } result } diff --git a/crates/paimon/src/table/compact/primary_key/lookup.rs b/crates/paimon/src/table/compact/primary_key/lookup.rs new file mode 100644 index 000000000..0b21a150a --- /dev/null +++ b/crates/paimon/src/table/compact/primary_key/lookup.rs @@ -0,0 +1,233 @@ +// 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. + +//! Per-key lookup changelog generation, matching Java's lookup merge wrapper. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +use arrow_array::{Int8Array, RecordBatch, UInt64Array}; +use arrow_schema::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}; +use crate::{Error, Result}; + +struct ChangelogOptions { + deduplicate: bool, + ignore_update_before: bool, + ignore_delete: bool, + ignored_fields: HashSet, +} + +impl ChangelogOptions { + fn parse(options: &HashMap) -> Result { + let boolean = |key: &str| match options.get(key).map(String::as_str) { + None => Ok(false), + Some(value) if value.eq_ignore_ascii_case("true") => Ok(true), + Some(value) if value.eq_ignore_ascii_case("false") => Ok(false), + Some(value) => Err(Error::ConfigInvalid { + message: format!("{key} must be true or false, got {value}"), + }), + }; + Ok(Self { + deduplicate: boolean("changelog-producer.row-deduplicate")?, + ignore_update_before: boolean("changelog-producer.ignore-update-before")?, + ignore_delete: boolean("changelog-producer.ignore-delete")?, + ignored_fields: options + .get("changelog-producer.row-deduplicate-ignore-fields") + .map(|fields| { + fields + .split(',') + .map(str::trim) + .filter(|field| !field.is_empty()) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + }) + } +} + +pub(in crate::table::compact) fn validate_options(options: &HashMap) -> Result<()> { + ChangelogOptions::parse(options).map(|_| ()) +} + +struct LookupChangelogMergeFunction { + inner: Box, + first_row: bool, + options: ChangelogOptions, + value_columns: Vec, +} + +pub(super) fn new_merge_function( + config: &KeyValueReadConfig, + fields: &[DataField], +) -> Result> { + let options = ChangelogOptions::parse(&config.table_options)?; + Ok(Box::new(LookupChangelogMergeFunction { + inner: super::new_merge_function(config, fields)?, + first_row: config.merge_engine == MergeEngine::FirstRow, + value_columns: fields + .iter() + .enumerate() + .filter(|(_, field)| { + config + .table_fields + .iter() + .any(|value| value.id() == field.id()) + && !options.ignored_fields.contains(field.name()) + }) + .map(|(index, _)| index) + .collect(), + options, + })) +} + +fn source_row( + batch_idx: usize, + row_idx: usize, + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, +) -> Result { + let indices = UInt64Array::from(vec![row_idx as u64]); + let values = (0..schema.fields().len()) + .map(|index| { + arrow_select::take::take( + batches[batch_idx].column_for_output(index, columns), + &indices, + None, + ) + .map_err(|error| invalid(format!("Failed to materialize lookup value: {error}"))) + }) + .collect::>>()?; + RecordBatch::try_new(schema.clone(), values) + .map_err(|error| invalid(format!("Invalid lookup value: {error}"))) +} + +fn result_row( + result: &MergeResult, + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, +) -> Result> { + match result { + MergeResult::SourceRow { batch_idx, row_idx } => { + source_row(*batch_idx, *row_idx, batches, columns, schema).map(Some) + } + MergeResult::MaterializedRow(batch) | MergeResult::MaterializedDeleteRow(batch) => { + Ok(Some(batch.clone())) + } + MergeResult::Omit => Ok(None), + } +} + +fn live_row(batch: Option) -> Result> { + match batch { + Some(batch) => { + let kinds = batch + .column_by_name(VALUE_KIND_FIELD_NAME) + .and_then(|column| column.as_any().downcast_ref::()) + .ok_or_else(|| invalid("Lookup value must carry its row kind"))?; + Ok(RowKind::from_value(kinds.value(0))? + .is_add() + .then_some(batch)) + } + None => Ok(None), + } +} + +fn with_kind(batch: &RecordBatch, kind: RowKind) -> Result { + let mut columns = batch.columns().to_vec(); + let index = batch + .schema() + .index_of(VALUE_KIND_FIELD_NAME) + .map_err(|error| invalid(error.to_string()))?; + columns[index] = Arc::new(Int8Array::from(vec![kind as i8])); + RecordBatch::try_new(batch.schema(), columns).map_err(|error| invalid(error.to_string())) +} + +impl MergeFunction for LookupChangelogMergeFunction { + fn merge( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> Result { + self.merge_with_changelog(rows, batches, columns, schema) + .map(|(result, _)| result) + } + + fn merge_with_changelog( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> Result<(MergeResult, Option)> { + let (base, has_l0, selected) = lookup_rows(rows, batches)?; + let result = self.inner.merge(&selected, batches, columns, schema)?; + if !has_l0 { + return Ok((result, None)); + } + let before = live_row( + base.as_ref() + .map(|row| source_row(row.batch_idx, row.row_idx, batches, columns, schema)) + .transpose()?, + )?; + let after = live_row(result_row(&result, batches, columns, schema)?)?; + let mut changes = Vec::with_capacity(2); + match (&before, &after) { + (None, Some(after)) => changes.push(with_kind(after, RowKind::Insert)?), + (Some(before), None) if !self.first_row && !self.options.ignore_delete => { + changes.push(with_kind(before, RowKind::Delete)?) + } + (Some(before), Some(after)) if !self.first_row => { + let equal = self.options.deduplicate + && self.value_columns.iter().all(|index| { + before.column(*index).to_data() == after.column(*index).to_data() + }); + if !equal { + if !self.options.ignore_update_before { + changes.push(with_kind(before, RowKind::UpdateBefore)?); + } + changes.push(with_kind(after, RowKind::UpdateAfter)?); + } + } + _ => {} + } + let changelog = if changes.is_empty() { + None + } else { + Some( + arrow_select::concat::concat_batches(schema, &changes) + .map_err(|error| invalid(error.to_string()))?, + ) + }; + Ok((result, changelog)) + } +} + +fn invalid(message: impl Into) -> Error { + Error::DataInvalid { + message: message.into(), + source: None, + } +} diff --git a/crates/paimon/src/table/incremental_scan.rs b/crates/paimon/src/table/incremental_scan.rs index d06b4c03b..48aa38002 100644 --- a/crates/paimon/src/table/incremental_scan.rs +++ b/crates/paimon/src/table/incremental_scan.rs @@ -63,8 +63,8 @@ pub enum IncrementalScanMode { /// Read existing changelog manifest files in the range. /// /// Skips [`OVERWRITE`](crate::spec::CommitKind::OVERWRITE) snapshots and - /// snapshots without a `changelog_manifest_list`. Does not generate - /// changelogs (no compact/lookup producer path). + /// snapshots without a `changelog_manifest_list`. Reads persisted input or + /// compaction-generated changelogs; it does not generate events during reads. Changelog, /// Resolve to [`Delta`](Self::Delta) when `changelog-producer=none`, /// otherwise to [`Changelog`](Self::Changelog). diff --git a/crates/paimon/src/table/kv_data_file_writer.rs b/crates/paimon/src/table/kv_data_file_writer.rs index 0943e3a7a..b96ebed5f 100644 --- a/crates/paimon/src/table/kv_data_file_writer.rs +++ b/crates/paimon/src/table/kv_data_file_writer.rs @@ -507,8 +507,12 @@ impl<'a> KeyValueRollingWriter<'a> { Ok(()) } + pub(super) async fn finish(&mut self) -> Result<()> { + self.close_current().await + } + pub(super) async fn prepare_commit(&mut self) -> Result> { - self.close_current().await?; + self.finish().await?; Ok(std::mem::take(&mut self.files)) } diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index e97d3a1bf..eae33f709 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -27,8 +27,9 @@ use super::data_file_reader::{file_index_selection_to_local_ranges, DataFileReader}; use super::sort_merge::{ - AggregateMergeFunction, DeduplicateMergeFunction, FirstRowMergeFunction, MergeFunction, - PartialUpdateMergeFunction, SortMergeReaderBuilder, + AggregateMergeFunction, DeduplicateMergeFunction, FirstRowMergeFunction, LookupMergeFunction, + MergeFunction, PartialUpdateMergeFunction, SortMergeReaderBuilder, + CHANGELOG_BATCH_METADATA_KEY, FILE_LEVEL_METADATA_KEY, }; use crate::arrow::format::MosaicPrefetchOptions; use crate::arrow::shredding::variant::assemble_variant_extraction_array; @@ -62,6 +63,7 @@ pub(crate) struct KeyValueFileReader { /// of a key survives); they are enforced by the post-merge residual /// filter using the full `config.predicates` instead. pushdown_predicates: Vec, + include_file_levels: bool, #[cfg(test)] input_batch_sizes: Option>>>, } @@ -350,9 +352,14 @@ 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")); Self { file_io, config, + include_file_levels, pushdown_predicates, #[cfg(test)] input_batch_sizes: None, @@ -371,6 +378,25 @@ impl KeyValueFileReader { pub(super) fn new_merge_function( config: &KeyValueReadConfig, merge_output_fields: &[DataField], + ) -> crate::Result> { + let inner = Self::new_unwrapped_merge_function(config, merge_output_fields)?; + Ok( + if config.merge_engine != MergeEngine::FirstRow + && config + .table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")) + { + Box::new(LookupMergeFunction(inner)) + } else { + inner + }, + ) + } + + pub(super) fn new_unwrapped_merge_function( + config: &KeyValueReadConfig, + merge_output_fields: &[DataField], ) -> crate::Result> { match config.merge_engine { MergeEngine::Deduplicate => Ok(Box::new(DeduplicateMergeFunction)), @@ -651,6 +677,7 @@ impl KeyValueFileReader { data_splits.into_iter().map(|split| vec![split]).collect() }; let file_io = self.file_io; + let include_file_levels = self.include_file_levels; let config = self.config; let table_schema_id = config.table_schema_id; let pushdown_predicates = self.pushdown_predicates; @@ -786,6 +813,7 @@ impl KeyValueFileReader { )), None => None, }; + let file_level = file_meta.level; let mut file_stream = reader.read_single_file_stream_with_schema_and_local_ranges( split.as_ref(), file_meta, @@ -795,7 +823,14 @@ impl KeyValueFileReader { local_row_ranges, )?; while let Some(batch) = file_stream.next().await { - yield batch?; + 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)) })?; + } else { yield batch; } } } }); @@ -892,7 +927,11 @@ impl KeyValueFileReader { let options = RecordBatchOptions::new().with_row_count(Some(batch.num_rows())); let reordered = RecordBatch::try_new_with_options( - output_schema.clone(), + if batch.schema().metadata().contains_key(CHANGELOG_BATCH_METADATA_KEY) { + let mut metadata = output_schema.metadata().clone(); + metadata.insert(CHANGELOG_BATCH_METADATA_KEY.into(), "true".into()); + Arc::new(output_schema.as_ref().clone().with_metadata(metadata)) + } else { output_schema.clone() }, columns, &options, ) diff --git a/crates/paimon/src/table/sort_merge.rs b/crates/paimon/src/table/sort_merge.rs index b0cdec9fa..bddcc33fd 100644 --- a/crates/paimon/src/table/sort_merge.rs +++ b/crates/paimon/src/table/sort_merge.rs @@ -40,7 +40,7 @@ use futures::StreamExt; use std::cmp::Ordering; use std::collections::HashMap; use std::collections::HashSet; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; // --------------------------------------------------------------------------- // MergeFunction @@ -57,7 +57,7 @@ pub(crate) enum BufferedBatch { } impl BufferedBatch { - fn column_for_output<'a>( + pub(super) fn column_for_output<'a>( &'a self, output_col_idx: usize, source_output_col_indices: &[usize], @@ -72,6 +72,7 @@ impl BufferedBatch { } /// A row reference as an index into the batch buffer. +#[derive(Clone)] pub(crate) struct MergeRow { /// Index into the shared batch buffer. pub batch_idx: usize, @@ -150,6 +151,81 @@ pub(crate) trait MergeFunction: Send + Sync { source_output_col_indices: &[usize], output_schema: &SchemaRef, ) -> crate::Result; + + /// Emit at most one key's changelog alongside its merged state. The reader + /// drains it before advancing to the next key, keeping memory bounded even + /// when all data rows are deleted. + fn merge_with_changelog( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> crate::Result<(MergeResult, Option)> { + Ok((self.merge(rows, batches, columns, schema)?, None)) + } +} + +/// 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"; +pub(super) const CHANGELOG_BATCH_METADATA_KEY: &str = "paimon.compact.changelog"; + +/// Lookup states at positive levels are already materialized. The lowest +/// positive level is the latest state; older levels must not be reduced again. +/// Mirrors Java LookupMergeFunction#pickHighLevel/getResult. +pub(super) fn lookup_rows( + rows: &[MergeRow], + batches: &[BufferedBatch], +) -> crate::Result<(Option, bool, Vec)> { + let mut base = None; + 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, + })?; + if level > 0 && (base.is_none() || level < base_level) { + base = Some(index); + base_level = level; + } + levels.push(level); + } + let has_l0 = levels.contains(&0); + let selected = rows + .iter() + .enumerate() + .filter(|(index, _)| levels[*index] <= 0 || base == Some(*index)) + .map(|(_, row)| row.clone()) + .collect(); + Ok((base.map(|index| rows[index].clone()), has_l0, selected)) +} + +pub(super) struct LookupMergeFunction(pub Box); + +impl MergeFunction for LookupMergeFunction { + fn merge( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> crate::Result { + let (_, _, selected) = lookup_rows(rows, batches)?; + self.0.merge(&selected, batches, columns, schema) + } } /// Deduplicate merge: keeps the row with the highest sequence. @@ -1518,12 +1594,10 @@ fn sort_merge_stream( tree.update(|a, b| compare_cursors(&cursors, a, b).then_with(|| a.cmp(&b)).is_gt()); } - match merge_function.merge( - &same_key_rows, - &batch_buffer, - &source_output_col_indices, - &output_schema, - )? { + let (merged, changelog) = merge_function.merge_with_changelog( + &same_key_rows, &batch_buffer, &source_output_col_indices, &output_schema, + )?; + match merged { MergeResult::SourceRow { batch_idx, row_idx } => { output_indices.push((batch_idx, row_idx)); } @@ -1550,6 +1624,20 @@ fn sort_merge_stream( MergeResult::Omit | MergeResult::MaterializedDeleteRow(_) => {} } + if let Some(changelog) = changelog { + if changelog.schema().as_ref() != output_schema.as_ref() { + Err(Error::UnexpectedError { message: "Changelog merge result schema does not match merge output schema".into(), source: None })?; + } + let mut metadata = output_schema.metadata().clone(); + metadata.insert(CHANGELOG_BATCH_METADATA_KEY.into(), "true".into()); + yield RecordBatch::try_new( + Arc::new(output_schema.as_ref().clone().with_metadata(metadata)), changelog.columns().to_vec(), + ).map_err(|error| Error::DataInvalid { message: format!("Invalid changelog batch: {error}"), source: Some(Box::new(error)) })?; + } + if output_indices.is_empty() { + compact_batch_buffer(&mut batch_buffer, &mut stream_batch_idx, &cursors); + } + // Yield a batch when we've accumulated enough rows. if output_indices.len() >= batch_size { let batch = build_output_interleave( diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 6769098b5..9d6c41874 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -442,6 +442,7 @@ impl TableCommit { .try_commit( CommitEntriesPlan::Compact { entries: prepared.entries, + changelog_entries: prepared.changelog_entries, schema_id: prepared.schema_id, read_snapshot: prepared.read_snapshot, planned_snapshot_id: prepared.snapshot_id, @@ -1252,8 +1253,14 @@ impl TableCommit { new_index_entries, .. } => (new_entries.as_slice(), &[][..], new_index_entries), - CommitEntriesPlan::Compact { entries, .. } => { - return self.check_recovery_entries(entries, &[], &[]).await; + CommitEntriesPlan::Compact { + entries, + changelog_entries, + .. + } => { + return self + .check_recovery_entries(entries, changelog_entries, &[]) + .await; } }; self.check_recovery_entries(entries, changelog_entries, index_entries) @@ -1900,6 +1907,7 @@ impl TableCommit { match plan { CommitEntriesPlan::Compact { entries, + changelog_entries, schema_id, read_snapshot, planned_snapshot_id, @@ -1928,6 +1936,7 @@ impl TableCommit { None => None, }; validate_file_entries(entries.iter())?; + validate_file_entries(changelog_entries.iter())?; // This runs after duplicate-commit filtering: an uncertain commit // can be retried even after the captured read snapshot expires. if let Some(captured) = read_snapshot { @@ -2042,9 +2051,54 @@ impl TableCommit { base_data_files.as_deref().unwrap_or_default(), entries, )?; + if self + .table + .schema() + .core_options() + .try_changelog_producer()? + == crate::spec::ChangelogProducer::Lookup + { + let buckets = entries + .iter() + .filter(|entry| *entry.kind() == FileKind::Delete) + .map(|entry| (entry.partition(), entry.bucket())) + .collect::>(); + for (partition, bucket) in buckets { + let expected = entries + .iter() + .filter(|entry| { + *entry.kind() == FileKind::Delete + && entry.partition() == partition + && entry.bucket() == bucket + && entry.file().level > 0 + }) + .collect::>(); + let current = base_data_files + .as_deref() + .unwrap_or_default() + .iter() + .filter(|entry| { + entry.partition() == partition + && entry.bucket() == bucket + && entry.file().level > 0 + }) + .collect::>(); + if expected.len() != current.len() + || expected.iter().any(|file| { + !current.iter().any(|entry| entry.file() == file.file()) + }) + { + return Err(crate::Error::DataInvalid { + message: "Lookup compaction history changed; create a new plan" + .into(), + source: None, + }); + } + } + } Ok(ResolvedCommit { entries: entries.clone(), - changelog_entries: vec![], + changelog_entries: changelog_entries.clone(), kind: CommitKind::COMPACT, index_manifest_name, index_manifest_changed: false, @@ -3868,9 +3922,10 @@ fn build_partition_stats_row(datums: &[Option], data_types: &[DataType]) /// Plan for resolving commit entries. enum CommitEntriesPlan { - /// Replace unchanged data files without publishing logical row changes. + /// Replace unchanged data files and atomically publish generated changelogs. Compact { entries: Vec, + changelog_entries: Vec, schema_id: i64, read_snapshot: Option>, planned_snapshot_id: i64, diff --git a/crates/paimon/tests/pk_compact/lookup_changelog.rs b/crates/paimon/tests/pk_compact/lookup_changelog.rs new file mode 100644 index 000000000..e43f00aba --- /dev/null +++ b/crates/paimon/tests/pk_compact/lookup_changelog.rs @@ -0,0 +1,464 @@ +// 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 lookup_table(options: &[(&str, &str)]) -> Table { + let mut configured = vec![("changelog-producer", "lookup")]; + configured.extend_from_slice(options); + table(&configured).await +} + +async fn compact(table: &Table) -> i64 { + 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(); + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap() +} + +pub(super) async fn events(table: &Table, start: i64, end: i64) -> Vec<(String, i32, i32)> { + let plan = table + .new_read_builder() + .new_incremental_scan(IncrementalScanMode::Auto, start, end) + .plan_combined() + .await + .unwrap(); + let batches: Vec = table + .new_read_builder() + .new_read() + .unwrap() + .to_arrow_with_row_kind(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + let mut events = Vec::new(); + for batch in batches { + let kinds = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let ids = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(2) + .as_any() + .downcast_ref::() + .unwrap(); + events.extend((0..batch.num_rows()).map(|row| { + ( + kinds.value(row).to_string(), + ids.value(row), + values.value(row), + ) + })); + } + events.sort(); + events +} + +fn expected(events: &[(&str, i32, i32)]) -> Vec<(String, i32, i32)> { + let mut events = events + .iter() + .map(|(kind, key, value)| (kind.to_string(), *key, *value)) + .collect::>(); + events.sort(); + events +} + +#[tokio::test] +async fn lookup_compaction_publishes_before_after_images_for_fixed_and_dynamic_tables() { + for dv in ["false"] { + for bucket in ["1", "-1"] { + let table = lookup_table(&[ + ("deletion-vectors.enabled", dv), + ("bucket", bucket), + ("read.batch-size", "1"), + ]) + .await; + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + assert!(events(&table, 0, 1).await.is_empty()); + let first = compact(&table).await; + assert_eq!( + events(&table, 0, first).await, + expected(&[("+I", 1, 10), ("+I", 2, 20)]) + ); + write_batch( + &table, + &make_batch_with_kinds(vec![1, 2, 3], vec![11, 999, 30], vec![0, 3, 0]), + ) + .await; + let second = compact(&table).await; + assert_eq!( + events(&table, first, second).await, + expected(&[("-U", 1, 10), ("+U", 1, 11), ("-D", 2, 20), ("+I", 3, 30)]) + ); + assert_eq!(rows(&table).await, vec![(1, 11), (3, 30)]); + assert_eq!( + compact(&table).await, + second, + "an already compacted bucket emits nothing" + ); + assert_eq!(events(&table, first, second).await.len(), 4); + } + } +} + +#[tokio::test] +async fn lookup_compaction_respects_java_changelog_flags_and_default_duplicate_updates() { + for (options, expected_events) in [ + ( + vec![], + expected(&[("-U", 1, 10), ("+U", 1, 10), ("-D", 2, 20)]), + ), + ( + vec![("changelog-producer.row-deduplicate", "true")], + expected(&[("-D", 2, 20)]), + ), + ( + vec![("changelog-producer.ignore-update-before", "true")], + expected(&[("+U", 1, 10), ("-D", 2, 20)]), + ), + ( + vec![("changelog-producer.ignore-delete", "true")], + expected(&[("-U", 1, 10), ("+U", 1, 10)]), + ), + ( + vec![ + ("changelog-producer.row-deduplicate", "true"), + ("changelog-producer.row-deduplicate-ignore-fields", "value"), + ("changelog-producer.ignore-delete", "true"), + ], + vec![], + ), + ] { + let table = lookup_table(&options).await; + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + let first = compact(&table).await; + let updated_value = if options + .iter() + .any(|(key, _)| *key == "changelog-producer.row-deduplicate-ignore-fields") + { + 11 + } else { + 10 + }; + write_batch( + &table, + &make_batch_with_kinds(vec![1, 2], vec![updated_value, 0], vec![0, 3]), + ) + .await; + let second = compact(&table).await; + assert_eq!( + events(&table, first, second).await, + expected_events, + "{options:?}" + ); + assert_eq!(rows(&table).await, vec![(1, updated_value)]); + } +} + +#[tokio::test] +async fn lookup_compaction_first_row_only_emits_new_keys() { + let table = lookup_table(&[("merge-engine", "first-row")]).await; + write_batch(&table, &make_batch(vec![1], vec![10])).await; + let first = compact(&table).await; + assert_eq!(events(&table, 0, first).await, expected(&[("+I", 1, 10)])); + write_batch(&table, &make_batch(vec![1, 2], vec![99, 20])).await; + let second = compact(&table).await; + assert_eq!( + events(&table, first, second).await, + expected(&[("+I", 2, 20)]) + ); + assert_eq!(rows(&table).await, vec![(1, 10), (2, 20)]); +} + +#[tokio::test] +async fn lookup_compaction_materializes_aggregation_history_once_and_ignores_high_level_only_keys() +{ + for dv in ["false"] { + let table = lookup_table(&[ + ("deletion-vectors.enabled", dv), + ("merge-engine", "aggregation"), + ("fields.value.aggregate-function", "sum"), + ]) + .await; + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + let first = compact(&table).await; + write_batch(&table, &make_batch(vec![1, 3], vec![5, 30])).await; + let second = compact(&table).await; + assert_eq!( + events(&table, first, second).await, + expected(&[("-U", 1, 10), ("+U", 1, 15), ("+I", 3, 30)]) + ); + write_batch(&table, &make_batch(vec![1], vec![7])).await; + let third = compact(&table).await; + assert_eq!( + events(&table, second, third).await, + expected(&[("-U", 1, 15), ("+U", 1, 22)]) + ); + assert_eq!(rows(&table).await, vec![(1, 22), (2, 20), (3, 30)]); + } +} + +async fn append_at_level(table: &Table, value: i32, level: i32) { + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + writer + .write_arrow_batch(&make_batch(vec![1], vec![value])) + .await + .unwrap(); + let mut messages = writer.prepare_commit().await.unwrap(); + for file in messages + .iter_mut() + .flat_map(|message| &mut message.new_files) + { + file.level = level; + } + builder.new_commit().commit(messages).await.unwrap(); + writer.close().await; +} + +#[tokio::test] +async fn lookup_compaction_subsets_and_retries_publish_each_changelog_once_with_concurrent_l0() { + for dv in ["false"] { + let table = lookup_table(&[ + ("bucket", "-1"), + ("dynamic-bucket.target-row-num", "1"), + ("deletion-vectors.enabled", dv), + ]) + .await; + write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await; + let first = compact(&table).await; + write_batch(&table, &make_batch(vec![1, 2], vec![11, 21])).await; + let builder = table + .new_compact_builder() + .with_commit_user("lookup-retry") + .unwrap(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks().len(), 2); + let results = execute(&table, &plan).await; + assert_eq!( + results + .iter() + .map(|result| result.metrics().changelog_rows_written) + .sum::(), + 4 + ); + write_batch(&table, &make_batch(vec![1], vec![12])).await; + let commit = builder.new_commit().unwrap(); + for result in results { + let metrics = commit.commit_compact(vec![result.clone()]).await.unwrap(); + let snapshot = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + assert_eq!(commit.commit_compact(vec![result]).await.unwrap(), metrics); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + snapshot + ); + } + let second = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + assert_eq!( + events(&table, first, second).await, + expected(&[("-U", 1, 10), ("+U", 1, 11), ("-U", 2, 20), ("+U", 2, 21)]) + ); + let third = compact(&table).await; + assert_eq!( + events(&table, second, third).await, + expected(&[("-U", 1, 11), ("+U", 1, 12)]) + ); + assert_eq!(rows(&table).await, vec![(1, 12), (2, 21)]); + } +} + +#[tokio::test] +async fn lookup_compaction_rejects_concurrent_positive_level_history_at_another_level() { + let table = lookup_table(&[]).await; + write_batch(&table, &make_batch(vec![1], vec![10])).await; + compact(&table).await; + write_batch(&table, &make_batch(vec![1], vec![11])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(&table, &plan).await; + append_at_level(&table, 12, 3).await; + let snapshot = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + assert!(builder + .new_commit() + .unwrap() + .commit_compact(results) + .await + .unwrap_err() + .to_string() + .contains("history changed")); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + snapshot + ); +} + +#[tokio::test] +async fn lookup_compaction_aborts_data_and_changelog_files_on_a_late_read_failure() { + let table = lookup_table(&[ + ("read.batch-size", "1"), + ("write-buffer-size", "1b"), + ("target-file-row-num", "1"), + ]) + .await; + write_batch(&table, &make_batch(vec![1], vec![10])).await; + write_batch(&table, &make_batch(vec![2], vec![20])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let scan = table.new_read_builder().new_scan().plan().await.unwrap(); + let second = plan.tasks()[0] + .files() + .iter() + .find(|file| file.max_sequence_number == 1) + .unwrap(); + table + .file_io() + .delete_file(&format!( + "{}/{}", + scan.splits()[0].bucket_path(), + second.file_name + )) + .await + .unwrap(); + let paths = || async { + table + .file_io() + .list_status_recursive(table.location()) + .await + .unwrap() + .into_iter() + .map(|file| file.path) + .collect::>() + }; + let before = paths().await; + assert!(builder + .new_executor() + .unwrap() + .execute(&plan.tasks()[0]) + .await + .is_err()); + assert_eq!(paths().await, before); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + Some(2) + ); +} + +#[tokio::test] +async fn lookup_compaction_validates_changelog_flags_before_execution() { + for key in [ + "changelog-producer.row-deduplicate", + "changelog-producer.ignore-update-before", + "changelog-producer.ignore-delete", + ] { + let table = lookup_table(&[(key, "invalid")]).await; + assert!(table + .new_compact_builder() + .new_planner() + .err() + .unwrap() + .to_string() + .contains(key)); + } +} + +#[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")); +} + +#[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)]); +} diff --git a/crates/paimon/tests/pk_compact_test.rs b/crates/paimon/tests/pk_compact_test.rs index 64541bcca..a79323a9f 100644 --- a/crates/paimon/tests/pk_compact_test.rs +++ b/crates/paimon/tests/pk_compact_test.rs @@ -20,13 +20,16 @@ mod common; #[path = "pk_compact/merge_engines.rs"] mod engine_tests; +#[path = "pk_compact/lookup_changelog.rs"] +mod lookup_changelog_tests; + #[path = "pk_compact/dynamic_bucket.rs"] mod dynamic_tests; use arrow_array::{Array, Int32Array, Int64Array, RecordBatch, StringArray}; use common::incremental_helpers::{ - make_batch, make_batch_with_kinds, make_partitioned_batch, memory_table, persist_table_schema, - pk_schema, setup_dirs, write_batch, + make_batch, make_batch_with_kinds, make_partitioned_batch, persist_table_schema, pk_schema, + setup_dirs, write_batch, }; use futures::TryStreamExt; use paimon::spec::{ @@ -35,8 +38,26 @@ use paimon::spec::{ use paimon::table::{CompactPlan, CompactResult, CompactTask, IncrementalScanMode, Table}; async fn table_with_schema(schema: TableSchema) -> Table { + table_with_cache(schema, &[]).await +} + +async fn table_with_cache(schema: TableSchema, cache_options: &[(&str, &str)]) -> Table { let path = format!("memory:/pk_compact/{}", uuid::Uuid::new_v4()); - let (io, table) = memory_table(&path, schema); + let mut options = paimon::Options::new(); + options.set("warehouse", &path); + options.set("local-cache.enabled", "true"); + for (name, value) in cache_options { + options.set(*name, *value); + } + let catalog = paimon::catalog::FileSystemCatalog::new(options).unwrap(); + let io = catalog.file_io().clone(); + let table = Table::new( + io.clone(), + paimon::catalog::Identifier::new("default", "pk_compact"), + path.clone(), + schema, + None, + ); setup_dirs(&io, &path).await; persist_table_schema(&io, &path, table.schema()).await; table @@ -532,10 +553,6 @@ async fn pk_compaction_rolls_ordered_files_and_rebuilds_indexes_with_level_stats #[tokio::test] async fn pk_compaction_refuses_partial_buckets_and_unsupported_layouts() { for options in [ - vec![ - ("merge-engine", "first-row"), - ("changelog-producer", "lookup"), - ], vec![("changelog-producer", "full-compaction")], vec![ ("merge-engine", "first-row"), diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 9e76a5a31..890c08659 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -114,7 +114,7 @@ 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` and `input` are supported; `first-row` requires `none`. +`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 @@ -175,6 +175,56 @@ publishing a snapshot. The captured snapshot must remain available until each task subset is successfully committed; retries of an already committed subset remain idempotent after its read snapshot expires. +## Lookup changelog producer + +Set `changelog-producer=lookup` and run the same compaction API to publish data +and changelog files atomically in a `COMPACT` snapshot. Ordinary writes produce +L0 data files; changelogs become available when those files are compacted. +Batch incremental `Changelog` / `Auto` scans and streaming follow-up scans read +the resulting changelog manifest. `Delta` scans continue to skip compaction. + +For each key, the lowest positive level is its latest materialized history. +Older positive levels already contributed to that state and are not reduced +again. Compaction merges this history with the captured L0 records and emits +changes only for keys with L0 input: + +- No live history and a live result: `INSERT`. +- Live history and a deleted result: `DELETE` carrying the history's values. +- Live history and a live result: `UPDATE_BEFORE` followed by `UPDATE_AFTER`. +- No live history and no live result: no changelog. + +Before images retain the historical sequence metadata; after images carry the +merged result's metadata. `first-row` emits only inserts for previously unseen +keys. `partial-update` and `aggregation` use their existing reducers and state +preservation restrictions. History selection precedes merging. + +Java-compatible options have these defaults: + +| Option | Default | Behavior | +| --- | --- | --- | +| `changelog-producer.row-deduplicate` | `false` | When true, skip update pairs whose user values are equal. | +| `changelog-producer.row-deduplicate-ignore-fields` | unset | Comma-separated fields excluded from that comparison. | +| `changelog-producer.ignore-update-before` | `false` | Omit `UPDATE_BEFORE` events. | +| `changelog-producer.ignore-delete` | `false` | Omit `DELETE` events. | + +With default options, even an unchanged value produces an update pair when its +key has L0 input. Ignoring changelog events does not change compacted table data. +`changelog-file.format`, `changelog-file.compression`, `changelog-file.prefix` +and changelog statistics options apply independently of data-file settings. +Changelog files roll at the ordinary size and row limits. `CompactResult` exposes +`changelog_files()`; `CompactMetrics` reports their file, row and byte counts in +`changelog_files_added`, `changelog_rows_written` and `changelog_bytes_written`. +The existing data metrics continue to count data files and rows only. + +The executor streams both outputs and aborts both writers on execution failure. +Prepared results include both file lists for serialization, subset commits and +idempotent retries. A concurrent positive-level addition in a selected bucket +rejects the commit because it may change the changelog's history. Concurrent L0 +appends remain eligible for a later compaction and produce no premature events. + +The executor requires an actual catalog/FileIO cache (`local-cache.enabled=true`); +setting the flag in table properties alone does not attach a cache. + ## Dynamic buckets Dynamic bucket compaction (`bucket=-1`) rewrites each existing partition/bucket @@ -241,7 +291,8 @@ deletions into reassigned row IDs are not provided. Setting ## Current scope -Compaction-generated changelogs (`full-compaction` / `lookup`) are unsupported. +The `full-compaction` changelog producer is unsupported; `lookup` is supported +for primary-key compaction without deletion vectors. Outside Data Evolution mode, row tracking, deletion vectors and dedicated blob/vector storage are unsupported. Query authorization, rowkind fields and postpone buckets are unsupported in all modes. diff --git a/docs/src/sql.md b/docs/src/sql.md index 8fff70bdb..819178bd0 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -2896,6 +2896,11 @@ plans, cross-partition dynamic bucket writes, or advanced aggregation options such as `ignore-retract`, `distinct`, `nested-key`, `count-limit`, and sequence groups. +Primary-key compaction supports `changelog-producer=lookup` without deletion +vectors, generating INSERT, UPDATE_BEFORE, UPDATE_AFTER and DELETE events when +L0 files are compacted. The executor requires `local-cache.enabled=true` on the +catalog/FileIO. See [Lookup changelog producer](compaction.md#lookup-changelog-producer). + ### Global Index Options Set these options when building global indexes with From c7233be82449e4b187adda220d673a15b6a612ca Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Thu, 29 Oct 2026 12:54:42 +0800 Subject: [PATCH 2/2] fix(table): align lookup changelog with Java reducers --- .../paimon/src/table/compact/primary_key.rs | 36 +- .../src/table/compact/primary_key/lookup.rs | 165 ++++++- crates/paimon/src/table/kv_file_reader.rs | 6 +- crates/paimon/src/table/sort_merge.rs | 72 ++- .../tests/pk_compact/lookup_changelog.rs | 430 +++++++++++++++++- docs/src/compaction.md | 17 +- 6 files changed, 681 insertions(+), 45 deletions(-) diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index f03282640..156c79858 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -103,12 +103,12 @@ impl MergeFunction for CompactDeduplicate { } } -/// The query reader omits singleton retracts and materialized DELETE rows. -/// Compaction persists these states so later inserts/retractions merge exactly -/// as they would against the original files. Singletons remain unreduced, as -/// in Java's reducer wrapper; compound records use the existing engine reducer. +/// Ordinary compaction preserves singleton states as Java's reducer wrapper +/// does. Lookup compaction always runs the reducer, including for one retract. +/// Persist synthesized DELETEs when their state can safely be replayed. struct CompactReducer { inner: Box, + reduce_singletons: bool, reject_delete_state: bool, reject_retract_group: bool, } @@ -121,14 +121,19 @@ impl MergeFunction for CompactReducer { columns: &[usize], schema: &SchemaRef, ) -> Result { - if let [row] = rows { - RowKind::from_value(row.value_kind)?; - return Ok(MergeResult::SourceRow { - batch_idx: row.batch_idx, - row_idx: row.row_idx, - }); + if !self.reduce_singletons { + if let [row] = rows { + RowKind::from_value(row.value_kind)?; + return Ok(MergeResult::SourceRow { + batch_idx: row.batch_idx, + row_idx: row.row_idx, + }); + } } - if self.reject_retract_group && rows.iter().any(|row| matches!(row.value_kind, 1 | 3)) { + if rows.len() > 1 + && self.reject_retract_group + && rows.iter().any(|row| matches!(row.value_kind, 1 | 3)) + { // Some aggregators preserve flags or raw DELETE payloads that a // single physical INSERT/DELETE cannot encode, even if the final // merged row is an INSERT. Leave this group in its source files. @@ -137,7 +142,7 @@ impl MergeFunction for CompactReducer { }); } Ok(match self.inner.merge(rows, batches, columns, schema)? { - MergeResult::MaterializedDeleteRow(_) if self.reject_delete_state => { + MergeResult::MaterializedDeleteRow(_) if self.reject_delete_state && rows.len() > 1 => { // A partial-update reducer initializes from the first retract // and then retracts its payload. Replaying an accumulated // DELETE can cancel its own state. First-value aggregators @@ -197,7 +202,14 @@ fn new_merge_function( } else { Box::new(CompactReducer { inner, + // LookupMergeFunction always invokes the underlying reducer, even + // for one retract. The ordinary reducer wrapper keeps singletons. + reduce_singletons: config + .table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")), reject_delete_state: config.merge_engine == MergeEngine::PartialUpdate + && !crate::spec::CoreOptions::new(&config.table_options).ignore_delete() && !aggregate_functions.is_empty(), reject_retract_group: !(config.merge_engine == MergeEngine::PartialUpdate && crate::spec::CoreOptions::new(&config.table_options).ignore_delete()) diff --git a/crates/paimon/src/table/compact/primary_key/lookup.rs b/crates/paimon/src/table/compact/primary_key/lookup.rs index 0b21a150a..b56c65eab 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup.rs @@ -20,12 +20,15 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; -use arrow_array::{Int8Array, RecordBatch, UInt64Array}; -use arrow_schema::SchemaRef; +use arrow_array::{cast::AsArray, Array, ArrayRef, Int8Array, MapArray, RecordBatch, UInt64Array}; +use arrow_row::{RowConverter, SortField}; +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}; +use crate::table::sort_merge::{ + lookup_rows, BufferedBatch, MergeFunction, MergeResult, MergeRow, FILE_LEVEL_METADATA_KEY, +}; use crate::{Error, Result}; struct ChangelogOptions { @@ -163,6 +166,125 @@ fn with_kind(batch: &RecordBatch, kind: RowKind) -> Result { RecordBatch::try_new(batch.schema(), columns).map_err(|error| invalid(error.to_string())) } +fn contains_float(data_type: &ArrowType) -> bool { + match data_type { + ArrowType::Float32 | ArrowType::Float64 => true, + ArrowType::List(field) | ArrowType::LargeList(field) | ArrowType::Map(field, _) => { + contains_float(field.data_type()) + } + ArrowType::FixedSizeList(field, _) => contains_float(field.data_type()), + ArrowType::Struct(fields) => fields.iter().any(|field| contains_float(field.data_type())), + _ => false, + } +} + +/// Java compares Maps by key and nested rows field by field. Binary arrays +/// retain their physical shortcut unless a floating-point descendant disables it. +fn values_equal(left: &ArrayRef, right: &ArrayRef) -> bool { + if left.to_data() == right.to_data() { + return true; + } + if left.len() != right.len() || left.data_type() != right.data_type() { + return false; + } + for row in 0..left.len() { + if left.is_null(row) || right.is_null(row) { + if left.is_null(row) != right.is_null(row) { + return false; + } + continue; + } + let equal = match left.data_type() { + ArrowType::Map(_, _) => { + let left = left.as_any().downcast_ref::().unwrap().value(row); + let right = right + .as_any() + .downcast_ref::() + .unwrap() + .value(row); + map_entries_equal(&left, &right) + } + ArrowType::Struct(_) => left + .as_struct() + .columns() + .iter() + .zip(right.as_struct().columns()) + .all(|(left, right)| values_equal(&left.slice(row, 1), &right.slice(row, 1))), + kind if contains_float(kind) => match kind { + ArrowType::List(_) => values_equal( + &left.as_list::().value(row), + &right.as_list::().value(row), + ), + ArrowType::LargeList(_) => values_equal( + &left.as_list::().value(row), + &right.as_list::().value(row), + ), + ArrowType::FixedSizeList(_, _) => values_equal( + &left.as_fixed_size_list().value(row), + &right.as_fixed_size_list().value(row), + ), + _ => left.slice(row, 1).to_data() == right.slice(row, 1).to_data(), + }, + _ => left.slice(row, 1).to_data() == right.slice(row, 1).to_data(), + }; + if !equal { + return false; + } + } + true +} + +fn map_entries_equal(left: &arrow_array::StructArray, right: &arrow_array::StructArray) -> bool { + if left.len() != right.len() { + return false; + } + let encoded = RowConverter::new(vec![SortField::new(left.column(0).data_type().clone())]) + .and_then(|converter| { + Ok(( + converter.convert_columns(&[left.column(0).clone()])?, + converter.convert_columns(&[right.column(0).clone()])?, + )) + }) + .ok(); + let mut by_key = HashMap::, Vec>::new(); + if let Some((_, keys)) = &encoded { + for row in 0..right.len() { + by_key + .entry(keys.row(row).as_ref().to_vec()) + .or_default() + .push(row); + } + } + let all = (0..right.len()).collect::>(); + let mut matched = vec![false; right.len()]; + for row in 0..left.len() { + let candidates = match &encoded { + Some((keys, _)) => by_key + .get(keys.row(row).as_ref()) + .map(Vec::as_slice) + .unwrap_or(&[]), + None => &all, + }; + let found = candidates.iter().copied().find(|&other| { + !matched[other] + && (encoded.is_some() + || values_equal( + &left.column(0).slice(row, 1), + &right.column(0).slice(other, 1), + )) + && values_equal( + &left.column(1).slice(row, 1), + &right.column(1).slice(other, 1), + ) + }); + let Some(other) = found else { + return false; + }; + matched[other] = true; + } + true +} + impl MergeFunction for LookupChangelogMergeFunction { fn merge( &self, @@ -183,10 +305,34 @@ impl MergeFunction for LookupChangelogMergeFunction { schema: &SchemaRef, ) -> Result<(MergeResult, Option)> { let (base, has_l0, selected) = lookup_rows(rows, batches)?; - let result = self.inner.merge(&selected, batches, columns, schema)?; + // Java deliberately excludes first-row from LookupMergeFunction: an + // older, higher-level row can still be the first accepted record. + let result = self.inner.merge( + if self.first_row { rows } else { &selected }, + batches, + columns, + schema, + )?; if !has_l0 { 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)) + }); + // FirstRowMergeFunctionWrapper emits the first add unchanged, + // including UPDATE_AFTER; an existing accepted history emits none. + let changelog = if contains_history { + None + } else { + live_row(result_row(&result, batches, columns, schema)?)? + }; + return Ok((result, changelog)); + } let before = live_row( base.as_ref() .map(|row| source_row(row.batch_idx, row.row_idx, batches, columns, schema)) @@ -196,14 +342,15 @@ impl MergeFunction for LookupChangelogMergeFunction { let mut changes = Vec::with_capacity(2); match (&before, &after) { (None, Some(after)) => changes.push(with_kind(after, RowKind::Insert)?), - (Some(before), None) if !self.first_row && !self.options.ignore_delete => { + (Some(before), None) if !self.options.ignore_delete => { changes.push(with_kind(before, RowKind::Delete)?) } - (Some(before), Some(after)) if !self.first_row => { + (Some(before), Some(after)) => { let equal = self.options.deduplicate - && self.value_columns.iter().all(|index| { - before.column(*index).to_data() == after.column(*index).to_data() - }); + && self + .value_columns + .iter() + .all(|index| values_equal(before.column(*index), after.column(*index))); if !equal { if !self.options.ignore_update_before { changes.push(with_kind(before, RowKind::UpdateBefore)?); diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index eae33f709..09c373bf0 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -28,7 +28,7 @@ use super::data_file_reader::{file_index_selection_to_local_ranges, DataFileReader}; use super::sort_merge::{ AggregateMergeFunction, DeduplicateMergeFunction, FirstRowMergeFunction, LookupMergeFunction, - MergeFunction, PartialUpdateMergeFunction, SortMergeReaderBuilder, + MergeFunction, PartialUpdateMergeFunction, QueryReducerMergeFunction, SortMergeReaderBuilder, CHANGELOG_BATCH_METADATA_KEY, FILE_LEVEL_METADATA_KEY, }; use crate::arrow::format::MosaicPrefetchOptions; @@ -387,7 +387,9 @@ impl KeyValueFileReader { .get("changelog-producer") .is_some_and(|value| value.eq_ignore_ascii_case("lookup")) { - Box::new(LookupMergeFunction(inner)) + Box::new(QueryReducerMergeFunction(Box::new(LookupMergeFunction( + inner, + )))) } else { inner }, diff --git a/crates/paimon/src/table/sort_merge.rs b/crates/paimon/src/table/sort_merge.rs index bddcc33fd..492dcd81a 100644 --- a/crates/paimon/src/table/sort_merge.rs +++ b/crates/paimon/src/table/sort_merge.rs @@ -228,6 +228,26 @@ impl MergeFunction for LookupMergeFunction { } } +/// Java queries wrap LookupMergeFunction in ReducerMergeFunctionWrapper, then +/// drop retract results. The shortcut counts the original inputs, before +/// lookup discards superseded high-level states. Compaction has no such wrapper. +pub(super) struct QueryReducerMergeFunction(pub Box); + +impl MergeFunction for QueryReducerMergeFunction { + fn merge( + &self, + rows: &[MergeRow], + batches: &[BufferedBatch], + columns: &[usize], + schema: &SchemaRef, + ) -> crate::Result { + if let Some(result) = singleton_reducer_result(rows)? { + return Ok(result); + } + self.0.merge(rows, batches, columns, schema) + } +} + /// Deduplicate merge: keeps the row with the highest sequence. /// When `sequence.field` is configured, compares its typed Arrow row encoding /// first, then falls back to system `_SEQUENCE_NUMBER` as tie-breaker. @@ -325,6 +345,7 @@ impl MergeFunction for FirstRowMergeFunction { #[derive(Debug)] pub(crate) struct PartialUpdateMergeFunction { ignore_delete: bool, + reduce_singletons: bool, remove_record_on_delete: bool, sequence_group_partial_delete: HashSet, sequence_group_enabled: bool, @@ -401,6 +422,9 @@ impl PartialUpdateMergeFunction { PartialUpdateConfig::new(table_options).validate_write_mode(true, table_name)?; Ok(Self { ignore_delete: CoreOptions::new(table_options).ignore_delete(), + reduce_singletons: table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")), remove_record_on_delete: PartialUpdateConfig::new(table_options) .remove_record_on_delete(), sequence_group_partial_delete: HashSet::new(), @@ -513,6 +537,9 @@ impl PartialUpdateMergeFunction { Ok(Self { ignore_delete: CoreOptions::new(table_options).ignore_delete(), + reduce_singletons: table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")), remove_record_on_delete: config.remove_record_on_delete(), sequence_group_partial_delete, sequence_group_enabled, @@ -534,8 +561,10 @@ impl MergeFunction for PartialUpdateMergeFunction { source_output_col_indices: &[usize], output_schema: &SchemaRef, ) -> crate::Result { - if let Some(result) = singleton_reducer_result(rows)? { - return Ok(result); + if !self.reduce_singletons { + if let Some(result) = singleton_reducer_result(rows)? { + return Ok(result); + } } self.merge_unreduced(rows, batch_buffer, source_output_col_indices, output_schema) } @@ -580,21 +609,20 @@ impl PartialUpdateMergeFunction { } let mut saw_add = false; let mut current_delete_row = false; - let mut saw_retract = false; + let mut initialized = false; let mut latest_row = None; for row_idx in ordered_row_indices { let row = &rows[row_idx]; let kind = RowKind::from_value(row.value_kind)?; if !kind.is_add() { - if self.ignore_delete { + if self.ignore_delete && !self.reduce_singletons { continue; } - latest_row = Some(row); - if !saw_add && !saw_retract { + if !initialized { // Java initializes the row from the first retract before - // applying sequence-group retractions. That row also - // becomes the DELETE payload when no insert follows. + // checking ignore-delete or applying sequence-group + // retractions. Keep that payload for later sparse updates. Self::replace_with_source( &mut selected_by_col, aggregators.as_mut().map(|guard| guard.as_mut_slice()), @@ -604,8 +632,12 @@ impl PartialUpdateMergeFunction { source_output_col_indices, )?; group_sequence_rows.fill(Some((row.batch_idx, row.row_idx))); + initialized = true; } - saw_retract = true; + if self.ignore_delete { + continue; + } + latest_row = Some(row); if self.sequence_group_enabled { let mut full_delete = false; for (group_idx, group) in self.sequence_groups.iter().enumerate() { @@ -693,6 +725,7 @@ impl PartialUpdateMergeFunction { }); } saw_add = true; + initialized = true; if !self.sequence_group_enabled { current_delete_row = false; } @@ -775,17 +808,24 @@ impl PartialUpdateMergeFunction { } } - if !saw_add && !saw_retract { + if !initialized { return Ok(MergeResult::Omit); } let deleted = current_delete_row || !saw_add; + // Ignored legacy retracts seed the value but do not advance Java's + // latestSequenceNumber, whose reset value is zero. + let ignored_retract_metadata = latest_row.is_none().then(|| MergeRow { + sequence_number: 0, + ..rows[0].clone() + }); + let latest_row = latest_row.or(ignored_retract_metadata.as_ref()).unwrap(); let output_columns: Vec = output_schema .fields() .iter() .enumerate() .map(|(output_col_idx, field)| { - if let Some(column) = merged_metadata_column(field, latest_row.unwrap(), deleted) { + if let Some(column) = merged_metadata_column(field, latest_row, deleted) { return Ok(column); } let column = match aggregators @@ -910,6 +950,7 @@ pub(crate) struct AggregateMergeFunction { /// `Some` holds the aggregator that owns a user value column. aggregators: Mutex>>>, remove_record_on_delete: bool, + reduce_singletons: bool, } impl AggregateMergeFunction { @@ -968,6 +1009,9 @@ impl AggregateMergeFunction { Ok(Self { aggregators: Mutex::new(aggregators), remove_record_on_delete: config.remove_record_on_delete(), + reduce_singletons: table_options + .get("changelog-producer") + .is_some_and(|value| value.eq_ignore_ascii_case("lookup")), }) } @@ -1103,8 +1147,10 @@ impl MergeFunction for AggregateMergeFunction { source_output_col_indices: &[usize], output_schema: &SchemaRef, ) -> crate::Result { - if let Some(result) = singleton_reducer_result(rows)? { - return Ok(result); + if !self.reduce_singletons { + if let Some(result) = singleton_reducer_result(rows)? { + return Ok(result); + } } let mut ordered: Vec<_> = rows.iter().collect(); // Stable sorting keeps input order for equal sequences. diff --git a/crates/paimon/tests/pk_compact/lookup_changelog.rs b/crates/paimon/tests/pk_compact/lookup_changelog.rs index e43f00aba..197311666 100644 --- a/crates/paimon/tests/pk_compact/lookup_changelog.rs +++ b/crates/paimon/tests/pk_compact/lookup_changelog.rs @@ -64,12 +64,14 @@ pub(super) async fn events(table: &Table, start: i64, end: i64) -> Vec<(String, .downcast_ref::() .unwrap(); let ids = batch - .column(1) + .column_by_name("id") + .unwrap() .as_any() .downcast_ref::() .unwrap(); let values = batch - .column(2) + .column_by_name("value") + .unwrap() .as_any() .downcast_ref::() .unwrap(); @@ -186,6 +188,163 @@ async fn lookup_compaction_respects_java_changelog_flags_and_default_duplicate_u } } +#[tokio::test] +async fn lookup_row_deduplication_compares_map_entries_by_key() { + use arrow_array::builder::{Int32Builder, MapBuilder}; + use paimon::spec::{DataField, MapType, RowType}; + for nested in [false, true] { + let map_type = DataType::Map(MapType::new( + DataType::Int(IntType::new()), + DataType::Int(IntType::new()), + )); + let value_type = if nested { + DataType::Row(RowType::new(vec![DataField::new( + 2, + "m".into(), + map_type.clone(), + )])) + } else { + map_type + }; + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", value_type) + .primary_key(["id"]) + .option("bucket", "1") + .option("changelog-producer", "lookup") + .option("changelog-producer.row-deduplicate", "true") + .option("changelog-producer.row-deduplicate-ignore-fields", "id") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |entries: &[(i32, Option)]| { + let mut maps = MapBuilder::new(None, Int32Builder::new(), Int32Builder::new()); + for (key, value) in entries { + maps.keys().append_value(*key); + maps.values().append_option(*value); + } + maps.append(true).unwrap(); + let value = std::sync::Arc::new(maps.finish()) as arrow_array::ArrayRef; + let value = if nested { + std::sync::Arc::new(arrow_array::StructArray::from(vec![( + std::sync::Arc::new(arrow_schema::Field::new( + "m", + value.data_type().clone(), + true, + )), + value, + )])) as arrow_array::ArrayRef + } else { + value + }; + RecordBatch::try_from_iter([ + ( + "id", + std::sync::Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef, + ), + ("value", value), + ]) + .unwrap() + }; + let payloads = |batch: &RecordBatch| { + let value = batch.column_by_name("value").unwrap(); + let value = if nested { + value + .as_any() + .downcast_ref::() + .unwrap() + .column(0) + } else { + value + }; + let maps = value + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|row| { + let entries = maps.value(row); + let keys = entries + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let values = entries + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + (0..entries.len()) + .map(|index| { + ( + keys.value(index), + (!values.is_null(index)).then(|| values.value(index)), + ) + }) + .collect::>() + }) + .collect::>() + }; + let mut previous = vec![(1, Some(10)), (2, None)]; + write_batch(&table, &batch(&previous)).await; + let mut start = compact(&table).await; + for (entries, count) in [ + (vec![(2, None), (1, Some(10))], 0), + (vec![(2, Some(20)), (1, Some(10))], 2), + (vec![(1, Some(10)), (2, Some(20))], 0), + (vec![(1, Some(10)), (3, Some(20))], 2), + (vec![], 2), + (vec![], 0), + ] { + write_batch(&table, &batch(&entries)).await; + let end = compact(&table).await; + let plan = table + .new_read_builder() + .new_incremental_scan(IncrementalScanMode::Changelog, start, end) + .plan_combined() + .await + .unwrap(); + let logs: Vec = table + .new_read_builder() + .new_read() + .unwrap() + .to_arrow_with_row_kind(plan.splits()) + .unwrap() + .try_collect() + .await + .unwrap(); + assert_eq!(logs.iter().map(RecordBatch::num_rows).sum::(), count); + if count != 0 { + let kinds = logs + .iter() + .flat_map(|batch| { + let kinds = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + (0..batch.num_rows()) + .map(|row| kinds.value(row).to_string()) + .collect::>() + }) + .collect::>(); + assert_eq!(kinds, vec!["-U", "+U"]); + assert_eq!( + logs.iter().flat_map(&payloads).collect::>(), + vec![previous.clone(), entries.clone()] + ); + } + let data = read(&table, Some(&["value"])).await; + assert_eq!( + data.iter().flat_map(&payloads).collect::>(), + vec![entries.clone()] + ); + previous = entries; + start = end; + } + } +} + #[tokio::test] async fn lookup_compaction_first_row_only_emits_new_keys() { let table = lookup_table(&[("merge-engine", "first-row")]).await; @@ -201,6 +360,267 @@ async fn lookup_compaction_first_row_only_emits_new_keys() { assert_eq!(rows(&table).await, vec![(1, 10), (2, 20)]); } +#[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; + append_at_level(&table, 10, 3).await; + append_at_level(&table, 20, 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![30])).await; + } + let end = compact(&table).await; + assert_eq!(rows(&table).await, vec![(1, 10)]); + assert!(events(&table, history, end).await.is_empty()); + } +} + +#[tokio::test] +async fn lookup_first_row_preserves_the_first_add_kind() { + let table = lookup_table(&[("merge-engine", "first-row")]).await; + write_batch(&table, &make_batch_with_kinds(vec![1], vec![10], vec![2])).await; + let end = compact(&table).await; + assert_eq!(events(&table, 0, end).await, expected(&[("+U", 1, 10)])); + assert_eq!(rows(&table).await, vec![(1, 10)]); +} + +#[tokio::test] +async fn lookup_first_row_ignores_retract_history_before_deciding_whether_a_key_is_new() { + for existing in [false, true] { + let table = lookup_table(&[("merge-engine", "first-row"), ("ignore-delete", "true")]).await; + if existing { + append_at_level(&table, 10, 3).await; + } + // Older Java first-row files can retain retracts even when deletes are + // ignored. A deduplicate writer creates that legacy physical input. + let legacy = table.copy_with_options(std::collections::HashMap::from([ + ("merge-engine".into(), "deduplicate".into()), + ("ignore-delete".into(), "false".into()), + ])); + append_kind_at_level(&legacy, 20, 2, 3).await; + let history = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap() + .unwrap(); + write_batch(&table, &make_batch(vec![1], vec![30])).await; + let end = compact(&table).await; + assert_eq!( + rows(&table).await, + vec![(1, if existing { 10 } else { 30 })] + ); + assert_eq!( + events(&table, history, end).await, + if existing { + vec![] + } else { + expected(&[("+I", 1, 30)]) + } + ); + } +} + +#[tokio::test] +async fn lookup_aggregation_reduces_singleton_retracts_like_java() { + for remove in ["false", "true"] { + for kind in [1, 3] { + let table = lookup_table(&[ + ("merge-engine", "aggregation"), + ("fields.value.aggregate-function", "sum"), + ("aggregation.remove-record-on-delete", remove), + ]) + .await; + write_batch(&table, &make_batch_with_kinds(vec![1], vec![5], vec![kind])).await; + let deleted = remove == "true" && kind == 3; + let values = if deleted { vec![] } else { vec![(1, -5)] }; + // Java query wraps the lookup function in a singleton reducer; + // lookup compact instead runs the raw aggregator. + assert!( + rows(&table).await.is_empty(), + "query: remove={remove}, kind={kind}" + ); + let end = compact(&table).await; + assert_eq!( + rows(&table).await, + values, + "compact: remove={remove}, kind={kind}" + ); + assert_eq!( + events(&table, 0, end).await, + if deleted { + vec![] + } else { + expected(&[("+I", 1, -5)]) + } + ); + } + } +} + +#[tokio::test] +async fn lookup_partial_update_preserves_a_single_sequence_group_retract() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("seq", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "partial-update") + .option("changelog-producer", "lookup") + .option("fields.seq.sequence-group", "value") + .option("fields.value.aggregate-function", "sum") + .build() + .unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |seq, value, kind| { + RecordBatch::try_from_iter([ + ( + "id", + std::sync::Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef, + ), + ( + "seq", + std::sync::Arc::new(Int32Array::from(vec![seq])) as arrow_array::ArrayRef, + ), + ( + "value", + std::sync::Arc::new(Int32Array::from(vec![value])) as arrow_array::ArrayRef, + ), + ( + "_VALUE_KIND", + std::sync::Arc::new(arrow_array::Int8Array::from(vec![kind])) + as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + write_batch(&table, &batch(10, 5, 3)).await; + let deleted = compact(&table).await; + assert!(read(&table, Some(&["value"])).await.is_empty()); + assert!(events(&table, 0, deleted).await.is_empty()); + write_batch(&table, &batch(11, 7, 0)).await; + let end = compact(&table).await; + assert_eq!( + events(&table, deleted, end).await, + expected(&[("+I", 1, 7)]) + ); + let actual = read(&table, Some(&["value"])).await; + assert_eq!(actual.iter().map(RecordBatch::num_rows).sum::(), 1); + assert_eq!( + actual[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 7 + ); +} + +#[tokio::test] +async fn lookup_partial_update_retains_ignored_legacy_delete_payload() { + for aggregate in [None, Some("sum"), Some("first_value")] { + let mut schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .column("b", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("merge-engine", "partial-update") + .option("changelog-producer", "lookup") + .option("ignore-delete", "true"); + if let Some(aggregate) = aggregate { + schema = schema + .option("fields.b.sequence-group", "value") + .option("fields.value.aggregate-function", aggregate); + } + let schema = schema.build().unwrap(); + let table = table_with_schema(TableSchema::new(0, &schema)).await; + let batch = |value, b, kind| { + RecordBatch::try_from_iter([ + ( + "id", + std::sync::Arc::new(Int32Array::from(vec![1])) as arrow_array::ArrayRef, + ), + ( + "value", + std::sync::Arc::new(Int32Array::from(vec![value])) as arrow_array::ArrayRef, + ), + ( + "b", + std::sync::Arc::new(Int32Array::from(vec![Some(b)])) as arrow_array::ArrayRef, + ), + ( + "_VALUE_KIND", + std::sync::Arc::new(arrow_array::Int8Array::from(vec![kind])) + as arrow_array::ArrayRef, + ), + ]) + .unwrap() + }; + let legacy = table.copy_with_options(std::collections::HashMap::from([ + ("merge-engine".into(), "deduplicate".into()), + ("ignore-delete".into(), "false".into()), + ])); + write_batch(&legacy, &batch(Some(10), 20, 3)).await; + write_batch(&legacy, &batch(Some(99), 21, 3)).await; + let deleted = compact(&table).await; + assert!(events(&table, 0, deleted).await.is_empty()); + let scan = table.new_read_builder().new_scan().plan().await.unwrap(); + let files = scan + .splits() + .iter() + .flat_map(|split| split.data_files()) + .collect::>(); + assert_eq!(files.len(), 1); + assert_eq!(files[0].delete_row_count, Some(1)); + assert_eq!( + (files[0].min_sequence_number, files[0].max_sequence_number), + (0, 0) + ); + write_batch(&table, &batch(aggregate.map(|_| 7), 30, 0)).await; + let end = compact(&table).await; + let actual = read(&table, Some(&["value", "b"])).await; + assert_eq!(actual.iter().map(RecordBatch::num_rows).sum::(), 1); + let value = actual[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!( + !value.is_null(0), + "ignored delete must seed the unchanged column" + ); + let expected_value = match aggregate { + None => 10, + Some("sum") => 17, + Some("first_value") => 7, + _ => unreachable!(), + }; + assert_eq!(value.value(0), expected_value, "aggregate={aggregate:?}"); + assert_eq!( + actual[0] + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + 30 + ); + assert_eq!( + events(&table, deleted, end).await, + expected(&[("+I", 1, expected_value)]) + ); + } +} + #[tokio::test] async fn lookup_compaction_materializes_aggregation_history_once_and_ignores_high_level_only_keys() { @@ -230,10 +650,14 @@ async fn lookup_compaction_materializes_aggregation_history_once_and_ignores_hig } async fn append_at_level(table: &Table, value: i32, level: i32) { + append_kind_at_level(table, value, level, 0).await; +} + +async fn append_kind_at_level(table: &Table, value: i32, level: i32, kind: i8) { let builder = table.new_write_builder(); let mut writer = builder.new_write().unwrap(); writer - .write_arrow_batch(&make_batch(vec![1], vec![value])) + .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![value], vec![kind])) .await .unwrap(); let mut messages = writer.prepare_commit().await.unwrap(); diff --git a/docs/src/compaction.md b/docs/src/compaction.md index 890c08659..2f8121f2d 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -133,9 +133,9 @@ state, including original sequence metadata and synthesized DELETE records: - `partial-update` combines sparse columns using the configured sequence groups and field aggregators. Delete payloads and group sequence values are preserved so subsequent sparse updates can continue from that state. -- `aggregation` writes the engine's field accumulators. Singleton records - remain unreduced, including retracts; compound DELETE states are preserved - rather than being discarded as they are by the query reader. +- `aggregation` writes the engine's field accumulators. With `none` or `input`, + singleton records remain unreduced, including retracts; compound DELETE states + are preserved rather than being discarded as they are by the query reader. The existing engine rules for ignored deletes, unsupported retracts and whole-record deletion also apply during compaction. If all records are @@ -194,9 +194,14 @@ changes only for keys with L0 input: - No live history and no live result: no changelog. Before images retain the historical sequence metadata; after images carry the -merged result's metadata. `first-row` emits only inserts for previously unseen -keys. `partial-update` and `aggregation` use their existing reducers and state -preservation restrictions. History selection precedes merging. +merged result's metadata. `first-row` emits the first accepted add for previously +unseen keys, preserving its `INSERT` or `UPDATE_AFTER` kind. It considers all +input levels when choosing the earliest record. `partial-update` and `aggregation` +always run the reducer in lookup mode, including singleton retracts, as Java does. +For example, lookup compaction turns a lone `DELETE(value=5)` with `sum` into +live state `-5` when `aggregation.remove-record-on-delete=false`; an ordinary +query still omits that singleton retract before compaction. Existing state +preservation restrictions still apply. History selection precedes merging. Java-compatible options have these defaults: