diff --git a/crates/paimon/src/deletion_vector/core.rs b/crates/paimon/src/deletion_vector/core.rs index e274e3017..57531fdad 100644 --- a/crates/paimon/src/deletion_vector/core.rs +++ b/crates/paimon/src/deletion_vector/core.rs @@ -65,6 +65,17 @@ impl DeletionVector { (*self.bitmap).clone() } + pub(crate) fn retained_bytes(&self) -> usize { + let stats = self.bitmap.statistics(); + let bytes = stats + .n_bytes_array_containers + .saturating_add(stats.n_bytes_bitset_containers) + .saturating_add(stats.n_bytes_run_containers) + .saturating_add(u64::from(stats.n_containers).saturating_mul(64)) + .saturating_add(64); + usize::try_from(bytes).unwrap_or(usize::MAX) + } + /// Number of deleted positions in this vector. pub fn cardinality(&self) -> u64 { self.bitmap.len() @@ -429,6 +440,19 @@ mod tests { use super::*; use roaring::RoaringBitmap; use std::env::current_dir; + + #[test] + fn dense_vectors_charge_container_memory_instead_of_one_integer_per_deletion() { + let mut bitmap = RoaringBitmap::new(); + bitmap.insert_range(0..8_500_000); + let vector = DeletionVector::from_bitmap(bitmap); + assert_eq!(vector.cardinality(), 8_500_000); + assert!(vector.retained_bytes() < 4 * 1024 * 1024); + assert!( + vector.retained_bytes() + >= vector.bitmap.statistics().n_bytes_bitset_containers as usize + ); + } #[test] fn test_read_deletion_vector() { let workdir = current_dir().unwrap(); diff --git a/crates/paimon/src/table/compact.rs b/crates/paimon/src/table/compact.rs index 3ec88e043..d3c56c9db 100644 --- a/crates/paimon/src/table/compact.rs +++ b/crates/paimon/src/table/compact.rs @@ -19,7 +19,9 @@ mod append; pub(super) mod data_evolution; -mod primary_key; +pub(super) mod deletion_vectors; +pub(super) mod primary_key; +use deletion_vectors::DeletionVectorUpdate; use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::Arc; @@ -112,12 +114,14 @@ impl<'a> CompactBuilder<'a> { pub fn new_executor(&self) -> Result { ensure_supported(self.table)?; let options = self.table.schema().core_options(); - let lookup = options.try_changelog_producer()? == ChangelogProducer::Lookup; + let lookup = options.try_changelog_producer()? == ChangelogProducer::Lookup + || (options.deletion_vectors_enabled() + && !self.table.schema().primary_keys().is_empty()); let local_cache = self.table.file_io().local_cache(); if lookup && local_cache.is_none() { return Err(Error::ConfigInvalid { message: - "Lookup compaction requires local-cache.enabled=true on the catalog/FileIO" + "Lookup or deletion-vector compaction requires local-cache.enabled=true on the catalog/FileIO" .into(), }); } @@ -260,6 +264,11 @@ impl CompactPlanner { .get_snapshot(snapshot_id) .await?; validate_snapshot_indexes(&self.table, &snapshot).await?; + validate_dv_schemas( + &self.table, + source.splits().iter().flat_map(|split| split.data_files()), + ) + .await?; let context = CompactContext { plan_id: uuid::Uuid::new_v4().to_string(), location: self.table.location().to_string(), @@ -297,6 +306,8 @@ pub struct CompactResult { files: Vec, #[serde(default)] changelog_files: Vec, + #[serde(default)] + deletion_vector_updates: Option>, } impl CompactResult { @@ -355,6 +366,7 @@ impl CompactExecutor { } validate_snapshot_indexes(&self.table, &snapshot).await?; let active = read_snapshot_entries(&manager, &snapshot, None).await?; + validate_dv_schemas(&self.table, active.iter().map(|entry| entry.file())).await?; task.validate_inputs(&active)?; let table = self .table @@ -383,6 +395,11 @@ impl CompactExecutor { .with_bucket_path(bucket_path) .with_data_files(task.files.clone()) .build()?; + let split = if options.deletion_vectors_enabled() && task.is_primary_key() { + primary_key::deletion_vectors::with_deletion_files(&table, &split, &snapshot).await? + } else { + split + }; if task.is_data_evolution() { let files = data_evolution::execute( &table, @@ -396,9 +413,11 @@ impl CompactExecutor { task: task.clone(), files, changelog_files: vec![], + deletion_vector_updates: None, }); } if task.is_primary_key() { + let mut deletions = None; let lookup = if let Some(cache) = &self.lookup_cache { let history = active .iter() @@ -413,26 +432,44 @@ impl CompactExecutor { if history.is_empty() { None } else { - Some( - primary_key::lookup_store::provider( - &table, - &split, - history, - cache.clone(), - self.resources.clone(), - ) - .await?, + let (provider, collected) = primary_key::lookup_store::provider( + &table, + &split, + history, + cache.clone(), + self.resources.clone(), ) + .await?; + deletions = collected; + Some(provider) } } else { None }; let (files, changelog_files) = primary_key::execute(&table, task, &split, self.resources.clone(), lookup).await?; + let mut deletion_vector_updates = Vec::new(); + if let Some(deletions) = deletions { + for (name, bitmap) in deletions.lock().unwrap().iter() { + let mut bytes = Vec::new(); + bitmap + .serialize_into(&mut bytes) + .map_err(|error| invalid(&error.to_string()))?; + deletion_vector_updates.push(DeletionVectorUpdate { + partition: task.partition.clone(), + bucket: task.bucket, + file_name: name.clone(), + bitmap: bytes, + }); + } + } return Ok(CompactResult { task: task.clone(), files, changelog_files, + deletion_vector_updates: options + .deletion_vectors_enabled() + .then_some(deletion_vector_updates), }); } let mut read = table.new_read_builder().new_read()?.with_ordered(true); @@ -484,6 +521,7 @@ impl CompactExecutor { task: task.clone(), files, changelog_files: vec![], + deletion_vector_updates: None, }), Err(error) => { writer.abort().await; @@ -618,6 +656,7 @@ impl CompactTask { pub(super) fn validate_primary_key_changes( schema: &TableSchema, changes: &[ManifestEntry], + changelog: &[ManifestEntry], captured: &[ManifestEntry], ) -> Result<()> { fn group(entry: &ManifestEntry) -> (&[u8], i32, i32) { @@ -678,6 +717,7 @@ pub(super) fn validate_primary_key_changes( if changes .iter() .filter(|entry| *entry.kind() == FileKind::Add) + .chain(changelog.iter()) .any(|entry| output_levels.get(&group(entry)) != Some(&entry.file().level)) { return Err(invalid( @@ -691,6 +731,7 @@ pub(super) fn validate_primary_key_changes( pub(super) struct PreparedCompact { pub entries: Vec, pub changelog_entries: Vec, + pub deletion_vector_updates: Option>, pub schema_id: i64, pub snapshot_id: i64, pub read_snapshot: Option>, @@ -708,6 +749,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul let mut outputs = HashSet::new(); let mut entries = Vec::new(); let mut changelog_entries = Vec::new(); + let mut deletion_vector_updates = Some(Vec::new()); let mut metrics = CompactMetrics::default(); for result in results { let task = &result.task; @@ -777,12 +819,39 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul return Err(invalid("Compaction must preserve the input row count")); } let options = task.context.schema.core_options(); - let ignores_retracts = task.is_primary_key() - && options.ignore_delete() - && matches!( - options.merge_engine()?, - MergeEngine::FirstRow | MergeEngine::PartialUpdate - ); + let dv = task.is_primary_key() && options.deletion_vectors_enabled(); + let ignores_retracts = dv + || (task.is_primary_key() + && options.ignore_delete() + && matches!( + options.merge_engine()?, + MergeEngine::FirstRow | MergeEngine::PartialUpdate + )); + if dv + && result.files.iter().any(|file| { + file.delete_row_count != Some(0) || file.row_count > i64::from(u32::MAX) + 1 + }) + { + return Err(invalid( + "Deletion-vector compaction data must contain only add records", + )); + } + let updates = result + .deletion_vector_updates + .as_deref() + .unwrap_or_default(); + if dv && result.deletion_vector_updates.is_none() { + // Require this only after already-committed results are recovered. + deletion_vector_updates = None; + } + if !dv && !updates.is_empty() { + return Err(invalid( + "Only deletion-vector compaction may publish row deletions", + )); + } + if let Some(collected) = &mut deletion_vector_updates { + collected.extend(updates.iter().cloned()); + } if task.is_primary_key() && (output_rows > input_rows || (input_rows > 0 && output_rows == 0 && !ignores_retracts)) @@ -798,17 +867,12 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul )); } let mut changelog_rows = 0u64; - let changelog_level = output_level.unwrap_or_else(|| { - result - .files - .first() - .map_or_else(|| task.output_level(), |file| file.level) - }); + let changelog_level = output_level.or_else(|| result.files.first().map(|file| file.level)); for file in &result.changelog_files { validate_file(file, true)?; if file.schema_id != task.context.schema.id() || file.file_source != Some(1) - || file.level != changelog_level + || changelog_level.is_some_and(|level| file.level != level) || file .delete_row_count .is_none_or(|count| count > file.row_count) @@ -860,6 +924,7 @@ pub(super) fn prepare_results(table: &Table, results: &[CompactResult]) -> Resul Ok(PreparedCompact { entries, changelog_entries, + deletion_vector_updates, schema_id: first.task.context.schema.id(), snapshot_id: first.task.snapshot_id(), read_snapshot: first @@ -909,7 +974,7 @@ pub(super) fn ensure_supported(table: &Table) -> Result<()> { if table.is_format_table() || options.row_tracking_enabled() || options.data_evolution_enabled() - || options.deletion_vectors_enabled() + || (options.deletion_vectors_enabled() && table.schema().primary_keys().is_empty()) || options.query_auth_enabled() || !options.blob_fields().is_empty() || table @@ -941,15 +1006,55 @@ pub(super) fn ensure_supported(table: &Table) -> Result<()> { message: "PK compaction requires fixed or dynamic buckets and changelog-producer=none, input or lookup; first-row requires changelog-producer=none or lookup".into(), }); } + if options.deletion_vectors_enabled() && options.merge_engine()? != MergeEngine::Deduplicate { + return Err(Error::Unsupported { + message: "Primary-key deletion-vector compaction requires deduplicate".into(), + }); + } if options.try_changelog_producer()? == ChangelogProducer::Lookup { primary_key::lookup::validate_options(properties)?; } options.validate_data_file_path_directory() } +/// Dropping DV tombstones relies on every historical version having used DV. +/// Switching an existing non-DV table requires a migration Rust does not offer. +pub(super) async fn validate_dv_schemas<'a>( + table: &Table, + files: impl Iterator, +) -> Result<()> { + if table.schema().primary_keys().is_empty() + || !table.schema().core_options().deletion_vectors_enabled() + { + return Ok(()); + } + let mut schemas = HashSet::new(); + for file in files { + if file.row_count > i64::from(u32::MAX) + 1 { + return Err(invalid( + "Deletion-vector compaction file exceeds u32 physical positions", + )); + } + if schemas.insert(file.schema_id) + && !table + .schema_manager() + .schema(file.schema_id) + .await? + .core_options() + .deletion_vectors_enabled() + { + return Err(Error::Unsupported { + message: "Deletion-vector compaction cannot migrate files written without deletion vectors".into(), + }); + } + } + Ok(()) +} + fn ensure_snapshot_supported(table: &Table, snapshot: &Snapshot) -> Result<()> { if snapshot.index_manifest().is_some() && !table.schema().core_options().data_evolution_enabled() + && !table.schema().core_options().deletion_vectors_enabled() && (table.schema().primary_keys().is_empty() || table.schema().core_options().bucket() != -1) { @@ -978,8 +1083,11 @@ pub(super) async fn validate_snapshot_indexes( return Ok(Some(name.to_string())); } if entries.iter().any(|entry| { - entry.index_file.index_type != "HASH" - || entry.index_file.deletion_vectors_ranges.is_some() + !(entry.index_file.index_type == "HASH" + || (table.schema().core_options().deletion_vectors_enabled() + && entry.index_file.index_type == "DELETION_VECTORS")) + || (entry.index_file.index_type == "HASH" + && entry.index_file.deletion_vectors_ranges.is_some()) || entry.index_file.global_index_meta.is_some() }) { return Err(Error::Unsupported { diff --git a/crates/paimon/src/table/compact/data_evolution.rs b/crates/paimon/src/table/compact/data_evolution.rs index 07ea02f6b..5bd71291d 100644 --- a/crates/paimon/src/table/compact/data_evolution.rs +++ b/crates/paimon/src/table/compact/data_evolution.rs @@ -22,20 +22,19 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::Arc; use futures::TryStreamExt; +use indexmap::IndexMap; use roaring::RoaringBitmap; use super::{invalid, BucketFiles, CompactContext, CompactTask}; use crate::resource::ResourceContext; use crate::spec::{ - bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, FileKind, - IndexManifest, IndexManifestEntry, ManifestEntry, PartitionComputer, Snapshot, + CoreOptions, DataField, DataFileMeta, DataType, FileKind, IndexManifestEntry, ManifestEntry, }; use crate::table::data_file_index_writer::FileIndexOptions; use crate::table::data_file_writer::DataFileWriter; -use crate::table::index_file_path::committed_index_file_path; use crate::table::source::{data_evolution_anchor_file, is_data_evolution_normal_file}; use crate::table::stats_filter::group_by_overlapping_row_id; -use crate::table::{DataEvolutionDeleteWriter, DataSplit, Table, TableScan}; +use crate::table::{DataSplit, Table, TableScan}; use crate::{Error, Result}; pub(super) fn ensure_supported(table: &Table) -> Result<()> { @@ -527,114 +526,52 @@ pub(super) fn validate_indexes(entries: &[IndexManifestEntry]) -> Result<()> { Ok(()) } -/// Rebase DVs on the latest snapshot on every CAS attempt. A concurrent DELETE -/// is carried forward, including deletions in untouched groups of this bucket. -pub(in crate::table) async fn rewrite_deletion_vectors( - table: &Table, - snapshot: &Snapshot, - delta: &[ManifestEntry], - created_files: &mut Vec, -) -> Result>> { - let Some(name) = snapshot.index_manifest() else { - return Ok(None); - }; - let mut indexes = IndexManifest::read( - table.file_io(), - &table.snapshot_manager().manifest_path(name), - ) - .await?; - let mut buckets = BTreeMap::<_, (Vec, Vec)>::new(); - for entry in delta { - let files = buckets - .entry((entry.partition().to_vec(), entry.bucket())) - .or_default(); - match entry.kind() { - FileKind::Delete => files.0.push(entry.file().clone()), - FileKind::Add => files.1.push(entry.file().clone()), - } - } - let writer = DataEvolutionDeleteWriter::new(table)?; - let mut changed = false; - for ((partition, bucket), (before, after)) in buckets { - let groups = group_by_overlapping_row_id(before); - let anchors = groups - .iter() - .map(|group| Ok(data_evolution_anchor_file(group)?.file_name.clone())) - .collect::>>()?; - // Like Java AppendDeleteFileMaintainer, replace only index files - // touched by selected anchors. Untouched files remain independent. - let (mut bitmaps, removed) = writer - .read_existing_bucket_deletion_vectors(&partition, bucket, snapshot, Some(&anchors)) - .await?; - let mut moved = false; - for group in groups { - let anchor = data_evolution_anchor_file(&group)?; - let Some(bitmap) = bitmaps.shift_remove(&anchor.file_name) else { - continue; - }; - moved = true; - let old_first = anchor.first_row_id.unwrap(); - for position in bitmap { - if i64::from(position) >= anchor.row_count { - return Err(invalid( - "Deletion vector position is outside its compact input", - )); - } - let absolute = old_first + i64::from(position); - let output = after - .iter() - .find(|file| { - file.row_id_range() - .is_some_and(|(start, end)| start <= absolute && absolute <= end) - }) - .ok_or_else(|| invalid("Compacted files do not cover a deleted row ID"))?; - let relative = u32::try_from(absolute - output.first_row_id.unwrap()) - .map_err(|_| invalid("Compacted deletion vector position exceeds u32::MAX"))?; - bitmaps - .entry(output.file_name.clone()) - .or_insert_with(RoaringBitmap::new) - .insert(relative); - } - } - if !moved { +/// Containers referencing these anchors participate in the row-ID remapping. +pub(super) fn deletion_vector_anchors(before: &[DataFileMeta]) -> Result> { + group_by_overlapping_row_id(before.to_vec()) + .iter() + .map(|group| Ok(data_evolution_anchor_file(group)?.file_name.clone())) + .collect() +} + +/// Move anchor-relative positions into the compacted files, retaining masks +/// belonging to other row-ID groups that share the selected containers. +pub(super) fn remap_deletion_vectors( + before: &[DataFileMeta], + after: &[DataFileMeta], + vectors: &mut IndexMap, +) -> Result { + let mut moved = false; + for group in group_by_overlapping_row_id(before.to_vec()) { + let anchor = data_evolution_anchor_file(&group)?; + let Some(bitmap) = vectors.shift_remove(&anchor.file_name) else { continue; + }; + moved = true; + let old_first = anchor.first_row_id.unwrap(); + for position in bitmap { + if i64::from(position) >= anchor.row_count { + return Err(invalid( + "Deletion vector position is outside its compact input", + )); + } + let absolute = old_first + i64::from(position); + let output = after + .iter() + .find(|file| { + file.row_id_range() + .is_some_and(|(start, end)| start <= absolute && absolute <= end) + }) + .ok_or_else(|| invalid("Compacted files do not cover a deleted row ID"))?; + let relative = u32::try_from(absolute - output.first_row_id.unwrap()) + .map_err(|_| invalid("Compacted deletion vector position exceeds u32::MAX"))?; + vectors + .entry(output.file_name.clone()) + .or_default() + .insert(relative); } - let file = writer - .write_deletion_vector_index_file(&partition, bucket, bitmaps) - .await?; - let computer = PartitionComputer::new( - table.schema().partition_keys(), - table.schema().fields(), - table.schema().core_options().partition_default_name(), - table.schema().core_options().legacy_partition_name(), - )?; - let bucket_dir = bucket_path( - &table.data_file_location(), - Some(&computer), - &BinaryRow::from_serialized_bytes(&partition)?, - bucket, - )?; - created_files.push(committed_index_file_path( - table.location(), - &bucket_dir, - table.schema().core_options().index_file_in_data_file_dir(), - &file, - )); - indexes.retain(|entry| { - !removed.iter().any(|old| { - entry.partition == partition && entry.bucket == bucket && entry.index_file == *old - }) - }); - indexes.push(IndexManifestEntry { - version: 1, - kind: FileKind::Add, - partition, - bucket, - index_file: file, - }); - changed = true; } - Ok(changed.then_some(indexes)) + Ok(moved) } #[cfg(test)] diff --git a/crates/paimon/src/table/compact/deletion_vectors.rs b/crates/paimon/src/table/compact/deletion_vectors.rs new file mode 100644 index 000000000..0f79681ff --- /dev/null +++ b/crates/paimon/src/table/compact/deletion_vectors.rs @@ -0,0 +1,160 @@ +// 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. + +//! Prepare compaction deletion-vector indexes against each commit attempt's snapshot. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +use crate::spec::{ + DataFileMeta, FileKind, IndexManifest, IndexManifestEntry, ManifestEntry, Snapshot, +}; +use crate::table::deletion_vector_index; +use crate::table::Table; +use crate::Result; + +/// Physical rows discovered by an executor, merged with the latest DV at commit. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(in crate::table) struct DeletionVectorUpdate { + pub partition: Vec, + pub bucket: i32, + pub file_name: String, + #[serde(with = "serde_bytes")] + pub bitmap: Vec, +} + +type Bucket = (Vec, i32); + +/// Both modes prepare the same index-manifest replacement. Their container +/// selection and physical-position transformations remain mode-specific. +pub(in crate::table) async fn rewrite_indexes( + table: &Table, + latest: &Snapshot, + delta: &[ManifestEntry], + updates: &[DeletionVectorUpdate], + created_files: &mut Vec, +) -> Result>> { + let data_evolution = table.schema().core_options().data_evolution_enabled(); + let mut buckets = BTreeMap::< + Bucket, + ( + Vec, + Vec, + Vec<&DeletionVectorUpdate>, + ), + >::new(); + for entry in delta { + let bucket = buckets + .entry((entry.partition().to_vec(), entry.bucket())) + .or_default(); + match entry.kind() { + FileKind::Delete => bucket.0.push(entry.file().clone()), + FileKind::Add => bucket.1.push(entry.file().clone()), + } + } + for update in updates { + buckets + .entry((update.partition.clone(), update.bucket)) + .or_default() + .2 + .push(update); + } + let indexes = match latest.index_manifest() { + Some(name) => { + IndexManifest::read( + table.file_io(), + &table.snapshot_manager().manifest_path(name), + ) + .await? + } + None if data_evolution => return Ok(None), + None => Vec::new(), + }; + // Group only metadata, then decode and release one bucket's masks at a time. + let mut retained = Vec::new(); + let mut containers = BTreeMap::>::new(); + for entry in indexes { + let key = (entry.partition.clone(), entry.bucket); + if entry.kind == FileKind::Add + && entry.index_file.index_type == "DELETION_VECTORS" + && buckets.contains_key(&key) + { + containers.entry(key).or_default().push(entry); + } else { + retained.push(entry); + } + } + let mut changed = false; + for ((partition, bucket), (before, after, additions)) in buckets { + let selected_files = if data_evolution { + Some(super::data_evolution::deletion_vector_anchors(&before)?) + } else { + None + }; + let mut selected = Vec::new(); + for entry in containers + .remove(&(partition.clone(), bucket)) + .into_iter() + .flatten() + { + if selected_files.as_ref().is_none_or(|files| { + entry + .index_file + .deletion_vectors_ranges + .as_ref() + .is_some_and(|ranges| ranges.keys().any(|name| files.contains(name))) + }) { + selected.push(entry); + } else { + retained.push(entry); + } + } + if !data_evolution && selected.is_empty() && additions.is_empty() { + continue; + } + let mut masks = + deletion_vector_index::read_containers(table, &partition, bucket, &selected).await?; + if data_evolution { + if !super::data_evolution::remap_deletion_vectors(&before, &after, &mut masks)? { + retained.extend(selected); + continue; + } + } else { + super::primary_key::deletion_vectors::apply_deletion_vectors( + &before, &additions, &mut masks, + )?; + } + // DE can explicitly replace an empty container; PK retires it entirely. + if data_evolution || masks.values().any(|mask| !mask.is_empty()) { + let file = + deletion_vector_index::write_bucket(table, &partition, bucket, masks).await?; + created_files.push(deletion_vector_index::file_path( + table, &partition, bucket, &file, + )?); + retained.push(IndexManifestEntry { + version: 1, + kind: FileKind::Add, + partition, + bucket, + index_file: file, + }); + } + changed = true; + } + Ok(changed.then_some(retained)) +} diff --git a/crates/paimon/src/table/compact/primary_key.rs b/crates/paimon/src/table/compact/primary_key.rs index 9f65eb6df..8320448c9 100644 --- a/crates/paimon/src/table/compact/primary_key.rs +++ b/crates/paimon/src/table/compact/primary_key.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +pub(in crate::table) mod deletion_vectors; pub(super) mod lookup; pub(super) mod lookup_file; pub(super) mod lookup_store; @@ -22,6 +23,7 @@ pub(in crate::table) mod planner; use std::sync::Arc; +use arrow_array::{BooleanArray, Int8Array}; use arrow_schema::SchemaRef; use futures::TryStreamExt; @@ -312,6 +314,7 @@ pub(super) async fn execute( let schema = table.schema(); let options = schema.core_options(); let lookup_changelog = options.try_changelog_producer()? == ChangelogProducer::Lookup; + let drop_delete = options.deletion_vectors_enabled() && task.output_level() > 0; let reader = new_reader(table, resources.clone())?.with_lookup(lookup); let mut batches = reader.read_with_merge_function(std::slice::from_ref(split), move |config, fields| { @@ -375,11 +378,12 @@ pub(super) async fn execute( managed_blob_enabled: false, resources: resources.as_ref(), }; - let mut writer = KeyValueRollingWriter::new( - file_options, - options.target_file_size(), - options.target_file_row_num()?, - ); + let mut row_limit = options.target_file_row_num()?; + if options.deletion_vectors_enabled() { + row_limit = row_limit.min(i64::from(u32::MAX) + 1); + } + let mut writer = + KeyValueRollingWriter::new(file_options, options.target_file_size(), row_limit); let changelog_format = options.changelog_file_format(); let mut changelog_writer = KeyValueRollingWriter::new( KeyValueFileOptions { @@ -412,6 +416,27 @@ pub(super) async fn execute( })?; changelog_writer.write(&batch).await?; } else { + // Generate lookup changelog from the winning KV first. In DV + // mode the history is physically deleted, so its winning + // retract need not remain in the new positive-level data. + let batch = if drop_delete { + let kinds = batch + .column_by_name(VALUE_KIND_FIELD_NAME) + .and_then(|column| column.as_any().downcast_ref::()) + .ok_or_else(|| super::invalid("Invalid compaction value-kind column"))?; + let live = kinds + .iter() + .map(|kind| { + let kind = + kind.ok_or_else(|| super::invalid("Null compaction value kind"))?; + Ok(RowKind::from_value(kind)?.is_add()) + }) + .collect::>>()?; + arrow_select::filter::filter_record_batch(&batch, &BooleanArray::from(live)) + .map_err(|error| super::invalid(&error.to_string()))? + } else { + batch + }; writer.write(&batch).await?; } } diff --git a/crates/paimon/src/table/compact/primary_key/deletion_vectors.rs b/crates/paimon/src/table/compact/primary_key/deletion_vectors.rs new file mode 100644 index 000000000..8f742844b --- /dev/null +++ b/crates/paimon/src/table/compact/primary_key/deletion_vectors.rs @@ -0,0 +1,305 @@ +// 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. + +//! Captured physical deletions and atomic retirement of compacted DV ranges. + +use std::collections::{HashMap, HashSet}; + +use indexmap::IndexMap; +use roaring::RoaringBitmap; + +use super::super::deletion_vectors::DeletionVectorUpdate; +use crate::deletion_vector::DeletionVectorFactory; +use crate::spec::{ + BinaryRow, DataFileMeta, DeletionVectorMeta, FileKind, IndexManifest, IndexManifestEntry, + ManifestEntry, PartitionComputer, Snapshot, +}; +use crate::table::index_file_path::IndexFileLocation; +use crate::table::{DataSplit, Table}; +use crate::{DeletionFile, Result}; + +type FileId = (Vec, i32, String); +type VectorReference = (String, Option, DeletionVectorMeta); + +async fn read_indexes(table: &Table, snapshot: &Snapshot) -> Result> { + match snapshot.index_manifest() { + Some(name) => { + IndexManifest::read( + table.file_io(), + &table.snapshot_manager().manifest_path(name), + ) + .await + } + None => Ok(vec![]), + } +} + +fn vector_references(indexes: &[IndexManifestEntry]) -> Result> { + let mut references = HashMap::new(); + for entry in indexes { + if entry.kind != FileKind::Add || entry.index_file.index_type != "DELETION_VECTORS" { + continue; + } + if let Some(ranges) = &entry.index_file.deletion_vectors_ranges { + for (name, meta) in ranges { + let id = (entry.partition.clone(), entry.bucket, name.clone()); + if references + .insert( + id, + ( + entry.index_file.file_name.clone(), + entry.index_file.external_path.clone(), + meta.clone(), + ), + ) + .is_some() + { + return Err(super::super::invalid( + "Multiple deletion vectors reference one data file", + )); + } + } + } + } + Ok(references) +} + +fn resolve_vector(table: &Table, bucket_path: &str, reference: &VectorReference) -> DeletionFile { + let location = IndexFileLocation::BucketLocal { + table_path: table.location().trim_end_matches('/'), + bucket_path, + index_file_in_data_file_dir: table.schema().core_options().index_file_in_data_file_dir(), + }; + DeletionFile::new( + location.resolve(&reference.0, reference.1.as_deref()), + i64::from(reference.2.offset), + i64::from(reference.2.length), + reference.2.cardinality, + ) +} + +async fn read_bitmap( + table: &Table, + path: &str, + reference: Option<&VectorReference>, +) -> Result { + match reference { + Some(reference) => Ok(DeletionVectorFactory::read( + table.file_io(), + &resolve_vector(table, path, reference), + ) + .await? + .to_bitmap()), + None => Ok(RoaringBitmap::new()), + } +} + +fn bucket_path(table: &Table, partition: &[u8], bucket: i32) -> Result { + let schema = table.schema(); + let options = schema.core_options(); + let computer = PartitionComputer::new( + schema.partition_keys(), + schema.fields(), + options.partition_default_name(), + options.legacy_partition_name(), + )?; + crate::spec::bucket_path( + &table.data_file_location(), + Some(&computer), + &BinaryRow::from_serialized_bytes(partition)?, + bucket, + ) +} + +/// Apply captured physical masks before merging, including masks on L0 files. +pub(in crate::table::compact) async fn with_deletion_files( + table: &Table, + split: &DataSplit, + snapshot: &Snapshot, +) -> Result { + let references = vector_references(&read_indexes(table, snapshot).await?)?; + let partition = split.partition().to_serialized_bytes(); + let files = split + .data_files() + .iter() + .map(|file| { + references + .get(&(partition.clone(), split.bucket(), file.file_name.clone())) + .map(|reference| resolve_vector(table, split.bucket_path(), reference)) + }) + .collect(); + DataSplit::builder() + .with_snapshot(split.snapshot_id()) + .with_partition(split.partition().clone()) + .with_bucket(split.bucket()) + .with_total_buckets(split.total_buckets()) + .with_bucket_path(split.bucket_path().to_string()) + .with_data_files(split.data_files().to_vec()) + .with_data_deletion_files(files) + .build() +} + +/// Selected masks must be unchanged. Actual point-lookup hits must still be +/// live; unrelated deletions in their files can safely be rebased by union. +pub(in crate::table) async fn validate_dependencies( + table: &Table, + captured: &Snapshot, + latest: &Snapshot, + delta: &[ManifestEntry], + updates: &[DeletionVectorUpdate], +) -> Result<()> { + let captured_vectors = vector_references(&read_indexes(table, captured).await?)?; + let same_snapshot = captured.has_same_read_metadata(latest); + let latest_references = if same_snapshot { + None + } else { + Some(vector_references(&read_indexes(table, latest).await?)?) + }; + let latest_vectors = latest_references.as_ref().unwrap_or(&captured_vectors); + let selected = delta + .iter() + .filter(|entry| *entry.kind() == FileKind::Delete) + .map(|entry| { + ( + entry.partition().to_vec(), + entry.bucket(), + entry.file().file_name.clone(), + ) + }) + .collect::>(); + for id in &selected { + let before = captured_vectors.get(id); + let after = latest_vectors.get(id); + if before == after { + continue; + } + // Updating another range can repack this mask into a new immutable + // container without changing any selected physical row. + let path = bucket_path(table, &id.0, id.1)?; + if read_bitmap(table, &path, before).await? != read_bitmap(table, &path, after).await? { + return Err(super::super::invalid( + "Compaction input deletion vectors changed; create a new plan", + )); + } + } + let manager = table.snapshot_manager(); + let captured_files = + crate::table::snapshot_manifest::read_snapshot_entries(&manager, captured, None).await?; + super::super::validate_dv_schemas(table, captured_files.iter().map(|entry| entry.file())) + .await?; + if updates.is_empty() { + return Ok(()); + } + let files = |entries: Vec| { + entries + .into_iter() + .map(|entry| { + ( + ( + entry.partition().to_vec(), + entry.bucket(), + entry.file().file_name.clone(), + ), + entry, + ) + }) + .collect::>() + }; + let captured_files = files(captured_files); + let latest_entries = if same_snapshot { + None + } else { + Some(files( + crate::table::snapshot_manifest::read_snapshot_entries(&manager, latest, None).await?, + )) + }; + let latest_files = latest_entries.as_ref().unwrap_or(&captured_files); + let buckets = selected + .iter() + .map(|id| (&id.0, id.1)) + .collect::>(); + let mut hits = HashMap::::new(); + for update in updates { + let id = ( + update.partition.clone(), + update.bucket, + update.file_name.clone(), + ); + let bitmap = RoaringBitmap::deserialize_from(update.bitmap.as_slice()) + .map_err(|error| super::super::invalid(&error.to_string()))?; + let captured_file = captured_files + .get(&id) + .ok_or_else(|| super::super::invalid("Lookup deletion target was not captured"))?; + let latest_file = latest_files.get(&id).ok_or_else(|| { + super::super::invalid("Lookup deletion target was removed; create a new plan") + })?; + if selected.contains(&id) + || !buckets.contains(&(&id.0, id.1)) + || captured_file.file().level <= 0 + || captured_file.file() != latest_file.file() + || captured_file.total_buckets() != latest_file.total_buckets() + || bitmap.is_empty() + || bitmap + .max() + .is_none_or(|position| i64::from(position) >= captured_file.file().row_count) + { + return Err(super::super::invalid( + "Invalid or changed lookup deletion target", + )); + } + *hits.entry(id).or_default() |= bitmap; + } + for (id, positions) in hits { + let path = bucket_path(table, &id.0, id.1)?; + let before = captured_vectors.get(&id); + let after = latest_vectors.get(&id); + for reference in [before, after].into_iter().flatten() { + let vector = DeletionVectorFactory::read( + table.file_io(), + &resolve_vector(table, &path, reference), + ) + .await?; + if !vector.to_bitmap().is_disjoint(&positions) { + return Err(super::super::invalid( + "Lookup dependency deletion vectors changed; create a new plan", + )); + } + if before == after { + break; + } + } + } + Ok(()) +} + +/// Retire rewritten files' masks and merge physical point-lookup deletions into +/// the retained files' latest masks. +pub(in crate::table::compact) fn apply_deletion_vectors( + before: &[DataFileMeta], + updates: &[&DeletionVectorUpdate], + vectors: &mut IndexMap, +) -> Result<()> { + for file in before { + vectors.shift_remove(&file.file_name); + } + for update in updates { + let bitmap = RoaringBitmap::deserialize_from(update.bitmap.as_slice()) + .map_err(|error| super::super::invalid(&error.to_string()))?; + *vectors.entry(update.file_name.clone()).or_default() |= bitmap; + } + Ok(()) +} diff --git a/crates/paimon/src/table/compact/primary_key/lookup_file.rs b/crates/paimon/src/table/compact/primary_key/lookup_file.rs index 2c8d94326..311a969f2 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup_file.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup_file.rs @@ -36,9 +36,9 @@ use std::collections::HashMap; use std::io::{Cursor, Write}; use std::sync::{Arc, Mutex}; -const PREFIX: &str = "__paimon_generated__/lookup-sst-v1"; +const PREFIX: &str = "__paimon_generated__/lookup-sst-v2"; const HEAD_LEN: usize = 24; -const POINTER_LEN: usize = 37; +const POINTER_LEN: usize = 45; struct Payload { path: String, @@ -64,6 +64,13 @@ pub(in crate::table::compact) struct LookupFileCache { index: Mutex>>, // One decoded IPC batch across all files, never one cache per file/key. payload: Mutex>, + vector: Mutex< + Option<( + crate::DeletionFile, + crate::deletion_vector::DeletionVector, + Option, + )>, + >, } impl LookupFileCache { @@ -115,6 +122,7 @@ impl LookupFileCache { working: tokio::sync::Mutex::new(()), index: Mutex::new(None), payload: Mutex::new(None), + vector: Mutex::new(None), }) } @@ -125,6 +133,21 @@ impl LookupFileCache { keys: &[usize], input: F, ) -> Result> + where + F: Fn() -> Result + Send, + { + self.lookup_with_position(identity, key, keys, input) + .await + .map(|row| row.map(|(batch, _)| batch)) + } + + pub(super) async fn lookup_with_position( + &self, + identity: &str, + key: &[u8], + keys: &[usize], + input: F, + ) -> Result> where F: Fn() -> Result + Send, { @@ -155,6 +178,46 @@ impl LookupFileCache { unreachable!() } + /// One snapshot-dependent bitmap shares the executor's working budget. + pub(super) async fn is_deleted( + &self, + io: &crate::io::FileIO, + file: &crate::DeletionFile, + position: u64, + resources: Option<&crate::resource::ResourceContext>, + ) -> Result { + let _working = self.working.lock().await; + { + let cached = self.vector.lock().unwrap(); + if let Some((_, vector, _)) = cached.as_ref().filter(|(key, _, _)| key == file) { + return Ok(vector.is_deleted(position)); + } + } + *self.vector.lock().unwrap() = None; + if usize::try_from(file.length()).map_or(true, |size| size > self.limit) { + return Err(exhausted( + "Lookup deletion vector exceeds the memory budget", + )); + } + let vector = crate::deletion_vector::DeletionVectorFactory::read(io, file).await?; + let size = vector.retained_bytes(); + if size > self.limit { + return Err(exhausted( + "Lookup deletion vector exceeds the memory budget", + )); + } + let reservation = resources + .map(|resources| { + let mut reservation = resources.reservation(); + reservation.try_grow(size)?; + Ok::<_, Error>(reservation) + }) + .transpose()?; + let deleted = vector.is_deleted(position); + *self.vector.lock().unwrap() = Some((file.clone(), vector, reservation)); + Ok(deleted) + } + async fn build( &self, head: &str, @@ -191,6 +254,7 @@ impl LookupFileCache { ); let mut previous = None::>; let mut converter = None; + let mut position = 0u64; while let Some(batch) = input.try_next().await? { if batch.num_rows() == 0 { continue; @@ -256,6 +320,7 @@ impl LookupFileCache { length: bytes.len() as u64, raw_length: ipc.bytes.len() as u64, row: 0, + position: 0, crc: crc32fast::hash(&bytes), compression, }; @@ -282,6 +347,7 @@ impl LookupFileCache { key.as_ref(), &Pointer { row: row as u64, + position: position + row as u64, ..pointer } .encode(), @@ -294,6 +360,9 @@ impl LookupFileCache { )); } } + position = position + .checked_add(batch.num_rows() as u64) + .ok_or_else(|| invalid("Lookup physical position overflow"))?; } writer.flush().await.map_err(exhausted)?; if writer.retained_bytes() > self.limit { @@ -370,7 +439,7 @@ impl LookupFileCache { Ok(index) } - async fn read(&self, path: &str, size: u64, key: &[u8]) -> Result> { + async fn read(&self, path: &str, size: u64, key: &[u8]) -> Result> { let index = self.parsed_index(path, size).await?; let footer_start = index.footer_start; let reader = GeneratedFileRead::new(self.local.clone(), path.into(), size); @@ -443,7 +512,7 @@ impl LookupFileCache { return Err(invalid("Lookup row is outside its IPC batch")); } arrow_select::take::take_record_batch(&batch, &UInt64Array::from(vec![pointer.row])) - .map(Some) + .map(|batch| Some((batch, pointer.position))) .map_err(invalid) } } @@ -454,13 +523,20 @@ struct Pointer { length: u64, raw_length: u64, row: u64, + position: u64, crc: u32, compression: BlockCompressionType, } impl Pointer { fn encode(self) -> Vec { let mut bytes = Vec::with_capacity(POINTER_LEN); - for value in [self.offset, self.length, self.raw_length, self.row] { + for value in [ + self.offset, + self.length, + self.raw_length, + self.row, + self.position, + ] { bytes.extend_from_slice(&value.to_le_bytes()); } bytes.extend_from_slice(&self.crc.to_le_bytes()); @@ -477,8 +553,9 @@ impl Pointer { length: number(8), raw_length: number(16), row: number(24), - crc: u32::from_le_bytes(bytes[32..36].try_into().unwrap()), - compression: BlockCompressionType::from_persistent_id(bytes[36]).map_err(invalid)?, + position: number(32), + crc: u32::from_le_bytes(bytes[40..44].try_into().unwrap()), + compression: BlockCompressionType::from_persistent_id(bytes[44]).map_err(invalid)?, }) } } diff --git a/crates/paimon/src/table/compact/primary_key/lookup_store.rs b/crates/paimon/src/table/compact/primary_key/lookup_store.rs index f215df20f..85b436e54 100644 --- a/crates/paimon/src/table/compact/primary_key/lookup_store.rs +++ b/crates/paimon/src/table/compact/primary_key/lookup_store.rs @@ -27,13 +27,14 @@ use arrow_array::RecordBatch; use arrow_row::{RowConverter, SortField}; use sha2::{Digest, Sha256}; use std::cmp::Ordering; -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::{Arc, Mutex}; struct HistoryFile { meta: DataFileMeta, range_compatible: bool, identity: tokio::sync::OnceCell, + vector: Option, } struct HistoryLevel { @@ -52,15 +53,18 @@ struct LevelLookup { comparator: Option, resources: Option, schema_identity: Vec, + deletions: Option, } +pub(super) type LookupDeletions = Arc>>; + pub(in crate::table::compact) async fn provider( table: &Table, split: &DataSplit, files: Vec, cache: Arc, resources: Option, -) -> Result> { +) -> Result<(Arc, Option)> { let fields = super::read_type(table); let keys = table .schema() @@ -87,9 +91,30 @@ pub(in crate::table::compact) async fn provider( ) .map_err(invalid)?; let comparator = KeyComparator::from_table_schema(table.schema()); + let dv = table.schema().core_options().deletion_vectors_enabled(); + let history = DataSplit::builder() + .with_snapshot(split.snapshot_id()) + .with_partition(split.partition().clone()) + .with_bucket(split.bucket()) + .with_total_buckets(split.total_buckets()) + .with_bucket_path(split.bucket_path().to_string()) + .with_data_files(files.clone()) + .build()?; + let history = if dv { + super::deletion_vectors::with_deletion_files( + table, + &history, + table + .travel_snapshot() + .ok_or_else(|| invalid("Lookup needs a captured snapshot"))?, + ) + .await? + } else { + history + }; let mut schemas = BTreeMap::from([(table.schema().id(), true)]); let mut levels = BTreeMap::<_, HistoryLevel>::new(); - for meta in files { + for (index, meta) in files.into_iter().enumerate() { let compatible = if let Some(&compatible) = schemas.get(&meta.schema_id) { compatible } else { @@ -126,6 +151,11 @@ pub(in crate::table::compact) async fn provider( meta, range_compatible, identity: tokio::sync::OnceCell::new(), + vector: history + .data_deletion_files() + .and_then(|files| files.get(index)) + .cloned() + .flatten(), }); } for level in levels.values_mut() { @@ -148,18 +178,23 @@ pub(in crate::table::compact) async fn provider( options, )) .map_err(invalid)?; - Ok(Arc::new(LevelLookup { - table: table.clone(), - split: split.clone(), - levels, - cache, - keys, - key_fields, - converter: Mutex::new(converter), - comparator, - resources, - schema_identity, - })) + let deletions = dv.then(|| Arc::new(Mutex::new(HashMap::new()))); + Ok(( + Arc::new(LevelLookup { + table: table.clone(), + split: split.clone(), + levels, + cache, + keys, + key_fields, + converter: Mutex::new(converter), + comparator, + resources, + schema_identity, + deletions: deletions.clone(), + }), + deletions, + )) } #[async_trait::async_trait] @@ -243,31 +278,59 @@ impl LookupProvider for LevelLookup { Ok::<_, Error>(hex::encode(digest.finalize())) }) .await?; - let row = self - .cache - .lookup(identity, key, &self.keys, || { - let split = DataSplit::builder() - .with_snapshot(self.split.snapshot_id()) - .with_partition(self.split.partition().clone()) - .with_bucket(self.split.bucket()) - .with_total_buckets(self.split.total_buckets()) - .with_bucket_path(self.split.bucket_path().to_string()) - .with_data_files(vec![file.meta.clone()]) - .build()?; - // Reuse the normal KV decoder for schema evolution, file - // formats, and preservation of system sequence/row kind. - super::new_reader(&self.table, self.resources.clone())? - .read_with_merge_function(&[split], |_, _| { - Ok(Box::new(super::CompactDeduplicate)) - }) - }) - .await?; - if let Some(batch) = row { + let input = || { + let split = DataSplit::builder() + .with_snapshot(self.split.snapshot_id()) + .with_partition(self.split.partition().clone()) + .with_bucket(self.split.bucket()) + .with_total_buckets(self.split.total_buckets()) + .with_bucket_path(self.split.bucket_path().to_string()) + .with_data_files(vec![file.meta.clone()]) + .build()?; + // Preserve physical ordinals before snapshot DV filtering. + super::new_reader(&self.table, self.resources.clone())? + .read_physical_file(split, file.meta.clone()) + }; + let row = if self.deletions.is_some() { + self.cache + .lookup_with_position(identity, key, &self.keys, &input) + .await? + } else { + self.cache + .lookup(identity, key, &self.keys, &input) + .await? + .map(|batch| (batch, 0)) + }; + if let Some((batch, position)) = row { + if let Some(vector) = &file.vector { + if self + .cache + .is_deleted( + self.table.file_io(), + vector, + position, + self.resources.as_ref(), + ) + .await? + { + continue; + } + } if found.is_some() { return Err(invalid( "Lookup history has overlapping keys within a positive level", )); } + if let Some(deletions) = &self.deletions { + let position = u32::try_from(position) + .map_err(|_| invalid("PK deletion vector position exceeds u32::MAX"))?; + deletions + .lock() + .unwrap() + .entry(file.meta.file_name.clone()) + .or_default() + .insert(position); + } let mut metadata = batch.schema().metadata().clone(); metadata.insert(FILE_LEVEL_METADATA_KEY.into(), file.meta.level.to_string()); found = Some( diff --git a/crates/paimon/src/table/compact/primary_key/planner.rs b/crates/paimon/src/table/compact/primary_key/planner.rs index 5a09538ab..102583984 100644 --- a/crates/paimon/src/table/compact/primary_key/planner.rs +++ b/crates/paimon/src/table/compact/primary_key/planner.rs @@ -26,7 +26,6 @@ use std::collections::BTreeMap; pub(in crate::table::compact) fn enabled(schema: &TableSchema) -> Result { let options = schema.core_options(); Ok(options.merge_engine()? == MergeEngine::Deduplicate - && !options.deletion_vectors_enabled() && matches!( options.try_changelog_producer()?, ChangelogProducer::None | ChangelogProducer::Input | ChangelogProducer::Lookup diff --git a/crates/paimon/src/table/data_evolution_writer.rs b/crates/paimon/src/table/data_evolution_writer.rs index e14e17816..330f04e23 100644 --- a/crates/paimon/src/table/data_evolution_writer.rs +++ b/crates/paimon/src/table/data_evolution_writer.rs @@ -26,16 +26,16 @@ //! This separation allows callers to compose multiple operations into a single commit, //! similar to Iceberg's Transaction/Action pattern. -use crate::deletion_vector::{DeletionVector, DeletionVectorFactory}; +use crate::deletion_vector::DeletionVectorFactory; use crate::io::FileIO; use crate::spec::{ - bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, DeletionVectorMeta, - FileKind, IndexFileMeta, IndexManifest, PartitionComputer, Snapshot, EMPTY_BINARY_ROW, + bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType, PartitionComputer, + Snapshot, }; use crate::table::commit_message::CommitMessage; use crate::table::data_file_index_writer::FileIndexOptions; use crate::table::data_file_writer::DataFileWriter; -use crate::table::index_file_path::IndexFileLocation; +use crate::table::deletion_vector_index; use crate::table::source::{data_evolution_anchor_file, is_data_evolution_normal_file}; use crate::table::stats_filter::group_by_overlapping_row_id; use crate::table::Table; @@ -46,21 +46,13 @@ use arrow_buffer::NullBuffer; use arrow_schema::DataType as ArrowDataType; use arrow_select::concat::concat_batches; use arrow_select::interleave::interleave; -use bytes::Bytes; use futures::TryStreamExt; -use indexmap::IndexMap; -use roaring::RoaringBitmap; use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::Arc; -use uuid::Uuid; #[path = "data_evolution_blob.rs"] mod blob_updates; -const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS"; -const DELETION_VECTORS_INDEX_VERSION_V1: u8 = 1; -const MANIFEST_DIR: &str = "manifest"; - /// Engine-agnostic writer for partial-column updates via `_ROW_ID`. /// /// Usage: @@ -510,26 +502,6 @@ impl DataEvolutionWriter { } } -/// One bucket's deletion-vector location, owning the strings the shared resolver -/// borrows. Built once per bucket so the read that merges existing vectors and the -/// write that follows it cannot disagree about where the file goes. -struct DeletionVectorLayout { - table_path: String, - relative_bucket_path: String, - bucket_path: String, - index_file_in_data_file_dir: bool, -} - -impl DeletionVectorLayout { - fn location(&self) -> IndexFileLocation<'_> { - IndexFileLocation::BucketLocal { - table_path: &self.table_path, - bucket_path: &self.bucket_path, - index_file_in_data_file_dir: self.index_file_in_data_file_dir, - } - } -} - /// Engine-agnostic DELETE writer for data evolution tables. /// /// DELETE is represented by a deletion-vector index file keyed by the normal @@ -746,9 +718,9 @@ impl DataEvolutionDeleteWriter { delete_plan: BucketDeletePlan, snapshot: &Snapshot, ) -> Result> { - let (mut bitmaps, deleted_index_files) = self - .read_existing_bucket_deletion_vectors(&partition, bucket, snapshot, None) - .await?; + let (mut bitmaps, deleted_index_files) = + deletion_vector_index::read_bucket(&self.table, &partition, bucket, snapshot, None) + .await?; let mut changed = false; for (anchor_file_name, positions) in delete_plan.deletes_by_anchor { @@ -762,204 +734,14 @@ impl DataEvolutionDeleteWriter { return Ok(None); } - let new_index_file = self - .write_deletion_vector_index_file(&partition, bucket, bitmaps) - .await?; + let new_index_file = + deletion_vector_index::write_bucket(&self.table, &partition, bucket, bitmaps).await?; let mut message = CommitMessage::new(partition, bucket, vec![]); message.check_from_snapshot = Some(delete_plan.check_from_snapshot); message.new_index_files = vec![new_index_file]; message.deleted_index_files = deleted_index_files; Ok(Some(message)) } - - /// Where this bucket's deletion vectors live. A deletion vector is an index - /// file, so it sits beside the bucket's data files when the table keeps index - /// files there, and under the table `index/` directory otherwise. Reads and - /// writes both go through this, so a file written here is found again. - fn deletion_vector_layout( - &self, - partition: &[u8], - bucket: i32, - ) -> Result { - let schema = self.table.schema(); - let partition_keys = schema.partition_keys(); - let core_options = CoreOptions::new(schema.options()); - let computer = if partition_keys.is_empty() { - None - } else { - Some(PartitionComputer::new( - partition_keys, - schema.fields(), - core_options.partition_default_name(), - core_options.legacy_partition_name(), - )?) - }; - let partition_row = if computer.is_some() { - BinaryRow::from_serialized_bytes(partition)? - } else { - EMPTY_BINARY_ROW - }; - let partition_path = match &computer { - Some(computer) => computer.generate_partition_path(&partition_row)?, - None => String::new(), - }; - Ok(DeletionVectorLayout { - relative_bucket_path: crate::spec::relative_bucket_path( - &partition_path, - bucket, - core_options.data_file_path_directory(), - ), - table_path: self.table.location().trim_end_matches('/').to_string(), - bucket_path: bucket_path( - &self.table.data_file_location(), - computer.as_ref(), - &partition_row, - bucket, - )?, - index_file_in_data_file_dir: core_options.index_file_in_data_file_dir(), - }) - } - - pub(in crate::table) async fn read_existing_bucket_deletion_vectors( - &self, - partition: &[u8], - bucket: i32, - snapshot: &Snapshot, - selected_anchors: Option<&HashSet>, - ) -> Result<(IndexMap, Vec)> { - let Some(index_manifest_name) = snapshot.index_manifest() else { - return Ok((IndexMap::new(), Vec::new())); - }; - let layout = self.deletion_vector_layout(partition, bucket)?; - - let manifest_path = format!( - "{}/{MANIFEST_DIR}/{}", - self.table.location().trim_end_matches('/'), - index_manifest_name - ); - let index_entries = IndexManifest::read(self.table.file_io(), &manifest_path).await?; - let mut bitmaps = IndexMap::new(); - let mut deleted_index_files = Vec::new(); - - for entry in index_entries { - if entry.kind != FileKind::Add - || entry.bucket != bucket - || entry.partition != partition - || entry.index_file.index_type != DELETION_VECTORS_INDEX_TYPE - || selected_anchors.is_some_and(|anchors| { - entry - .index_file - .deletion_vectors_ranges - .as_ref() - .is_none_or(|ranges| !ranges.keys().any(|name| anchors.contains(name))) - }) - { - continue; - } - deleted_index_files.push(entry.index_file.clone()); - let Some(ranges) = entry.index_file.deletion_vectors_ranges.as_ref() else { - continue; - }; - let index_path = layout.location().resolve( - &entry.index_file.file_name, - entry.index_file.external_path.as_deref(), - ); - for (data_file_name, meta) in ranges { - let deletion_file = crate::DeletionFile::new( - index_path.clone(), - meta.offset as i64, - meta.length as i64, - meta.cardinality, - ); - let bitmap = DeletionVectorFactory::read(self.table.file_io(), &deletion_file) - .await? - .to_bitmap(); - bitmaps.insert(data_file_name.clone(), bitmap); - } - } - - Ok((bitmaps, deleted_index_files)) - } - - pub(in crate::table) async fn write_deletion_vector_index_file( - &self, - partition: &[u8], - bucket: i32, - mut bitmaps: IndexMap, - ) -> Result { - bitmaps.sort_keys(); - - let file_name = format!("index-{}-1", Uuid::new_v4()); - let layout = self.deletion_vector_layout(partition, bucket)?; - let external_path = super::external_path::new_index_external_path( - self.table.schema().options(), - layout.index_file_in_data_file_dir, - &layout.relative_bucket_path, - &file_name, - )?; - let path = layout - .location() - .resolve(&file_name, external_path.as_deref()); - let directory = path.rsplit_once('/').expect("index file has a parent").0; - self.table.file_io().mkdirs(directory).await?; - - let mut bytes = vec![DELETION_VECTORS_INDEX_VERSION_V1]; - let mut ranges = IndexMap::new(); - for (data_file_name, bitmap) in bitmaps { - if bitmap.is_empty() { - continue; - } - let offset = i32::try_from(bytes.len()).map_err(|_| crate::Error::DataInvalid { - message: "Deletion-vector index file is too large".to_string(), - source: None, - })?; - let deletion_vector = DeletionVector::from_bitmap(bitmap); - let serialized = deletion_vector.serialize_to_bytes()?; - let length = i32::from_be_bytes( - serialized[0..4] - .try_into() - .expect("serialized deletion vector has length prefix"), - ); - ranges.insert( - data_file_name, - DeletionVectorMeta { - offset, - length, - cardinality: Some(deletion_vector.cardinality() as i64), - }, - ); - bytes.extend_from_slice(&serialized); - } - - let file_size = i64::try_from(bytes.len()).map_err(|_| crate::Error::DataInvalid { - message: "Deletion-vector index file is too large".to_string(), - source: None, - })?; - let row_count = i32::try_from(ranges.len()).map_err(|_| crate::Error::DataInvalid { - message: "Deletion-vector index file has too many entries".to_string(), - source: None, - })?; - if let Err(error) = self - .table - .file_io() - .new_output(&path)? - .write(Bytes::from(bytes)) - .await - { - let _ = self.table.file_io().delete_file(&path).await; - return Err(error); - } - - Ok(IndexFileMeta { - index_type: DELETION_VECTORS_INDEX_TYPE.to_string(), - file_name, - file_size, - row_count: i64::from(row_count), - deletion_vectors_ranges: Some(ranges), - external_path, - global_index_meta: None, - }) - } } struct DeleteFileRowRange { @@ -2028,6 +1810,7 @@ mod tests { use arrow_array::{Int32Array, StringArray, StructArray}; use arrow_buffer::NullBuffer; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; + use bytes::Bytes; use std::sync::Arc; fn test_file_io() -> FileIO { diff --git a/crates/paimon/src/table/deletion_vector_index.rs b/crates/paimon/src/table/deletion_vector_index.rs new file mode 100644 index 000000000..1471fd142 --- /dev/null +++ b/crates/paimon/src/table/deletion_vector_index.rs @@ -0,0 +1,273 @@ +// 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. + +//! Deletion-vector index I/O shared by table writers and compaction. + +use std::collections::HashSet; + +use bytes::Bytes; +use indexmap::IndexMap; +use roaring::RoaringBitmap; +use uuid::Uuid; + +use crate::deletion_vector::{DeletionVector, DeletionVectorFactory}; +use crate::spec::{ + bucket_path, BinaryRow, CoreOptions, DeletionVectorMeta, FileKind, IndexFileMeta, + IndexManifest, IndexManifestEntry, PartitionComputer, Snapshot, EMPTY_BINARY_ROW, +}; +use crate::table::index_file_path::IndexFileLocation; +use crate::table::Table; +use crate::Result; + +const DELETION_VECTORS_INDEX_TYPE: &str = "DELETION_VECTORS"; +const DELETION_VECTORS_INDEX_VERSION_V1: u8 = 1; +const MANIFEST_DIR: &str = "manifest"; + +/// One bucket's deletion-vector location, owning the strings the shared resolver +/// borrows. Built once per bucket so the read that merges existing vectors and the +/// write that follows it cannot disagree about where the file goes. +struct DeletionVectorLayout { + table_path: String, + relative_bucket_path: String, + bucket_path: String, + index_file_in_data_file_dir: bool, +} + +impl DeletionVectorLayout { + fn location(&self) -> IndexFileLocation<'_> { + IndexFileLocation::BucketLocal { + table_path: &self.table_path, + bucket_path: &self.bucket_path, + index_file_in_data_file_dir: self.index_file_in_data_file_dir, + } + } +} + +/// Where this bucket's deletion vectors live. A deletion vector is an index +/// file, so it sits beside the bucket's data files when the table keeps index +/// files there, and under the table `index/` directory otherwise. Reads and +/// writes both go through this, so a file written here is found again. +fn deletion_vector_layout( + table: &Table, + partition: &[u8], + bucket: i32, +) -> Result { + let schema = table.schema(); + let partition_keys = schema.partition_keys(); + let core_options = CoreOptions::new(schema.options()); + let computer = if partition_keys.is_empty() { + None + } else { + Some(PartitionComputer::new( + partition_keys, + schema.fields(), + core_options.partition_default_name(), + core_options.legacy_partition_name(), + )?) + }; + let partition_row = if computer.is_some() { + BinaryRow::from_serialized_bytes(partition)? + } else { + EMPTY_BINARY_ROW + }; + let partition_path = match &computer { + Some(computer) => computer.generate_partition_path(&partition_row)?, + None => String::new(), + }; + Ok(DeletionVectorLayout { + relative_bucket_path: crate::spec::relative_bucket_path( + &partition_path, + bucket, + core_options.data_file_path_directory(), + ), + table_path: table.location().trim_end_matches('/').to_string(), + bucket_path: bucket_path( + &table.data_file_location(), + computer.as_ref(), + &partition_row, + bucket, + )?, + index_file_in_data_file_dir: core_options.index_file_in_data_file_dir(), + }) +} + +/// Read selected containers in a bucket, retaining every range in each container. +/// `selected_files` filters containers by the data files they reference. +pub(super) async fn read_bucket( + table: &Table, + partition: &[u8], + bucket: i32, + snapshot: &Snapshot, + selected_files: Option<&HashSet>, +) -> Result<(IndexMap, Vec)> { + let Some(index_manifest_name) = snapshot.index_manifest() else { + return Ok((IndexMap::new(), Vec::new())); + }; + let manifest_path = format!( + "{}/{MANIFEST_DIR}/{}", + table.location().trim_end_matches('/'), + index_manifest_name + ); + let index_entries = IndexManifest::read(table.file_io(), &manifest_path).await?; + let containers = index_entries + .into_iter() + .filter(|entry| { + entry.kind == FileKind::Add + && entry.bucket == bucket + && entry.partition == partition + && entry.index_file.index_type == DELETION_VECTORS_INDEX_TYPE + && selected_files.is_none_or(|files| { + entry + .index_file + .deletion_vectors_ranges + .as_ref() + .is_some_and(|ranges| ranges.keys().any(|name| files.contains(name))) + }) + }) + .collect::>(); + let bitmaps = read_containers(table, partition, bucket, &containers).await?; + let deleted_index_files = containers + .into_iter() + .map(|entry| entry.index_file) + .collect(); + Ok((bitmaps, deleted_index_files)) +} + +/// Decode already selected containers without rereading their index manifest. +pub(super) async fn read_containers( + table: &Table, + partition: &[u8], + bucket: i32, + containers: &[IndexManifestEntry], +) -> Result> { + let layout = deletion_vector_layout(table, partition, bucket)?; + let mut bitmaps = IndexMap::new(); + for entry in containers { + let Some(ranges) = entry.index_file.deletion_vectors_ranges.as_ref() else { + continue; + }; + let index_path = layout.location().resolve( + &entry.index_file.file_name, + entry.index_file.external_path.as_deref(), + ); + for (data_file_name, meta) in ranges { + let deletion_file = crate::DeletionFile::new( + index_path.clone(), + meta.offset as i64, + meta.length as i64, + meta.cardinality, + ); + let bitmap = DeletionVectorFactory::read(table.file_io(), &deletion_file) + .await? + .to_bitmap(); + bitmaps.insert(data_file_name.clone(), bitmap); + } + } + + Ok(bitmaps) +} + +pub(super) fn file_path( + table: &Table, + partition: &[u8], + bucket: i32, + file: &IndexFileMeta, +) -> Result { + Ok(deletion_vector_layout(table, partition, bucket)? + .location() + .resolve(&file.file_name, file.external_path.as_deref())) +} + +/// Write one immutable container. The caller decides which masks to combine. +pub(super) async fn write_bucket( + table: &Table, + partition: &[u8], + bucket: i32, + mut bitmaps: IndexMap, +) -> Result { + bitmaps.sort_keys(); + + let file_name = format!("index-{}-1", Uuid::new_v4()); + let layout = deletion_vector_layout(table, partition, bucket)?; + let external_path = super::external_path::new_index_external_path( + table.schema().options(), + layout.index_file_in_data_file_dir, + &layout.relative_bucket_path, + &file_name, + )?; + let path = layout + .location() + .resolve(&file_name, external_path.as_deref()); + let directory = path.rsplit_once('/').expect("index file has a parent").0; + table.file_io().mkdirs(directory).await?; + + let mut bytes = vec![DELETION_VECTORS_INDEX_VERSION_V1]; + let mut ranges = IndexMap::new(); + for (data_file_name, bitmap) in bitmaps { + if bitmap.is_empty() { + continue; + } + let offset = i32::try_from(bytes.len()).map_err(|_| crate::Error::DataInvalid { + message: "Deletion-vector index file is too large".to_string(), + source: None, + })?; + let deletion_vector = DeletionVector::from_bitmap(bitmap); + let serialized = deletion_vector.serialize_to_bytes()?; + let length = i32::from_be_bytes( + serialized[0..4] + .try_into() + .expect("serialized deletion vector has length prefix"), + ); + ranges.insert( + data_file_name, + DeletionVectorMeta { + offset, + length, + cardinality: Some(deletion_vector.cardinality() as i64), + }, + ); + bytes.extend_from_slice(&serialized); + } + + let file_size = i64::try_from(bytes.len()).map_err(|_| crate::Error::DataInvalid { + message: "Deletion-vector index file is too large".to_string(), + source: None, + })?; + let row_count = i32::try_from(ranges.len()).map_err(|_| crate::Error::DataInvalid { + message: "Deletion-vector index file has too many entries".to_string(), + source: None, + })?; + if let Err(error) = table + .file_io() + .new_output(&path)? + .write(Bytes::from(bytes)) + .await + { + let _ = table.file_io().delete_file(&path).await; + return Err(error); + } + + Ok(IndexFileMeta { + index_type: DELETION_VECTORS_INDEX_TYPE.to_string(), + file_name, + file_size, + row_count: i64::from(row_count), + deletion_vectors_ranges: Some(ranges), + external_path, + global_index_meta: None, + }) +} diff --git a/crates/paimon/src/table/kv_file_reader.rs b/crates/paimon/src/table/kv_file_reader.rs index 6b2bda833..2eab24f3a 100644 --- a/crates/paimon/src/table/kv_file_reader.rs +++ b/crates/paimon/src/table/kv_file_reader.rs @@ -377,6 +377,49 @@ impl KeyValueFileReader { self } + /// Decode an immutable file without filtering or merging physical rows. + /// Lookup files need these ordinals even when a snapshot deletes some rows. + pub(super) fn read_physical_file( + self, + split: DataSplit, + file: DataFileMeta, + ) -> crate::Result { + Ok(Box::pin(try_stream! { + let config = self.config; + let old = if file.schema_id != config.table_schema_id { + Some(config.schema_manager.schema(file.schema_id).await?) + } else { None }; + let fields = old.as_ref().map_or(config.table_fields.as_slice(), |s| s.fields()); + let keys = old.as_ref().map(|s| s.trimmed_primary_keys()); + let data_schema = key_value_data_schema_fields( + fields, keys.as_deref().unwrap_or(&config.primary_keys), + )?; + let reader = DataFileReader::new( + self.file_io, config.schema_manager.clone(), config.table_schema_id, + config.table_fields.clone(), + key_value_fields_extractor(&config.table_fields, &config.primary_keys), + config.read_type, vec![], + ) + .with_batch_size(Some(config.read_batch_size)) + .with_parquet_read_budget(config.parquet_read_budget) + .with_table_options(config.table_options) + .with_mosaic_prefetch(config.mosaic_prefetch); + let expected = file.row_count; + let mut stream = reader.read_single_file_stream_with_schema_and_local_ranges( + &split, file, old.map(|s| s.fields().to_vec()), data_schema, None, None, + )?; + let mut rows = 0i64; + while let Some(batch) = stream.next().await { + let batch = batch?; + rows += batch.num_rows() as i64; + yield batch; + } + if rows != expected { + Err(Error::DataInvalid { message: "Lookup physical row count differs from file metadata".into(), source: None })?; + } + })) + } + #[cfg(test)] fn with_input_batch_sizes( mut self, diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index a56f6e022..b82a9c9df 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -58,6 +58,7 @@ mod data_file_writer; mod de_vector_read; mod de_vector_scan; mod dedicated_format_file_writer; +mod deletion_vector_index; mod dynamic_bucket_assigner; mod dynamic_bucket_assigner_router; mod dynamic_bucket_index; diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index ae8bb4ddd..60e0f6066 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -443,6 +443,7 @@ impl TableCommit { CommitEntriesPlan::Compact { entries: prepared.entries, changelog_entries: prepared.changelog_entries, + deletion_vector_updates: prepared.deletion_vector_updates, schema_id: prepared.schema_id, read_snapshot: prepared.read_snapshot, planned_snapshot_id: prepared.snapshot_id, @@ -1351,7 +1352,40 @@ impl TableCommit { } return Ok(CommitAttemptResult::Success); } - Ok(false) => None, + Ok(false) => { + if self.deletion_vectors_enabled && resolved.kind == CommitKind::COMPACT { + // Only this attempt's generated DV files are definitely + // unpublished. Prepared data/changelogs and old shared + // containers still belong to the caller or prior snapshots. + let created = created_files + .iter() + .map(String::as_str) + .collect::>(); + for entry in resolved + .new_index_manifest_entries + .iter() + .flatten() + .filter(|entry| entry.index_file.index_type == DELETION_VECTORS_INDEX_TYPE) + { + let Ok(bucket) = self.bucket_path(&entry.partition, entry.bucket) else { + continue; + }; + let path = committed_index_file_path( + self.table.location(), + &bucket, + self.table + .schema() + .core_options() + .index_file_in_data_file_dir(), + &entry.index_file, + ); + if created.contains(path.as_str()) { + let _ = self.table.file_io().delete_file(&path).await; + } + } + } + None + } Err(error) => Some(error), }; Ok(CommitAttemptResult::Retry(Box::new(RetryState { @@ -1370,15 +1404,13 @@ impl TableCommit { ) -> Result<(Snapshot, Vec)> { let new_snapshot_id = latest_snapshot.as_ref().map(|s| s.id() + 1).unwrap_or(1); - if self.data_evolution_enabled - && resolved.kind == CommitKind::COMPACT - && self.deletion_vectors_enabled - { + if self.deletion_vectors_enabled && resolved.kind == CommitKind::COMPACT { if let Some(snapshot) = latest_snapshot { - if let Some(indexes) = super::compact::data_evolution::rewrite_deletion_vectors( + if let Some(indexes) = super::compact::deletion_vectors::rewrite_indexes( &self.table, snapshot, &resolved.entries, + &resolved.deletion_vector_updates, created_files, ) .await? @@ -1908,6 +1940,7 @@ impl TableCommit { CommitEntriesPlan::Compact { entries, changelog_entries, + deletion_vector_updates, schema_id, read_snapshot, planned_snapshot_id, @@ -1927,6 +1960,19 @@ impl TableCommit { source: None, }); } + // Duplicate recovery runs before resolve_commit. An already + // published legacy result can omit this field, but a first + // publication must retain every historical row deletion. + let deletion_vector_updates = match deletion_vector_updates.as_deref() { + Some(updates) => updates, + None if !self.data_evolution_enabled && self.deletion_vectors_enabled => { + return Err(crate::Error::DataInvalid { + message: "Deletion-vector compaction results are missing row deletions; execute the plan again".into(), + source: None, + }); + } + None => &[], + }; // Concurrent writes may replace HASH indexes, deletion vectors // or global indexes. Rebase on the latest manifest every retry. let index_manifest_name = match latest_snapshot { @@ -1965,6 +2011,7 @@ impl TableCommit { super::compact::validate_primary_key_changes( self.table.schema(), entries, + changelog_entries, &captured_files, )?; } @@ -2057,9 +2104,29 @@ impl TableCommit { base_data_files.as_deref().unwrap_or_default(), entries, )?; + if !self.data_evolution_enabled && self.deletion_vectors_enabled { + let captured = + read_snapshot + .as_deref() + .ok_or_else(|| crate::Error::DataInvalid { + message: + "PK deletion-vector compaction requires its captured snapshot" + .into(), + source: None, + })?; + super::compact::primary_key::deletion_vectors::validate_dependencies( + &self.table, + captured, + latest_snapshot.as_ref().expect("validated latest snapshot"), + entries, + deletion_vector_updates, + ) + .await?; + } Ok(ResolvedCommit { entries: entries.clone(), changelog_entries: changelog_entries.clone(), + deletion_vector_updates: deletion_vector_updates.to_vec(), kind: CommitKind::COMPACT, index_manifest_name, index_manifest_changed: false, @@ -2133,6 +2200,7 @@ impl TableCommit { Ok(ResolvedCommit { entries: entries.clone(), changelog_entries: changelog_entries.clone(), + deletion_vector_updates: vec![], kind, index_manifest_name, index_manifest_changed, @@ -2205,6 +2273,7 @@ impl TableCommit { Ok(ResolvedCommit { entries, changelog_entries: vec![], + deletion_vector_updates: vec![], kind: CommitKind::OVERWRITE, index_manifest_name, index_manifest_changed, @@ -3887,6 +3956,8 @@ enum CommitEntriesPlan { Compact { entries: Vec, changelog_entries: Vec, + deletion_vector_updates: + Option>, schema_id: i64, read_snapshot: Option>, planned_snapshot_id: i64, @@ -3947,6 +4018,7 @@ fn direct_commit_kind( struct ResolvedCommit { entries: Vec, changelog_entries: Vec, + deletion_vector_updates: Vec, kind: CommitKind, index_manifest_name: Option, index_manifest_changed: bool, @@ -4121,6 +4193,7 @@ mod tests { mod parity { use super::*; include!("table_commit/parity_tests.rs"); + include!("table_commit/pk_dv_retry_tests.rs"); include!("table_commit/recovery_tests.rs"); include!("table_commit/row_tracking_tests.rs"); include!("table_commit/abort_tests.rs"); diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs b/crates/paimon/src/table/table_commit/parity_tests.rs index 33e090a5e..3e7a8784c 100644 --- a/crates/paimon/src/table/table_commit/parity_tests.rs +++ b/crates/paimon/src/table/table_commit/parity_tests.rs @@ -362,11 +362,20 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe use arrow_array::{Int32Array, RecordBatch}; use futures::TryStreamExt; + async fn dv_paths(table: &Table, snapshot: &Snapshot) -> Result> { + let entries = IndexManifest::read(table.file_io(), &table.snapshot_manager().manifest_path(snapshot.index_manifest().unwrap())).await?; + let commit = table.new_write_builder().new_commit(); + entries.into_iter().filter(|entry| entry.index_file.index_type == DELETION_VECTORS_INDEX_TYPE).map(|entry| { + Ok(committed_index_file_path(table.location(), &commit.bucket_path(&entry.partition, entry.bucket)?, table.schema().core_options().index_file_in_data_file_dir(), &entry.index_file)) + }).collect() + } + struct RacingCompactCommit { table: Table, calls: std::sync::atomic::AtomicUsize, publish_first: bool, delete_after_publish: bool, + paths: std::sync::Mutex>>, } #[async_trait::async_trait] @@ -378,6 +387,8 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe _: &str, _: &[PartitionStatistics], ) -> Result { + let paths = dv_paths(&self.table, snapshot).await?; + self.paths.lock().unwrap().push(paths); if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 { if self.publish_first { assert!(self.table.snapshot_manager().commit_snapshot(snapshot).await?); @@ -424,11 +435,14 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe let mut delete = crate::table::DataEvolutionDeleteWriter::new(&table).unwrap(); delete.add_row_ids([1]).unwrap(); table.new_write_builder().new_commit().commit(delete.prepare_commit().await.unwrap()).await.unwrap(); + let captured = table.snapshot_manager().get_latest_snapshot().await.unwrap().unwrap(); + let original_paths = dv_paths(&table, &captured).await.unwrap(); let builder = table.new_compact_builder(); let plan = builder.new_planner().unwrap().plan().await.unwrap(); let result = builder.new_executor().unwrap().execute(&plan.tasks()[0]).await.unwrap(); let publisher = Arc::new(RacingCompactCommit { table: table.clone(), calls: std::sync::atomic::AtomicUsize::new(0), publish_first, delete_after_publish, + paths: std::sync::Mutex::new(Vec::new()), }); let mut commit = builder.new_commit().unwrap(); commit.snapshot_commit = publisher.clone(); @@ -439,6 +453,15 @@ async fn data_evolution_compact_retries_preserve_concurrent_deletes_and_publishe let after = table.snapshot_manager().get_latest_snapshot().await.unwrap().unwrap(); assert_eq!(after.id(), if publish_first && !delete_after_publish { 4 } else { 5 }); assert_eq!(after.commit_kind(), if delete_after_publish { &CommitKind::OVERWRITE } else { &CommitKind::COMPACT }); + let attempts = publisher.paths.lock().unwrap().clone(); + assert_eq!(attempts[0].len(), 1); + assert!(!original_paths.contains(&attempts[0][0])); + assert_eq!(io.exists(&attempts[0][0]).await.unwrap(), publish_first, + "CAS false must remove its unpublished DV; publish-then-error must retain it"); + let current_paths = dv_paths(&table, &after).await.unwrap(); + for path in original_paths.iter().chain(attempts.last().unwrap()).chain(¤t_paths) { + assert!(io.exists(path).await.unwrap(), "deleted old-snapshot or published DV: {path}"); + } assert_eq!(commit.commit_compact(vec![result.clone()]).await.unwrap(), metrics); // Recovery after coordinator restart must not need the captured source // snapshot once the COMPACT identity is already present in history. diff --git a/crates/paimon/src/table/table_commit/pk_dv_retry_tests.rs b/crates/paimon/src/table/table_commit/pk_dv_retry_tests.rs new file mode 100644 index 000000000..7b75a5526 --- /dev/null +++ b/crates/paimon/src/table/table_commit/pk_dv_retry_tests.rs @@ -0,0 +1,370 @@ +// 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. + +mod pk_dv_retry { + use super::*; + use arrow_array::{Int32Array, RecordBatch}; + use indexmap::IndexMap; + use roaring::RoaringBitmap; + use std::sync::atomic::{AtomicUsize, Ordering}; + + async fn write(table: &Table, ids: Vec, level: i32) -> (Vec, DataFileMeta) { + let batch = RecordBatch::try_from_iter([ + ( + "id", + Arc::new(Int32Array::from(ids.clone())) as arrow_array::ArrayRef, + ), + ( + "value", + Arc::new(Int32Array::from( + ids.iter() + .map(|id| id * if level == 0 { 100 } else { 10 }) + .collect::>(), + )), + ), + ]) + .unwrap(); + let builder = table.new_write_builder(); + let mut writer = builder.new_write().unwrap(); + writer.write_arrow_batch(&batch).await.unwrap(); + let mut messages = writer.prepare_commit().await.unwrap(); + for file in &mut messages[0].new_files { + file.level = level; + } + let output = ( + messages[0].partition.clone(), + messages[0].new_files[0].clone(), + ); + builder.new_commit().commit(messages).await.unwrap(); + output + } + + async fn vectors( + table: &Table, + ) -> ( + IndexMap, + Vec, + ) { + let snapshot = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + crate::table::deletion_vector_index::read_bucket( + table, + &EMPTY_SERIALIZED_ROW, + 0, + &snapshot, + None, + ) + .await + .unwrap() + } + + async fn replace_vectors(table: &Table, masks: &[(&str, &[u32])]) { + let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.to_vec(), 0, vec![]); + message.deleted_index_files = vectors(table).await.1; + message.new_index_files.push( + crate::table::deletion_vector_index::write_bucket( + table, + &message.partition, + 0, + masks + .iter() + .map(|(name, positions)| { + (name.to_string(), positions.iter().copied().collect()) + }) + .collect(), + ) + .await + .unwrap(), + ); + table + .new_write_builder() + .new_commit() + .commit(vec![message]) + .await + .unwrap(); + } + + async fn fixture() -> (Table, String, String) { + let path = format!("memory:/pk-dv-retry/{}", uuid::Uuid::new_v4()); + let mut options = crate::Options::new(); + options.set("warehouse", &path); + options.set("local-cache.enabled", "true"); + let io = crate::catalog::FileSystemCatalog::new(options) + .unwrap() + .file_io() + .clone(); + setup_dirs(&io, &path).await; + let schema = crate::spec::Schema::builder() + .column("id", DataType::Int(crate::spec::IntType::new())) + .column("value", DataType::Int(crate::spec::IntType::new())) + .primary_key(["id"]) + .option("bucket", "1") + .option("deletion-vectors.enabled", "true") + .option("changelog-producer", "lookup") + .option("read.batch-size", "7") + .build() + .unwrap(); + let table = Table::new( + io.clone(), + Identifier::new("default", "pk_dv_retry"), + path.clone(), + TableSchema::new(0, &schema), + None, + ); + io.mkdirs(&format!("{path}/schema")).await.unwrap(); + io.new_output(&format!("{path}/schema/schema-0")) + .unwrap() + .write(serde_json::to_vec(table.schema()).unwrap().into()) + .await + .unwrap(); + let (partition, anchor) = write(&table, (0..1000).collect(), 5).await; + assert_eq!(partition.as_slice(), EMPTY_SERIALIZED_ROW.as_slice()); + let (_, unrelated) = write(&table, vec![2000, 2001], 5).await; + replace_vectors( + &table, + &[(&anchor.file_name, &[500]), (&unrelated.file_name, &[0])], + ) + .await; + (table, anchor.file_name, unrelated.file_name) + } + + async fn dv_paths(table: &Table, snapshot: &Snapshot) -> Vec { + let entries = IndexManifest::read( + table.file_io(), + &table + .snapshot_manager() + .manifest_path(snapshot.index_manifest().unwrap()), + ) + .await + .unwrap(); + let commit = table.new_write_builder().new_commit(); + entries + .into_iter() + .filter(|entry| entry.index_file.index_type == DELETION_VECTORS_INDEX_TYPE) + .map(|entry| { + committed_index_file_path( + table.location(), + &commit.bucket_path(&entry.partition, entry.bucket).unwrap(), + table.schema().core_options().index_file_in_data_file_dir(), + &entry.index_file, + ) + }) + .collect() + } + + struct RacingCommit { + table: Table, + calls: AtomicUsize, + publish_first: bool, + paths: std::sync::Mutex>>, + } + + #[async_trait::async_trait] + impl SnapshotCommit for RacingCommit { + async fn commit( + &self, + _: Option<&str>, + snapshot: &Snapshot, + _: &str, + _: &[PartitionStatistics], + ) -> Result { + let paths = dv_paths(&self.table, snapshot).await; + self.paths.lock().unwrap().push(paths); + if self.calls.fetch_add(1, Ordering::SeqCst) == 0 { + if self.publish_first { + assert!( + self.table + .snapshot_manager() + .commit_snapshot(snapshot) + .await? + ); + return Err(crate::Error::IoUnsupported { + message: "lost compact response".into(), + }); + } + write(&self.table, vec![4000], 0).await; + return Ok(false); + } + self.table + .snapshot_manager() + .commit_snapshot(snapshot) + .await + } + } + + #[tokio::test] + async fn compact_cas_loss_cleans_only_new_dv_but_publication_error_preserves_it() { + for publish_first in [false, true] { + let (table, anchor, unrelated) = fixture().await; + let old = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let original_paths = dv_paths(&table, &old).await; + write(&table, vec![42], 0).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks().len(), 1); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let result = builder + .new_executor() + .unwrap() + .execute(&plan.tasks()[0]) + .await + .unwrap(); + assert_eq!(result.files()[0].level, 4); + assert!(!result.changelog_files().is_empty()); + let publisher = Arc::new(RacingCommit { + table: table.clone(), + calls: AtomicUsize::new(0), + publish_first, + paths: std::sync::Mutex::new(vec![]), + }); + let mut commit = builder.new_commit().unwrap(); + commit.snapshot_commit = publisher.clone(); + commit.commit_min_retry_wait_ms = 0; + commit.commit_max_retry_wait_ms = 0; + let metrics = commit.commit_compact(vec![result.clone()]).await.unwrap(); + assert_eq!( + publisher.calls.load(Ordering::SeqCst), + if publish_first { 1 } else { 2 } + ); + let attempts = publisher.paths.lock().unwrap().clone(); + assert_eq!(attempts[0].len(), 1); + assert!(!original_paths.contains(&attempts[0][0])); + assert_eq!( + table.file_io().exists(&attempts[0][0]).await.unwrap(), + publish_first + ); + for path in original_paths.iter().chain(attempts.last().unwrap()) { + assert!( + table.file_io().exists(path).await.unwrap(), + "deleted retained or published DV: {path}" + ); + } + let bucket = commit.bucket_path(plan.tasks()[0].partition(), 0).unwrap(); + for file in result.files().iter().chain(result.changelog_files()) { + for path in file.collect_files(&bucket) { + assert!(table.file_io().exists(&path).await.unwrap()); + } + } + let masks = vectors(&table).await.0; + assert_eq!(masks[&anchor], [42, 500].into_iter().collect()); + assert_eq!(masks[&unrelated], [0].into_iter().collect()); + let after = table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(); + assert_eq!(commit.commit_compact(vec![result]).await.unwrap(), metrics); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + after + ); + } + } + + #[tokio::test] + async fn selected_dv_container_repacking_compares_masks_and_rebases_unrelated_rows() { + for selected_changed in [false, true] { + let (table, anchor, unrelated) = fixture().await; + let (_, selected) = write(&table, vec![42, 43], 0).await; + replace_vectors( + &table, + &[ + (&anchor, &[500]), + (&unrelated, &[0]), + (&selected.file_name, &[0]), + ], + ) + .await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let result = builder + .new_executor() + .unwrap() + .execute(&plan.tasks()[0]) + .await + .unwrap(); + assert_eq!(result.files()[0].row_count, 1); + let captured = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + let old_paths = dv_paths(&table, &captured).await; + replace_vectors( + &table, + &[ + (&anchor, &[500, 501]), + (&unrelated, &[0]), + ( + &selected.file_name, + if selected_changed { &[0, 1] } else { &[0] }, + ), + ], + ) + .await; + let latest = table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap(); + assert_ne!(dv_paths(&table, &latest).await, old_paths); + let published = builder + .new_commit() + .unwrap() + .commit_compact(vec![result]) + .await; + if selected_changed { + assert!(published + .unwrap_err() + .to_string() + .contains("input deletion vectors changed")); + assert_eq!( + table + .snapshot_manager() + .get_latest_snapshot_id() + .await + .unwrap(), + Some(latest.id()) + ); + } else { + published.unwrap(); + let masks = vectors(&table).await.0; + assert_eq!(masks[&anchor], [43, 500, 501].into_iter().collect()); + assert_eq!(masks[&unrelated], [0].into_iter().collect()); + assert!(!masks.contains_key(&selected.file_name)); + } + for path in old_paths { + assert!(table.file_io().exists(&path).await.unwrap()); + } + } + } +} diff --git a/crates/paimon/tests/pk_compact/deletion_vectors.rs b/crates/paimon/tests/pk_compact/deletion_vectors.rs new file mode 100644 index 000000000..cc81493a6 --- /dev/null +++ b/crates/paimon/tests/pk_compact/deletion_vectors.rs @@ -0,0 +1,592 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use bytes::Bytes; +use indexmap::IndexMap; +use paimon::spec::{DataFileMeta, IndexFileMeta, IndexManifest, IndexManifestEntry, Snapshot}; +use paimon::table::CommitMessage; +use roaring::RoaringBitmap; + +async fn dv_table(options: &[(&str, &str)]) -> Table { + let mut properties = vec![ + ("deletion-vectors.enabled", "true"), + ("deletion-vectors.merge-on-read", "false"), + ("read.batch-size", "7"), + ]; + properties.extend_from_slice(options); + table(&properties).await +} + +async fn latest(table: &Table) -> Snapshot { + table + .snapshot_manager() + .get_latest_snapshot() + .await + .unwrap() + .unwrap() +} + +async fn compact(table: &Table) -> Vec { + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(table, &plan).await; + builder + .new_commit() + .unwrap() + .commit_compact(results.clone()) + .await + .unwrap(); + results +} + +async fn seed(table: &Table) -> DataFileMeta { + write_batch(table, &make_batch((0..1000).collect(), (0..1000).collect())).await; + let results = compact(table).await; + assert_eq!(results.len(), 1); + let file = results[0].files()[0].clone(); + assert_eq!(file.level, 5); + file +} + +async fn all_files(table: &Table) -> Vec { + table + .new_read_builder() + .new_scan() + .with_scan_all_files() + .plan() + .await + .unwrap() + .splits() + .iter() + .flat_map(|split| split.data_files()) + .cloned() + .collect() +} + +async fn indexes(table: &Table) -> Vec { + match latest(table).await.index_manifest() { + Some(name) => IndexManifest::read( + table.file_io(), + &table.snapshot_manager().manifest_path(name), + ) + .await + .unwrap(), + None => vec![], + } +} + +fn index_path(table: &Table, bucket: i32, file: &IndexFileMeta) -> String { + if table.schema().core_options().index_file_in_data_file_dir() { + format!("{}/bucket-{bucket}/{}", table.location(), file.file_name) + } else { + format!("{}/index/{}", table.location(), file.file_name) + } +} + +// Java v1 envelope: version, then [length, magic+roaring, checksum] per file. +async fn write_vectors(table: &Table, bucket: i32, deletions: &[(&str, &[u32])]) -> IndexFileMeta { + let mut bytes = vec![1]; + let mut ranges = IndexMap::new(); + for (name, positions) in deletions { + let bitmap: RoaringBitmap = positions.iter().copied().collect(); + let mut payload = 1581511376u32.to_be_bytes().to_vec(); + bitmap.serialize_into(&mut payload).unwrap(); + ranges.insert( + name.to_string(), + paimon::spec::DeletionVectorMeta { + offset: bytes.len() as i32, + length: payload.len() as i32, + cardinality: Some(bitmap.len() as i64), + }, + ); + bytes.extend_from_slice(&(payload.len() as i32).to_be_bytes()); + bytes.extend_from_slice(&payload); + bytes.extend_from_slice(&crc32fast::hash(&payload).to_be_bytes()); + } + let file = IndexFileMeta { + index_type: "DELETION_VECTORS".into(), + file_name: format!("index-{}-1", uuid::Uuid::new_v4()), + file_size: bytes.len() as i64, + row_count: ranges.len() as i64, + deletion_vectors_ranges: Some(ranges), + external_path: None, + global_index_meta: None, + }; + let path = index_path(table, bucket, &file); + table + .file_io() + .mkdirs(path.rsplit_once('/').unwrap().0) + .await + .unwrap(); + table + .file_io() + .new_output(&path) + .unwrap() + .write(Bytes::from(bytes)) + .await + .unwrap(); + file +} + +async fn set_vectors(table: &Table, task: &CompactTask, deletions: &[(&str, &[u32])]) { + let mut message = CommitMessage::new(task.partition().to_vec(), task.bucket(), vec![]); + message + .new_index_files + .push(write_vectors(table, task.bucket(), deletions).await); + message.deleted_index_files = indexes(table) + .await + .into_iter() + .filter(|entry| { + let ranges = &entry.index_file.deletion_vectors_ranges; + entry.partition == task.partition() + && entry.bucket == task.bucket() + && ranges.as_ref().is_some_and(|ranges| { + deletions.iter().any(|(name, _)| ranges.contains_key(*name)) + }) + }) + .map(|entry| entry.index_file) + .collect(); + table + .new_write_builder() + .new_commit() + .commit(vec![message]) + .await + .unwrap(); +} + +async fn positions(table: &Table, name: &str) -> RoaringBitmap { + for entry in indexes(table).await { + let ranges = entry.index_file.deletion_vectors_ranges.as_ref(); + if let Some(meta) = ranges.and_then(|ranges| ranges.get(name)) { + let bytes = table + .file_io() + .new_input(&index_path(table, entry.bucket, &entry.index_file)) + .unwrap() + .read() + .await + .unwrap(); + let offset = meta.offset as usize; + assert_eq!(&bytes[offset + 4..offset + 8], &1581511376u32.to_be_bytes()); + return RoaringBitmap::deserialize_from( + &bytes[offset + 8..offset + 4 + meta.length as usize], + ) + .unwrap(); + } + } + RoaringBitmap::new() +} + +#[tokio::test] +async fn dv_incremental_compact_keeps_history_unique_for_all_changelog_modes() { + for producer in ["none", "input", "lookup"] { + for bucket in ["1", "-1"] { + let table = dv_table(&[ + ("changelog-producer", producer), + ("bucket", bucket), + ( + "index-file-in-data-file-dir", + if bucket == "1" { "true" } else { "false" }, + ), + ]) + .await; + let anchor = seed(&table).await; + let before = latest(&table).await; + let original = (0..1000).map(|key| (key, key)).collect::>(); + write_batch( + &table, + &make_batch_with_kinds( + vec![257, 258, 2000], + vec![25700, 999, 20000], + vec![0, 3, 0], + ), + ) + .await; + assert_eq!(rows(&table).await, original, "default DV scan excludes L0"); + let hashes = indexes(&table) + .await + .into_iter() + .filter(|entry| entry.index_file.index_type == "HASH") + .collect::>(); + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + assert_eq!(plan.tasks().len(), 1); + assert!(plan.tasks()[0].files().iter().all(|file| file.level == 0)); + let results = execute(&table, &plan).await; + assert!(results[0] + .files() + .iter() + .all(|file| file.level == 4 && file.delete_row_count == Some(0))); + let commit = builder.new_commit().unwrap(); + let mut missing_updates = serde_json::to_value(&results).unwrap(); + for result in missing_updates.as_array_mut().unwrap() { + assert!(result + .as_object_mut() + .unwrap() + .remove("deletion_vector_updates") + .is_some()); + } + assert!(commit + .commit_compact(serde_json::from_value(missing_updates.clone()).unwrap()) + .await + .is_err()); + assert_eq!(latest(&table).await.id(), plan.snapshot_id().unwrap()); + let mut legacy = serde_json::to_value(&results).unwrap(); + for result in legacy.as_array_mut().unwrap() { + result["task"] + .as_object_mut() + .unwrap() + .remove("output_level"); + } + commit + .commit_compact(serde_json::from_value(legacy).unwrap()) + .await + .unwrap(); + let after = latest(&table).await; + commit + .commit_compact(serde_json::from_value(missing_updates).unwrap()) + .await + .unwrap(); + commit.commit_compact(results).await.unwrap(); + assert_eq!(latest(&table).await.id(), after.id()); + let mut expected = original + .iter() + .copied() + .filter(|(key, _)| *key != 258) + .collect::>(); + expected[257].1 = 25700; + expected.push((2000, 20000)); + assert_eq!( + rows(&table).await, + expected, + "{producer}/{bucket}: no duplicate history keys" + ); + assert_eq!( + positions(&table, &anchor.file_name) + .await + .iter() + .collect::>(), + vec![257, 258] + ); + assert!(all_files(&table).await.contains(&anchor)); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&before))).await, + original + ); + assert_eq!( + indexes(&table) + .await + .into_iter() + .filter(|entry| entry.index_file.index_type == "HASH") + .collect::>(), + hashes + ); + if producer == "lookup" { + let mut expected = vec![ + ("-U".into(), 257, 257), + ("+U".into(), 257, 25700), + ("-D".into(), 258, 258), + ("+I".into(), 2000, 20000), + ]; + expected.sort(); + assert_eq!( + super::lookup_changelog_tests::events(&table, before.id(), after.id()).await, + expected + ); + } else { + assert_eq!(after.changelog_record_count(), None); + } + } + } +} + +#[tokio::test] +async fn dv_winning_delete_can_publish_empty_legacy_output_and_retry() { + let table = dv_table(&[("changelog-producer", "lookup")]).await; + let anchor = seed(&table).await; + let before = latest(&table).await; + write_batch( + &table, + &make_batch_with_kinds(vec![257], vec![999], vec![3]), + ) + .await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(&table, &plan).await; + assert!(results[0].files().is_empty()); + let changelog = &results[0].changelog_files()[0]; + assert_eq!( + ( + changelog.level, + changelog.row_count, + changelog.min_sequence_number, + changelog.max_sequence_number + ), + (4, 1, 257, 257) + ); + let mut legacy = serde_json::to_value(&results).unwrap(); + legacy[0]["task"] + .as_object_mut() + .unwrap() + .remove("output_level"); + let commit = builder.new_commit().unwrap(); + let mut wrong_level = legacy.clone(); + wrong_level[0]["changelog_files"][0]["_LEVEL"] = 1.into(); + let wrong_level: Vec = serde_json::from_value(wrong_level).unwrap(); + assert_eq!(wrong_level[0].changelog_files()[0].level, 1); + let captured = latest(&table).await.id(); + assert!(commit.commit_compact(wrong_level).await.is_err()); + assert_eq!(latest(&table).await.id(), captured); + commit + .commit_compact(serde_json::from_value(legacy).unwrap()) + .await + .unwrap(); + let after = latest(&table).await; + commit.commit_compact(results).await.unwrap(); + assert_eq!(latest(&table).await.id(), after.id()); + assert_eq!( + super::lookup_changelog_tests::events(&table, before.id(), after.id()).await, + vec![("-D".into(), 257, 257)] + ); + assert_eq!( + positions(&table, &anchor.file_name) + .await + .iter() + .collect::>(), + vec![257] + ); + assert_eq!( + rows(&table).await, + (0..1000) + .filter(|key| *key != 257) + .map(|key| (key, key)) + .collect::>() + ); +} + +#[tokio::test] +async fn dv_filters_selected_rows_and_keeps_other_ranges_in_a_shared_container() { + let table = dv_table(&[]).await; + let anchor = seed(&table).await; + write_batch(&table, &make_batch(vec![42, 43], vec![4200, 4300])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let selected = plan.tasks()[0].files()[0].file_name.clone(); + set_vectors( + &table, + &plan.tasks()[0], + &[(&selected, &[0]), (&anchor.file_name, &[500])], + ) + .await; + let before = latest(&table).await; + let old_rows = rows(&table).await; + compact(&table).await; + let expected = (0..1000) + .filter(|key| *key != 500) + .map(|key| (key, if key == 43 { 4300 } else { key })) + .collect::>(); + assert_eq!(rows(&table).await, expected); + assert_eq!( + positions(&table, &anchor.file_name) + .await + .iter() + .collect::>(), + vec![43, 500] + ); + assert!(positions(&table, &selected).await.is_empty()); + assert!(!all_files(&table) + .await + .iter() + .any(|file| file.file_name == selected)); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&before))).await, + old_rows + ); +} + +#[tokio::test] +async fn dv_commit_accepts_concurrent_l0_but_rejects_changed_inputs_or_missing_targets() { + for change in [ + "l0", + "unrelated-target-dv", + "target-dv", + "input-dv", + "missing-target", + ] { + let table = dv_table(&[]).await; + let anchor = seed(&table).await; + let anchor_name = anchor.file_name.clone(); + write_batch(&table, &make_batch(vec![42], vec![9000])).await; + let builder = table.new_compact_builder(); + let plan = builder.new_planner().unwrap().plan().await.unwrap(); + let results = execute(&table, &plan).await; + match change { + "l0" => write_batch(&table, &make_batch(vec![42], vec![9001])).await, + "unrelated-target-dv" | "target-dv" => { + let position = if change == "target-dv" { 42 } else { 500 }; + set_vectors(&table, &plan.tasks()[0], &[(&anchor_name, &[position])]).await + } + "input-dv" => { + set_vectors( + &table, + &plan.tasks()[0], + &[(&plan.tasks()[0].files()[0].file_name, &[0])], + ) + .await + } + "missing-target" => { + let mut removal = CommitMessage::new( + plan.tasks()[0].partition().to_vec(), + plan.tasks()[0].bucket(), + vec![], + ); + removal.deleted_files.push(anchor); + table + .new_write_builder() + .new_commit() + .commit(vec![removal]) + .await + .unwrap(); + } + _ => unreachable!(), + } + let before = latest(&table).await.id(); + let commit = builder.new_commit().unwrap(); + let result = commit.commit_compact(results.clone()).await; + if matches!(change, "l0" | "unrelated-target-dv") { + result.unwrap(); + let published = latest(&table).await.id(); + commit.commit_compact(results).await.unwrap(); + assert_eq!(latest(&table).await.id(), published); + assert_eq!( + rows(&table).await.iter().find(|(key, _)| *key == 42), + Some(&(42, 9000)) + ); + if change == "l0" { + compact(&table).await; + assert_eq!( + rows(&table).await.iter().find(|(key, _)| *key == 42), + Some(&(42, 9001)) + ); + } else { + assert_eq!( + positions(&table, &anchor_name) + .await + .iter() + .collect::>(), + vec![42, 500] + ); + let visible = rows(&table).await; + assert_eq!(visible.len(), 999); + assert!(!visible.iter().any(|(key, _)| *key == 500)); + } + } else { + assert!(result.is_err(), "{change}"); + assert_eq!(latest(&table).await.id(), before); + } + } +} + +#[tokio::test] +async fn dv_planning_checks_persisted_schema_even_when_its_id_matches_current_options() { + let original = table(&[]).await; + write_batch(&original, &make_batch(vec![1], vec![10])).await; + let dv = original.copy_with_options(std::collections::HashMap::from([( + "deletion-vectors.enabled".into(), + "true".into(), + )])); + assert_eq!(dv.schema().id(), original.schema().id()); + let error = dv + .new_compact_builder() + .new_planner() + .unwrap() + .plan() + .await + .unwrap_err(); + assert!( + error.to_string().contains("without deletion vectors"), + "{error}" + ); +} + +#[tokio::test] +async fn dv_repeated_updates_keep_one_bucket_container_and_old_snapshot_masks() { + for bucket in ["1", "-1"] { + let table = dv_table(&[("bucket", bucket)]).await; + let anchor = seed(&table).await; + // Keep this intermediate run larger than the next L0 file, so the + // planner looks it up rather than selecting it for a rewrite. + write_batch( + &table, + &make_batch((0..256).collect(), (4200..4456).collect()), + ) + .await; + let first = compact(&table).await; + let intermediate = first[0].files()[0].clone(); + assert_eq!(intermediate.row_count, 256); + let previous = latest(&table).await; + let previous_rows = rows(&table).await; + write_batch(&table, &make_batch(vec![42], vec![4201])).await; + compact(&table).await; + + let vectors = indexes(&table) + .await + .into_iter() + .filter(|entry| entry.index_file.index_type == "DELETION_VECTORS") + .collect::>(); + assert_eq!( + vectors.len(), + 1, + "Java combines DV entries by bucket: {bucket}" + ); + assert_eq!( + positions(&table, &anchor.file_name) + .await + .iter() + .collect::>(), + (0..256).collect::>() + ); + assert_eq!( + positions(&table, &intermediate.file_name) + .await + .iter() + .collect::>(), + vec![42] + ); + assert_eq!( + rows(&table).await, + (0..1000) + .map(|key| { + ( + key, + if key == 42 { + 4201 + } else if key < 256 { + key + 4200 + } else { + key + }, + ) + }) + .collect::>() + ); + assert_eq!( + rows(&table.copy_with_pinned_snapshot(Some(&previous))).await, + previous_rows + ); + } +} diff --git a/crates/paimon/tests/pk_compact_test.rs b/crates/paimon/tests/pk_compact_test.rs index 776aaddcf..fe45814be 100644 --- a/crates/paimon/tests/pk_compact_test.rs +++ b/crates/paimon/tests/pk_compact_test.rs @@ -32,6 +32,9 @@ mod incremental_lookup_tests; #[path = "pk_compact/dynamic_bucket.rs"] mod dynamic_tests; +#[path = "pk_compact/deletion_vectors.rs"] +mod deletion_vector_tests; + use arrow_array::{Array, Int32Array, Int64Array, RecordBatch, StringArray}; use common::incremental_helpers::{ make_batch, make_batch_with_kinds, make_partitioned_batch, persist_table_schema, pk_schema, diff --git a/docs/src/compaction.md b/docs/src/compaction.md index b936547bc..e76928b91 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -130,9 +130,25 @@ The policy uses `num-sorted-run.compaction-trigger` (default `5`), `num-levels` `compaction.max-size-amplification-percent` (default `200`). When selected runs reach the oldest run, output goes to the highest configured or existing level; otherwise it goes immediately below the next unselected run. Explicit full -compaction is not exposed. Outputs retain winning retracts, so older values +compaction is not exposed. Without deletion vectors, outputs retain winning retracts, so older values in untouched files do not reappear after a delete or out-of-order update. +With deletion vectors, `deduplicate` supports the same incremental selection for +`none`, `input` and `lookup`. All three require actual catalog/FileIO local-cache. +Lookup Files preserve unfiltered physical row positions; captured deletion vectors +filter selected inputs and history before merging. A history hit moves its winning +state into the output and marks the original position deleted, even when an +out-of-order input loses. Winning retracts are omitted from data after changelog +generation, so a DELETE may publish a changelog without a new data file. + +Commit atomically unions historical row deletions with the latest DV indexes and +retires rewritten inputs' ranges, preserving unrelated ranges and HASH indexes. +It rejects changed input masks, removed lookup targets and concurrently deleted +hit positions; unrelated new DV bits can be merged. Uncommitted serialized DV results must +retain their physical deletion metadata, including an empty list when appropriate. +Each output file is limited to 4,294,967,296 physical rows. Migrating files written +without DV is unsupported, including enabling DV through temporary table options. + `first-row`, `partial-update` and `aggregation` retain their current selection behavior: every active file in a selected bucket participates, and output uses the highest input level or level 1 for all-L0 inputs. Their state @@ -346,16 +362,16 @@ deletions into reassigned row IDs are not provided. Setting ## Current scope -The `full-compaction` changelog producer is unsupported; `lookup` is supported -for primary-key compaction without deletion vectors. -Outside Data Evolution mode, row tracking, deletion vectors and dedicated +The `full-compaction` changelog producer is unsupported. Primary-key deletion-vector +compaction supports `deduplicate` with `none`, `input` or `lookup`. +Outside Data Evolution mode, row tracking and dedicated blob/vector storage are unsupported. Query authorization, rowkind fields and postpone buckets are unsupported in all modes. Ordered bucketed append, append incremental clustering, primary-key clustering override (`pk-clustering-override`) and record-level expiration (`record-level.expire-time`) are also unsupported. -Snapshot index manifests may contain dynamic-bucket HASH indexes, or deletion -vectors and global indexes in Data Evolution mode. +Snapshot index manifests may contain dynamic-bucket HASH indexes and primary-key +deletion vectors, or deletion vectors and global indexes in Data Evolution mode. Embedded and sidecar file indexes are regenerated according to the captured table options. Language bindings and background compaction scheduling are not yet provided. diff --git a/docs/src/sql.md b/docs/src/sql.md index 208407451..6d99357a2 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -2949,8 +2949,8 @@ plans, cross-partition dynamic bucket writes, or advanced aggregation options such as `ignore-retract`, `distinct`, `nested-key`, `count-limit`, and sequence groups. -Primary-key compaction supports `changelog-producer=lookup` without deletion -vectors, generating INSERT, UPDATE_BEFORE, UPDATE_AFTER and DELETE events when +Primary-key compaction supports `changelog-producer=lookup`, including +deletion-vector-enabled `deduplicate` tables, generating INSERT, UPDATE_BEFORE, UPDATE_AFTER and DELETE events when L0 files are compacted. The executor requires `local-cache.enabled=true` on the catalog/FileIO. See [Lookup changelog producer](compaction.md#lookup-changelog-producer).