From 52d8b507eddc7871f50a5de7c31bce03ce06b9b7 Mon Sep 17 00:00:00 2001 From: mingfeng Date: Wed, 16 Sep 2026 21:44:36 -0700 Subject: [PATCH 1/4] feat(manifest): add bounded sorted compaction foundation --- crates/paimon/src/spec/core_options.rs | 149 +++ crates/paimon/src/spec/schema.rs | 97 +- crates/paimon/src/table/manifest_sort.rs | 1113 ++++++++++++++++++++++ crates/paimon/src/table/mod.rs | 1 + crates/paimon/src/table/table_commit.rs | 924 ++++++++++++++++++ 5 files changed, 2281 insertions(+), 3 deletions(-) create mode 100644 crates/paimon/src/table/manifest_sort.rs diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index d3114fd4c..70bacfe2e 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -88,6 +88,17 @@ const MANIFEST_COMPRESSION_OPTION: &str = "manifest.compression"; const MANIFEST_TARGET_FILE_SIZE_OPTION: &str = "manifest.target-file-size"; const MANIFEST_TARGET_SIZE_OPTION: &str = "manifest.target-size"; const MANIFEST_MERGE_MIN_COUNT_OPTION: &str = "manifest.merge-min-count"; +pub(crate) const MANIFEST_SORT_ENABLED_OPTION: &str = "manifest-sort.enabled"; +pub(crate) const MANIFEST_SORT_PARTITION_FIELD_OPTION: &str = "manifest-sort.partition-field"; +const MANIFEST_SORT_MAX_REWRITE_SIZE_OPTION: &str = "manifest-sort.max-rewrite-size"; +const MANIFEST_FULL_COMPACTION_THRESHOLD_SIZE_OPTION: &str = + "manifest.full-compaction-threshold-size"; +const COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT_OPTION: &str = + "compaction.max-size-amplification-percent"; +const COMPACTION_SIZE_RATIO_OPTION: &str = "compaction.size-ratio"; +const SORT_SPILL_BUFFER_SIZE_OPTION: &str = "sort-spill-buffer-size"; +const LOCAL_SORT_MAX_NUM_FILE_HANDLES_OPTION: &str = "local-sort.max-num-file-handles"; +const WRITE_BUFFER_SPILL_MAX_DISK_SIZE_OPTION: &str = "write-buffer-spill.max-disk-size"; const WRITE_PARQUET_BUFFER_SIZE_OPTION: &str = "write.parquet-buffer-size"; const READ_BATCH_SIZE_OPTION: &str = "read.batch-size"; const PARQUET_ROW_GROUP_PARALLELISM_OPTION: &str = "read.parquet.row-group.parallelism"; @@ -133,6 +144,12 @@ const DEFAULT_SOURCE_SPLIT_OPEN_FILE_COST: i64 = 4 * 1024 * 1024; const DEFAULT_MANIFEST_COMPRESSION: &str = "zstd"; const DEFAULT_MANIFEST_TARGET_FILE_SIZE: i64 = 8 * 1024 * 1024; const DEFAULT_MANIFEST_MERGE_MIN_COUNT: usize = 30; +const DEFAULT_MANIFEST_SORT_MAX_REWRITE_SIZE: i64 = 256 * 1024 * 1024; +const DEFAULT_MANIFEST_FULL_COMPACTION_THRESHOLD_SIZE: i64 = 16 * 1024 * 1024; +const DEFAULT_COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT: i32 = 200; +const DEFAULT_COMPACTION_SIZE_RATIO: i32 = 1; +const DEFAULT_SORT_SPILL_BUFFER_SIZE: i64 = 64 * 1024 * 1024; +const DEFAULT_LOCAL_SORT_MAX_NUM_FILE_HANDLES: usize = 128; const DEFAULT_PARTITION_DEFAULT_NAME: &str = "__DEFAULT_PARTITION__"; const DEFAULT_CHANGELOG_FILE_PREFIX: &str = "changelog-"; const DEFAULT_TARGET_FILE_SIZE: i64 = 256 * 1024 * 1024; @@ -1165,6 +1182,87 @@ impl<'a> CoreOptions<'a> { .unwrap_or(DEFAULT_MANIFEST_MERGE_MIN_COUNT) } + /// Whether commit-time manifest compaction uses the sorted rewrite path. + pub fn manifest_sort_enabled(&self) -> bool { + self.options + .get(MANIFEST_SORT_ENABLED_OPTION) + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + + /// Partition field used as the manifest sort key. + /// + /// When absent, the first partition field is used. + pub fn manifest_sort_partition_field(&self) -> Option<&str> { + self.options + .get(MANIFEST_SORT_PARTITION_FIELD_OPTION) + .map(String::as_str) + .filter(|value| !value.is_empty()) + } + + /// Maximum input bytes consumed by the aggressive sorted rewrite portion. + pub fn manifest_sort_max_rewrite_size(&self) -> i64 { + self.options + .get(MANIFEST_SORT_MAX_REWRITE_SIZE_OPTION) + .and_then(|v| parse_memory_size(v)) + .filter(|v| *v >= 0) + .unwrap_or(DEFAULT_MANIFEST_SORT_MAX_REWRITE_SIZE) + } + + /// Delta manifest bytes which trigger full manifest compaction. + pub fn manifest_full_compaction_threshold_size(&self) -> i64 { + self.options + .get(MANIFEST_FULL_COMPACTION_THRESHOLD_SIZE_OPTION) + .and_then(|v| parse_memory_size(v)) + .filter(|v| *v > 0) + .unwrap_or(DEFAULT_MANIFEST_FULL_COMPACTION_THRESHOLD_SIZE) + } + + /// Maximum acceptable size amplification percentage for manifest sorted runs. + pub fn compaction_max_size_amplification_percent(&self) -> i32 { + self.options + .get(COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT_OPTION) + .and_then(|v| v.parse().ok()) + .filter(|v| *v > 0) + .unwrap_or(DEFAULT_COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT) + } + + /// Size-ratio flexibility used when selecting adjacent manifest sorted runs. + pub fn compaction_size_ratio(&self) -> i32 { + self.options + .get(COMPACTION_SIZE_RATIO_OPTION) + .and_then(|v| v.parse().ok()) + .filter(|v| *v > 0) + .unwrap_or(DEFAULT_COMPACTION_SIZE_RATIO) + } + + /// Maximum in-memory bytes held by one manifest external-sort run. + pub fn sort_spill_buffer_size(&self) -> i64 { + self.options + .get(SORT_SPILL_BUFFER_SIZE_OPTION) + .and_then(|v| parse_memory_size(v)) + .filter(|v| *v > 0) + .unwrap_or(DEFAULT_SORT_SPILL_BUFFER_SIZE) + } + + /// Maximum number of spill runs opened by one external merge. + pub fn local_sort_max_num_file_handles(&self) -> usize { + self.options + .get(LOCAL_SORT_MAX_NUM_FILE_HANDLES_OPTION) + .and_then(|v| v.parse().ok()) + .filter(|v| *v >= 2) + .unwrap_or(DEFAULT_LOCAL_SORT_MAX_NUM_FILE_HANDLES) + } + + /// Maximum local disk bytes used by one spillable write buffer. + pub fn write_buffer_spill_max_disk_size(&self) -> i64 { + self.options + .get(WRITE_BUFFER_SPILL_MAX_DISK_SIZE_OPTION) + .and_then(|v| parse_memory_size(v)) + .filter(|v| *v > 0) + .unwrap_or(i64::MAX) + } + /// Number of buckets for the table. Default is -1 (dynamic bucket). pub fn bucket(&self) -> i32 { self.options @@ -2562,6 +2660,18 @@ mod tests { assert_eq!(core.manifest_compression(), "zstd"); assert_eq!(core.manifest_target_size(), 8 * 1024 * 1024); assert_eq!(core.manifest_merge_min_count(), 30); + assert!(!core.manifest_sort_enabled()); + assert_eq!(core.manifest_sort_partition_field(), None); + assert_eq!(core.manifest_sort_max_rewrite_size(), 256 * 1024 * 1024); + assert_eq!( + core.manifest_full_compaction_threshold_size(), + 16 * 1024 * 1024 + ); + assert_eq!(core.compaction_max_size_amplification_percent(), 200); + assert_eq!(core.compaction_size_ratio(), 1); + assert_eq!(core.sort_spill_buffer_size(), 64 * 1024 * 1024); + assert_eq!(core.local_sort_max_num_file_handles(), 128); + assert_eq!(core.write_buffer_spill_max_disk_size(), i64::MAX); } #[test] @@ -2579,6 +2689,33 @@ mod tests { ), (MANIFEST_COMPRESSION_OPTION.to_string(), "null".to_string()), (MANIFEST_MERGE_MIN_COUNT_OPTION.to_string(), "3".to_string()), + (MANIFEST_SORT_ENABLED_OPTION.to_string(), "true".to_string()), + ( + MANIFEST_SORT_PARTITION_FIELD_OPTION.to_string(), + "pt".to_string(), + ), + ( + MANIFEST_SORT_MAX_REWRITE_SIZE_OPTION.to_string(), + "64mb".to_string(), + ), + ( + MANIFEST_FULL_COMPACTION_THRESHOLD_SIZE_OPTION.to_string(), + "32mb".to_string(), + ), + ( + COMPACTION_MAX_SIZE_AMPLIFICATION_PERCENT_OPTION.to_string(), + "300".to_string(), + ), + (COMPACTION_SIZE_RATIO_OPTION.to_string(), "5".to_string()), + (SORT_SPILL_BUFFER_SIZE_OPTION.to_string(), "8mb".to_string()), + ( + LOCAL_SORT_MAX_NUM_FILE_HANDLES_OPTION.to_string(), + "16".to_string(), + ), + ( + WRITE_BUFFER_SPILL_MAX_DISK_SIZE_OPTION.to_string(), + "1gb".to_string(), + ), ]); let core = CoreOptions::new(&options); assert_eq!(core.bucket(), 4); @@ -2590,6 +2727,18 @@ mod tests { assert_eq!(core.manifest_compression(), "null"); assert_eq!(core.manifest_target_size(), 1024); assert_eq!(core.manifest_merge_min_count(), 3); + assert!(core.manifest_sort_enabled()); + assert_eq!(core.manifest_sort_partition_field(), Some("pt")); + assert_eq!(core.manifest_sort_max_rewrite_size(), 64 * 1024 * 1024); + assert_eq!( + core.manifest_full_compaction_threshold_size(), + 32 * 1024 * 1024 + ); + assert_eq!(core.compaction_max_size_amplification_percent(), 300); + assert_eq!(core.compaction_size_ratio(), 5); + assert_eq!(core.sort_spill_buffer_size(), 8 * 1024 * 1024); + assert_eq!(core.local_sort_max_num_file_handles(), 16); + assert_eq!(core.write_buffer_spill_max_disk_size(), 1024 * 1024 * 1024); } #[test] diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 1e020effb..51ea1508a 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -18,9 +18,9 @@ use crate::spec::core_options::{ first_row_supports_changelog_producer, ChangelogProducer, CoreOptions, MergeEngine, BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, - CHANGELOG_PRODUCER_OPTION, INDEX_FILE_IN_DATA_FILE_DIR_OPTION, POSTPONE_BUCKET, - QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, - TABLE_TYPE_OPTION, + CHANGELOG_PRODUCER_OPTION, INDEX_FILE_IN_DATA_FILE_DIR_OPTION, MANIFEST_SORT_ENABLED_OPTION, + MANIFEST_SORT_PARTITION_FIELD_OPTION, POSTPONE_BUCKET, QUERY_AUTH_ENABLED_OPTION, + SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, TABLE_TYPE_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType}; use crate::spec::{ @@ -1194,6 +1194,7 @@ impl Schema { Self::validate_rowkind_field(options, primary_keys, fields)?; Self::validate_deletion_vectors(options)?; Self::validate_bucket_keys(options, fields, partition_keys, primary_keys)?; + Self::validate_manifest_sort(options, fields, partition_keys)?; Self::validate_sequence_field(options, fields, partition_keys, primary_keys)?; Self::validate_read_batch_size(options)?; Self::validate_primary_key_vector_index(fields, primary_keys, options)?; @@ -1777,6 +1778,69 @@ impl Schema { Ok(()) } + fn validate_manifest_sort( + options: &HashMap, + fields: &[DataField], + partition_keys: &[String], + ) -> crate::Result<()> { + let core_options = CoreOptions::new(options); + if !core_options.manifest_sort_enabled() { + return Ok(()); + } + if partition_keys.is_empty() { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Cannot enable '{MANIFEST_SORT_ENABLED_OPTION}' for non-partition table." + ), + }); + } + let field_name = core_options + .manifest_sort_partition_field() + .unwrap_or(&partition_keys[0]); + if !partition_keys + .iter() + .any(|partition| partition == field_name) + { + return Err(crate::Error::ConfigInvalid { + message: format!( + "'{MANIFEST_SORT_PARTITION_FIELD_OPTION}' = '{field_name}' is not a partition \ + field. Available partition fields: {partition_keys:?}." + ), + }); + } + let field = fields + .iter() + .find(|field| field.name() == field_name) + .expect("partition field existence is validated before manifest sort"); + if !matches!( + field.data_type(), + DataType::Boolean(_) + | DataType::TinyInt(_) + | DataType::SmallInt(_) + | DataType::Int(_) + | DataType::BigInt(_) + | DataType::Float(_) + | DataType::Double(_) + | DataType::Char(_) + | DataType::VarChar(_) + | DataType::Date(_) + | DataType::Time(_) + | DataType::Timestamp(_) + | DataType::LocalZonedTimestamp(_) + | DataType::Decimal(_) + | DataType::Binary(_) + | DataType::VarBinary(_) + ) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Partition field '{field_name}' with type {:?} cannot be used for manifest sorting.", + field.data_type() + ), + }); + } + Ok(()) + } + fn options_error_to_config_invalid(error: crate::Error) -> crate::Error { match error { crate::Error::Unsupported { message } => crate::Error::ConfigInvalid { message }, @@ -3452,6 +3516,33 @@ mod tests { ); } + #[test] + fn test_manifest_sort_requires_partition_and_existing_sort_field() { + let non_partitioned = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option(MANIFEST_SORT_ENABLED_OPTION, "true") + .build(); + assert_config_invalid(non_partitioned, "non-partition table"); + + let invalid_field = Schema::builder() + .column("pt", DataType::Int(IntType::new())) + .column("id", DataType::Int(IntType::new())) + .partition_keys(["pt"]) + .option(MANIFEST_SORT_ENABLED_OPTION, "true") + .option(MANIFEST_SORT_PARTITION_FIELD_OPTION, "id") + .build(); + assert_config_invalid(invalid_field, "is not a partition field"); + + let valid = Schema::builder() + .column("pt", DataType::Int(IntType::new())) + .partition_keys(["pt"]) + .option(MANIFEST_SORT_ENABLED_OPTION, "true") + .option(MANIFEST_SORT_PARTITION_FIELD_OPTION, "pt") + .build() + .unwrap(); + assert_eq!(valid.partition_keys(), &["pt".to_string()]); + } + fn cast_test_schema(options: &[(&str, &str)]) -> TableSchema { let mut builder = Schema::builder() .column("a", DataType::Int(IntType::new())) diff --git a/crates/paimon/src/table/manifest_sort.rs b/crates/paimon/src/table/manifest_sort.rs new file mode 100644 index 000000000..329204932 --- /dev/null +++ b/crates/paimon/src/table/manifest_sort.rs @@ -0,0 +1,1113 @@ +// 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. + +//! Planning primitives for commit-time manifest sorting. +//! +//! This mirrors the partition-sort foundation in Java Paimon's +//! `ManifestFileSorter`: manifests are organized into non-overlapping sorted +//! runs, runs are selected with the universal-compaction size strategy, and +//! selected files are split into independently rewritable overlap sections. + +use crate::spec::{ + datum_cmp, extract_datum, BinaryRow, CoreOptions, DataField, Datum, FileKind, ManifestEntry, + ManifestFileMeta, +}; +use crate::Result; +use std::cmp::Ordering; +use std::collections::{BinaryHeap, HashMap, HashSet}; +use std::fs::{File, OpenOptions}; +use std::io::{BufReader, BufWriter, Read, Write}; +use std::path::{Path, PathBuf}; + +pub(crate) const MAX_SORTED_RUN_LEVEL: i32 = 4; + +#[derive(Debug, Clone)] +pub(crate) struct ManifestSortConfig { + pub(crate) target_size: i64, + pub(crate) merge_min_count: usize, + pub(crate) full_compaction_threshold_size: i64, + pub(crate) max_rewrite_size: i64, + pub(crate) max_size_amplification_percent: i32, + pub(crate) sorted_run_size_ratio: i32, + pub(crate) spill_buffer_size: usize, + pub(crate) spill_max_disk_size: u64, + pub(crate) max_file_handles: usize, +} + +impl ManifestSortConfig { + pub(crate) fn from_options(options: &CoreOptions<'_>) -> Self { + Self { + target_size: options.manifest_target_size().max(1), + merge_min_count: options.manifest_merge_min_count(), + full_compaction_threshold_size: options.manifest_full_compaction_threshold_size(), + max_rewrite_size: options.manifest_sort_max_rewrite_size(), + max_size_amplification_percent: options.compaction_max_size_amplification_percent(), + sorted_run_size_ratio: options.compaction_size_ratio(), + spill_buffer_size: usize::try_from(options.sort_spill_buffer_size()) + .unwrap_or(usize::MAX) + .max(1), + spill_max_disk_size: u64::try_from(options.write_buffer_spill_max_disk_size()) + .unwrap_or(u64::MAX), + max_file_handles: options.local_sort_max_num_file_handles().max(2), + } + } +} + +#[derive(Debug, Clone)] +pub(crate) struct PartitionSortKey { + field_index: usize, + data_type: crate::spec::DataType, +} + +impl PartitionSortKey { + pub(crate) fn new( + partition_fields: &[DataField], + configured_field: Option<&str>, + ) -> Result { + let field_index = match configured_field { + Some(name) => partition_fields + .iter() + .position(|field| field.name() == name) + .ok_or_else(|| crate::Error::ConfigInvalid { + message: format!( + "Cannot resolve manifest sort partition field '{name}' from {:?}", + partition_fields + .iter() + .map(|field| field.name()) + .collect::>() + ), + })?, + None => { + if partition_fields.is_empty() { + return Err(crate::Error::ConfigInvalid { + message: "Cannot enable manifest-sort.enabled for non-partition table." + .to_string(), + }); + } + 0 + } + }; + Ok(Self { + field_index, + data_type: partition_fields[field_index].data_type().clone(), + }) + } + + pub(crate) fn entry_key(&self, entry: &ManifestEntry) -> Result> { + self.key_from_serialized_row(entry.partition()) + } + + pub(crate) fn range(&self, meta: ManifestFileMeta) -> Result { + let min = self.key_from_serialized_row(meta.partition_stats().min_values())?; + let max = self.key_from_serialized_row(meta.partition_stats().max_values())?; + Ok(ManifestRange { meta, min, max }) + } + + fn key_from_serialized_row(&self, bytes: &[u8]) -> Result> { + let row = BinaryRow::from_serialized_bytes(bytes)?; + if self.field_index >= row.arity() as usize { + return Err(crate::Error::DataInvalid { + message: format!( + "Manifest sort field index {} is outside partition row arity {}", + self.field_index, + row.arity() + ), + source: None, + }); + } + extract_datum(&row, self.field_index, &self.data_type) + } +} + +#[derive(Debug, Clone)] +pub(crate) struct ManifestRange { + pub(crate) meta: ManifestFileMeta, + min: Option, + max: Option, +} + +#[derive(Debug, Clone)] +pub(crate) struct ManifestAdjacentSortedRun { + pub(crate) level: i32, + pub(crate) files: Vec, + pub(crate) total_size: i64, +} + +impl ManifestAdjacentSortedRun { + fn from_sorted(files: Vec) -> Self { + let total_size = files.iter().map(|file| file.meta.file_size()).sum(); + Self { + level: -1, + files, + total_size, + } + } +} + +#[derive(Debug, Clone)] +pub(crate) struct ManifestSection { + pub(crate) files: Vec, + pub(crate) total_size: i64, + pub(crate) has_default_compaction_file: bool, +} + +impl ManifestSection { + fn merge(mut left: Self, right: Self) -> Self { + left.files.extend(right.files); + left.total_size += right.total_size; + left.has_default_compaction_file |= right.has_default_compaction_file; + left + } +} + +#[derive(Debug)] +pub(crate) struct ManifestRewritePlan { + pub(crate) picked_file_names: HashSet, + pub(crate) sections: Vec, +} + +pub(crate) fn compare_optional_datums(left: &Option, right: &Option) -> Ordering { + match (left, right) { + (None, None) => Ordering::Equal, + (None, Some(_)) => Ordering::Less, + (Some(_), None) => Ordering::Greater, + // Java's Float.compare / Double.compare canonicalize NaNs and put + // -0.0 before +0.0; Rust's partial_cmp has no order for NaN. + (Some(Datum::Float(left)), Some(Datum::Float(right))) => { + canonical_float(*left).total_cmp(&canonical_float(*right)) + } + (Some(Datum::Double(left)), Some(Datum::Double(right))) => { + canonical_double(*left).total_cmp(&canonical_double(*right)) + } + (Some(left), Some(right)) => { + datum_cmp(left, right).expect("validated manifest sort field must have an ordering") + } + } +} + +fn canonical_float(value: f32) -> f32 { + if value.is_nan() { + f32::NAN + } else { + value + } +} + +fn canonical_double(value: f64) -> f64 { + if value.is_nan() { + f64::NAN + } else { + value + } +} + +#[derive(Debug)] +struct SortRecord { + key: Option, + encoded: Vec, + entry: ManifestEntry, +} + +impl SortRecord { + fn new(entry: ManifestEntry, sort_key: &PartitionSortKey) -> Result { + let key = sort_key.entry_key(&entry)?; + let encoded = + serde_json::to_vec(&entry).map_err(|error| crate::Error::UnexpectedError { + message: "Failed to encode manifest entry for external sort".to_string(), + source: Some(Box::new(error)), + })?; + Ok(Self { + key, + encoded, + entry, + }) + } + + fn from_encoded(encoded: Vec, sort_key: &PartitionSortKey) -> Result { + let entry = + serde_json::from_slice(&encoded).map_err(|error| crate::Error::UnexpectedError { + message: "Failed to decode spilled manifest entry".to_string(), + source: Some(Box::new(error)), + })?; + let key = sort_key.entry_key(&entry)?; + Ok(Self { + key, + encoded, + entry, + }) + } +} + +fn compare_sort_records(left: &SortRecord, right: &SortRecord) -> Ordering { + compare_optional_datums(&left.key, &right.key) + .then_with(|| file_kind_order(left.entry.kind()).cmp(&file_kind_order(right.entry.kind()))) + .then_with(|| { + left.entry + .file() + .file_name + .cmp(&right.entry.file().file_name) + }) +} + +#[derive(Debug)] +struct SpillRun { + path: PathBuf, + bytes: u64, +} + +/// Bounded manifest-entry sorter. +/// +/// Entries are sorted in memory up to `spill_buffer_size`, then written as +/// length-prefixed JSON records under an owned temporary directory. The JSON +/// representation is an internal spill format only; persisted Paimon +/// manifests remain Avro. Runs are merged until the final fan-in is bounded by +/// `max_file_handles`. +pub(crate) struct SpillableManifestSorter { + sort_key: PartitionSortKey, + buffer: Vec, + buffer_bytes: usize, + spill_buffer_size: usize, + spill_max_disk_size: u64, + max_file_handles: usize, + temp_dir: tempfile::TempDir, + runs: Vec, + disk_bytes: u64, + next_run_id: u64, +} + +impl SpillableManifestSorter { + pub(crate) fn new(sort_key: &PartitionSortKey, config: &ManifestSortConfig) -> Result { + let temp_dir = tempfile::tempdir().map_err(|error| crate::Error::UnexpectedError { + message: "Failed to create manifest-sort spill directory".to_string(), + source: Some(Box::new(error)), + })?; + Ok(Self { + sort_key: sort_key.clone(), + buffer: Vec::new(), + buffer_bytes: 0, + spill_buffer_size: config.spill_buffer_size.max(1), + spill_max_disk_size: config.spill_max_disk_size, + max_file_handles: config.max_file_handles.max(2), + temp_dir, + runs: Vec::new(), + disk_bytes: 0, + next_run_id: 0, + }) + } + + pub(crate) fn push(&mut self, entry: ManifestEntry) -> Result<()> { + let record = SortRecord::new(entry, &self.sort_key)?; + self.buffer_bytes = self + .buffer_bytes + .saturating_add(record.encoded.len().saturating_add(8)); + self.buffer.push(record); + if self.buffer_bytes >= self.spill_buffer_size { + self.spill_buffer()?; + } + Ok(()) + } + + pub(crate) fn finish(mut self) -> Result { + if self.runs.is_empty() { + self.buffer.sort_by(compare_sort_records); + return Ok(SortedManifestEntries::Memory( + self.buffer + .into_iter() + .map(|record| record.entry) + .collect::>() + .into_iter(), + )); + } + self.spill_buffer()?; + self.reduce_run_count()?; + SpilledManifestEntries::new(self.temp_dir, self.runs, self.sort_key) + .map(SortedManifestEntries::Spilled) + } + + fn spill_buffer(&mut self) -> Result<()> { + if self.buffer.is_empty() { + return Ok(()); + } + self.buffer.sort_by(compare_sort_records); + let path = self.next_run_path(); + let records = std::mem::take(&mut self.buffer); + self.buffer_bytes = 0; + let run = self.write_run(&path, records.into_iter())?; + self.runs.push(run); + Ok(()) + } + + fn reduce_run_count(&mut self) -> Result<()> { + while self.runs.len() > self.max_file_handles { + let old_runs = std::mem::take(&mut self.runs); + let mut merged = Vec::new(); + let mut iter = old_runs.into_iter(); + loop { + let group = iter + .by_ref() + .take(self.max_file_handles) + .collect::>(); + if group.is_empty() { + break; + } + if group.len() == 1 { + merged.push(group.into_iter().next().expect("one run")); + continue; + } + let path = self.next_run_path(); + let run = self.merge_runs(&path, &group)?; + for input in group { + std::fs::remove_file(&input.path).map_err(|error| { + crate::Error::UnexpectedError { + message: format!( + "Failed to remove merged manifest-sort spill run '{}'", + input.path.display() + ), + source: Some(Box::new(error)), + } + })?; + self.disk_bytes = self.disk_bytes.saturating_sub(input.bytes); + } + merged.push(run); + } + self.runs = merged; + } + Ok(()) + } + + fn merge_runs(&mut self, path: &Path, runs: &[SpillRun]) -> Result { + let mut readers = runs + .iter() + .map(|run| SpillRunReader::open(&run.path)) + .collect::>>()?; + let mut heap = BinaryHeap::new(); + for (source, reader) in readers.iter_mut().enumerate() { + if let Some(record) = reader.next_record(&self.sort_key)? { + heap.push(HeapRecord { record, source }); + } + } + + let sort_key = self.sort_key.clone(); + let records = std::iter::from_fn(|| { + let item = heap.pop()?; + let source = item.source; + let next = readers[source].next_record(&sort_key); + match next { + Ok(Some(record)) => heap.push(HeapRecord { record, source }), + Ok(None) => {} + Err(error) => return Some(Err(error)), + } + Some(Ok(item.record)) + }); + self.write_run_results(path, records) + } + + fn write_run(&mut self, path: &Path, records: I) -> Result + where + I: Iterator, + { + self.write_run_results(path, records.map(Ok)) + } + + fn write_run_results(&mut self, path: &Path, records: I) -> Result + where + I: Iterator>, + { + let file = OpenOptions::new() + .create_new(true) + .write(true) + .open(path) + .map_err(|error| crate::Error::UnexpectedError { + message: format!( + "Failed to create manifest-sort spill run '{}'", + path.display() + ), + source: Some(Box::new(error)), + })?; + let mut writer = BufWriter::new(file); + let mut written = 0u64; + for record in records { + let record = record?; + let encoded_len = + u64::try_from(record.encoded.len()).map_err(|_| crate::Error::UnexpectedError { + message: "Manifest-sort spill record length overflows u64".to_string(), + source: None, + })?; + let record_size = 8u64.saturating_add(encoded_len); + if self + .disk_bytes + .saturating_add(written) + .saturating_add(record_size) + > self.spill_max_disk_size + { + return Err(crate::Error::UnexpectedError { + message: format!( + "Manifest sort exceeds local spill disk limit of {} bytes", + self.spill_max_disk_size + ), + source: None, + }); + } + writer + .write_all(&encoded_len.to_le_bytes()) + .and_then(|_| writer.write_all(&record.encoded)) + .map_err(|error| crate::Error::UnexpectedError { + message: format!( + "Failed to write manifest-sort spill run '{}'", + path.display() + ), + source: Some(Box::new(error)), + })?; + written = written.saturating_add(record_size); + } + writer + .flush() + .map_err(|error| crate::Error::UnexpectedError { + message: format!( + "Failed to flush manifest-sort spill run '{}'", + path.display() + ), + source: Some(Box::new(error)), + })?; + self.disk_bytes = self.disk_bytes.saturating_add(written); + Ok(SpillRun { + path: path.to_path_buf(), + bytes: written, + }) + } + + fn next_run_path(&mut self) -> PathBuf { + let path = self + .temp_dir + .path() + .join(format!("manifest-sort-run-{}", self.next_run_id)); + self.next_run_id += 1; + path + } +} + +pub(crate) enum SortedManifestEntries { + Memory(std::vec::IntoIter), + Spilled(SpilledManifestEntries), +} + +impl SortedManifestEntries { + pub(crate) fn next_entry(&mut self) -> Result> { + match self { + Self::Memory(entries) => Ok(entries.next()), + Self::Spilled(entries) => entries.next_entry(), + } + } +} + +pub(crate) struct SpilledManifestEntries { + _temp_dir: tempfile::TempDir, + readers: Vec, + heap: BinaryHeap, + sort_key: PartitionSortKey, +} + +impl SpilledManifestEntries { + fn new( + temp_dir: tempfile::TempDir, + runs: Vec, + sort_key: PartitionSortKey, + ) -> Result { + let mut readers = runs + .iter() + .map(|run| SpillRunReader::open(&run.path)) + .collect::>>()?; + let mut heap = BinaryHeap::new(); + for (source, reader) in readers.iter_mut().enumerate() { + if let Some(record) = reader.next_record(&sort_key)? { + heap.push(HeapRecord { record, source }); + } + } + Ok(Self { + _temp_dir: temp_dir, + readers, + heap, + sort_key, + }) + } + + fn next_entry(&mut self) -> Result> { + let Some(item) = self.heap.pop() else { + return Ok(None); + }; + if let Some(record) = self.readers[item.source].next_record(&self.sort_key)? { + self.heap.push(HeapRecord { + record, + source: item.source, + }); + } + Ok(Some(item.record.entry)) + } +} + +struct SpillRunReader { + reader: BufReader, +} + +impl SpillRunReader { + fn open(path: &Path) -> Result { + let file = File::open(path).map_err(|error| crate::Error::UnexpectedError { + message: format!( + "Failed to open manifest-sort spill run '{}'", + path.display() + ), + source: Some(Box::new(error)), + })?; + Ok(Self { + reader: BufReader::new(file), + }) + } + + fn next_record(&mut self, sort_key: &PartitionSortKey) -> Result> { + let mut length = [0u8; 8]; + let first = + self.reader + .read(&mut length[..1]) + .map_err(|error| crate::Error::UnexpectedError { + message: "Failed to read manifest-sort spill record length".to_string(), + source: Some(Box::new(error)), + })?; + if first == 0 { + return Ok(None); + } + self.reader.read_exact(&mut length[1..]).map_err(|error| { + crate::Error::UnexpectedError { + message: "Truncated manifest-sort spill record length".to_string(), + source: Some(Box::new(error)), + } + })?; + let length = usize::try_from(u64::from_le_bytes(length)).map_err(|_| { + crate::Error::UnexpectedError { + message: "Manifest-sort spill record length overflows usize".to_string(), + source: None, + } + })?; + let mut encoded = vec![0u8; length]; + self.reader + .read_exact(&mut encoded) + .map_err(|error| crate::Error::UnexpectedError { + message: "Truncated manifest-sort spill record".to_string(), + source: Some(Box::new(error)), + })?; + SortRecord::from_encoded(encoded, sort_key).map(Some) + } +} + +struct HeapRecord { + record: SortRecord, + source: usize, +} + +impl PartialEq for HeapRecord { + fn eq(&self, other: &Self) -> bool { + compare_sort_records(&self.record, &other.record) == Ordering::Equal + && self.source == other.source + } +} + +impl Eq for HeapRecord {} + +impl PartialOrd for HeapRecord { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for HeapRecord { + fn cmp(&self, other: &Self) -> Ordering { + compare_sort_records(&other.record, &self.record) + .then_with(|| other.source.cmp(&self.source)) + } +} + +pub(crate) fn reaches_full_compaction_threshold( + input: &[ManifestFileMeta], + target_size: i64, + threshold_size: i64, +) -> bool { + input + .iter() + .filter(|file| file.num_deleted_files() > 0 || file.file_size() < target_size) + .map(ManifestFileMeta::file_size) + .sum::() + >= threshold_size +} + +/// Return whether a concrete partition may be covered by a manifest's +/// partition statistics. +/// +/// Missing or legacy statistics are treated conservatively as overlapping so +/// full compaction never drops a DELETE while retaining its ADD in an +/// unre-written manifest. +pub(crate) fn manifest_may_contain_partition( + meta: &ManifestFileMeta, + partition: &[u8], + partition_fields: &[DataField], +) -> Result { + if partition_fields.is_empty() { + return Ok(true); + } + let partition = BinaryRow::from_serialized_bytes(partition)?; + let min = match BinaryRow::from_serialized_bytes(meta.partition_stats().min_values()) { + Ok(row) if row.arity() as usize >= partition_fields.len() => row, + _ => return Ok(true), + }; + let max = match BinaryRow::from_serialized_bytes(meta.partition_stats().max_values()) { + Ok(row) if row.arity() as usize >= partition_fields.len() => row, + _ => return Ok(true), + }; + + for (index, field) in partition_fields.iter().enumerate() { + let point = extract_datum(&partition, index, field.data_type())?; + let min_value = extract_datum(&min, index, field.data_type())?; + let max_value = extract_datum(&max, index, field.data_type())?; + match point { + None => match meta.partition_stats().null_counts().get(index) { + Some(Some(count)) if *count == 0 => return Ok(false), + _ => {} + }, + Some(point) => match (min_value, max_value) { + (Some(min_value), Some(max_value)) => { + if datum_cmp(&point, &min_value) == Some(Ordering::Less) + || datum_cmp(&point, &max_value) == Some(Ordering::Greater) + { + return Ok(false); + } + } + _ => match meta.partition_stats().null_counts().get(index) { + // Only a known all-null field can exclude a non-null value. + Some(Some(count)) + if *count == meta.num_added_files() + meta.num_deleted_files() => + { + return Ok(false) + } + // Unknown legacy stats must not be used for exclusion. + _ => {} + }, + }, + } + } + Ok(true) +} + +pub(crate) fn plan_rewrite( + input: &[ManifestFileMeta], + default_compaction: &HashMap, + sort_key: &PartitionSortKey, + config: &ManifestSortConfig, +) -> Result> { + let lsm_files = input + .iter() + .filter(|file| !default_compaction.contains_key(file.file_name())) + .cloned() + .map(|file| sort_key.range(file)) + .collect::>>()?; + let level_runs = build_level_sorted_runs(lsm_files); + let picked_runs = ManifestPickStrategy::new( + config.max_size_amplification_percent, + config.sorted_run_size_ratio, + ) + .pick(&level_runs); + + let mut picked_file_names = HashSet::new(); + let mut picked_files = Vec::new(); + for run_index in picked_runs { + for file in &level_runs[run_index].files { + picked_file_names.insert(file.meta.file_name().to_string()); + picked_files.push(file.clone()); + } + } + for file in input { + if default_compaction.contains_key(file.file_name()) { + picked_file_names.insert(file.file_name().to_string()); + picked_files.push(sort_key.range(file.clone())?); + } + } + if picked_files.is_empty() { + return Ok(None); + } + + let sections = merge_small_adjacent_sections( + split_into_sections(picked_files, default_compaction), + config.target_size, + ); + Ok(Some(ManifestRewritePlan { + picked_file_names, + sections, + })) +} + +fn build_level_sorted_runs(mut input: Vec) -> Vec { + input.sort_by(compare_ranges); + let mut runs: Vec> = Vec::new(); + for file in input { + let earliest = runs + .iter() + .enumerate() + .min_by(|(_, left), (_, right)| { + compare_optional_datums( + &left.last().expect("non-empty run").max, + &right.last().expect("non-empty run").max, + ) + }) + .map(|(index, _)| index); + match earliest { + Some(index) + if compare_optional_datums( + &file.min, + &runs[index].last().expect("non-empty run").max, + ) != Ordering::Less => + { + runs[index].push(file); + } + _ => runs.push(vec![file]), + } + } + + let mut result = runs + .into_iter() + .map(ManifestAdjacentSortedRun::from_sorted) + .collect::>(); + result.sort_by_key(|run| run.total_size); + let first_leveled = result.len().saturating_sub(MAX_SORTED_RUN_LEVEL as usize); + for (index, run) in result.iter_mut().enumerate() { + run.level = if index >= first_leveled { + (index - first_leveled + 1) as i32 + } else { + 0 + }; + } + result +} + +fn split_into_sections( + mut picked_files: Vec, + default_compaction: &HashMap, +) -> Vec { + picked_files.sort_by(compare_ranges); + let mut sections = Vec::new(); + let mut current_files = Vec::new(); + let mut current_size = 0; + let mut current_has_default = false; + let mut current_max: Option> = None; + + for file in picked_files { + let starts_new = current_max + .as_ref() + .is_some_and(|max| compare_optional_datums(&file.min, max) != Ordering::Less); + if starts_new { + sections.push(ManifestSection { + files: current_files, + total_size: current_size, + has_default_compaction_file: current_has_default, + }); + current_files = Vec::new(); + current_size = 0; + current_has_default = false; + current_max = None; + } + + current_size += file.meta.file_size(); + current_has_default |= default_compaction.contains_key(file.meta.file_name()); + if current_max + .as_ref() + .is_none_or(|max| compare_optional_datums(&file.max, max) == Ordering::Greater) + { + current_max = Some(file.max.clone()); + } + current_files.push(file.meta); + } + + if !current_files.is_empty() { + sections.push(ManifestSection { + files: current_files, + total_size: current_size, + has_default_compaction_file: current_has_default, + }); + } + sections +} + +fn merge_small_adjacent_sections( + sections: Vec, + target_size: i64, +) -> Vec { + let mut result = Vec::new(); + let mut pending: Option = None; + for section in sections { + pending = match pending { + None => Some(section), + Some(previous) + if previous.total_size < target_size || section.total_size < target_size => + { + Some(ManifestSection::merge(previous, section)) + } + Some(previous) => { + result.push(previous); + Some(section) + } + }; + } + if let Some(pending) = pending { + result.push(pending); + } + result +} + +struct ManifestPickStrategy { + size_amplification_threshold: i32, + size_ratio_threshold: i32, +} + +impl ManifestPickStrategy { + fn new(size_amplification_threshold: i32, size_ratio_threshold: i32) -> Self { + Self { + size_amplification_threshold, + size_ratio_threshold, + } + } + + /// Return indexes into `level_runs`. + fn pick(&self, level_runs: &[ManifestAdjacentSortedRun]) -> Vec { + if level_runs.len() <= MAX_SORTED_RUN_LEVEL as usize { + return Vec::new(); + } + + let highest = level_runs.last().expect("non-empty runs"); + let lower_total = level_runs + .iter() + .filter(|run| run.level < highest.level) + .map(|run| run.total_size) + .sum::(); + if highest.level > 0 + && lower_total.saturating_mul(100) + > i64::from(self.size_amplification_threshold).saturating_mul(highest.total_size) + { + return (0..level_runs.len()).collect(); + } + + let mut picked = vec![0]; + let mut picked_size = level_runs[0].total_size; + for (index, run) in level_runs.iter().enumerate().skip(1) { + if run.level <= 1 + || picked_size.saturating_mul(i64::from(100 + self.size_ratio_threshold)) + >= run.total_size.saturating_mul(100) + { + picked.push(index); + picked_size += run.total_size; + } + } + if picked.len() == 1 { + Vec::new() + } else { + picked + } + } +} + +fn compare_ranges(left: &ManifestRange, right: &ManifestRange) -> Ordering { + compare_optional_datums(&left.min, &right.min) + .then_with(|| compare_optional_datums(&left.max, &right.max)) + .then_with(|| left.meta.file_name().cmp(right.meta.file_name())) +} + +fn file_kind_order(kind: &FileKind) -> u8 { + match kind { + FileKind::Add => 0, + FileKind::Delete => 1, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::spec::stats::BinaryTableStats; + use crate::spec::{BinaryRowBuilder, DataFileMeta, DataType, IntType}; + + fn row(value: i32) -> Vec { + let mut builder = BinaryRowBuilder::new(1); + builder.write_int(0, value); + builder.build().to_serialized_bytes() + } + + fn meta(name: &str, size: i64, min: i32, max: i32) -> ManifestFileMeta { + ManifestFileMeta::new( + name.to_string(), + size, + 1, + 0, + BinaryTableStats::new(row(min), row(max), vec![Some(0)]), + 0, + ) + } + + fn sort_key() -> PartitionSortKey { + PartitionSortKey::new( + &[DataField::new( + 0, + "pt".to_string(), + DataType::Int(IntType::new()), + )], + None, + ) + .unwrap() + } + + fn entry(kind: FileKind, partition: i32, file_name: &str) -> ManifestEntry { + ManifestEntry::new( + kind, + row(partition), + 0, + 1, + DataFileMeta { + file_name: file_name.to_string(), + file_size: 1, + row_count: 1, + min_key: Vec::new(), + max_key: Vec::new(), + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 0, + schema_id: 0, + level: 0, + extra_files: Vec::new(), + creation_time: None, + delete_row_count: None, + embedded_index: None, + file_source: None, + value_stats_cols: None, + external_path: None, + first_row_id: None, + write_cols: None, + column_max_sequence_numbers: None, + }, + 2, + ) + } + + fn spill_config( + spill_buffer_size: usize, + max_file_handles: usize, + spill_max_disk_size: u64, + ) -> ManifestSortConfig { + ManifestSortConfig { + target_size: 1024, + merge_min_count: 2, + full_compaction_threshold_size: 1024, + max_rewrite_size: 1024, + max_size_amplification_percent: 200, + sorted_run_size_ratio: 1, + spill_buffer_size, + spill_max_disk_size, + max_file_handles, + } + } + + #[test] + fn boundary_equality_shares_run_but_splits_sections() { + let key = sort_key(); + let ranges = vec![ + key.range(meta("a", 100, 0, 10)).unwrap(), + key.range(meta("b", 100, 10, 20)).unwrap(), + ]; + let runs = build_level_sorted_runs(ranges.clone()); + assert_eq!(runs.len(), 1); + assert_eq!(runs[0].files.len(), 2); + + let sections = split_into_sections(ranges, &HashMap::new()); + assert_eq!(sections.len(), 2); + } + + #[test] + fn pick_strategy_forces_low_levels_once_run_count_exceeds_four() { + let key = sort_key(); + let files = [100, 200, 1_000, 2_000, 4_000] + .into_iter() + .enumerate() + .map(|(index, size)| key.range(meta(&format!("m-{index}"), size, 0, 10)).unwrap()) + .collect(); + let runs = build_level_sorted_runs(files); + assert_eq!(runs.len(), 5); + let picked = ManifestPickStrategy::new(1_000, 1).pick(&runs); + assert_eq!(picked, vec![0, 1]); + } + + #[test] + fn small_adjacent_sections_are_merged() { + let sections = vec![ + ManifestSection { + files: vec![meta("a", 40, 0, 1)], + total_size: 40, + has_default_compaction_file: false, + }, + ManifestSection { + files: vec![meta("b", 200, 2, 3)], + total_size: 200, + has_default_compaction_file: true, + }, + ]; + let merged = merge_small_adjacent_sections(sections, 100); + assert_eq!(merged.len(), 1); + assert_eq!(merged[0].files.len(), 2); + assert!(merged[0].has_default_compaction_file); + } + + #[test] + fn spill_sort_is_bounded_and_merges_runs_in_order() { + let key = sort_key(); + let mut sorter = SpillableManifestSorter::new(&key, &spill_config(1, 2, u64::MAX)).unwrap(); + sorter.push(entry(FileKind::Delete, 1, "b")).unwrap(); + sorter.push(entry(FileKind::Add, 2, "c")).unwrap(); + sorter.push(entry(FileKind::Add, 1, "a")).unwrap(); + sorter.push(entry(FileKind::Add, 1, "b")).unwrap(); + + let mut sorted = sorter.finish().unwrap(); + let mut actual = Vec::new(); + while let Some(entry) = sorted.next_entry().unwrap() { + actual.push(( + entry.partition().to_vec(), + *entry.kind(), + entry.file().file_name.clone(), + )); + } + assert_eq!( + actual, + vec![ + (row(1), FileKind::Add, "a".to_string()), + (row(1), FileKind::Add, "b".to_string()), + (row(1), FileKind::Delete, "b".to_string()), + (row(2), FileKind::Add, "c".to_string()), + ] + ); + } + + #[test] + fn spill_sort_enforces_disk_limit() { + let key = sort_key(); + let mut sorter = SpillableManifestSorter::new(&key, &spill_config(1, 2, 1)).unwrap(); + let error = sorter + .push(entry(FileKind::Add, 1, "too-large")) + .unwrap_err(); + assert!(error.to_string().contains("spill disk limit")); + } +} diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index f46c835f2..536e9b3b2 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -63,6 +63,7 @@ pub(crate) mod index_file_path; mod kv_file_reader; mod kv_file_writer; mod lumina_index_build_builder; +mod manifest_sort; pub(crate) mod merge_tree_split_generator; mod object_table; mod partition_filter; diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 676d53587..c2a832ecb 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -33,6 +33,11 @@ use crate::spec::{ use crate::table::commit_message::CommitMessage; use crate::table::global_index_build_common::same_extra_field_ids; use crate::table::index_file_path::committed_index_file_path; +use crate::table::manifest_sort::{ + manifest_may_contain_partition, plan_rewrite, reaches_full_compaction_threshold, + ManifestRewritePlan, ManifestSection, ManifestSortConfig, PartitionSortKey, + SpillableManifestSorter, +}; use crate::table::partition_filter::PartitionFilter; use crate::table::snapshot_commit::SnapshotCommit; use crate::table::{SnapshotManager, Table, TableScan}; @@ -116,6 +121,9 @@ pub struct TableCommit { manifest_compression: String, manifest_target_size: i64, manifest_merge_min_count: usize, + manifest_sort_enabled: bool, + manifest_sort_partition_field: Option, + manifest_sort_config: ManifestSortConfig, row_tracking_enabled: bool, data_evolution_enabled: bool, partition_default_name: String, @@ -140,6 +148,11 @@ impl TableCommit { let manifest_compression = core_options.manifest_compression().to_string(); let manifest_target_size = core_options.manifest_target_size(); let manifest_merge_min_count = core_options.manifest_merge_min_count(); + let manifest_sort_enabled = core_options.manifest_sort_enabled(); + let manifest_sort_partition_field = core_options + .manifest_sort_partition_field() + .map(str::to_string); + let manifest_sort_config = ManifestSortConfig::from_options(&core_options); let row_tracking_enabled = core_options.row_tracking_enabled(); let data_evolution_enabled = core_options.data_evolution_enabled(); let partition_default_name = core_options.partition_default_name().to_string(); @@ -156,6 +169,9 @@ impl TableCommit { manifest_compression, manifest_target_size, manifest_merge_min_count, + manifest_sort_enabled, + manifest_sort_partition_field, + manifest_sort_config, row_tracking_enabled, data_evolution_enabled, partition_default_name, @@ -1129,6 +1145,11 @@ impl TableCommit { if manifest_files.is_empty() { return Ok((vec![], vec![])); } + if self.manifest_sort_enabled { + return self + .sort_manifest_files(file_io, manifest_dir, manifest_files) + .await; + } let target_size = self.manifest_target_size.max(1); let mut result = Vec::new(); @@ -1168,6 +1189,555 @@ impl TableCommit { Ok((result, new_files)) } + async fn sort_manifest_files( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: Vec, + ) -> Result<(Vec, Vec)> { + let partition_fields = self.table.schema().partition_fields(); + let sort_key = PartitionSortKey::new( + &partition_fields, + self.manifest_sort_partition_field.as_deref(), + )?; + let mut new_files = Vec::new(); + let result = self + .sort_manifest_files_inner( + file_io, + manifest_dir, + &manifest_files, + &partition_fields, + &sort_key, + &mut new_files, + ) + .await; + if result.is_err() { + for file in &new_files { + let path = format!("{manifest_dir}/{}", file.file_name()); + let _ = file_io.delete_file(&path).await; + } + } + result.map(|files| (files, new_files)) + } + + async fn sort_manifest_files_inner( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: &[ManifestFileMeta], + partition_fields: &[crate::spec::DataField], + sort_key: &PartitionSortKey, + new_files: &mut Vec, + ) -> Result> { + if reaches_full_compaction_threshold( + manifest_files, + self.manifest_sort_config.target_size, + self.manifest_sort_config.full_compaction_threshold_size, + ) { + if let Some(files) = self + .try_full_manifest_sort( + file_io, + manifest_dir, + manifest_files, + partition_fields, + sort_key, + new_files, + ) + .await? + { + return Ok(files); + } + } + self.try_minor_manifest_sort(file_io, manifest_dir, manifest_files, sort_key, new_files) + .await + } + + async fn try_full_manifest_sort( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: &[ManifestFileMeta], + partition_fields: &[crate::spec::DataField], + sort_key: &PartitionSortKey, + new_files: &mut Vec, + ) -> Result>> { + let mut deleted_identifiers = HashSet::new(); + let mut deleted_partitions = Vec::new(); + for meta in manifest_files + .iter() + .filter(|meta| meta.num_deleted_files() > 0) + { + let path = format!("{manifest_dir}/{}", meta.file_name()); + for entry in Manifest::read(file_io, &path).await? { + if entry.kind() == &FileKind::Delete { + deleted_identifiers.insert(entry.identifier()); + deleted_partitions.push(entry.partition().to_vec()); + } + } + } + + let mut default_compaction = HashMap::new(); + for meta in manifest_files { + let small = meta.file_size() < self.manifest_sort_config.target_size; + let mut in_delete_range = false; + for partition in &deleted_partitions { + if manifest_may_contain_partition(meta, partition, partition_fields)? { + in_delete_range = true; + break; + } + } + if small || in_delete_range { + default_compaction.insert(meta.file_name().to_string(), in_delete_range); + } + } + + let Some(plan) = plan_rewrite( + manifest_files, + &default_compaction, + sort_key, + &self.manifest_sort_config, + )? + else { + return Ok(None); + }; + + let mut output = ManifestRewriteOutput::full( + manifest_files + .iter() + .filter(|file| !plan.picked_file_names.contains(file.file_name())) + .cloned() + .collect(), + ); + self.rewrite_manifest_sections( + file_io, + manifest_dir, + plan, + &default_compaction, + &deleted_identifiers, + true, + sort_key, + new_files, + &mut output, + ) + .await?; + Ok(Some(output.into_files())) + } + + async fn try_minor_manifest_sort( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: &[ManifestFileMeta], + sort_key: &PartitionSortKey, + new_files: &mut Vec, + ) -> Result> { + let default_compaction = manifest_files + .iter() + .filter(|file| file.file_size() < self.manifest_sort_config.target_size) + .map(|file| (file.file_name().to_string(), false)) + .collect::>(); + let Some(plan) = plan_rewrite( + manifest_files, + &default_compaction, + sort_key, + &self.manifest_sort_config, + )? + else { + return Ok(manifest_files.to_vec()); + }; + + let mut output = ManifestRewriteOutput::minor(manifest_files, &plan.picked_file_names)?; + for file in manifest_files + .iter() + .filter(|file| !plan.picked_file_names.contains(file.file_name())) + { + output.add_unchanged(file.clone())?; + } + self.rewrite_manifest_sections( + file_io, + manifest_dir, + plan, + &default_compaction, + &HashSet::new(), + false, + sort_key, + new_files, + &mut output, + ) + .await?; + Ok(output.into_files()) + } + + #[allow(clippy::too_many_arguments)] + async fn rewrite_manifest_sections( + &self, + file_io: &FileIO, + manifest_dir: &str, + plan: ManifestRewritePlan, + default_compaction: &HashMap, + deleted_identifiers: &HashSet, + full_compaction: bool, + sort_key: &PartitionSortKey, + new_files: &mut Vec, + output: &mut ManifestRewriteOutput, + ) -> Result<()> { + let mut sections = plan.sections; + let mut processed_size = 0i64; + let mut reached_limit = false; + let mut index = 0usize; + while index < sections.len() { + let section = sections[index].clone(); + if section.files.len() == 1 { + self.rewrite_manifest_section( + file_io, + manifest_dir, + §ion.files, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + index += 1; + continue; + } + + if processed_size.saturating_add(section.total_size) + <= self.manifest_sort_config.max_rewrite_size + { + processed_size += section.total_size; + self.rewrite_manifest_section( + file_io, + manifest_dir, + §ion.files, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + } else if !reached_limit { + let available = self + .manifest_sort_config + .max_rewrite_size + .saturating_sub(processed_size); + let mut rewrite_files = Vec::new(); + let mut remaining_files = Vec::new(); + let mut rewrite_size = 0i64; + let mut remaining_size = 0i64; + let mut remaining_has_default = false; + for file in section.files { + // The budget is a file-granularity soft limit: include the + // first file that crosses it and at least two files, even + // when the limit is smaller than the target file size. + // Otherwise repeated commits could make no progress. + if rewrite_size <= available || rewrite_files.len() < 2 { + rewrite_size += file.file_size(); + rewrite_files.push(file); + } else { + remaining_size += file.file_size(); + remaining_has_default |= default_compaction.contains_key(file.file_name()); + remaining_files.push(file); + } + } + if !rewrite_files.is_empty() { + self.rewrite_manifest_section( + file_io, + manifest_dir, + &rewrite_files, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + } + if !remaining_files.is_empty() { + sections.push(ManifestSection { + files: remaining_files, + total_size: remaining_size, + has_default_compaction_file: remaining_has_default, + }); + } + reached_limit = true; + } else if section.has_default_compaction_file { + self.rewrite_manifest_subsegments( + file_io, + manifest_dir, + §ion.files, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + } else { + output.add_all_unchanged(section.files)?; + } + index += 1; + } + Ok(()) + } + + #[allow(clippy::too_many_arguments)] + async fn rewrite_manifest_subsegments( + &self, + file_io: &FileIO, + manifest_dir: &str, + files: &[ManifestFileMeta], + default_compaction: &HashMap, + deleted_identifiers: &HashSet, + full_compaction: bool, + sort_key: &PartitionSortKey, + new_files: &mut Vec, + output: &mut ManifestRewriteOutput, + ) -> Result<()> { + let mut candidates = Vec::new(); + let mut candidate_size = 0i64; + for file in files { + candidate_size += file.file_size(); + candidates.push(file.clone()); + if candidate_size >= self.manifest_sort_config.target_size { + self.rewrite_manifest_section( + file_io, + manifest_dir, + &candidates, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + candidates.clear(); + candidate_size = 0; + } + } + if !candidates.is_empty() { + if !deleted_identifiers.is_empty() + || candidates.len() >= self.manifest_sort_config.merge_min_count + { + self.rewrite_manifest_section( + file_io, + manifest_dir, + &candidates, + default_compaction, + deleted_identifiers, + full_compaction, + sort_key, + new_files, + output, + ) + .await?; + } else { + output.add_all_unchanged(candidates)?; + } + } + Ok(()) + } + + #[allow(clippy::too_many_arguments)] + async fn rewrite_manifest_section( + &self, + file_io: &FileIO, + manifest_dir: &str, + section: &[ManifestFileMeta], + default_compaction: &HashMap, + deleted_identifiers: &HashSet, + full_compaction: bool, + sort_key: &PartitionSortKey, + new_files: &mut Vec, + output: &mut ManifestRewriteOutput, + ) -> Result<()> { + if section.is_empty() { + return Ok(()); + } + if section.len() == 1 + && !default_compaction + .get(section[0].file_name()) + .copied() + .unwrap_or(false) + { + output.add_unchanged(section[0].clone())?; + return Ok(()); + } + + if full_compaction { + let mut sorter = SpillableManifestSorter::new(sort_key, &self.manifest_sort_config)?; + for meta in section { + let path = format!("{manifest_dir}/{}", meta.file_name()); + for entry in Manifest::read(file_io, &path).await? { + if entry.kind() == &FileKind::Add + && !deleted_identifiers.contains(&entry.identifier()) + { + sorter.push(entry)?; + } + } + } + let sorted = self + .write_sorted_manifest_entries(file_io, manifest_dir, sorter, new_files) + .await?; + output.add_sorted_files(sorted)?; + return Ok(()); + } + + let mut sorter = SpillableManifestSorter::new(sort_key, &self.manifest_sort_config)?; + let mut delete_identifiers = HashSet::new(); + for meta in section { + let path = format!("{manifest_dir}/{}", meta.file_name()); + for entry in Manifest::read(file_io, &path).await? { + if entry.kind() == &FileKind::Delete { + delete_identifiers.insert(entry.identifier()); + } + sorter.push(entry)?; + } + } + let (add_files, delete_files) = self + .write_minor_sorted_manifest_entries( + file_io, + manifest_dir, + sorter, + delete_identifiers, + new_files, + ) + .await?; + output.add_sorted_files(add_files)?; + output.add_delete_files(delete_files)?; + Ok(()) + } + + async fn write_sorted_manifest_entries( + &self, + file_io: &FileIO, + manifest_dir: &str, + sorter: SpillableManifestSorter, + new_files: &mut Vec, + ) -> Result> { + let mut sorted = sorter.finish()?; + let mut chunk = Vec::new(); + let mut chunk_bytes = 0usize; + let target = usize::try_from(self.manifest_target_size.max(1)).unwrap_or(usize::MAX); + let mut result = Vec::new(); + while let Some(entry) = sorted.next_entry()? { + chunk_bytes = chunk_bytes.saturating_add(manifest_entry_sort_size(&entry)?); + chunk.push(entry); + if chunk_bytes >= target { + let files = self + .write_manifest_sort_chunk(file_io, manifest_dir, &mut chunk, new_files) + .await?; + result.extend(files); + chunk_bytes = 0; + } + } + result.extend( + self.write_manifest_sort_chunk(file_io, manifest_dir, &mut chunk, new_files) + .await?, + ); + Ok(result) + } + + async fn write_minor_sorted_manifest_entries( + &self, + file_io: &FileIO, + manifest_dir: &str, + sorter: SpillableManifestSorter, + delete_identifiers: HashSet, + new_files: &mut Vec, + ) -> Result<(Vec, Vec)> { + let mut sorted = sorter.finish()?; + let mut matched = HashSet::new(); + let mut emitted_deletes = HashSet::new(); + let mut add_chunk = Vec::new(); + let mut delete_chunk = Vec::new(); + let mut add_bytes = 0usize; + let mut delete_bytes = 0usize; + let target = usize::try_from(self.manifest_target_size.max(1)).unwrap_or(usize::MAX); + let mut add_files = Vec::new(); + let mut delete_files = Vec::new(); + + while let Some(entry) = sorted.next_entry()? { + let identifier = entry.identifier(); + match entry.kind() { + FileKind::Add if delete_identifiers.contains(&identifier) => { + matched.insert(identifier); + } + FileKind::Add => { + add_bytes = add_bytes.saturating_add(manifest_entry_sort_size(&entry)?); + add_chunk.push(entry); + if add_bytes >= target { + add_files.extend( + self.write_manifest_sort_chunk( + file_io, + manifest_dir, + &mut add_chunk, + new_files, + ) + .await?, + ); + add_bytes = 0; + } + } + FileKind::Delete + if !matched.contains(&identifier) && emitted_deletes.insert(identifier) => + { + delete_bytes = delete_bytes.saturating_add(manifest_entry_sort_size(&entry)?); + delete_chunk.push(entry); + if delete_bytes >= target { + delete_files.extend( + self.write_manifest_sort_chunk( + file_io, + manifest_dir, + &mut delete_chunk, + new_files, + ) + .await?, + ); + delete_bytes = 0; + } + } + FileKind::Delete => {} + } + } + + add_files.extend( + self.write_manifest_sort_chunk(file_io, manifest_dir, &mut add_chunk, new_files) + .await?, + ); + delete_files.extend( + self.write_manifest_sort_chunk(file_io, manifest_dir, &mut delete_chunk, new_files) + .await?, + ); + Ok((add_files, delete_files)) + } + + async fn write_manifest_sort_chunk( + &self, + file_io: &FileIO, + manifest_dir: &str, + entries: &mut Vec, + new_files: &mut Vec, + ) -> Result> { + if entries.is_empty() { + return Ok(Vec::new()); + } + let prefix = format!("manifest-{}", uuid::Uuid::new_v4()); + let entries = std::mem::take(entries); + let files = self + .write_manifest_files(file_io, manifest_dir, &prefix, &entries) + .await?; + new_files.extend(files.clone()); + Ok(files) + } + async fn merge_manifest_candidates( &self, file_io: &FileIO, @@ -2931,6 +3501,121 @@ impl TableCommit { } } +fn manifest_entry_sort_size(entry: &ManifestEntry) -> Result { + serde_json::to_vec(entry) + .map(|encoded| encoded.len().saturating_add(8)) + .map_err(|error| crate::Error::UnexpectedError { + message: "Failed to size manifest entry for sorted output".to_string(), + source: Some(Box::new(error)), + }) +} + +enum ManifestRewriteOutput { + Full { + files: Vec, + }, + Minor { + slots: Vec>, + positions: HashMap, + min_picked_index: usize, + max_picked_index: usize, + }, +} + +impl ManifestRewriteOutput { + fn full(files: Vec) -> Self { + Self::Full { files } + } + + fn minor(input: &[ManifestFileMeta], picked_file_names: &HashSet) -> Result { + let positions = input + .iter() + .enumerate() + .map(|(index, file)| (file.file_name().to_string(), index)) + .collect::>(); + let mut picked_indexes = input.iter().enumerate().filter_map(|(index, file)| { + picked_file_names + .contains(file.file_name()) + .then_some(index) + }); + let Some(first) = picked_indexes.next() else { + return Err(crate::Error::UnexpectedError { + message: "Manifest sort plan selected no input files".to_string(), + source: None, + }); + }; + let (min_picked_index, max_picked_index) = picked_indexes + .fold((first, first), |(min, max), index| { + (min.min(index), max.max(index)) + }); + Ok(Self::Minor { + slots: vec![Vec::new(); input.len()], + positions, + min_picked_index, + max_picked_index, + }) + } + + fn add_unchanged(&mut self, file: ManifestFileMeta) -> Result<()> { + match self { + Self::Full { files } => files.push(file), + Self::Minor { + slots, positions, .. + } => { + let Some(index) = positions.get(file.file_name()).copied() else { + return Err(crate::Error::UnexpectedError { + message: format!( + "Manifest sort output cannot place unknown file '{}'", + file.file_name() + ), + source: None, + }); + }; + slots[index].push(file); + } + } + Ok(()) + } + + fn add_all_unchanged(&mut self, files: Vec) -> Result<()> { + for file in files { + self.add_unchanged(file)?; + } + Ok(()) + } + + fn add_sorted_files(&mut self, added: Vec) -> Result<()> { + match self { + Self::Full { files } => files.extend(added), + Self::Minor { + slots, + min_picked_index, + .. + } => slots[*min_picked_index].extend(added), + } + Ok(()) + } + + fn add_delete_files(&mut self, deleted: Vec) -> Result<()> { + match self { + Self::Full { files } => files.extend(deleted), + Self::Minor { + slots, + max_picked_index, + .. + } => slots[*max_picked_index].extend(deleted), + } + Ok(()) + } + + fn into_files(self) -> Vec { + match self { + Self::Full { files } => files, + Self::Minor { slots, .. } => slots.into_iter().flatten().collect(), + } + } +} + /// Serialized BinaryRow for partition stats; unlike `datums_to_binary_row`, returns a /// valid arity-N row even when every datum is `None` (the all-null case must still /// decode on the Java side). @@ -6048,6 +6733,245 @@ mod tests { ); } + fn sorted_partition_commit(file_io: &FileIO, path: &str, full: bool) -> TableCommit { + let mut options = HashMap::from([ + ("manifest-sort.enabled".to_string(), "true".to_string()), + ("manifest.target-file-size".to_string(), "1 mb".to_string()), + ]); + if full { + options.insert( + "manifest.full-compaction-threshold-size".to_string(), + "1 b".to_string(), + ); + } + let schema = test_partitioned_schema().copy_with_replaced_options(options); + let table = Table::new( + file_io.clone(), + Identifier::new("default", "sort_test"), + path.to_string(), + schema, + None, + ); + TableCommit::new(table, "test-user".to_string()) + } + + #[tokio::test] + async fn test_manifest_sort_minor_keeps_unmatched_delete_after_adds() { + let file_io = test_file_io(); + let path = "memory:/test_manifest_sort_minor_deletes"; + setup_dirs(&file_io, path).await; + let commit = sorted_partition_commit(&file_io, path, false); + let dir = format!("{path}/manifest"); + let partition = partition_bytes("a"); + let original = commit + .write_manifest_files( + &file_io, + &dir, + "manifest-minor-old", + &[ + ManifestEntry::new( + FileKind::Add, + partition.clone(), + 0, + 1, + test_data_file("remove.parquet", 1), + 2, + ), + ManifestEntry::new( + FileKind::Add, + partition.clone(), + 0, + 1, + test_data_file("keep.parquet", 1), + 2, + ), + ], + ) + .await + .unwrap(); + let later = commit + .write_manifest_files( + &file_io, + &dir, + "manifest-minor-new", + &[ + ManifestEntry::new( + FileKind::Delete, + partition.clone(), + 0, + 1, + test_data_file("remove.parquet", 1), + 2, + ), + ManifestEntry::new( + FileKind::Delete, + partition, + 0, + 1, + test_data_file("unmatched.parquet", 1), + 2, + ), + ], + ) + .await + .unwrap(); + let (merged, created) = commit + .sort_manifest_files(&file_io, &dir, [original, later].concat()) + .await + .unwrap(); + assert_eq!(created.len(), 2); + assert_eq!(merged.len(), 2); + let entries = [ + Manifest::read(&file_io, &format!("{dir}/{}", merged[0].file_name())) + .await + .unwrap(), + Manifest::read(&file_io, &format!("{dir}/{}", merged[1].file_name())) + .await + .unwrap(), + ] + .concat(); + assert_eq!( + entries + .iter() + .map(|entry| (entry.file().file_name.as_str(), *entry.kind())) + .collect::>(), + vec![ + ("keep.parquet", FileKind::Add), + ("unmatched.parquet", FileKind::Delete), + ] + ); + } + + #[tokio::test] + async fn test_manifest_sort_full_nets_cross_manifest_deletes() { + let file_io = test_file_io(); + let path = "memory:/test_manifest_sort_full_deletes"; + setup_dirs(&file_io, path).await; + let commit = sorted_partition_commit(&file_io, path, true); + let dir = format!("{path}/manifest"); + let partition = partition_bytes("a"); + let original = commit + .write_manifest_files( + &file_io, + &dir, + "manifest-full-old", + &[ + ManifestEntry::new( + FileKind::Add, + partition.clone(), + 0, + 1, + test_data_file("remove.parquet", 1), + 2, + ), + ManifestEntry::new( + FileKind::Add, + partition.clone(), + 0, + 1, + test_data_file("keep.parquet", 1), + 2, + ), + ], + ) + .await + .unwrap(); + let later = commit + .write_manifest_files( + &file_io, + &dir, + "manifest-full-new", + &[ManifestEntry::new( + FileKind::Delete, + partition, + 0, + 1, + test_data_file("remove.parquet", 1), + 2, + )], + ) + .await + .unwrap(); + let (merged, created) = commit + .sort_manifest_files(&file_io, &dir, [original, later].concat()) + .await + .unwrap(); + assert_eq!(created.len(), 1); + assert_eq!(merged, created); + let entries = Manifest::read(&file_io, &format!("{dir}/{}", merged[0].file_name())) + .await + .unwrap(); + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].file().file_name, "keep.parquet"); + assert_eq!(*entries[0].kind(), FileKind::Add); + } + + #[tokio::test] + async fn test_manifest_sort_commit_keeps_delta_separate_and_orders_base() { + let file_io = test_file_io(); + let path = "memory:/test_manifest_sort_commit"; + setup_dirs(&file_io, path).await; + let commit = sorted_partition_commit(&file_io, path, false); + let messages = ["z", "a", "m"] + .into_iter() + .map(|partition| { + CommitMessage::new( + partition_bytes(partition), + 0, + vec![test_data_file(&format!("{partition}.parquet"), 1)], + ) + }) + .collect(); + commit.commit(messages).await.unwrap(); + commit + .commit(vec![CommitMessage::new( + partition_bytes("b"), + 0, + vec![test_data_file("b.parquet", 1)], + )]) + .await + .unwrap(); + commit + .commit(vec![CommitMessage::new( + partition_bytes("q"), + 0, + vec![test_data_file("q.parquet", 1)], + )]) + .await + .unwrap(); + let snapshot = latest_snapshot(&file_io, path).await.unwrap(); + let dir = format!("{path}/manifest"); + let base = ManifestList::read( + &file_io, + &format!("{dir}/{}", snapshot.base_manifest_list()), + ) + .await + .unwrap(); + assert_eq!(base.len(), 1, "two under-target manifests are sorted"); + let base_entries = Manifest::read(&file_io, &format!("{dir}/{}", base[0].file_name())) + .await + .unwrap(); + assert_eq!( + base_entries + .iter() + .map(|entry| entry.file().file_name.as_str()) + .collect::>(), + vec!["a.parquet", "b.parquet", "m.parquet", "z.parquet"] + ); + let delta = ManifestList::read( + &file_io, + &format!("{dir}/{}", snapshot.delta_manifest_list()), + ) + .await + .unwrap(); + assert_eq!(delta.len(), 1); + let delta_entries = Manifest::read(&file_io, &format!("{dir}/{}", delta[0].file_name())) + .await + .unwrap(); + assert_eq!(delta_entries[0].file().file_name, "q.parquet"); + assert_eq!(active_entries(&file_io, path, &snapshot).await.len(), 5); + } + /// `write_manifest_file` must aggregate min/max bucket and level across entries so the /// Java reader can prune manifests by bucket / level (see apache/paimon#5345). This /// drives a real commit so all the call-site plumbing is exercised end to end. From c680ecd21e8dbb271a01678a15df6c4ab1b5e801 Mon Sep 17 00:00:00 2001 From: mingfeng Date: Wed, 16 Sep 2026 22:04:28 -0700 Subject: [PATCH 2/4] feat(manifest): sort data evolution entries by row id --- crates/paimon/src/spec/schema.rs | 16 +- crates/paimon/src/table/manifest_sort.rs | 489 +++++++++++++++++++---- crates/paimon/src/table/table_commit.rs | 34 +- 3 files changed, 448 insertions(+), 91 deletions(-) diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 51ea1508a..0f50533bb 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -1788,9 +1788,15 @@ impl Schema { return Ok(()); } if partition_keys.is_empty() { + if core_options.data_evolution_enabled() + && core_options.manifest_sort_partition_field().is_none() + { + return Ok(()); + } return Err(crate::Error::ConfigInvalid { message: format!( - "Cannot enable '{MANIFEST_SORT_ENABLED_OPTION}' for non-partition table." + "Cannot enable '{MANIFEST_SORT_ENABLED_OPTION}' for a non-partition table \ + without Data Evolution RowID sorting." ), }); } @@ -3524,6 +3530,14 @@ mod tests { .build(); assert_config_invalid(non_partitioned, "non-partition table"); + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option(MANIFEST_SORT_ENABLED_OPTION, "true") + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .build() + .unwrap(); + let invalid_field = Schema::builder() .column("pt", DataType::Int(IntType::new())) .column("id", DataType::Int(IntType::new())) diff --git a/crates/paimon/src/table/manifest_sort.rs b/crates/paimon/src/table/manifest_sort.rs index 329204932..d36351b9b 100644 --- a/crates/paimon/src/table/manifest_sort.rs +++ b/crates/paimon/src/table/manifest_sort.rs @@ -68,76 +68,261 @@ impl ManifestSortConfig { } #[derive(Debug, Clone)] -pub(crate) struct PartitionSortKey { +pub(crate) struct PartitionSortField { field_index: usize, data_type: crate::spec::DataType, } -impl PartitionSortKey { - pub(crate) fn new( +#[derive(Debug, Clone)] +pub(crate) enum ManifestSortKey { + Partition { + field: PartitionSortField, + }, + RowId { + partition_fields: Vec, + }, +} + +impl ManifestSortKey { + pub(crate) fn create( partition_fields: &[DataField], configured_field: Option<&str>, - ) -> Result { - let field_index = match configured_field { - Some(name) => partition_fields + data_evolution_enabled: bool, + manifests: &[ManifestFileMeta], + ) -> Result> { + if data_evolution_enabled + && !manifests.is_empty() + && manifests .iter() - .position(|field| field.name() == name) - .ok_or_else(|| crate::Error::ConfigInvalid { - message: format!( - "Cannot resolve manifest sort partition field '{name}' from {:?}", - partition_fields - .iter() - .map(|field| field.name()) - .collect::>() - ), - })?, - None => { - if partition_fields.is_empty() { - return Err(crate::Error::ConfigInvalid { - message: "Cannot enable manifest-sort.enabled for non-partition table." - .to_string(), - }); - } - 0 - } + .all(|meta| meta.min_row_id().is_some() && meta.max_row_id().is_some()) + { + let partition_fields = match configured_field { + Some(name) => vec![resolve_partition_field(partition_fields, name)?], + None => partition_fields + .iter() + .enumerate() + .map(|(field_index, field)| PartitionSortField { + field_index, + data_type: field.data_type().clone(), + }) + .collect(), + }; + return Ok(Some(Self::RowId { partition_fields })); + } + + if partition_fields.is_empty() { + return Ok(None); + } + let field = match configured_field { + Some(name) => resolve_partition_field(partition_fields, name)?, + None => PartitionSortField { + field_index: 0, + data_type: partition_fields[0].data_type().clone(), + }, }; - Ok(Self { - field_index, - data_type: partition_fields[field_index].data_type().clone(), - }) + Ok(Some(Self::Partition { field })) } - pub(crate) fn entry_key(&self, entry: &ManifestEntry) -> Result> { - self.key_from_serialized_row(entry.partition()) + fn entry_key(&self, entry: &ManifestEntry) -> Result { + match self { + Self::Partition { field } => Ok(ManifestEntrySortKey::Partition( + read_partition_values(entry.partition(), std::slice::from_ref(field))?, + )), + Self::RowId { partition_fields } => { + let first_row_id = + entry + .file() + .first_row_id + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "Manifest entry '{}' has no first RowID", + entry.file().file_name + ), + source: None, + })?; + let row_count = entry.file().row_count; + if row_count <= 0 { + return Err(crate::Error::DataInvalid { + message: format!( + "Manifest entry '{}' has invalid row count {row_count}", + entry.file().file_name + ), + source: None, + }); + } + let range_end = first_row_id.checked_add(row_count - 1).ok_or_else(|| { + crate::Error::DataInvalid { + message: format!( + "Manifest entry '{}' RowID range overflows i64", + entry.file().file_name + ), + source: None, + } + })?; + Ok(ManifestEntrySortKey::RowId { + partition: read_partition_values(entry.partition(), partition_fields)?, + first_row_id, + range_end, + max_sequence_number: entry.file().max_sequence_number, + }) + } + } } pub(crate) fn range(&self, meta: ManifestFileMeta) -> Result { - let min = self.key_from_serialized_row(meta.partition_stats().min_values())?; - let max = self.key_from_serialized_row(meta.partition_stats().max_values())?; + let (min, max) = match self { + Self::Partition { field } => ( + ManifestBound::Partition(read_partition_values( + meta.partition_stats().min_values(), + std::slice::from_ref(field), + )?), + ManifestBound::Partition(read_partition_values( + meta.partition_stats().max_values(), + std::slice::from_ref(field), + )?), + ), + Self::RowId { partition_fields } => ( + ManifestBound::RowId { + partition: read_partition_values( + meta.partition_stats().min_values(), + partition_fields, + )?, + row_id: meta.min_row_id().ok_or_else(|| crate::Error::DataInvalid { + message: format!("Manifest '{}' has no minimum RowID", meta.file_name()), + source: None, + })?, + }, + ManifestBound::RowId { + partition: read_partition_values( + meta.partition_stats().max_values(), + partition_fields, + )?, + row_id: meta.max_row_id().ok_or_else(|| crate::Error::DataInvalid { + message: format!("Manifest '{}' has no maximum RowID", meta.file_name()), + source: None, + })?, + }, + ), + }; Ok(ManifestRange { meta, min, max }) } - fn key_from_serialized_row(&self, bytes: &[u8]) -> Result> { - let row = BinaryRow::from_serialized_bytes(bytes)?; - if self.field_index >= row.arity() as usize { - return Err(crate::Error::DataInvalid { - message: format!( - "Manifest sort field index {} is outside partition row arity {}", - self.field_index, - row.arity() - ), - source: None, - }); + fn compare_bounds(&self, left: &ManifestBound, right: &ManifestBound) -> Ordering { + match (self, left, right) { + ( + Self::Partition { .. }, + ManifestBound::Partition(left), + ManifestBound::Partition(right), + ) => compare_partition_values(left, right), + ( + Self::RowId { .. }, + ManifestBound::RowId { + partition: left_partition, + row_id: left_row_id, + }, + ManifestBound::RowId { + partition: right_partition, + row_id: right_row_id, + }, + ) => compare_partition_values(left_partition, right_partition) + .then_with(|| left_row_id.cmp(right_row_id)), + _ => unreachable!("manifest bounds are built by one whole-pass sort key"), } - extract_datum(&row, self.field_index, &self.data_type) } + + fn is_after_max(&self, min: &ManifestBound, max: &ManifestBound) -> bool { + let ordering = self.compare_bounds(min, max); + match self { + // Partition ranges preserve Java's historical boundary-equality + // behavior. RowID ranges are inclusive, so equality overlaps. + Self::Partition { .. } => ordering != Ordering::Less, + Self::RowId { .. } => ordering == Ordering::Greater, + } + } +} + +fn resolve_partition_field( + partition_fields: &[DataField], + name: &str, +) -> Result { + partition_fields + .iter() + .enumerate() + .find(|(_, field)| field.name() == name) + .map(|(field_index, field)| PartitionSortField { + field_index, + data_type: field.data_type().clone(), + }) + .ok_or_else(|| crate::Error::ConfigInvalid { + message: format!( + "Cannot resolve manifest sort partition field '{name}' from {:?}", + partition_fields + .iter() + .map(|field| field.name()) + .collect::>() + ), + }) +} + +fn read_partition_values( + bytes: &[u8], + fields: &[PartitionSortField], +) -> Result>> { + if fields.is_empty() { + return Ok(Vec::new()); + } + let row = BinaryRow::from_serialized_bytes(bytes)?; + fields + .iter() + .map(|field| { + if field.field_index >= row.arity() as usize { + return Err(crate::Error::DataInvalid { + message: format!( + "Manifest sort field index {} is outside partition row arity {}", + field.field_index, + row.arity() + ), + source: None, + }); + } + extract_datum(&row, field.field_index, &field.data_type) + }) + .collect() +} + +fn compare_partition_values(left: &[Option], right: &[Option]) -> Ordering { + left.iter() + .zip(right) + .map(|(left, right)| compare_optional_datums(left, right)) + .find(|ordering| *ordering != Ordering::Equal) + .unwrap_or_else(|| left.len().cmp(&right.len())) +} + +#[derive(Debug)] +enum ManifestEntrySortKey { + Partition(Vec>), + RowId { + partition: Vec>, + first_row_id: i64, + range_end: i64, + max_sequence_number: i64, + }, +} + +#[derive(Debug, Clone)] +enum ManifestBound { + Partition(Vec>), + RowId { + partition: Vec>, + row_id: i64, + }, } #[derive(Debug, Clone)] pub(crate) struct ManifestRange { pub(crate) meta: ManifestFileMeta, - min: Option, - max: Option, + min: ManifestBound, + max: ManifestBound, } #[derive(Debug, Clone)] @@ -217,13 +402,13 @@ fn canonical_double(value: f64) -> f64 { #[derive(Debug)] struct SortRecord { - key: Option, + key: ManifestEntrySortKey, encoded: Vec, entry: ManifestEntry, } impl SortRecord { - fn new(entry: ManifestEntry, sort_key: &PartitionSortKey) -> Result { + fn new(entry: ManifestEntry, sort_key: &ManifestSortKey) -> Result { let key = sort_key.entry_key(&entry)?; let encoded = serde_json::to_vec(&entry).map_err(|error| crate::Error::UnexpectedError { @@ -237,7 +422,7 @@ impl SortRecord { }) } - fn from_encoded(encoded: Vec, sort_key: &PartitionSortKey) -> Result { + fn from_encoded(encoded: Vec, sort_key: &ManifestSortKey) -> Result { let entry = serde_json::from_slice(&encoded).map_err(|error| crate::Error::UnexpectedError { message: "Failed to decode spilled manifest entry".to_string(), @@ -253,8 +438,42 @@ impl SortRecord { } fn compare_sort_records(left: &SortRecord, right: &SortRecord) -> Ordering { - compare_optional_datums(&left.key, &right.key) + let partition_order = match (&left.key, &right.key) { + (ManifestEntrySortKey::Partition(left), ManifestEntrySortKey::Partition(right)) => { + compare_partition_values(left, right) + } + ( + ManifestEntrySortKey::RowId { + partition: left, .. + }, + ManifestEntrySortKey::RowId { + partition: right, .. + }, + ) => compare_partition_values(left, right), + _ => unreachable!("entries in one sorter use one whole-pass sort key"), + }; + partition_order .then_with(|| file_kind_order(left.entry.kind()).cmp(&file_kind_order(right.entry.kind()))) + .then_with(|| match (&left.key, &right.key) { + ( + ManifestEntrySortKey::RowId { + first_row_id: left_first, + range_end: left_end, + max_sequence_number: left_sequence, + .. + }, + ManifestEntrySortKey::RowId { + first_row_id: right_first, + range_end: right_end, + max_sequence_number: right_sequence, + .. + }, + ) => left_first + .cmp(right_first) + .then_with(|| left_end.cmp(right_end)) + .then_with(|| right_sequence.cmp(left_sequence)), + _ => Ordering::Equal, + }) .then_with(|| { left.entry .file() @@ -277,7 +496,7 @@ struct SpillRun { /// manifests remain Avro. Runs are merged until the final fan-in is bounded by /// `max_file_handles`. pub(crate) struct SpillableManifestSorter { - sort_key: PartitionSortKey, + sort_key: ManifestSortKey, buffer: Vec, buffer_bytes: usize, spill_buffer_size: usize, @@ -290,7 +509,7 @@ pub(crate) struct SpillableManifestSorter { } impl SpillableManifestSorter { - pub(crate) fn new(sort_key: &PartitionSortKey, config: &ManifestSortConfig) -> Result { + pub(crate) fn new(sort_key: &ManifestSortKey, config: &ManifestSortConfig) -> Result { let temp_dir = tempfile::tempdir().map_err(|error| crate::Error::UnexpectedError { message: "Failed to create manifest-sort spill directory".to_string(), source: Some(Box::new(error)), @@ -518,14 +737,14 @@ pub(crate) struct SpilledManifestEntries { _temp_dir: tempfile::TempDir, readers: Vec, heap: BinaryHeap, - sort_key: PartitionSortKey, + sort_key: ManifestSortKey, } impl SpilledManifestEntries { fn new( temp_dir: tempfile::TempDir, runs: Vec, - sort_key: PartitionSortKey, + sort_key: ManifestSortKey, ) -> Result { let mut readers = runs .iter() @@ -577,7 +796,7 @@ impl SpillRunReader { }) } - fn next_record(&mut self, sort_key: &PartitionSortKey) -> Result> { + fn next_record(&mut self, sort_key: &ManifestSortKey) -> Result> { let mut length = [0u8; 8]; let first = self.reader @@ -712,7 +931,7 @@ pub(crate) fn manifest_may_contain_partition( pub(crate) fn plan_rewrite( input: &[ManifestFileMeta], default_compaction: &HashMap, - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, config: &ManifestSortConfig, ) -> Result> { let lsm_files = input @@ -721,7 +940,7 @@ pub(crate) fn plan_rewrite( .cloned() .map(|file| sort_key.range(file)) .collect::>>()?; - let level_runs = build_level_sorted_runs(lsm_files); + let level_runs = build_level_sorted_runs(lsm_files, sort_key); let picked_runs = ManifestPickStrategy::new( config.max_size_amplification_percent, config.sorted_run_size_ratio, @@ -747,7 +966,7 @@ pub(crate) fn plan_rewrite( } let sections = merge_small_adjacent_sections( - split_into_sections(picked_files, default_compaction), + split_into_sections(picked_files, default_compaction, sort_key), config.target_size, ); Ok(Some(ManifestRewritePlan { @@ -756,15 +975,18 @@ pub(crate) fn plan_rewrite( })) } -fn build_level_sorted_runs(mut input: Vec) -> Vec { - input.sort_by(compare_ranges); +fn build_level_sorted_runs( + mut input: Vec, + sort_key: &ManifestSortKey, +) -> Vec { + input.sort_by(|left, right| compare_ranges(left, right, sort_key)); let mut runs: Vec> = Vec::new(); for file in input { let earliest = runs .iter() .enumerate() .min_by(|(_, left), (_, right)| { - compare_optional_datums( + sort_key.compare_bounds( &left.last().expect("non-empty run").max, &right.last().expect("non-empty run").max, ) @@ -772,10 +994,8 @@ fn build_level_sorted_runs(mut input: Vec) -> Vec + if sort_key + .is_after_max(&file.min, &runs[index].last().expect("non-empty run").max) => { runs[index].push(file); } @@ -802,18 +1022,19 @@ fn build_level_sorted_runs(mut input: Vec) -> Vec, default_compaction: &HashMap, + sort_key: &ManifestSortKey, ) -> Vec { - picked_files.sort_by(compare_ranges); + picked_files.sort_by(|left, right| compare_ranges(left, right, sort_key)); let mut sections = Vec::new(); let mut current_files = Vec::new(); let mut current_size = 0; let mut current_has_default = false; - let mut current_max: Option> = None; + let mut current_max: Option = None; for file in picked_files { let starts_new = current_max .as_ref() - .is_some_and(|max| compare_optional_datums(&file.min, max) != Ordering::Less); + .is_some_and(|max| sort_key.is_after_max(&file.min, max)); if starts_new { sections.push(ManifestSection { files: current_files, @@ -830,7 +1051,7 @@ fn split_into_sections( current_has_default |= default_compaction.contains_key(file.meta.file_name()); if current_max .as_ref() - .is_none_or(|max| compare_optional_datums(&file.max, max) == Ordering::Greater) + .is_none_or(|max| sort_key.compare_bounds(&file.max, max) == Ordering::Greater) { current_max = Some(file.max.clone()); } @@ -924,9 +1145,14 @@ impl ManifestPickStrategy { } } -fn compare_ranges(left: &ManifestRange, right: &ManifestRange) -> Ordering { - compare_optional_datums(&left.min, &right.min) - .then_with(|| compare_optional_datums(&left.max, &right.max)) +fn compare_ranges( + left: &ManifestRange, + right: &ManifestRange, + sort_key: &ManifestSortKey, +) -> Ordering { + sort_key + .compare_bounds(&left.min, &right.min) + .then_with(|| sort_key.compare_bounds(&left.max, &right.max)) .then_with(|| left.meta.file_name().cmp(right.meta.file_name())) } @@ -960,19 +1186,43 @@ mod tests { ) } - fn sort_key() -> PartitionSortKey { - PartitionSortKey::new( + fn row_id_meta( + name: &str, + size: i64, + partition: i32, + min_row_id: i64, + max_row_id: i64, + ) -> ManifestFileMeta { + meta(name, size, partition, partition).with_row_id_stats(Some(min_row_id), Some(max_row_id)) + } + + fn sort_key() -> ManifestSortKey { + ManifestSortKey::create( &[DataField::new( 0, "pt".to_string(), DataType::Int(IntType::new()), )], None, + false, + &[], ) .unwrap() + .unwrap() } fn entry(kind: FileKind, partition: i32, file_name: &str) -> ManifestEntry { + row_id_entry(kind, partition, file_name, None, 1, 0) + } + + fn row_id_entry( + kind: FileKind, + partition: i32, + file_name: &str, + first_row_id: Option, + row_count: i64, + max_sequence_number: i64, + ) -> ManifestEntry { ManifestEntry::new( kind, row(partition), @@ -981,13 +1231,13 @@ mod tests { DataFileMeta { file_name: file_name.to_string(), file_size: 1, - row_count: 1, + row_count, min_key: Vec::new(), max_key: Vec::new(), key_stats: BinaryTableStats::empty(), value_stats: BinaryTableStats::empty(), min_sequence_number: 0, - max_sequence_number: 0, + max_sequence_number, schema_id: 0, level: 0, extra_files: Vec::new(), @@ -997,7 +1247,7 @@ mod tests { file_source: None, value_stats_cols: None, external_path: None, - first_row_id: None, + first_row_id, write_cols: None, column_max_sequence_numbers: None, }, @@ -1030,14 +1280,91 @@ mod tests { key.range(meta("a", 100, 0, 10)).unwrap(), key.range(meta("b", 100, 10, 20)).unwrap(), ]; - let runs = build_level_sorted_runs(ranges.clone()); + let runs = build_level_sorted_runs(ranges.clone(), &key); assert_eq!(runs.len(), 1); assert_eq!(runs[0].files.len(), 2); - let sections = split_into_sections(ranges, &HashMap::new()); + let sections = split_into_sections(ranges, &HashMap::new(), &key); assert_eq!(sections.len(), 2); } + #[test] + fn row_id_boundary_equality_overlaps_runs_and_sections() { + let manifests = vec![ + row_id_meta("a", 100, 0, 0, 10), + row_id_meta("b", 100, 0, 10, 20), + ]; + let key = ManifestSortKey::create( + &[DataField::new( + 0, + "pt".to_string(), + DataType::Int(IntType::new()), + )], + None, + true, + &manifests, + ) + .unwrap() + .unwrap(); + assert!(matches!(key, ManifestSortKey::RowId { .. })); + + let ranges = manifests + .into_iter() + .map(|meta| key.range(meta)) + .collect::>>() + .unwrap(); + let runs = build_level_sorted_runs(ranges.clone(), &key); + assert_eq!(runs.len(), 2); + + let sections = split_into_sections(ranges, &HashMap::new(), &key); + assert_eq!(sections.len(), 1); + } + + #[test] + fn row_id_sort_orders_kind_range_sequence_and_name() { + let manifests = vec![row_id_meta("m", 100, 0, 10, 20)]; + let key = ManifestSortKey::create( + &[DataField::new( + 0, + "pt".to_string(), + DataType::Int(IntType::new()), + )], + None, + true, + &manifests, + ) + .unwrap() + .unwrap(); + let mut sorter = SpillableManifestSorter::new(&key, &spill_config(1, 2, u64::MAX)).unwrap(); + sorter + .push(row_id_entry(FileKind::Delete, 0, "delete", Some(1), 1, 0)) + .unwrap(); + sorter + .push(row_id_entry(FileKind::Add, 0, "later", Some(2), 1, 0)) + .unwrap(); + sorter + .push(row_id_entry(FileKind::Add, 0, "old", Some(1), 1, 3)) + .unwrap(); + sorter + .push(row_id_entry(FileKind::Add, 0, "new", Some(1), 1, 5)) + .unwrap(); + + let mut sorted = sorter.finish().unwrap(); + let mut actual = Vec::new(); + while let Some(entry) = sorted.next_entry().unwrap() { + actual.push((*entry.kind(), entry.file().file_name.clone())); + } + assert_eq!( + actual, + vec![ + (FileKind::Add, "new".to_string()), + (FileKind::Add, "old".to_string()), + (FileKind::Add, "later".to_string()), + (FileKind::Delete, "delete".to_string()), + ] + ); + } + #[test] fn pick_strategy_forces_low_levels_once_run_count_exceeds_four() { let key = sort_key(); @@ -1046,7 +1373,7 @@ mod tests { .enumerate() .map(|(index, size)| key.range(meta(&format!("m-{index}"), size, 0, 10)).unwrap()) .collect(); - let runs = build_level_sorted_runs(files); + let runs = build_level_sorted_runs(files, &key); assert_eq!(runs.len(), 5); let picked = ManifestPickStrategy::new(1_000, 1).pick(&runs); assert_eq!(picked, vec![0, 1]); diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index c2a832ecb..17088a014 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -35,7 +35,7 @@ use crate::table::global_index_build_common::same_extra_field_ids; use crate::table::index_file_path::committed_index_file_path; use crate::table::manifest_sort::{ manifest_may_contain_partition, plan_rewrite, reaches_full_compaction_threshold, - ManifestRewritePlan, ManifestSection, ManifestSortConfig, PartitionSortKey, + ManifestRewritePlan, ManifestSection, ManifestSortConfig, ManifestSortKey, SpillableManifestSorter, }; use crate::table::partition_filter::PartitionFilter; @@ -1150,7 +1150,16 @@ impl TableCommit { .sort_manifest_files(file_io, manifest_dir, manifest_files) .await; } + self.merge_manifest_files_legacy(file_io, manifest_dir, manifest_files) + .await + } + async fn merge_manifest_files_legacy( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: Vec, + ) -> Result<(Vec, Vec)> { let target_size = self.manifest_target_size.max(1); let mut result = Vec::new(); let mut new_files = Vec::new(); @@ -1196,10 +1205,17 @@ impl TableCommit { manifest_files: Vec, ) -> Result<(Vec, Vec)> { let partition_fields = self.table.schema().partition_fields(); - let sort_key = PartitionSortKey::new( + let Some(sort_key) = ManifestSortKey::create( &partition_fields, self.manifest_sort_partition_field.as_deref(), - )?; + self.data_evolution_enabled, + &manifest_files, + )? + else { + return self + .merge_manifest_files_legacy(file_io, manifest_dir, manifest_files) + .await; + }; let mut new_files = Vec::new(); let result = self .sort_manifest_files_inner( @@ -1226,7 +1242,7 @@ impl TableCommit { manifest_dir: &str, manifest_files: &[ManifestFileMeta], partition_fields: &[crate::spec::DataField], - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, ) -> Result> { if reaches_full_compaction_threshold( @@ -1258,7 +1274,7 @@ impl TableCommit { manifest_dir: &str, manifest_files: &[ManifestFileMeta], partition_fields: &[crate::spec::DataField], - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, ) -> Result>> { let mut deleted_identifiers = HashSet::new(); @@ -1328,7 +1344,7 @@ impl TableCommit { file_io: &FileIO, manifest_dir: &str, manifest_files: &[ManifestFileMeta], - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, ) -> Result> { let default_compaction = manifest_files @@ -1377,7 +1393,7 @@ impl TableCommit { default_compaction: &HashMap, deleted_identifiers: &HashSet, full_compaction: bool, - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, output: &mut ManifestRewriteOutput, ) -> Result<()> { @@ -1496,7 +1512,7 @@ impl TableCommit { default_compaction: &HashMap, deleted_identifiers: &HashSet, full_compaction: bool, - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, output: &mut ManifestRewriteOutput, ) -> Result<()> { @@ -1554,7 +1570,7 @@ impl TableCommit { default_compaction: &HashMap, deleted_identifiers: &HashSet, full_compaction: bool, - sort_key: &PartitionSortKey, + sort_key: &ManifestSortKey, new_files: &mut Vec, output: &mut ManifestRewriteOutput, ) -> Result<()> { From e94e349b4dc2928bd4446fa1095d84fa57b937b6 Mon Sep 17 00:00:00 2001 From: mingfeng Date: Wed, 16 Sep 2026 22:11:33 -0700 Subject: [PATCH 3/4] feat(manifest): sort bucketed tables by bucket first --- crates/paimon/src/spec/schema.rs | 18 +- crates/paimon/src/table/manifest_sort.rs | 245 ++++++++++++++++++++++- crates/paimon/src/table/table_commit.rs | 1 + 3 files changed, 251 insertions(+), 13 deletions(-) diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 0f50533bb..6be5dadb2 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -1788,15 +1788,16 @@ impl Schema { return Ok(()); } if partition_keys.is_empty() { - if core_options.data_evolution_enabled() - && core_options.manifest_sort_partition_field().is_none() - { + let has_implicit_sort_key = core_options.data_evolution_enabled() + || core_options.bucket() > 0 + || core_options.bucket() == POSTPONE_BUCKET; + if has_implicit_sort_key && core_options.manifest_sort_partition_field().is_none() { return Ok(()); } return Err(crate::Error::ConfigInvalid { message: format!( "Cannot enable '{MANIFEST_SORT_ENABLED_OPTION}' for a non-partition table \ - without Data Evolution RowID sorting." + without fixed/postponed buckets or Data Evolution RowID sorting." ), }); } @@ -3538,6 +3539,15 @@ mod tests { .build() .unwrap(); + for bucket in ["2", "-2"] { + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option(MANIFEST_SORT_ENABLED_OPTION, "true") + .option("bucket", bucket) + .build() + .unwrap(); + } + let invalid_field = Schema::builder() .column("pt", DataType::Int(IntType::new())) .column("id", DataType::Int(IntType::new())) diff --git a/crates/paimon/src/table/manifest_sort.rs b/crates/paimon/src/table/manifest_sort.rs index d36351b9b..182054917 100644 --- a/crates/paimon/src/table/manifest_sort.rs +++ b/crates/paimon/src/table/manifest_sort.rs @@ -78,6 +78,10 @@ pub(crate) enum ManifestSortKey { Partition { field: PartitionSortField, }, + Bucket { + field: Option, + compare_manifest_buckets: bool, + }, RowId { partition_fields: Vec, }, @@ -89,6 +93,7 @@ impl ManifestSortKey { configured_field: Option<&str>, data_evolution_enabled: bool, manifests: &[ManifestFileMeta], + bucketed: bool, ) -> Result> { if data_evolution_enabled && !manifests.is_empty() @@ -110,17 +115,23 @@ impl ManifestSortKey { return Ok(Some(Self::RowId { partition_fields })); } - if partition_fields.is_empty() { - return Ok(None); - } - let field = match configured_field { - Some(name) => resolve_partition_field(partition_fields, name)?, - None => PartitionSortField { + let field = match (configured_field, partition_fields.first()) { + (Some(name), _) => Some(resolve_partition_field(partition_fields, name)?), + (None, Some(field)) => Some(PartitionSortField { field_index: 0, - data_type: partition_fields[0].data_type().clone(), - }, + data_type: field.data_type().clone(), + }), + (None, None) => None, }; - Ok(Some(Self::Partition { field })) + if bucketed { + return Ok(Some(Self::Bucket { + field, + compare_manifest_buckets: manifests + .iter() + .all(|meta| meta.min_bucket().is_some() && meta.max_bucket().is_some()), + })); + } + Ok(field.map(|field| Self::Partition { field })) } fn entry_key(&self, entry: &ManifestEntry) -> Result { @@ -128,6 +139,13 @@ impl ManifestSortKey { Self::Partition { field } => Ok(ManifestEntrySortKey::Partition( read_partition_values(entry.partition(), std::slice::from_ref(field))?, )), + Self::Bucket { field, .. } => Ok(ManifestEntrySortKey::Bucket { + bucket: entry.bucket(), + partition: read_partition_values( + entry.partition(), + field.as_ref().map(std::slice::from_ref).unwrap_or_default(), + )?, + }), Self::RowId { partition_fields } => { let first_row_id = entry @@ -181,6 +199,34 @@ impl ManifestSortKey { std::slice::from_ref(field), )?), ), + Self::Bucket { + field, + compare_manifest_buckets, + } => { + let fields = field.as_ref().map(std::slice::from_ref).unwrap_or_default(); + ( + ManifestBound::Bucket { + bucket: compare_manifest_buckets.then(|| { + meta.min_bucket() + .expect("bucket stats were validated for the whole input") + }), + partition: read_partition_values( + meta.partition_stats().min_values(), + fields, + )?, + }, + ManifestBound::Bucket { + bucket: compare_manifest_buckets.then(|| { + meta.max_bucket() + .expect("bucket stats were validated for the whole input") + }), + partition: read_partition_values( + meta.partition_stats().max_values(), + fields, + )?, + }, + ) + } Self::RowId { partition_fields } => ( ManifestBound::RowId { partition: read_partition_values( @@ -214,6 +260,19 @@ impl ManifestSortKey { ManifestBound::Partition(left), ManifestBound::Partition(right), ) => compare_partition_values(left, right), + ( + Self::Bucket { .. }, + ManifestBound::Bucket { + bucket: left_bucket, + partition: left_partition, + }, + ManifestBound::Bucket { + bucket: right_bucket, + partition: right_partition, + }, + ) => left_bucket + .cmp(right_bucket) + .then_with(|| compare_partition_values(left_partition, right_partition)), ( Self::RowId { .. }, ManifestBound::RowId { @@ -236,6 +295,32 @@ impl ManifestSortKey { // Partition ranges preserve Java's historical boundary-equality // behavior. RowID ranges are inclusive, so equality overlaps. Self::Partition { .. } => ordering != Ordering::Less, + Self::Bucket { + field, + compare_manifest_buckets, + } => match (min, max) { + ( + ManifestBound::Bucket { + bucket: min_bucket, + partition: min_partition, + }, + ManifestBound::Bucket { + bucket: max_bucket, + partition: max_partition, + }, + ) => { + if *compare_manifest_buckets { + let bucket_order = min_bucket.cmp(max_bucket); + if bucket_order != Ordering::Equal { + return bucket_order == Ordering::Greater; + } + } + field.as_ref().is_some_and(|_| { + compare_partition_values(min_partition, max_partition) != Ordering::Less + }) || (field.is_none() && *compare_manifest_buckets) + } + _ => unreachable!("bucket sort builds bucket manifest bounds"), + }, Self::RowId { .. } => ordering == Ordering::Greater, } } @@ -301,6 +386,10 @@ fn compare_partition_values(left: &[Option], right: &[Option]) -> #[derive(Debug)] enum ManifestEntrySortKey { Partition(Vec>), + Bucket { + bucket: i32, + partition: Vec>, + }, RowId { partition: Vec>, first_row_id: i64, @@ -312,6 +401,10 @@ enum ManifestEntrySortKey { #[derive(Debug, Clone)] enum ManifestBound { Partition(Vec>), + Bucket { + bucket: Option, + partition: Vec>, + }, RowId { partition: Vec>, row_id: i64, @@ -442,6 +535,18 @@ fn compare_sort_records(left: &SortRecord, right: &SortRecord) -> Ordering { (ManifestEntrySortKey::Partition(left), ManifestEntrySortKey::Partition(right)) => { compare_partition_values(left, right) } + ( + ManifestEntrySortKey::Bucket { + bucket: left_bucket, + partition: left_partition, + }, + ManifestEntrySortKey::Bucket { + bucket: right_bucket, + partition: right_partition, + }, + ) => left_bucket + .cmp(right_bucket) + .then_with(|| compare_partition_values(left_partition, right_partition)), ( ManifestEntrySortKey::RowId { partition: left, .. @@ -1196,6 +1301,21 @@ mod tests { meta(name, size, partition, partition).with_row_id_stats(Some(min_row_id), Some(max_row_id)) } + fn bucket_meta( + name: &str, + size: i64, + partition: i32, + min_bucket: i32, + max_bucket: i32, + ) -> ManifestFileMeta { + meta(name, size, partition, partition).with_bucket_level_stats( + Some(min_bucket), + Some(max_bucket), + None, + None, + ) + } + fn sort_key() -> ManifestSortKey { ManifestSortKey::create( &[DataField::new( @@ -1206,6 +1326,7 @@ mod tests { None, false, &[], + false, ) .unwrap() .unwrap() @@ -1215,6 +1336,39 @@ mod tests { row_id_entry(kind, partition, file_name, None, 1, 0) } + fn bucket_entry(partition: i32, bucket: i32, file_name: &str) -> ManifestEntry { + ManifestEntry::new( + FileKind::Add, + row(partition), + bucket, + 4, + DataFileMeta { + file_name: file_name.to_string(), + file_size: 1, + row_count: 1, + min_key: Vec::new(), + max_key: Vec::new(), + key_stats: BinaryTableStats::empty(), + value_stats: BinaryTableStats::empty(), + min_sequence_number: 0, + max_sequence_number: 0, + schema_id: 0, + level: 0, + extra_files: Vec::new(), + creation_time: None, + delete_row_count: None, + embedded_index: None, + file_source: None, + value_stats_cols: None, + external_path: None, + first_row_id: None, + write_cols: None, + column_max_sequence_numbers: None, + }, + 2, + ) + } + fn row_id_entry( kind: FileKind, partition: i32, @@ -1303,6 +1457,7 @@ mod tests { None, true, &manifests, + false, ) .unwrap() .unwrap(); @@ -1332,6 +1487,7 @@ mod tests { None, true, &manifests, + false, ) .unwrap() .unwrap(); @@ -1365,6 +1521,77 @@ mod tests { ); } + #[test] + fn bucket_sort_orders_bucket_before_partition() { + let manifests = vec![bucket_meta("m", 100, 0, 0, 3)]; + let key = ManifestSortKey::create( + &[DataField::new( + 0, + "pt".to_string(), + DataType::Int(IntType::new()), + )], + None, + false, + &manifests, + true, + ) + .unwrap() + .unwrap(); + assert!(matches!(key, ManifestSortKey::Bucket { .. })); + + let mut sorter = SpillableManifestSorter::new(&key, &spill_config(1, 2, u64::MAX)).unwrap(); + sorter.push(bucket_entry(0, 2, "bucket-2")).unwrap(); + sorter + .push(bucket_entry(10, 1, "bucket-1-partition-10")) + .unwrap(); + sorter + .push(bucket_entry(0, 1, "bucket-1-partition-0")) + .unwrap(); + + let mut sorted = sorter.finish().unwrap(); + let mut actual = Vec::new(); + while let Some(entry) = sorted.next_entry().unwrap() { + actual.push(entry.file().file_name.clone()); + } + assert_eq!( + actual, + vec![ + "bucket-1-partition-0".to_string(), + "bucket-1-partition-10".to_string(), + "bucket-2".to_string(), + ] + ); + } + + #[test] + fn bucket_ranges_fall_back_when_legacy_manifests_lack_bucket_stats() { + let manifests = vec![meta("a", 100, 0, 0), meta("b", 100, 0, 0)]; + let key = ManifestSortKey::create(&[], None, false, &manifests, true) + .unwrap() + .unwrap(); + let ranges = manifests + .into_iter() + .map(|meta| key.range(meta)) + .collect::>>() + .unwrap(); + assert_eq!(build_level_sorted_runs(ranges.clone(), &key).len(), 2); + assert_eq!(split_into_sections(ranges, &HashMap::new(), &key).len(), 1); + } + + #[test] + fn data_evolution_row_id_sort_precedes_bucket_sort() { + let manifests = vec![row_id_meta("m", 100, 0, 10, 20).with_bucket_level_stats( + Some(0), + Some(3), + None, + None, + )]; + let key = ManifestSortKey::create(&[], None, true, &manifests, true) + .unwrap() + .unwrap(); + assert!(matches!(key, ManifestSortKey::RowId { .. })); + } + #[test] fn pick_strategy_forces_low_levels_once_run_count_exceeds_four() { let key = sort_key(); diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 17088a014..67c907d7d 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -1210,6 +1210,7 @@ impl TableCommit { self.manifest_sort_partition_field.as_deref(), self.data_evolution_enabled, &manifest_files, + self.total_buckets > 0 || self.total_buckets == POSTPONE_BUCKET, )? else { return self From a109094e2ea83754f5745847466a5796cd31bea3 Mon Sep 17 00:00:00 2001 From: mingfeng Date: Wed, 16 Sep 2026 22:27:04 -0700 Subject: [PATCH 4/4] feat(table): add manifest compaction API --- crates/paimon/src/table/table_commit.rs | 344 ++++++++++++++++++++++++ 1 file changed, 344 insertions(+) diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 67c907d7d..14f89f999 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -214,6 +214,158 @@ impl TableCommit { .await } + /// Compact the latest snapshot's data manifests and commit a metadata-only + /// [`CommitKind::COMPACT`] snapshot when the manifest set changes. + /// + /// This operation retries optimistic commit conflicts and preserves the + /// latest snapshot's data, index, row-tracking and user metadata. + pub async fn compact_manifests(&self) -> Result<()> { + CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?; + self.table.ensure_not_branch_reference_for_write()?; + + let compactor = self.with_manifest_compaction_options(); + let mut retry_count = 0u32; + let start_time_ms = current_time_millis(); + loop { + let Some(latest_snapshot) = compactor.snapshot_manager.get_latest_snapshot().await? + else { + return Ok(()); + }; + if compactor.compact_manifests_once(&latest_snapshot).await? { + return Ok(()); + } + + let elapsed_ms = current_time_millis().saturating_sub(start_time_ms); + if elapsed_ms > compactor.commit_timeout_ms + || retry_count >= compactor.commit_max_retries + { + return Err(crate::Error::DataInvalid { + message: format!( + "Manifest compaction failed for table '{}' after {} millis with {} retries; \ + there may be commit conflicts between multiple writers.", + compactor.table.identifier().full_name(), + elapsed_ms, + retry_count + ), + source: None, + }); + } + + compactor.commit_retry_wait(retry_count).await; + retry_count += 1; + } + } + + fn with_manifest_compaction_options(&self) -> Self { + let mut manifest_sort_config = self.manifest_sort_config.clone(); + manifest_sort_config.merge_min_count = 1; + manifest_sort_config.full_compaction_threshold_size = 1; + Self { + table: self.table.clone(), + snapshot_manager: self.snapshot_manager.clone(), + snapshot_commit: Arc::clone(&self.snapshot_commit), + commit_user: self.commit_user.clone(), + total_buckets: self.total_buckets, + commit_max_retries: self.commit_max_retries, + commit_timeout_ms: self.commit_timeout_ms, + commit_min_retry_wait_ms: self.commit_min_retry_wait_ms, + commit_max_retry_wait_ms: self.commit_max_retry_wait_ms, + manifest_compression: self.manifest_compression.clone(), + manifest_target_size: self.manifest_target_size, + manifest_merge_min_count: 1, + manifest_sort_enabled: self.manifest_sort_enabled, + manifest_sort_partition_field: self.manifest_sort_partition_field.clone(), + manifest_sort_config, + row_tracking_enabled: self.row_tracking_enabled, + data_evolution_enabled: self.data_evolution_enabled, + partition_default_name: self.partition_default_name.clone(), + } + } + + async fn compact_manifests_once(&self, latest_snapshot: &Snapshot) -> Result { + let file_io = self.snapshot_manager.file_io(); + let manifest_dir = self.snapshot_manager.manifest_dir(); + let base_path = format!("{manifest_dir}/{}", latest_snapshot.base_manifest_list()); + let delta_path = format!("{manifest_dir}/{}", latest_snapshot.delta_manifest_list()); + let mut before = ManifestList::read(file_io, &base_path).await?; + before.extend(ManifestList::read(file_io, &delta_path).await?); + + let (after, new_manifest_files) = if self.manifest_sort_enabled { + self.sort_manifest_files(file_io, &manifest_dir, before.clone()) + .await? + } else { + self.compact_manifest_files_legacy(file_io, &manifest_dir, before.clone()) + .await? + }; + if same_manifest_file_set(&before, &after) { + self.delete_manifest_files(file_io, &manifest_dir, &new_manifest_files) + .await; + return Ok(true); + } + + let unique_id = uuid::Uuid::new_v4(); + let base_manifest_list_name = format!("manifest-list-{unique_id}-0"); + let delta_manifest_list_name = format!("manifest-list-{unique_id}-1"); + let base_manifest_list_path = format!("{manifest_dir}/{base_manifest_list_name}"); + let delta_manifest_list_path = format!("{manifest_dir}/{delta_manifest_list_name}"); + ManifestList::write_with_compression( + file_io, + &base_manifest_list_path, + &after, + &self.manifest_compression, + ) + .await?; + ManifestList::write_with_compression( + file_io, + &delta_manifest_list_path, + &[], + &self.manifest_compression, + ) + .await?; + + let snapshot = Snapshot::builder() + .version(3) + .id(latest_snapshot.id() + 1) + .schema_id(latest_snapshot.schema_id()) + .base_manifest_list(base_manifest_list_name) + .delta_manifest_list(delta_manifest_list_name) + .index_manifest(latest_snapshot.index_manifest().map(str::to_string)) + .commit_user(self.commit_user.clone()) + .commit_identifier(BATCH_COMMIT_IDENTIFIER) + .commit_kind(CommitKind::COMPACT) + .time_millis(current_time_millis()) + .total_record_count(latest_snapshot.total_record_count()) + .delta_record_count(Some(0)) + .watermark(latest_snapshot.watermark()) + .statistics(latest_snapshot.statistics().map(str::to_string)) + .properties(latest_snapshot.properties().cloned()) + .next_row_id(latest_snapshot.next_row_id()) + .build(); + + if self.snapshot_commit.commit(&snapshot, &[]).await? { + return Ok(true); + } + + self.delete_manifest_files(file_io, &manifest_dir, &new_manifest_files) + .await; + let _ = file_io.delete_file(&base_manifest_list_path).await; + let _ = file_io.delete_file(&delta_manifest_list_path).await; + Ok(false) + } + + async fn delete_manifest_files( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifests: &[ManifestFileMeta], + ) { + for manifest in manifests { + let _ = file_io + .delete_file(&format!("{manifest_dir}/{}", manifest.file_name())) + .await; + } + } + async fn commit_with_identifier_impl( &self, commit_messages: Vec, @@ -1198,6 +1350,89 @@ impl TableCommit { Ok((result, new_files)) } + async fn compact_manifest_files_legacy( + &self, + file_io: &FileIO, + manifest_dir: &str, + manifest_files: Vec, + ) -> Result<(Vec, Vec)> { + if !reaches_full_compaction_threshold( + &manifest_files, + self.manifest_target_size, + self.manifest_sort_config.full_compaction_threshold_size, + ) { + return self + .merge_manifest_files_legacy(file_io, manifest_dir, manifest_files) + .await; + } + + let original_manifest_files = manifest_files.clone(); + let partition_fields = self.table.schema().partition_fields(); + let mut deleted_partitions = Vec::new(); + for meta in manifest_files.iter().filter(|meta| { + meta.num_deleted_files() > 0 || meta.file_size() < self.manifest_target_size + }) { + if meta.num_deleted_files() == 0 { + continue; + } + let path = format!("{manifest_dir}/{}", meta.file_name()); + for entry in Manifest::read(file_io, &path).await? { + if entry.kind() == &FileKind::Delete { + deleted_partitions.push(entry.partition().to_vec()); + } + } + } + + let mut reused = Vec::new(); + let mut to_compact = Vec::new(); + for meta in manifest_files { + let must_change = + meta.num_deleted_files() > 0 || meta.file_size() < self.manifest_target_size; + let mut overlaps_delete = false; + if !must_change { + for partition in &deleted_partitions { + if manifest_may_contain_partition(&meta, partition, &partition_fields)? { + overlaps_delete = true; + break; + } + } + } + if must_change || overlaps_delete { + to_compact.push(meta); + } else { + reused.push(meta); + } + } + + // Match the ordinary merger's singleton shortcut. A maintenance call + // must not publish a new snapshot when there is no useful rewrite. + if to_compact.len() <= 1 { + return self + .merge_manifest_files_legacy(file_io, manifest_dir, original_manifest_files) + .await; + } + + let mut entries = Vec::new(); + for meta in &to_compact { + let path = format!("{manifest_dir}/{}", meta.file_name()); + entries.extend(Manifest::read(file_io, &path).await?); + } + let active_entries = merge_active_entries(entries); + let new_files = if active_entries.is_empty() { + Vec::new() + } else { + self.write_manifest_files( + file_io, + manifest_dir, + &format!("manifest-{}", uuid::Uuid::new_v4()), + &active_entries, + ) + .await? + }; + reused.extend(new_files.iter().cloned()); + Ok((reused, new_files)) + } + async fn sort_manifest_files( &self, file_io: &FileIO, @@ -3527,6 +3762,10 @@ fn manifest_entry_sort_size(entry: &ManifestEntry) -> Result { }) } +fn same_manifest_file_set(left: &[ManifestFileMeta], right: &[ManifestFileMeta]) -> bool { + left.len() == right.len() && left.iter().all(|file| right.contains(file)) +} + enum ManifestRewriteOutput { Full { files: Vec, @@ -6750,6 +6989,111 @@ mod tests { ); } + #[tokio::test] + async fn test_compact_manifests_commits_compact_snapshot_and_is_idempotent() { + let file_io = test_file_io(); + let table_path = "memory:/test_compact_manifests"; + setup_dirs(&file_io, table_path).await; + let commit = setup_commit(&file_io, table_path); + + commit + .commit(vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("data-0.parquet", 10)], + )]) + .await + .unwrap(); + commit + .commit(vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("data-1.parquet", 20)], + )]) + .await + .unwrap(); + + let before = latest_snapshot(&file_io, table_path).await.unwrap(); + assert_eq!(before.id(), 2); + let before_files = active_entries(&file_io, table_path, &before) + .await + .into_iter() + .map(|entry| entry.file().file_name.clone()) + .collect::>(); + + commit.compact_manifests().await.unwrap(); + + let compacted = latest_snapshot(&file_io, table_path).await.unwrap(); + assert_eq!(compacted.id(), 3); + assert_eq!(compacted.commit_kind(), &CommitKind::COMPACT); + assert_eq!(compacted.total_record_count(), Some(30)); + assert_eq!(compacted.delta_record_count(), Some(0)); + let manifest_dir = format!("{table_path}/manifest"); + let base = ManifestList::read( + &file_io, + &format!("{manifest_dir}/{}", compacted.base_manifest_list()), + ) + .await + .unwrap(); + let delta = ManifestList::read( + &file_io, + &format!("{manifest_dir}/{}", compacted.delta_manifest_list()), + ) + .await + .unwrap(); + assert_eq!(base.len(), 1); + assert!(delta.is_empty()); + let after_files = active_entries(&file_io, table_path, &compacted) + .await + .into_iter() + .map(|entry| entry.file().file_name.clone()) + .collect::>(); + assert_eq!(after_files, before_files); + + commit.compact_manifests().await.unwrap(); + assert_eq!( + latest_snapshot(&file_io, table_path).await.unwrap().id(), + compacted.id(), + "a compacted singleton must not publish another snapshot" + ); + } + + #[tokio::test] + async fn test_compact_manifests_eliminates_cross_manifest_deletes() { + let file_io = test_file_io(); + let table_path = "memory:/test_compact_manifests_deletes"; + setup_dirs(&file_io, table_path).await; + let commit = setup_commit(&file_io, table_path); + + commit + .commit(vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("old.parquet", 10)], + )]) + .await + .unwrap(); + commit + .overwrite( + vec![CommitMessage::new( + vec![], + 0, + vec![test_data_file("new.parquet", 12)], + )], + None, + ) + .await + .unwrap(); + + commit.compact_manifests().await.unwrap(); + let snapshot = latest_snapshot(&file_io, table_path).await.unwrap(); + assert_eq!(snapshot.commit_kind(), &CommitKind::COMPACT); + let entries = active_entries(&file_io, table_path, &snapshot).await; + assert_eq!(entries.len(), 1); + assert_eq!(entries[0].file().file_name, "new.parquet"); + assert_eq!(*entries[0].kind(), FileKind::Add); + } + fn sorted_partition_commit(file_io: &FileIO, path: &str, full: bool) -> TableCommit { let mut options = HashMap::from([ ("manifest-sort.enabled".to_string(), "true".to_string()),