This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 5aeb1144 fix(commit): close remaining TableCommit parity gaps (#915)
5aeb1144 is described below
commit 5aeb11444a34b18ae6ea3d6d6b5362769aaaa998
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 19:31:03 2026 +0800
fix(commit): close remaining TableCommit parity gaps (#915)
---
bindings/python/tests/test_table_commit.py | 3 +-
.../datafusion/tests/partition_count_pushdown.rs | 25 +-
crates/paimon/src/table/data_evolution_writer.rs | 29 +-
crates/paimon/src/table/table_commit.rs | 334 ++++++++---
.../paimon/src/table/table_commit/parity_tests.rs | 13 +-
.../src/table/table_commit/recovery_tests.rs | 615 +++++++++++++++++++++
crates/paimon/src/table/table_write.rs | 72 +++
7 files changed, 983 insertions(+), 108 deletions(-)
diff --git a/bindings/python/tests/test_table_commit.py
b/bindings/python/tests/test_table_commit.py
index b168b224..12b6f666 100644
--- a/bindings/python/tests/test_table_commit.py
+++ b/bindings/python/tests/test_table_commit.py
@@ -404,7 +404,8 @@ def
test_close_cleans_unprepared_output_but_preserves_prepared_files(tmp_path, b
assert prepared_paths
_write(writer, [2], [20])
_write(writer, [3], [30])
- assert set(tmp_path.rglob("*.parquet")) - prepared_paths
+ # Rolled files may still be closing in the background. close() must await
+ # that work and clean outstanding output before we inspect the directory.
writer.close()
assert set(tmp_path.rglob("*.parquet")) == prepared_paths
builder.new_commit().commit(1, messages)
diff --git a/crates/integrations/datafusion/tests/partition_count_pushdown.rs
b/crates/integrations/datafusion/tests/partition_count_pushdown.rs
index 357d7282..f20674f2 100644
--- a/crates/integrations/datafusion/tests/partition_count_pushdown.rs
+++ b/crates/integrations/datafusion/tests/partition_count_pushdown.rs
@@ -706,10 +706,11 @@ async fn
test_removed_file_deletion_vector_does_not_reduce_count() {
.clone();
assert_eq!(removed.row_count, 2);
- // The public commit API can remove a data file while retaining its DV
index.
+ // Remove the data file and its DV together, as required by commit
validation.
let mut message =
paimon::table::CommitMessage::new(entry.partition.clone(),
entry.bucket, vec![]);
message.deleted_files.push(removed);
+ message.deleted_index_files.push(entry.index_file.clone());
table
.new_write_builder()
.new_commit()
@@ -717,7 +718,27 @@ async fn
test_removed_file_deletion_vector_does_not_reduce_count() {
.await
.unwrap();
let after = snapshots.get_latest_snapshot().await.unwrap().unwrap();
- assert_eq!(after.index_manifest(), snapshot.index_manifest());
+ // Model a legacy snapshot that retained the removed file's DV. Build this
+ // reader-compatibility fixture directly; new commits must reject that
state.
+ let mut metadata = serde_json::to_value(&after).unwrap();
+ metadata["indexManifest"] = serde_json::json!(snapshot.index_manifest());
+ let snapshot_path = snapshots.snapshot_path(after.id());
+ table.file_io().delete_file(&snapshot_path).await.unwrap();
+ table
+ .file_io()
+ .new_output(&snapshot_path)
+ .unwrap()
+ .write(serde_json::to_vec(&metadata).unwrap().into())
+ .await
+ .unwrap();
+ assert_eq!(
+ snapshots
+ .get_snapshot(after.id())
+ .await
+ .unwrap()
+ .index_manifest(),
+ snapshot.index_manifest()
+ );
let plan = table.new_read_builder().new_scan().plan().await.unwrap();
assert!(plan
.splits()
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index d2855348..9373dcae 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -30,7 +30,7 @@ use crate::deletion_vector::{DeletionVector,
DeletionVectorFactory};
use crate::io::FileIO;
use crate::spec::{
bucket_path, BinaryRow, CoreOptions, DataField, DataFileMeta, DataType,
DeletionVectorMeta,
- FileKind, IndexFileMeta, IndexManifest, PartitionComputer,
EMPTY_BINARY_ROW,
+ FileKind, IndexFileMeta, IndexManifest, PartitionComputer, Snapshot,
EMPTY_BINARY_ROW,
};
use crate::table::commit_message::CommitMessage;
use crate::table::data_file_writer::DataFileWriter;
@@ -38,7 +38,6 @@ use crate::table::index_file_path::IndexFileLocation;
use crate::table::source::data_evolution_anchor_file;
use crate::table::stats_filter::group_by_overlapping_row_id;
use crate::table::DataSplitBuilder;
-use crate::table::SnapshotManager;
use crate::table::Table;
use crate::Result;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch};
@@ -485,8 +484,14 @@ impl DataEvolutionDeleteWriter {
return Ok(Vec::new());
}
- let scan = self
- .table
+ let snapshot = super::time_travel::resolve_snapshot(&self.table)
+ .await?
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "No files with row tracking found in target
table".into(),
+ source: None,
+ })?;
+ let scan_table = self.table.copy_with_pinned_snapshot(&snapshot);
+ let scan = scan_table
.new_read_builder()
.new_scan()
.with_scan_all_files();
@@ -572,7 +577,7 @@ impl DataEvolutionDeleteWriter {
let mut messages = Vec::new();
for ((partition, bucket), delete_plan) in deletes_by_bucket {
if let Some(message) = self
- .prepare_bucket_delete_message(partition, bucket, delete_plan)
+ .prepare_bucket_delete_message(partition, bucket, delete_plan,
&snapshot)
.await?
{
messages.push(message);
@@ -587,13 +592,10 @@ impl DataEvolutionDeleteWriter {
partition: Vec<u8>,
bucket: i32,
delete_plan: BucketDeletePlan,
+ snapshot: &Snapshot,
) -> Result<Option<CommitMessage>> {
let (mut bitmaps, deleted_index_files) = self
- .read_existing_bucket_deletion_vectors(
- &partition,
- bucket,
- delete_plan.check_from_snapshot,
- )
+ .read_existing_bucket_deletion_vectors(&partition, bucket,
snapshot)
.await?;
let mut changed = false;
@@ -661,13 +663,8 @@ impl DataEvolutionDeleteWriter {
&self,
partition: &[u8],
bucket: i32,
- snapshot_id: i64,
+ snapshot: &Snapshot,
) -> Result<(IndexMap<String, RoaringBitmap>, Vec<IndexFileMeta>)> {
- let snapshot_manager = SnapshotManager::new(
- self.table.file_io().clone(),
- self.table.location().to_string(),
- );
- let snapshot = snapshot_manager.get_snapshot(snapshot_id).await?;
let Some(index_manifest_name) = snapshot.index_manifest() else {
return Ok((IndexMap::new(), Vec::new()));
};
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index b0b8ec2c..b1d399b2 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -52,6 +52,14 @@ type PartitionBucketKey = (Vec<u8>, i32);
type RowIdRange = (i64, i64);
type ExistingRowIdRanges = HashMap<PartitionBucketKey, Vec<RowIdRange>>;
+fn commit_file_identifier(entry: &ManifestEntry) -> crate::spec::Identifier {
+ let mut identifier = entry.identifier();
+ if is_empty_partition(&identifier.partition) {
+ identifier.partition.clear();
+ }
+ identifier
+}
+
fn validate_file_entries<'a>(entries: impl IntoIterator<Item = &'a
ManifestEntry>) -> Result<()> {
// Mirror Java FileEntry.mergeEntries while also rejecting repeated entries
// of the same kind instead of letting two DELETEs cancel each other.
@@ -65,7 +73,7 @@ fn validate_file_entries<'a>(entries: impl IntoIterator<Item
= &'a ManifestEntry
let mut files = HashMap::new();
for entry in entries {
let state = files
- .entry(entry.identifier())
+ .entry(commit_file_identifier(entry))
.or_insert_with(State::default);
let duplicate = match entry.kind() {
FileKind::Add => {
@@ -252,6 +260,17 @@ impl TableCommit {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
commits.sort_by_key(|(id, _)| *id);
+ for pair in commits.windows(2) {
+ if pair[0].0 == pair[1].0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Duplicate commit identifier {} in recovery batch",
+ pair[0].0
+ ),
+ source: None,
+ });
+ }
+ }
let latest = self.snapshot_manager.get_latest_snapshot().await?;
let mut pending = Vec::new();
for (id, messages) in commits {
@@ -266,48 +285,19 @@ impl TableCommit {
}
let count = pending.len();
for (id, messages) in pending {
- self.commit_with_identifier(messages, id).await?;
+ self.filter_and_commit_with_identifier(messages, id).await?;
}
Ok(count)
}
async fn check_recovery_files(&self, messages: &[CommitMessage]) ->
Result<()> {
- let index_in_bucket =
-
CoreOptions::new(self.table.schema().options()).index_file_in_data_file_dir();
- for message in messages {
- let bucket_path = self.bucket_path(&message.partition,
message.bucket)?;
- let mut paths = Vec::new();
- for file in message
- .new_files
- .iter()
- .chain(&message.new_changelog_files)
- .chain(&message.compact_after)
- .chain(&message.compact_changelog_files)
- {
- paths.extend(file.collect_files(&bucket_path));
- }
- for file in message
- .new_index_files
- .iter()
- .chain(&message.compact_new_index_files)
- {
- paths.push(committed_index_file_path(
- self.table.location().trim_end_matches('/'),
- &bucket_path,
- index_in_bucket,
- file,
- ));
- }
- for path in paths {
- if !self.table.file_io().exists(&path).await? {
- return Err(crate::Error::DataInvalid {
- message: format!("Cannot recover commit: file '{path}'
does not exist"),
- source: None,
- });
- }
- }
- }
- Ok(())
+ reject_compact_increment(messages)?;
+ self.check_recovery_entries(
+ &self.messages_to_entries(messages),
+ &self.messages_to_changelog_entries(messages),
+ &self.messages_to_index_entries(messages),
+ )
+ .await
}
/// Commit new files in APPEND mode.
@@ -493,7 +483,10 @@ impl TableCommit {
validate_fixed_bucket_commit_mode(&commit_messages, true)?;
validate_bucket_ownership(&commit_messages)?;
- if commit_messages.is_empty() && static_partitions.is_none() {
+ if commit_messages.is_empty()
+ && static_partitions.is_none()
+ && !self.table.schema().partition_fields().is_empty()
+ {
return Ok(());
}
@@ -958,6 +951,7 @@ impl TableCommit {
let commit_empty_overwrite = plan.commit_kind_hint() ==
CommitKind::OVERWRITE;
let commit_empty_append =
!self.ignore_empty_commit && matches!(plan,
CommitEntriesPlan::Direct { .. });
+ let check_append_files = filter_committed;
let mut filter_committed = filter_committed;
let mut publication_uncertain = false;
@@ -972,6 +966,7 @@ impl TableCommit {
{
break;
}
+ self.check_recovered_files_exist(&plan).await?;
filter_committed = false;
}
if let Some(start_snapshot_id) =
duplicate_check_start_snapshot_id {
@@ -989,7 +984,12 @@ impl TableCommit {
}
validate_expected_latest_snapshot(expected_snapshot_id,
&latest_snapshot)?;
let resolved = self
- .resolve_commit(&mut plan, &latest_snapshot,
retry_state.as_deref())
+ .resolve_commit(
+ &mut plan,
+ &latest_snapshot,
+ retry_state.as_deref(),
+ check_append_files,
+ )
.await?;
if resolved.entries.is_empty()
@@ -1048,6 +1048,68 @@ impl TableCommit {
}
}
+ /// Validate files before replaying a commit whose identity is no longer
visible.
+ async fn check_recovered_files_exist(&self, plan: &CommitEntriesPlan) ->
Result<()> {
+ let (entries, changelog_entries, index_entries) = match plan {
+ CommitEntriesPlan::Direct {
+ entries,
+ changelog_entries,
+ new_index_entries,
+ ..
+ } => (
+ entries.as_slice(),
+ changelog_entries.as_slice(),
+ new_index_entries,
+ ),
+ CommitEntriesPlan::Overwrite {
+ new_entries,
+ new_index_entries,
+ ..
+ } => (new_entries.as_slice(), &[][..], new_index_entries),
+ };
+ self.check_recovery_entries(entries, changelog_entries, index_entries)
+ .await
+ }
+
+ async fn check_recovery_entries(
+ &self,
+ entries: &[ManifestEntry],
+ changelog_entries: &[ManifestEntry],
+ index_entries: &[IndexManifestEntry],
+ ) -> Result<()> {
+ let mut paths = HashSet::new();
+ for entry in entries
+ .iter()
+ .chain(changelog_entries)
+ .filter(|e| *e.kind() == FileKind::Add)
+ {
+ paths.extend(
+ entry
+ .file()
+ .collect_files(&self.bucket_path(entry.partition(),
entry.bucket())?),
+ );
+ }
+ let index_in_data_dir =
+
CoreOptions::new(self.table.schema().options()).index_file_in_data_file_dir();
+ for entry in index_entries.iter().filter(|e| e.kind == FileKind::Add) {
+ paths.insert(committed_index_file_path(
+ self.table.location().trim_end_matches('/'),
+ &self.bucket_path(&entry.partition, entry.bucket)?,
+ index_in_data_dir,
+ &entry.index_file,
+ ));
+ }
+ for path in paths {
+ if !self.table.file_io().exists(&path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot recover commit: file '{path}'
does not exist"),
+ source: None,
+ });
+ }
+ }
+ Ok(())
+ }
+
/// Single commit attempt.
async fn try_commit_once(
&self,
@@ -1565,6 +1627,7 @@ impl TableCommit {
plan: &mut CommitEntriesPlan,
latest_snapshot: &Option<Snapshot>,
retry_state: Option<&RetryState>,
+ check_append_files: bool,
) -> Result<ResolvedCommit> {
let file_io = self.snapshot_manager.file_io();
let manifest_dir = self.snapshot_manager.manifest_dir();
@@ -1578,7 +1641,6 @@ impl TableCommit {
} => {
validate_file_entries(entries.iter())?;
- let has_delete = entries.iter().any(|e| *e.kind() ==
FileKind::Delete);
let kind = direct_commit_kind(entries, new_index_entries);
let has_partition_bucket_counts = entries
.iter()
@@ -1587,16 +1649,12 @@ impl TableCommit {
&& entries.iter().any(|entry| {
*entry.kind() == FileKind::Add && entry.bucket() ==
POSTPONE_BUCKET
});
- let detect_conflicts = has_delete
+ let detect_conflicts = check_append_files
+ || kind == CommitKind::OVERWRITE
|| check_from_snapshot.is_some()
|| has_partition_bucket_counts
|| has_postpone_entries;
let base_data_files = if detect_conflicts {
- self.check_deletion_vector_conflicts(
- latest_snapshot.as_ref(),
- new_index_entries,
- *check_from_snapshot,
- )?;
self.detect_commit_conflicts(
latest_snapshot,
retry_state,
@@ -1624,6 +1682,13 @@ impl TableCommit {
new_index_entries,
)?);
let all = Self::merge_index_entries(&previous, &index_entries,
false)?;
+ self.check_deletion_vector_references(
+ latest_snapshot,
+ entries,
+ &index_entries,
+ &all,
+ )
+ .await?;
let index_manifest_changed = all != previous;
let index_manifest_name = latest_snapshot
.as_ref()
@@ -1685,6 +1750,13 @@ impl TableCommit {
}
}
let all = Self::merge_index_entries(&all, &new_index_entries,
false)?;
+ self.check_deletion_vector_references(
+ latest_snapshot,
+ &entries,
+ &new_index_entries,
+ &all,
+ )
+ .await?;
let index_manifest_changed = all != previous;
let index_manifest_name = latest_snapshot
.as_ref()
@@ -1806,6 +1878,15 @@ impl TableCommit {
let Some(write_cols) = entry.file().write_cols.as_ref() else {
continue;
};
+ // Dedicated normal/vector/blob INSERT files also carry write_cols.
+ // Only files targeting existing row IDs can invalidate an
existing index.
+ if entry.file().first_row_id.is_none()
+ && !write_cols
+ .iter()
+ .any(|col| col == crate::spec::ROW_ID_FIELD_NAME)
+ {
+ continue;
+ }
for col in write_cols {
if !is_system_field(col) {
updated_cols.insert(col.clone());
@@ -2395,40 +2476,103 @@ impl TableCommit {
Ok(())
}
- fn check_deletion_vector_conflicts(
+ /// Validate DV references against the state this attempt will publish.
Index
+ /// merging separately verifies replacement identities and one DV per file.
+ async fn check_deletion_vector_references(
&self,
- latest_snapshot: Option<&Snapshot>,
+ latest_snapshot: &Option<Snapshot>,
+ data_entries: &[ManifestEntry],
index_entries: &[IndexManifestEntry],
- check_from_snapshot: Option<i64>,
+ merged_indexes: &[IndexManifestEntry],
) -> Result<()> {
- if !self.data_evolution_enabled {
- return Ok(());
- }
- let Some(check_from_snapshot) = check_from_snapshot else {
- return Ok(());
+ let file_key = |partition: &[u8], bucket: i32, name: &str| {
+ (
+ if is_empty_partition(partition) {
+ vec![]
+ } else {
+ partition.to_vec()
+ },
+ bucket,
+ name.to_string(),
+ )
};
- let has_deletion_vector_index_change = index_entries
+ let deleted_files = data_entries
.iter()
- .any(|entry| entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE);
- if !has_deletion_vector_index_change {
- return Ok(());
+ .filter(|entry| *entry.kind() == FileKind::Delete)
+ .map(|entry| file_key(entry.partition(), entry.bucket(),
&entry.file().file_name))
+ .collect::<HashSet<_>>();
+ let added_dvs = index_entries
+ .iter()
+ .filter(|entry| {
+ entry.kind == FileKind::Add
+ && entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE
+ })
+ .map(|entry| file_key(&entry.partition, entry.bucket,
&entry.index_file.file_name))
+ .collect::<HashSet<_>>();
+ let mut referenced_files = HashSet::new();
+ for entry in merged_indexes
+ .iter()
+ .filter(|entry| entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE)
+ {
+ let is_added = added_dvs.contains(&file_key(
+ &entry.partition,
+ entry.bucket,
+ &entry.index_file.file_name,
+ ));
+ if let Some(ranges) = &entry.index_file.deletion_vectors_ranges {
+ for name in ranges.keys() {
+ let key = file_key(&entry.partition, entry.bucket, name);
+ // Also forbid retaining a DV after deleting its data file.
+ if is_added || deleted_files.contains(&key) {
+ referenced_files.insert(key);
+ }
+ }
+ }
}
- let Some(latest_snapshot) = latest_snapshot else {
+ if referenced_files.is_empty() {
return Ok(());
+ }
+ // DV partitions need not be the partitions of the data entries in a
+ // mixed commit. Do not reuse the data conflict scan's narrower cache.
+ let fields = self.table.schema().partition_fields();
+ let filter = if fields.is_empty() {
+ None
+ } else {
+ Some(PartitionFilter::from_partition_set(
+ referenced_files
+ .iter()
+ .map(|(partition, _, _)| partition.clone())
+ .collect(),
+ &fields,
+ )?)
};
- if latest_snapshot.id() <= check_from_snapshot {
- return Ok(());
+ let base = self
+ .scan_snapshot_entries(latest_snapshot, filter.as_ref())
+ .await?;
+ let mut active_files = base
+ .iter()
+ .map(|entry| file_key(entry.partition(), entry.bucket(),
&entry.file().file_name))
+ .collect::<HashSet<_>>();
+ for entry in data_entries {
+ let key = file_key(entry.partition(), entry.bucket(),
&entry.file().file_name);
+ match entry.kind() {
+ FileKind::Add => {
+ active_files.insert(key);
+ }
+ FileKind::Delete => {
+ active_files.remove(&key);
+ }
+ }
}
-
- Err(crate::Error::DataInvalid {
- message: format!(
- "Row ID conflict: deletion-vector DELETE was prepared from
snapshot \
- {check_from_snapshot}, but latest snapshot is {}. Retry with
the latest \
- deletion vectors.",
- latest_snapshot.id()
- ),
- source: None,
- })
+ for key in referenced_files {
+ if !active_files.contains(&key) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Deletion vector references missing data
file '{}' in bucket {}; prepare the DELETE again from the latest snapshot",
key.2, key.1),
+ source: None,
+ });
+ }
+ }
+ Ok(())
}
fn check_delete_entries_against_base(
@@ -2438,10 +2582,10 @@ impl TableCommit {
) -> Result<()> {
let mut active_identifiers = base_entries
.iter()
- .map(ManifestEntry::identifier)
+ .map(commit_file_identifier)
.collect::<HashSet<_>>();
for entry in delta_entries {
- let identifier = entry.identifier();
+ let identifier = commit_file_identifier(entry);
match entry.kind() {
FileKind::Add => {
active_identifiers.insert(identifier);
@@ -2602,10 +2746,17 @@ impl TableCommit {
return Ok(());
};
- let source_snapshot = self
- .snapshot_manager
- .get_snapshot(check_from_snapshot)
- .await?;
+ let write_ranges =
self.build_row_id_write_ranges(delta_entries).await?;
+ if write_ranges.is_empty() {
+ return Ok(());
+ }
+ let source_snapshot = if check_from_snapshot == latest_snapshot.id() {
+ latest_snapshot.clone()
+ } else {
+ self.snapshot_manager
+ .get_snapshot(check_from_snapshot)
+ .await?
+ };
let check_next_row_id =
source_snapshot
.next_row_id()
@@ -2616,15 +2767,14 @@ impl TableCommit {
source: None,
})?;
- let write_ranges =
self.build_row_id_write_ranges(delta_entries).await?;
- if write_ranges.is_empty() {
- return Ok(());
- }
-
let delta_entry_refs = delta_entries.iter().collect::<Vec<_>>();
let partition_filter =
self.build_entries_partition_filter(&delta_entry_refs)?;
for snapshot_id in check_from_snapshot + 1..=latest_snapshot.id() {
- let snapshot =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
+ let snapshot = if snapshot_id == latest_snapshot.id() {
+ latest_snapshot.clone()
+ } else {
+ self.snapshot_manager.get_snapshot(snapshot_id).await?
+ };
if snapshot.commit_kind() == &CommitKind::COMPACT {
continue;
}
@@ -3478,6 +3628,7 @@ mod tests {
mod parity {
use super::*;
include!("table_commit/parity_tests.rs");
+ include!("table_commit/recovery_tests.rs");
}
#[tokio::test]
@@ -4163,6 +4314,12 @@ mod tests {
let overwrite =
CommitMessage::new(vec![], 0,
vec![test_data_file("overwrite.parquet", 100)]);
+ file_io
+ .new_output(&format!("{table_path}/bucket-0/overwrite.parquet"))
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared data"))
+ .await
+ .unwrap();
commit
.overwrite_with_identifier(vec![overwrite.clone()], None, 2)
.await
@@ -4497,8 +4654,8 @@ mod tests {
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
assert!(
- err_msg.contains("Row ID conflict"),
- "expected row-id conflict for stale DV commit, got: {err_msg}"
+ err_msg.contains("Conflicting deletion vectors"),
+ "expected conflicting vectors for stale DV commit, got: {err_msg}"
);
let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
@@ -4771,6 +4928,7 @@ mod tests {
commit.commit(vec![first]).await.unwrap();
let mut data_file = test_data_file("data-update-id.parquet", 10);
+ data_file.first_row_id = Some(0);
data_file.write_cols = Some(vec!["id".to_string()]);
let result = commit
.commit(vec![CommitMessage::new(vec![], 0, vec![data_file])])
@@ -4807,6 +4965,7 @@ mod tests {
commit.commit(vec![first]).await.unwrap();
let mut data_file = test_data_file("data-update-id.parquet", 10);
+ data_file.first_row_id = Some(0);
data_file.write_cols = Some(vec!["id".to_string()]);
commit
.commit(vec![CommitMessage::new(vec![], 0, vec![data_file])])
@@ -4837,6 +4996,7 @@ mod tests {
commit.commit(vec![first]).await.unwrap();
let mut data_file = test_data_file("data-update-name.parquet", 10);
+ data_file.first_row_id = Some(0);
data_file.write_cols = Some(vec!["name".to_string()]);
let result = commit
.commit(vec![CommitMessage::new(vec![], 0, vec![data_file])])
@@ -4879,6 +5039,7 @@ mod tests {
commit.commit(vec![first]).await.unwrap();
let mut data_file = test_data_file("data-update-name.parquet", 10);
+ data_file.first_row_id = Some(0);
data_file.write_cols = Some(vec!["name".to_string()]);
commit
.commit(vec![CommitMessage::new(vec![], 0, vec![data_file])])
@@ -4902,6 +5063,7 @@ mod tests {
commit.commit(vec![first]).await.unwrap();
let mut data_file = test_data_file("data-update-name.parquet", 10);
+ data_file.first_row_id = Some(0);
data_file.write_cols = Some(vec!["name".to_string()]);
commit
.commit(vec![CommitMessage::new(vec![], 0, vec![data_file])])
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
index 972530c6..8799eddd 100644
--- a/crates/paimon/src/table/table_commit/parity_tests.rs
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -123,7 +123,7 @@ async fn dv_replacement_rejects_missing_old_index() {
setup_dirs(&io, path).await;
let commit = setup_commit(&io, path);
commit
- .commit(vec![dv_message("current-dv", "data.parquet")])
+ .commit(vec![append_message("data.parquet"), dv_message("current-dv",
"data.parquet")])
.await
.unwrap();
let mut message = dv_message("new-dv", "data.parquet");
@@ -142,6 +142,8 @@ async fn
dv_replacement_is_overwrite_and_preserves_unrelated_vectors() {
let commit = setup_commit(&io, path);
commit
.commit(vec![
+ append_message("data-a"),
+ append_message("data-b"),
dv_message("old", "data-a"),
dv_message("other", "data-b"),
])
@@ -181,12 +183,14 @@ async fn
unrelated_dv_delete_does_not_bypass_indexed_column_policy() {
);
let commit = TableCommit::new(table, "test-user".into());
let mut initial = dv_message("old-dv", "data");
+ initial.new_files.push(test_data_file("data", 10));
initial
.new_index_files
.push(test_global_index_file("global", 0, 0, 9));
commit.commit(vec![initial]).await.unwrap();
let mut update = append_message("update.parquet");
update.new_files[0].write_cols = Some(vec!["id".into()]);
+ update.new_files[0].first_row_id = Some(0);
let mut delete = dv_message("new-dv", "data");
delete
.deleted_index_files
@@ -356,6 +360,9 @@ async fn mixed_append_cannot_bypass_stale_dv_check() {
for name in ["data", "concurrent"] {
let mut message = append_message(name);
message.new_files[0].file_source = Some(0);
+ if name == "concurrent" {
+ message.new_index_files =
vec![test_deletion_vector_index_file("first-dv", "data")];
+ }
commit.commit(vec![message]).await.unwrap();
}
let mut delete = dv_message("dv", "data");
@@ -364,7 +371,7 @@ async fn mixed_append_cannot_bypass_stale_dv_check() {
append.new_files[0].file_source = Some(0);
let error = commit.commit(vec![append, delete]).await.unwrap_err();
assert!(
- error.to_string().contains("deletion-vector DELETE"),
+ error.to_string().contains("Conflicting deletion vectors"),
"{error}"
);
assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 2);
@@ -545,7 +552,7 @@ async fn
dv_publication_response_loss_uses_overwrite_identity() {
});
commit.snapshot_commit = publisher.clone();
commit
- .commit_with_identifier(vec![dv_message("dv", "data")], 7)
+ .commit_with_identifier(vec![append_message("data"), dv_message("dv",
"data")], 7)
.await
.unwrap();
assert_eq!(
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
new file mode 100644
index 00000000..631a5ce9
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -0,0 +1,615 @@
+// 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.
+
+#[tokio::test]
+async fn empty_unpartitioned_overwrite_clears_old_rows() {
+ let io = test_file_io();
+ let path = "memory:/audit-empty-overwrite";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit(vec![append_message("old.parquet")])
+ .await
+ .unwrap();
+ commit.overwrite(Vec::new(), None).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(
+ active_entries(&io, path, &snapshot).await.len(),
+ 0,
+ "Empty unpartitioned overwrite must remove old data; snapshot={}
count={:?}",
+ snapshot.id(),
+ snapshot.total_record_count()
+ );
+}
+
+#[tokio::test]
+async fn recovery_rejects_duplicate_file_after_identity_expiry() {
+ let io = test_file_io();
+ let path = "memory:/audit-expired-identity";
+ setup_dirs(&io, path).await;
+ let a = TableCommit::new(test_table(&io, path), "writer-a".into());
+ let b = TableCommit::new(test_table(&io, path), "writer-b".into());
+ let first = append_message("a.parquet");
+ for name in ["a.parquet", "b.parquet"] {
+ io.new_output(&format!("{path}/bucket-0/{name}"))
+ .unwrap()
+ .write(bytes::Bytes::from_static(
+ b"physical file retained by latest snapshot",
+ ))
+ .await
+ .unwrap();
+ }
+ a.commit_with_identifier(vec![first.clone()], 7)
+ .await
+ .unwrap();
+ b.commit_with_identifier(vec![append_message("b.parquet")], 1)
+ .await
+ .unwrap();
+ // Expiration removes the old snapshot, but both data files remain
referenced by snapshot 2.
+ let manager = a.table.snapshot_manager();
+ manager.delete_snapshot(1).await.unwrap();
+ manager.write_earliest_hint(2).await.unwrap();
+ assert!(io
+ .exists(&format!("{path}/bucket-0/a.parquet"))
+ .await
+ .unwrap());
+ let result = a
+ .filter_and_commit_with_identifier(vec![first.clone()], 7)
+ .await;
+ let snapshot = manager.get_latest_snapshot().await.unwrap().unwrap();
+ let active_rows: i64 = active_entries(&io, path, &snapshot)
+ .await
+ .iter()
+ .map(|e| e.file().row_count)
+ .sum();
+ assert!(result.is_err(), "Duplicate replay must be rejected;
result={result:?}, latest={}, snapshot count={:?}, actual active
rows={active_rows}", snapshot.id(), snapshot.total_record_count());
+ assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.total_record_count(), Some(active_rows));
+ let result = a.filter_and_commit(vec![(7, vec![first])]).await;
+ assert!(
+ result.is_err(),
+ "Batch recovery must reject duplicate replay: {result:?}"
+ );
+ assert_eq!(
+ manager.get_latest_snapshot().await.unwrap().unwrap().id(),
+ 2
+ );
+}
+
+#[tokio::test]
+async fn recovery_rejects_missing_file_after_identity_expiry() {
+ let io = test_file_io();
+ let path = "memory:/audit-expired-missing-file";
+ setup_dirs(&io, path).await;
+ let a = TableCommit::new(test_table(&io, path), "writer-a".into());
+ let b = TableCommit::new(test_table(&io, path), "writer-b".into());
+ let first = append_message("a.parquet");
+ let data_path = format!("{path}/bucket-0/a.parquet");
+ io.new_output(&data_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"data"))
+ .await
+ .unwrap();
+ a.commit_with_identifier(vec![first.clone()], 7)
+ .await
+ .unwrap();
+ b.truncate_table().await.unwrap();
+ let manager = a.table.snapshot_manager();
+ manager.delete_snapshot(1).await.unwrap();
+ manager.write_earliest_hint(2).await.unwrap();
+ io.delete_file(&data_path).await.unwrap();
+ let result = a
+ .filter_and_commit_with_identifier(vec![first.clone()], 7)
+ .await;
+ let snapshot = manager.get_latest_snapshot().await.unwrap().unwrap();
+ assert!(result.is_err(), "Replay referencing cleaned file must fail;
result={result:?}, latest={}, count={:?}, physical_exists={}", snapshot.id(),
snapshot.total_record_count(), io.exists(&data_path).await.unwrap());
+ let error = a
+ .filter_and_commit(vec![(7, vec![first])])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains(&data_path), "{error}");
+ assert_eq!(
+ manager.get_latest_snapshot().await.unwrap().unwrap().id(),
+ 2
+ );
+}
+
+#[tokio::test]
+async fn recovery_checks_data_changelog_extra_and_index_paths() {
+ let io = test_file_io();
+ let path = "memory:/recovery-paths";
+ setup_dirs(&io, path).await;
+ let table = test_table_with_options(
+ &io,
+ path,
+ HashMap::from([("index-file-in-data-file-dir".into(), "true".into())]),
+ );
+ let commit = TableCommit::new(table, "recover".into());
+ let mut message = append_message("data");
+ message.new_files[0].external_path = Some(format!("{path}/external/data"));
+ message.new_files[0].extra_files = vec!["data.index".into()];
+ let mut changelog = test_data_file("changelog", 10);
+ changelog.extra_files = vec!["changelog.index".into()];
+ message.new_changelog_files.push(changelog);
+ let mut bucket_index = test_global_index_file("hash", 0, 0, 9);
+ bucket_index.index_type = "HASH".into();
+ bucket_index.global_index_meta = None;
+ let mut external_index = test_global_index_file("external-index", 1, 0, 9);
+ external_index.external_path = Some(format!("{path}/external/index"));
+ message.new_index_files = vec![
+ bucket_index,
+ test_global_index_file("global", 0, 0, 9),
+ external_index,
+ ];
+ let paths = [
+ "external/data",
+ "external/data.index",
+ "bucket-0/changelog",
+ "bucket-0/changelog.index",
+ "bucket-0/hash",
+ "index/global",
+ "external/index",
+ ]
+ .map(|name| format!("{path}/{name}"));
+ for file in &paths {
+ io.new_output(file)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ }
+ for missing in &paths {
+ io.delete_file(missing).await.unwrap();
+ let error = commit
+ .filter_and_commit_with_identifier(vec![message.clone()], 1)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains(missing), "{error}");
+ let error = commit
+ .filter_and_commit(vec![(1, vec![message.clone()])])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains(missing), "{error}");
+ assert!(latest_snapshot(&io, path).await.is_none());
+ io.new_output(missing)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ }
+ commit
+ .filter_and_commit_with_identifier(vec![message.clone()], 1)
+ .await
+ .unwrap();
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 1);
+ // A known committed identity is filtered before files are checked.
+ io.delete_file(&paths[0]).await.unwrap();
+ commit
+ .filter_and_commit_with_identifier(vec![message], 1)
+ .await
+ .unwrap();
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 1);
+}
+
+#[tokio::test]
+async fn batch_recovery_validates_all_groups_before_publication() {
+ let io = test_file_io();
+ let path = "memory:/batch-recovery";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ let first = append_message("first.parquet");
+ let second = append_message("second.parquet");
+ let first_path = format!("{path}/bucket-0/first.parquet");
+ let second_path = format!("{path}/bucket-0/second.parquet");
+ io.new_output(&first_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ let error = commit
+ .filter_and_commit(vec![(7, vec![first.clone()]), (7,
vec![second.clone()])])
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("Duplicate commit identifier 7"),
+ "{error}"
+ );
+ assert!(latest_snapshot(&io, path).await.is_none());
+
+ let pending = vec![(8, vec![second]), (7, vec![first])];
+ let error = commit.filter_and_commit(pending.clone()).await.unwrap_err();
+ assert!(error.to_string().contains(&second_path), "{error}");
+ assert!(latest_snapshot(&io, path).await.is_none());
+ io.new_output(&second_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ assert_eq!(commit.filter_and_commit(pending.clone()).await.unwrap(), 2);
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.commit_identifier(), 8);
+ assert_eq!(snapshot.total_record_count(), Some(20));
+ io.delete_file(&first_path).await.unwrap();
+ io.delete_file(&second_path).await.unwrap();
+ assert_eq!(commit.filter_and_commit(pending).await.unwrap(), 0);
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 2);
+}
+
+#[tokio::test]
+async fn dv_commits_allow_unrelated_writes_but_reject_replaced_vectors() {
+ let io = test_file_io();
+ let path = "memory:/dv-concurrency";
+ setup_dirs(&io, path).await;
+ let table = Table::new(
+ io.clone(),
+ Identifier::new("default", "dv"),
+ path.into(),
+ test_partitioned_schema().copy_with_options(HashMap::from([
+ ("data-evolution.enabled".into(), "true".into()),
+ ("row-tracking.enabled".into(), "true".into()),
+ ])),
+ None,
+ );
+ let commit = TableCommit::new(table, "dv".into());
+ let part_a = partition_bytes("a");
+ let part_b = partition_bytes("b");
+ let append = |partition, name| {
+ let mut file = test_data_file(name, 10);
+ file.file_source = Some(0);
+ CommitMessage::new(partition, 0, vec![file])
+ };
+ commit
+ .commit(vec![
+ append(part_a.clone(), "data-a"),
+ append(part_b.clone(), "data-b"),
+ ])
+ .await
+ .unwrap();
+ let mut delete_a = dv_message("dv-a", "data-a");
+ delete_a.partition = part_a.clone();
+ delete_a.check_from_snapshot = Some(1);
+ let mut delete_b = dv_message("dv-b", "data-b");
+ delete_b.partition = part_b.clone();
+ delete_b.check_from_snapshot = Some(1);
+ commit.commit(vec![delete_a]).await.unwrap();
+ // Mixed APPEND in A and DELETE in B must scan both relevant partitions.
+ commit
+ .commit(vec![append(part_a.clone(), "append-a"), delete_b])
+ .await
+ .unwrap();
+ let mut replace = dv_message("replacement", "data-a");
+ replace.partition = part_a.clone();
+ replace.deleted_index_files = vec![test_deletion_vector_index_file("dv-a",
"data-a")];
+ replace.check_from_snapshot = Some(2);
+ commit.commit(vec![replace.clone()]).await.unwrap();
+ replace.new_index_files[0].file_name = "stale-replacement".into();
+ let error = commit.commit(vec![replace]).await.unwrap_err();
+ assert!(
+ error.to_string().contains("missing deletion vector"),
+ "{error}"
+ );
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 4);
+ let snapshot = latest_snapshot(&io, path).await;
+ let indexes = TableCommit::read_prev_index_entries(&io,
&format!("{path}/manifest"), &snapshot)
+ .await
+ .unwrap();
+ assert_eq!(
+ indexes
+ .iter()
+ .map(|entry| entry.index_file.file_name.as_str())
+ .collect::<HashSet<_>>(),
+ HashSet::from(["replacement", "dv-b"])
+ );
+}
+
+#[tokio::test]
+async fn dv_references_must_survive_concurrent_and_same_commit_data_deletion()
{
+ for concurrent in [true, false] {
+ let io = test_file_io();
+ let path = format!("memory:/dv-removed-data-{concurrent}");
+ setup_dirs(&io, &path).await;
+ let commit = setup_commit(&io, &path);
+ let data = test_data_file("data", 10);
+ commit
+ .commit(vec![CommitMessage::new(vec![], 0, vec![data.clone()])])
+ .await
+ .unwrap();
+ let mut dv = dv_message("dv", "data");
+ dv.check_from_snapshot = Some(1);
+ if concurrent {
+ commit.truncate_table().await.unwrap();
+ } else {
+ dv.deleted_files = vec![data];
+ }
+ let error = commit.commit(vec![dv]).await.unwrap_err();
+ assert!(
+ error.to_string().contains("references missing data file"),
+ "{error}"
+ );
+ assert_eq!(
+ latest_snapshot(&io, &path).await.unwrap().id(),
+ if concurrent { 2 } else { 1 }
+ );
+ }
+ let io = test_file_io();
+ let path = "memory:/dv-retained-on-deleted-file";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ let data = test_data_file("data", 10);
+ commit
+ .commit(vec![
+ CommitMessage::new(vec![], 0, vec![data.clone()]),
+ dv_message("dv", "data"),
+ ])
+ .await
+ .unwrap();
+ let mut deletion = CommitMessage::new(vec![], 0, vec![]);
+ deletion.deleted_files = vec![data];
+ assert!(commit
+ .commit(vec![deletion.clone()])
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("references missing data file"));
+ deletion.deleted_index_files = vec![test_deletion_vector_index_file("dv",
"data")];
+ commit.commit(vec![deletion]).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert!(snapshot.index_manifest().is_none());
+ assert!(active_entries(&io, path, &snapshot).await.is_empty());
+}
+
+#[tokio::test]
+async fn explicit_row_id_update_still_checks_indexed_columns() {
+ let io = test_file_io();
+ let path = "memory:/explicit-row-id-update";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ let mut index = CommitMessage::new(vec![], 0, vec![]);
+ index.new_index_files = vec![test_global_index_file("global", 0, 0, 9)];
+ commit.commit(vec![index]).await.unwrap();
+ let mut update = append_message("update");
+ update.new_files[0].write_cols = Some(vec!["id".into(),
crate::spec::ROW_ID_FIELD_NAME.into()]);
+ let error = commit.commit(vec![update]).await.unwrap_err();
+ assert!(
+ error.to_string().contains("globally indexed columns"),
+ "{error}"
+ );
+}
+
+struct ConcurrentDvCommit {
+ table: Table,
+ change: &'static str,
+ calls: std::sync::atomic::AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for ConcurrentDvCommit {
+ async fn commit(&self, snapshot: &Snapshot, _: &[PartitionStatistics]) ->
Result<bool> {
+ if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
+ let other = TableCommit::new(self.table.clone(),
"concurrent".into());
+ match self.change {
+ "append" =>
other.commit(vec![append_message("unrelated")]).await?,
+ "delete" => other.truncate_table().await?,
+ "dv" => {
+ other
+ .commit(vec![dv_message("competing-dv", "data")])
+ .await?
+ }
+ _ => unreachable!(),
+ }
+ return Ok(false);
+ }
+ self.table
+ .snapshot_manager()
+ .commit_snapshot(snapshot)
+ .await
+ }
+}
+
+#[tokio::test]
+async fn dv_retry_revalidates_data_files_and_vectors() {
+ for change in ["append", "delete", "dv"] {
+ let io = test_file_io();
+ let path = format!("memory:/dv-retry-{change}");
+ setup_dirs(&io, &path).await;
+ let mut commit = setup_commit(&io, &path);
+ commit.commit(vec![append_message("data")]).await.unwrap();
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(ConcurrentDvCommit {
+ table: commit.table.clone(),
+ change,
+ calls: std::sync::atomic::AtomicUsize::new(0),
+ });
+ commit.snapshot_commit = publisher.clone();
+ let mut deletion = dv_message("dv", "data");
+ deletion.check_from_snapshot = Some(1);
+ let result = commit.commit(vec![deletion]).await;
+ let snapshot = latest_snapshot(&io, &path).await.unwrap();
+ if change == "append" {
+ result.unwrap();
+ assert_eq!(snapshot.id(), 3);
+ assert_eq!(snapshot.total_record_count(), Some(20));
+
assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), 2);
+ } else {
+ let error = result.unwrap_err().to_string();
+ assert!(
+ error.contains(if change == "delete" {
+ "references missing data file"
+ } else {
+ "Conflicting deletion vectors"
+ }),
+ "{error}"
+ );
+ assert_eq!(snapshot.id(), 2);
+
assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), 1);
+ }
+ }
+}
+
+#[tokio::test]
+async fn rest_delete_writer_pins_catalog_snapshot_and_preserves_vectors() {
+ use crate::api::rest_api::RESTApi;
+ use crate::common::Options;
+ use arrow_array::{Array, Int32Array, RecordBatch, StringArray};
+ use axum::{
+ body::Bytes,
+ http::{Method, Uri},
+ Json, Router,
+ };
+ use futures::TryStreamExt;
+ use std::sync::Mutex;
+ let io = test_file_io();
+ let path = "memory:/audit-rest-dv";
+ setup_dirs(&io, path).await;
+ let schema =
test_data_evolution_schema().copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".into(),
+ "true".into(),
+ )]));
+ let seed_table = Table::new(
+ io.clone(),
+ Identifier::new("database", "table"),
+ path.into(),
+ schema.clone(),
+ None,
+ );
+ let mut writer = crate::table::TableWrite::new(&seed_table,
"seed".into()).unwrap();
+ let batch = RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2])) as arrow_array::ArrayRef,
+ ),
+ (
+ "name",
+ Arc::new(StringArray::from(vec!["zero", "one", "two"])) as
arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(seed_table, "seed".into())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ let mut value = serde_json::to_value(latest_snapshot(&io,
path).await.unwrap()).unwrap();
+ value["id"] = 7.into();
+ let snapshot = Arc::new(Mutex::new(
+ serde_json::from_value::<Snapshot>(value).unwrap(),
+ ));
+ let posts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let loads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let handler_loads = loads.clone();
+ let handler_snapshot = snapshot.clone();
+ let handler_posts = posts.clone();
+ let app = Router::new().fallback(move |method: Method, uri: Uri, body:
Bytes| {
+ let loads = handler_loads.clone();
+ let snapshot = handler_snapshot.clone();
+ let posts = handler_posts.clone();
+ async move {
+ let response = if method == Method::POST &&
uri.path().ends_with("/commit") {
+ posts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ let request: serde_json::Value =
serde_json::from_slice(&body).unwrap();
+ let next: Snapshot =
serde_json::from_value(request["snapshot"].clone()).unwrap();
+ assert_eq!(next.id(), snapshot.lock().unwrap().id() + 1);
+ *snapshot.lock().unwrap() = next;
+ // Force the retry/deduplication path after catalog
publication.
+ serde_json::json!({"success": false})
+ } else if uri.path().ends_with("/snapshot") {
+ loads.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ serde_json::json!({"snapshot": {"snapshot":
*snapshot.lock().unwrap(), "recordCount": 10}})
+ } else {
+ serde_json::json!({"schemaId": 0})
+ };
+ Json(response)
+ }
+ });
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let mut options = Options::new();
+ options.set("uri", format!("http://{}";, listener.local_addr().unwrap()));
+ options.set("prefix", "test");
+ options.set("token.provider", "bear");
+ options.set("token", "test-token");
+ let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
+ let api = Arc::new(RESTApi::new(options.clone(), false).await.unwrap());
+ let identifier = Identifier::new("database", "table");
+ let env =
+ crate::table::RESTEnv::new(identifier.clone(), "uuid".into(), api,
options, false, None);
+ let table = Table::new(io.clone(), identifier, path.into(), schema,
Some(env));
+ let mut commit = TableCommit::new(table.clone(), "rest-writer".into());
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ async fn remaining_ids(table: &Table) -> Vec<i32> {
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let read = builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut ids = Vec::new();
+ for batch in batches {
+ let column = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ ids.extend((0..column.len()).map(|i| column.value(i)));
+ }
+ ids.sort();
+ ids
+ }
+ for row_id in [0, 1] {
+ let mut deletion = table.new_write_builder().new_delete().unwrap();
+ deletion.add_row_ids([row_id]).unwrap();
+ let before = loads.load(std::sync::atomic::Ordering::SeqCst);
+ let messages = deletion.prepare_commit().await.unwrap();
+ assert_eq!(loads.load(std::sync::atomic::Ordering::SeqCst) - before,
1);
+ assert_eq!(messages[0].check_from_snapshot, Some(7 + row_id));
+ commit
+ .commit_with_identifier(messages, 42 + row_id)
+ .await
+ .unwrap();
+ assert_eq!(
+ remaining_ids(&table).await,
+ if row_id == 0 { vec![1, 2] } else { vec![2] }
+ );
+ }
+ let mut left = table.new_write_builder().new_delete().unwrap();
+ left.add_row_ids([2]).unwrap();
+ let mut right = table.new_write_builder().new_delete().unwrap();
+ right.add_row_ids([2]).unwrap();
+ let left = left.prepare_commit().await.unwrap();
+ let right = right.prepare_commit().await.unwrap();
+ commit.commit_with_identifier(left, 44).await.unwrap();
+ let error = commit.commit_with_identifier(right, 45).await.unwrap_err();
+ assert!(
+ error.to_string().contains("missing deletion vector"),
+ "{error}"
+ );
+ assert!(remaining_ids(&table).await.is_empty());
+ assert_eq!(snapshot.lock().unwrap().id(), 10);
+ assert_eq!(posts.load(std::sync::atomic::Ordering::SeqCst), 3);
+ for id in 7..=10 {
+ assert!(!io
+ .exists(&format!("{path}/snapshot/snapshot-{id}"))
+ .await
+ .unwrap());
+ }
+ server.abort();
+}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index f23c83f3..fa819f13 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -4675,4 +4675,76 @@ pub(in crate::table) mod tests {
"append tables keep file-level bin pack"
);
}
+ #[tokio::test]
+ async fn dedicated_insert_preserves_existing_global_index() {
+ for action in ["THROW_ERROR", "DROP_PARTITION_INDEX"] {
+ let file_io = test_file_io();
+ let path = "memory:/audit-dedicated-index-append";
+ setup_dirs(&file_io, path).await;
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "audit"),
+ path.into(),
+ test_vector_table_schema("parquet").copy_with_options(
+ std::collections::HashMap::from([(
+ "global-index.column-update-action".into(),
+ action.into(),
+ )]),
+ ),
+ None,
+ );
+ async fn append(table: &Table, id: i32) -> crate::Result<()> {
+ let mut writer = TableWrite::new(table, "audit".into())?;
+ writer
+ .write_arrow_batch(&make_vector_batch(vec![id],
vec![vec![1.0, 0.0]]))
+ .await?;
+ let messages = writer.prepare_commit().await?;
+ assert!(messages[0]
+ .new_files
+ .iter()
+ .all(|f| f.first_row_id.is_none() &&
f.write_cols.is_some()));
+ TableCommit::new(table.clone(), "audit".into())
+ .commit(messages)
+ .await
+ }
+ append(&table, 1).await.unwrap();
+ let mut message = crate::table::CommitMessage::new(vec![], 0,
vec![]);
+ message.new_index_files.push(crate::spec::IndexFileMeta {
+ index_type: "lumina".into(),
+ file_name: "audit-index".into(),
+ file_size: 1,
+ row_count: 1,
+ deletion_vectors_ranges: None,
+ external_path: None,
+ global_index_meta: Some(crate::spec::GlobalIndexMeta {
+ index_field_id: 1,
+ row_range_start: 0,
+ row_range_end: 0,
+ extra_field_ids: None,
+ index_meta: None,
+ source_meta: None,
+ }),
+ });
+ TableCommit::new(table.clone(), "audit-index".into())
+ .commit(vec![message])
+ .await
+ .unwrap();
+ append(&table, 2).await.unwrap();
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.next_row_id(), Some(2));
+ let entries = crate::spec::IndexManifest::read(
+ &file_io,
+ &format!("{path}/manifest/{}",
snapshot.index_manifest().unwrap()),
+ )
+ .await
+ .unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].index_file.file_name, "audit-index");
+ }
+ }
}
