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 ef540b1e fix(commit): align TableCommit validation and retries with
Java (#913)
ef540b1e is described below
commit ef540b1e94d439551a3e1f27d46483743f1a7ce9
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 15:59:04 2026 +0800
fix(commit): align TableCommit validation and retries with Java (#913)
---
crates/paimon/src/spec/core_options.rs | 14 -
crates/paimon/src/table/table_commit.rs | 418 ++++++++++----
.../paimon/src/table/table_commit/parity_tests.rs | 642 +++++++++++++++++++++
3 files changed, 933 insertions(+), 141 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index bd20c09b..41a9664c 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -87,7 +87,6 @@ pub(crate) const PATH_OPTION: &str = "path";
const MANIFEST_COMPRESSION_OPTION: &str = "manifest.compression";
const MANIFEST_TARGET_FILE_SIZE_OPTION: &str = "manifest.target-file-size";
const MANIFEST_TARGET_SIZE_OPTION: &str = "manifest.target-size";
-const MANIFEST_MERGE_MIN_COUNT_OPTION: &str = "manifest.merge-min-count";
const MANIFEST_SIDECAR_ENABLED_OPTION: &str = "manifest.sidecar.enabled";
const MANIFEST_SORT_ENABLED_OPTION: &str = "manifest-sort.enabled";
const WRITE_PARQUET_BUFFER_SIZE_OPTION: &str = "write.parquet-buffer-size";
@@ -138,7 +137,6 @@ const DEFAULT_SOURCE_SPLIT_TARGET_SIZE: i64 = 128 * 1024 *
1024;
const DEFAULT_SOURCE_SPLIT_OPEN_FILE_COST: i64 = 4 * 1024 * 1024;
const DEFAULT_MANIFEST_COMPRESSION: &str = "zstd";
const DEFAULT_MANIFEST_TARGET_FILE_SIZE: i64 = 8 * 1024 * 1024;
-const DEFAULT_MANIFEST_MERGE_MIN_COUNT: usize = 30;
const DEFAULT_PARTITION_DEFAULT_NAME: &str = "__DEFAULT_PARTITION__";
const DEFAULT_CHANGELOG_FILE_PREFIX: &str = "changelog-";
const DEFAULT_TARGET_FILE_SIZE: i64 = 256 * 1024 * 1024;
@@ -1233,15 +1231,6 @@ impl<'a> CoreOptions<'a> {
.unwrap_or(DEFAULT_MANIFEST_TARGET_FILE_SIZE)
}
- /// Compatibility option; Rust commits do not currently compact manifests.
- pub fn manifest_merge_min_count(&self) -> usize {
- self.options
- .get(MANIFEST_MERGE_MIN_COUNT_OPTION)
- .and_then(|v| v.parse().ok())
- .filter(|v| *v > 0)
- .unwrap_or(DEFAULT_MANIFEST_MERGE_MIN_COUNT)
- }
-
/// Whether manifest block sidecars are read and written.
///
/// An explicit `manifest.sidecar.enabled` value wins. When it is absent,
@@ -2729,7 +2718,6 @@ mod tests {
assert!(!core.row_tracking_enabled());
assert_eq!(core.manifest_compression(), "zstd");
assert_eq!(core.manifest_target_size(), 8 * 1024 * 1024);
- assert_eq!(core.manifest_merge_min_count(), 30);
assert!(!core.manifest_sidecar_enabled());
}
@@ -2747,7 +2735,6 @@ mod tests {
"1kb".to_string(),
),
(MANIFEST_COMPRESSION_OPTION.to_string(), "null".to_string()),
- (MANIFEST_MERGE_MIN_COUNT_OPTION.to_string(), "3".to_string()),
]);
let core = CoreOptions::new(&options);
assert_eq!(core.bucket(), 4);
@@ -2758,7 +2745,6 @@ mod tests {
assert!(core.row_tracking_enabled());
assert_eq!(core.manifest_compression(), "null");
assert_eq!(core.manifest_target_size(), 1024);
- assert_eq!(core.manifest_merge_min_count(), 3);
}
#[test]
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index d25a6547..8ce382f0 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -180,7 +180,7 @@ pub struct TableCommit {
impl TableCommit {
pub fn new(table: Table, commit_user: String) -> Self {
- let snapshot_manager = SnapshotManager::new(table.file_io.clone(),
table.location.clone());
+ let snapshot_manager = table.snapshot_manager();
let snapshot_commit = if let Some(env) = &table.rest_env {
env.snapshot_commit()
} else {
@@ -860,75 +860,91 @@ impl TableCommit {
filter_committed && plan.commit_kind_hint() ==
CommitKind::OVERWRITE;
let mut filter_committed = filter_committed;
- loop {
- let latest_snapshot =
self.snapshot_manager.get_latest_snapshot().await?;
- if filter_committed {
- if self
- .is_committed_identifier(&latest_snapshot,
commit_identifier)
- .await?
- {
- break;
+ let mut publication_uncertain = false;
+ let mut last_publication_error = None;
+ let result = async {
+ loop {
+ let latest_snapshot =
self.snapshot_manager.get_latest_snapshot().await?;
+ if filter_committed {
+ if self
+ .is_committed_identifier(&latest_snapshot,
commit_identifier)
+ .await?
+ {
+ break;
+ }
+ filter_committed = false;
}
- filter_committed = false;
- }
- if let Some(start_snapshot_id) = duplicate_check_start_snapshot_id
{
- if self
- .is_duplicate_commit(
- start_snapshot_id,
- &latest_snapshot,
- commit_identifier,
- &plan.commit_kind_hint(),
- )
- .await?
+ if let Some(start_snapshot_id) =
duplicate_check_start_snapshot_id {
+ if self
+ .is_duplicate_commit(
+ start_snapshot_id,
+ &latest_snapshot,
+ commit_identifier,
+ &plan.commit_kind_hint(),
+ )
+ .await?
+ {
+ break;
+ }
+ }
+ validate_expected_latest_snapshot(expected_snapshot_id,
&latest_snapshot)?;
+ let resolved = self
+ .resolve_commit(&mut plan, &latest_snapshot,
retry_state.as_deref())
+ .await?;
+
+ if resolved.entries.is_empty()
+ && resolved.changelog_entries.is_empty()
+ && !resolved.index_manifest_changed
+ && !commit_empty_overwrite
{
break;
}
- }
- validate_expected_latest_snapshot(expected_snapshot_id,
&latest_snapshot)?;
- let resolved = self
- .resolve_commit(&mut plan, &latest_snapshot,
retry_state.as_deref())
- .await?;
- if resolved.entries.is_empty()
- && resolved.changelog_entries.is_empty()
- && !resolved.index_manifest_changed
- && !commit_empty_overwrite
- {
- break;
- }
+ let result = self
+ .try_commit_once(resolved, &latest_snapshot,
commit_identifier)
+ .await?;
- let result = self
- .try_commit_once(resolved, &latest_snapshot, commit_identifier)
- .await?;
+ match result {
+ CommitAttemptResult::Success => break,
+ CommitAttemptResult::Retry(mut state) => {
+ if let Some(error) = state.publication_error.take() {
+ publication_uncertain = true;
+ last_publication_error = Some(error);
+ }
+
duplicate_check_start_snapshot_id.get_or_insert_with(|| {
+ latest_snapshot.as_ref().map(|s| s.id() +
1).unwrap_or(1)
+ });
+ retry_state = Some(state);
+ }
+ }
- match result {
- CommitAttemptResult::Success => break,
- CommitAttemptResult::Retry(state) => {
- duplicate_check_start_snapshot_id.get_or_insert_with(|| {
- latest_snapshot.as_ref().map(|s| s.id() +
1).unwrap_or(1)
+ let elapsed_ms = current_time_millis() - start_time_ms;
+ if elapsed_ms > self.commit_timeout_ms || retry_count >=
self.commit_max_retries {
+ let snap_id =
duplicate_check_start_snapshot_id.unwrap_or(1);
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Commit failed for snapshot {} after {} millis
with {} retries, \
+ there may exist commit conflicts between multiple
jobs.",
+ snap_id, elapsed_ms, retry_count
+ ),
+ source: last_publication_error.map(|error|
Box::new(error) as _),
});
- retry_state = Some(state);
}
- }
- let elapsed_ms = current_time_millis() - start_time_ms;
- if elapsed_ms > self.commit_timeout_ms || retry_count >=
self.commit_max_retries {
- let snap_id = duplicate_check_start_snapshot_id.unwrap_or(1);
- return Err(crate::Error::DataInvalid {
- message: format!(
- "Commit failed for snapshot {} after {} millis with {}
retries, \
- there may exist commit conflicts between multiple
jobs.",
- snap_id, elapsed_ms, retry_count
- ),
- source: None,
- });
+ self.commit_retry_wait(retry_count).await;
+ retry_count += 1;
}
- self.commit_retry_wait(retry_count).await;
- retry_count += 1;
+ Ok(())
+ }
+ .await;
+ match result {
+ Err(error) if publication_uncertain =>
Err(crate::Error::UnexpectedError {
+ message: "Commit outcome may be unknown; retain prepared files
and retry with the same commit identifier".into(),
+ source: Some(Box::new(error)),
+ }),
+ result => result,
}
-
- Ok(())
}
/// Single commit attempt.
@@ -938,6 +954,45 @@ impl TableCommit {
latest_snapshot: &Option<Snapshot>,
commit_identifier: i64,
) -> Result<CommitAttemptResult> {
+ let mut created_files = Vec::new();
+ let prepared = self
+ .prepare_snapshot(
+ &mut resolved,
+ latest_snapshot,
+ commit_identifier,
+ &mut created_files,
+ )
+ .await;
+ let (snapshot, statistics) = match prepared {
+ Ok(prepared) => prepared,
+ Err(error) => {
+ for path in created_files {
+ let _ =
self.snapshot_manager.file_io().delete_file(&path).await;
+ }
+ return Err(error);
+ }
+ };
+ // Once publication starts its outcome may be unknown. These files must
+ // remain available even if the response is lost or a later retry
fails.
+ let publication_error = match self.snapshot_commit.commit(&snapshot,
&statistics).await {
+ Ok(true) => return Ok(CommitAttemptResult::Success),
+ Ok(false) => None,
+ Err(error) => Some(error),
+ };
+ Ok(CommitAttemptResult::Retry(Box::new(RetryState {
+ latest_snapshot: latest_snapshot.clone(),
+ base_data_files: resolved.base_data_files.take(),
+ publication_error,
+ })))
+ }
+
+ async fn prepare_snapshot(
+ &self,
+ resolved: &mut ResolvedCommit,
+ latest_snapshot: &Option<Snapshot>,
+ commit_identifier: i64,
+ created_files: &mut Vec<String>,
+ ) -> Result<(Snapshot, Vec<PartitionStatistics>)> {
let new_snapshot_id = latest_snapshot.as_ref().map(|s| s.id() +
1).unwrap_or(1);
// Row tracking
@@ -956,7 +1011,7 @@ impl TableCommit {
let (assigned, nrid) = self.assign_row_tracking_meta(
new_snapshot_id,
first_row_id_start,
- resolved.entries,
+ std::mem::take(&mut resolved.entries),
)?;
resolved.entries = assigned;
next_row_id = Some(nrid);
@@ -984,10 +1039,12 @@ impl TableCommit {
&manifest_dir,
&new_manifest_prefix,
&resolved.entries,
+ created_files,
)
.await?;
// Write delta manifest list
+ created_files.push(delta_manifest_list_path.clone());
ManifestList::write_with_compression(
file_io,
&delta_manifest_list_path,
@@ -1006,8 +1063,10 @@ impl TableCommit {
&manifest_dir,
&changelog_manifest_prefix,
&resolved.changelog_entries,
+ created_files,
)
.await?;
+ created_files.push(changelog_manifest_list_path.clone());
ManifestList::write_with_compression(
file_io,
&changelog_manifest_list_path,
@@ -1045,6 +1104,7 @@ impl TableCommit {
vec![]
};
+ created_files.push(base_manifest_list_path.clone());
ManifestList::write_with_compression(
file_io,
&base_manifest_list_path,
@@ -1063,35 +1123,89 @@ impl TableCommit {
}
total_record_count += delta_record_count;
+ if let Some(entries) = &resolved.new_index_manifest_entries {
+ resolved.index_manifest_name = self
+ .write_index_manifest(file_io, &manifest_dir, entries,
created_files)
+ .await?;
+ }
+ let schema_id = self.latest_schema_id().await?;
+ let statistics_file = self
+ .inherited_statistics(latest_snapshot.as_ref(), schema_id)
+ .await?;
let snapshot = Snapshot::builder()
.version(3)
.id(new_snapshot_id)
- .schema_id(self.table.schema().id())
+ .schema_id(schema_id)
.base_manifest_list(base_manifest_list_name)
.delta_manifest_list(delta_manifest_list_name)
.commit_user(self.commit_user.clone())
.commit_identifier(commit_identifier)
- .commit_kind(resolved.kind)
+ .commit_kind(resolved.kind.clone())
.time_millis(current_time_millis())
.total_record_count(Some(total_record_count))
.delta_record_count(Some(delta_record_count))
.changelog_manifest_list(changelog_record_count.map(|_|
changelog_manifest_list_name))
.changelog_manifest_list_size(changelog_manifest_list_size)
.changelog_record_count(changelog_record_count)
+ .watermark(latest_snapshot.as_ref().and_then(Snapshot::watermark))
+ .statistics(statistics_file)
.next_row_id(next_row_id)
- .index_manifest(resolved.index_manifest_name)
+ .index_manifest(resolved.index_manifest_name.clone())
.build();
let statistics =
self.generate_partition_statistics(&resolved.entries)?;
- if self.snapshot_commit.commit(&snapshot, &statistics).await? {
- Ok(CommitAttemptResult::Success)
- } else {
- Ok(CommitAttemptResult::Retry(Box::new(RetryState {
- latest_snapshot: latest_snapshot.clone(),
- base_data_files: resolved.base_data_files.take(),
- })))
+ Ok((snapshot, statistics))
+ }
+
+ async fn latest_schema_id(&self) -> Result<i64> {
+ if let Some(env) = &self.table.rest_env {
+ return env
+ .api()
+ .get_table(env.identifier())
+ .await?
+ .schema_id
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "REST table response is missing schemaId".into(),
+ source: None,
+ });
}
+ // Tables constructed directly by callers need not have schema files.
+ Ok(self
+ .table
+ .schema_manager()
+ .latest()
+ .await?
+ .map(|schema| schema.id())
+ .unwrap_or(self.table.schema().id()))
+ }
+
+ async fn inherited_statistics(
+ &self,
+ snapshot: Option<&Snapshot>,
+ schema_id: i64,
+ ) -> Result<Option<String>> {
+ let Some(name) = snapshot.and_then(Snapshot::statistics) else {
+ return Ok(None);
+ };
+ // Read the statistics' own schema ID: the snapshot's schema is not a
+ // substitute when older writers have carried incompatible statistics.
+ #[derive(serde::Deserialize)]
+ struct StatisticsSchema {
+ #[serde(rename = "schemaId")]
+ schema_id: i64,
+ }
+ let path = format!(
+ "{}/statistics/{name}",
+ self.table.location().trim_end_matches('/')
+ );
+ let bytes = self.table.file_io().new_input(&path)?.read().await?;
+ let statistics: StatisticsSchema =
+ serde_json::from_slice(&bytes).map_err(|error|
crate::Error::DataInvalid {
+ message: format!("Invalid statistics metadata: {path}"),
+ source: Some(Box::new(error)),
+ })?;
+ Ok((statistics.schema_id == schema_id).then(|| name.to_string()))
}
/// Write an index manifest file from already-merged entries.
@@ -1102,12 +1216,14 @@ impl TableCommit {
file_io: &FileIO,
manifest_dir: &str,
merged_index_entries: &[IndexManifestEntry],
+ created_files: &mut Vec<String>,
) -> Result<Option<String>> {
if merged_index_entries.is_empty() {
return Ok(None);
}
let name = format!("index-manifest-{}-0", uuid::Uuid::new_v4());
let path = format!("{manifest_dir}/{name}");
+ created_files.push(path.clone());
IndexManifest::write_with_compression(
file_io,
&path,
@@ -1125,6 +1241,7 @@ impl TableCommit {
manifest_dir: &str,
name_prefix: &str,
entries: &[ManifestEntry],
+ created_files: &mut Vec<String>,
) -> Result<Vec<ManifestFileMeta>> {
if entries.is_empty() {
return Ok(vec![]);
@@ -1158,6 +1275,7 @@ impl TableCommit {
&file_name,
&entries[chunk_start..chunk_end],
bytes,
+ created_files,
)
.await?;
result.push(meta);
@@ -1181,6 +1299,7 @@ impl TableCommit {
&file_name,
&entries[chunk_start..],
bytes,
+ created_files,
)
.await?;
result.push(meta);
@@ -1197,6 +1316,7 @@ impl TableCommit {
file_name: &str,
entries: &[ManifestEntry],
bytes: Vec<u8>,
+ created_files: &mut Vec<String>,
) -> Result<ManifestFileMeta> {
let file_size = bytes.len() as i64;
let sidecar = if self.manifest_sidecar_enabled {
@@ -1209,6 +1329,7 @@ impl TableCommit {
} else {
None
};
+ created_files.push(path.to_string());
let output = file_io.new_output(path)?;
output.write(bytes::Bytes::from(bytes)).await?;
let sidecar_name = sidecar
@@ -1216,6 +1337,7 @@ impl TableCommit {
.map(|_| format!("{}{}", file_name,
crate::spec::MANIFEST_SIDECAR_SUFFIX));
if let (Some(sidecar), Some(_)) = (sidecar, sidecar_name.as_ref()) {
let sidecar_path = crate::spec::ManifestSidecar::path(path);
+ created_files.push(sidecar_path.clone());
let write_result = async {
let output = file_io.new_output(&sidecar_path)?;
output.write(bytes::Bytes::from(sidecar)).await
@@ -1289,6 +1411,9 @@ impl TableCommit {
let Some(latest) = latest_snapshot else {
return Ok(false);
};
+ if latest.commit_user() == self.commit_user {
+ return Ok(commit_identifier <= latest.commit_identifier());
+ }
let earliest_snapshot_id = self
.snapshot_manager
.earliest_snapshot_id()
@@ -1316,8 +1441,12 @@ impl TableCommit {
commit_kind: &CommitKind,
) -> Result<bool> {
if let Some(latest) = latest_snapshot {
- for snapshot_id in start_snapshot_id..=latest.id() {
- let snap =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
+ for snapshot_id in (start_snapshot_id..=latest.id()).rev() {
+ let snap = if snapshot_id == latest.id() {
+ latest.clone()
+ } else {
+ self.snapshot_manager.get_snapshot(snapshot_id).await?
+ };
if snap.commit_user() == self.commit_user
&& snap.commit_identifier() == commit_identifier
&& snap.commit_kind() == commit_kind
@@ -1348,14 +1477,8 @@ impl TableCommit {
} => {
validate_file_entries(entries.iter())?;
- // Auto-promote to OVERWRITE when CoW rewrites produce Delete
entries.
- // This ensures the snapshot correctly reflects file
replacements.
let has_delete = entries.iter().any(|e| *e.kind() ==
FileKind::Delete);
- let kind = if has_delete {
- CommitKind::OVERWRITE
- } else {
- CommitKind::APPEND
- };
+ let kind = direct_commit_kind(entries, new_index_entries);
let has_partition_bucket_counts = entries
.iter()
.any(|entry| entry.total_buckets() != self.total_buckets);
@@ -1368,9 +1491,8 @@ impl TableCommit {
|| has_partition_bucket_counts
|| has_postpone_entries;
let base_data_files = if detect_conflicts {
- self.check_deletion_vector_index_only_conflict(
+ self.check_deletion_vector_conflicts(
latest_snapshot.as_ref(),
- entries,
new_index_entries,
*check_from_snapshot,
)?;
@@ -1402,14 +1524,10 @@ impl TableCommit {
)?);
let all = Self::merge_index_entries(&previous, &index_entries,
false)?;
let index_manifest_changed = all != previous;
- let index_manifest_name = if index_manifest_changed {
- self.write_index_manifest(file_io, &manifest_dir, &all)
- .await?
- } else {
- latest_snapshot
- .as_ref()
- .and_then(|s| s.index_manifest().map(|s|
s.to_string()))
- };
+ let index_manifest_name = latest_snapshot
+ .as_ref()
+ .and_then(|s| s.index_manifest().map(str::to_string));
+ let new_index_manifest_entries =
index_manifest_changed.then_some(all);
Ok(ResolvedCommit {
entries: entries.clone(),
@@ -1417,6 +1535,7 @@ impl TableCommit {
kind,
index_manifest_name,
index_manifest_changed,
+ new_index_manifest_entries,
base_data_files,
})
}
@@ -1466,14 +1585,10 @@ impl TableCommit {
}
let all = Self::merge_index_entries(&all, &new_index_entries,
false)?;
let index_manifest_changed = all != previous;
- let index_manifest_name = if index_manifest_changed {
- self.write_index_manifest(file_io, &manifest_dir, &all)
- .await?
- } else {
- latest_snapshot
- .as_ref()
- .and_then(|s| s.index_manifest().map(|s|
s.to_string()))
- };
+ let index_manifest_name = latest_snapshot
+ .as_ref()
+ .and_then(|s| s.index_manifest().map(str::to_string));
+ let new_index_manifest_entries =
index_manifest_changed.then_some(all);
Ok(ResolvedCommit {
entries,
@@ -1481,6 +1596,7 @@ impl TableCommit {
kind: CommitKind::OVERWRITE,
index_manifest_name,
index_manifest_changed,
+ new_index_manifest_entries,
base_data_files,
})
}
@@ -1505,12 +1621,18 @@ impl TableCommit {
.iter()
.filter(|entry| entry.kind == FileKind::Delete)
.collect::<Vec<_>>();
- if !deletions.is_empty() {
- all.retain(|entry| {
- !deletions
- .iter()
- .any(|delete| same_index_file_entry(entry, delete))
- });
+ for deletion in deletions {
+ let position = all
+ .iter()
+ .position(|entry| same_index_file_entry(entry, deletion));
+ if let Some(position) = position {
+ all.remove(position);
+ } else if deletion.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot replace missing deletion vector
index '{}'; prepare the DELETE again from the latest snapshot",
deletion.index_file.file_name),
+ source: None,
+ });
+ }
}
let additions = new_index_entries
@@ -1532,6 +1654,38 @@ impl TableCommit {
});
Self::validate_global_index_overlap(&all, &additions)?;
Self::validate_added_global_index_overlap(&additions)?;
+ let mut dv_files = HashSet::new();
+ let mut dv_names = HashSet::new();
+ for entry in all
+ .iter()
+ .chain(&additions)
+ .filter(|entry| entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE)
+ {
+ let partition = if is_empty_partition(&entry.partition) {
+ &[][..]
+ } else {
+ entry.partition.as_slice()
+ };
+ if !dv_names.insert((partition, entry.bucket,
entry.index_file.file_name.as_str())) {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Duplicate deletion vector index '{}'",
+ entry.index_file.file_name
+ ),
+ source: None,
+ });
+ }
+ if let Some(ranges) = &entry.index_file.deletion_vectors_ranges {
+ for file in ranges.keys() {
+ if !dv_files.insert((partition, entry.bucket,
file.as_str())) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Conflicting deletion vectors for
data file '{file}'; prepare the DELETE again from the latest snapshot"),
+ source: None,
+ });
+ }
+ }
+ }
+ }
all.extend(additions);
Ok(all)
}
@@ -1542,13 +1696,6 @@ impl TableCommit {
commit_entries: &[ManifestEntry],
new_index_entries: &[IndexManifestEntry],
) -> Result<Vec<IndexManifestEntry>> {
- if new_index_entries
- .iter()
- .any(|entry| entry.kind == FileKind::Delete)
- {
- return Ok(vec![]);
- }
-
let mut updated_cols = HashSet::new();
let mut written_partitions: Vec<Vec<u8>> = Vec::new();
for entry in commit_entries
@@ -1586,6 +1733,9 @@ impl TableCommit {
let mut conflicted_cols = HashSet::new();
for entry in previous_entries {
if entry.kind != FileKind::Add
+ || new_index_entries.iter().any(|deleted| {
+ deleted.kind == FileKind::Delete &&
same_index_file_entry(entry, deleted)
+ })
|| !written_partitions
.iter()
.any(|partition| same_index_partition(partition,
&entry.partition))
@@ -1961,6 +2111,7 @@ impl TableCommit {
if let Some(RetryState {
latest_snapshot: Some(previous_snapshot),
base_data_files: Some(previous_base),
+ ..
}) = retry_state
{
if let Some(incremental) = self
@@ -2143,14 +2294,13 @@ impl TableCommit {
Ok(())
}
- fn check_deletion_vector_index_only_conflict(
+ fn check_deletion_vector_conflicts(
&self,
latest_snapshot: Option<&Snapshot>,
- data_entries: &[ManifestEntry],
index_entries: &[IndexManifestEntry],
check_from_snapshot: Option<i64>,
) -> Result<()> {
- if !self.data_evolution_enabled || !data_entries.is_empty() {
+ if !self.data_evolution_enabled {
return Ok(());
}
let Some(check_from_snapshot) = check_from_snapshot else {
@@ -2992,21 +3142,33 @@ enum CommitEntriesPlan {
impl CommitEntriesPlan {
fn commit_kind_hint(&self) -> CommitKind {
match self {
- CommitEntriesPlan::Direct { entries, .. } => {
- if entries
- .iter()
- .any(|entry| *entry.kind() == FileKind::Delete)
- {
- CommitKind::OVERWRITE
- } else {
- CommitKind::APPEND
- }
- }
+ CommitEntriesPlan::Direct {
+ entries,
+ new_index_entries,
+ ..
+ } => direct_commit_kind(entries, new_index_entries),
CommitEntriesPlan::Overwrite { .. } => CommitKind::OVERWRITE,
}
}
}
+fn direct_commit_kind(
+ entries: &[ManifestEntry],
+ index_entries: &[IndexManifestEntry],
+) -> CommitKind {
+ if entries
+ .iter()
+ .any(|entry| *entry.kind() == FileKind::Delete)
+ || index_entries
+ .iter()
+ .any(|entry| entry.index_file.index_type ==
DELETION_VECTORS_INDEX_TYPE)
+ {
+ CommitKind::OVERWRITE
+ } else {
+ CommitKind::APPEND
+ }
+}
+
/// Fully resolved commit ready for writing.
struct ResolvedCommit {
entries: Vec<ManifestEntry>,
@@ -3014,6 +3176,7 @@ struct ResolvedCommit {
kind: CommitKind,
index_manifest_name: Option<String>,
index_manifest_changed: bool,
+ new_index_manifest_entries: Option<Vec<IndexManifestEntry>>,
base_data_files: Option<Vec<ManifestEntry>>,
}
@@ -3025,6 +3188,7 @@ enum CommitAttemptResult {
struct RetryState {
latest_snapshot: Option<Snapshot>,
base_data_files: Option<Vec<ManifestEntry>>,
+ publication_error: Option<crate::Error>,
}
struct RowIdWriteRange {
@@ -3180,6 +3344,11 @@ fn rand_f64() -> f64 {
mod tests {
use super::*;
+ mod parity {
+ use super::*;
+ include!("table_commit/parity_tests.rs");
+ }
+
#[tokio::test]
async fn abort_still_cleans_up_for_a_query_auth_table() {
let table = crate::table::query_auth_table();
@@ -6117,12 +6286,7 @@ mod tests {
let table_path = "memory:/test_commit_preserves_delete";
setup_dirs(&file_io, table_path).await;
- let table = test_table_with_options(
- &file_io,
- table_path,
- HashMap::from([("manifest.merge-min-count".to_string(),
"2".to_string())]),
- );
- let mut commit = TableCommit::new(table, "test-user".to_string());
+ let mut commit = setup_commit(&file_io, table_path);
let partition = vec![0, 0, 0, 0];
let old_file = test_data_file("old.parquet", 1);
let mut initial_files = vec![old_file.clone()];
diff --git a/crates/paimon/src/table/table_commit/parity_tests.rs
b/crates/paimon/src/table/table_commit/parity_tests.rs
new file mode 100644
index 00000000..972530c6
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/parity_tests.rs
@@ -0,0 +1,642 @@
+// 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.
+
+fn append_message(name: &str) -> CommitMessage {
+ CommitMessage::new(vec![], 0, vec![test_data_file(name, 10)])
+}
+
+fn dv_message(name: &str, data_file: &str) -> CommitMessage {
+ let mut message = CommitMessage::new(vec![], 0, vec![]);
+ message.new_index_files = vec![test_deletion_vector_index_file(name,
data_file)];
+ message
+}
+
+#[tokio::test]
+async fn small_manifests_remain_unchanged_across_many_commits() {
+ let io = test_file_io();
+ let path = "memory:/no-small-manifest-merge";
+ setup_dirs(&io, path).await;
+ let table = test_table_with_options(
+ &io,
+ path,
+ HashMap::from([("manifest.target-file-size".into(), "8 mb".into())]),
+ );
+ let commit = TableCommit::new(table, "test-user".into());
+ let manifest_dir = format!("{path}/manifest");
+ let mut original_manifests = Vec::new();
+ let mut original_bytes = HashMap::new();
+
+ // Every previous delta must remain an unchanged base manifest across
+ // repeated commits, even though all these files fit within the target
size.
+ for id in 0..35 {
+ commit
+ .commit(vec![append_message(&format!("data-{id}.parquet"))])
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ let base = ManifestList::read(
+ &io,
+ &format!("{manifest_dir}/{}", snapshot.base_manifest_list()),
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ base, original_manifests,
+ "commit {id} rewrote historical manifests"
+ );
+ let delta = ManifestList::read(
+ &io,
+ &format!("{manifest_dir}/{}", snapshot.delta_manifest_list()),
+ )
+ .await
+ .unwrap();
+ assert_eq!(delta.len(), 1);
+ let manifest_path = format!("{manifest_dir}/{}", delta[0].file_name());
+ let bytes =
io.new_input(&manifest_path).unwrap().read().await.unwrap();
+ assert!(original_bytes.insert(manifest_path, bytes).is_none());
+ original_manifests.extend(delta);
+ }
+
+ assert!(
+ original_manifests
+ .iter()
+ .map(ManifestFileMeta::file_size)
+ .sum::<i64>()
+ < commit.manifest_target_size
+ );
+ let stored_manifests = manifest_paths(&io, path)
+ .await
+ .into_iter()
+ .filter(|path| {
+ let name = path.rsplit('/').next().unwrap();
+ name.starts_with("manifest-") &&
!name.starts_with("manifest-list-")
+ })
+ .collect::<HashSet<_>>();
+ assert_eq!(stored_manifests, original_bytes.keys().cloned().collect());
+ for (path, expected) in original_bytes {
+ assert_eq!(io.new_input(&path).unwrap().read().await.unwrap(),
expected);
+ }
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.total_record_count(), Some(350));
+ assert_eq!(active_entries(&io, path, &snapshot).await.len(), 35);
+}
+
+#[tokio::test]
+async fn duplicate_dv_for_one_data_file_is_rejected() {
+ let io = test_file_io();
+ let path = "memory:/duplicate-dv";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit(vec![append_message("data.parquet")])
+ .await
+ .unwrap();
+ let error = commit
+ .commit(vec![
+ dv_message("dv-a", "data.parquet"),
+ dv_message("dv-b", "data.parquet"),
+ ])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("deletion vector"), "{error}");
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 1);
+}
+
+#[tokio::test]
+async fn dv_replacement_rejects_missing_old_index() {
+ let io = test_file_io();
+ let path = "memory:/stale-dv-replacement";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit(vec![dv_message("current-dv", "data.parquet")])
+ .await
+ .unwrap();
+ let mut message = dv_message("new-dv", "data.parquet");
+ message
+ .deleted_index_files
+ .push(test_deletion_vector_index_file("stale-dv", "data.parquet"));
+ assert!(commit.commit(vec![message]).await.is_err());
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 1);
+}
+
+#[tokio::test]
+async fn dv_replacement_is_overwrite_and_preserves_unrelated_vectors() {
+ let io = test_file_io();
+ let path = "memory:/dv-replacement";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit(vec![
+ dv_message("old", "data-a"),
+ dv_message("other", "data-b"),
+ ])
+ .await
+ .unwrap();
+ let mut message = dv_message("new", "data-a");
+ message
+ .deleted_index_files
+ .push(test_deletion_vector_index_file("old", "data-a"));
+ commit.commit(vec![message]).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.commit_kind(), &CommitKind::OVERWRITE);
+ let entries =
+ TableCommit::read_prev_index_entries(&io, &format!("{path}/manifest"),
&Some(snapshot))
+ .await
+ .unwrap();
+ let names = entries
+ .iter()
+ .map(|entry| entry.index_file.file_name.as_str())
+ .collect::<HashSet<_>>();
+ assert_eq!(names, HashSet::from(["new", "other"]));
+}
+
+#[tokio::test]
+async fn unrelated_dv_delete_does_not_bypass_indexed_column_policy() {
+ for action in ["THROW_ERROR", "DROP_PARTITION_INDEX"] {
+ let io = test_file_io();
+ let path = format!("memory:/indexed-update-{action}");
+ setup_dirs(&io, &path).await;
+ let table = test_table_with_options(
+ &io,
+ &path,
+ HashMap::from([(
+ "global-index.column-update-action".to_string(),
+ action.to_string(),
+ )]),
+ );
+ let commit = TableCommit::new(table, "test-user".into());
+ let mut initial = dv_message("old-dv", "data");
+ 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()]);
+ let mut delete = dv_message("new-dv", "data");
+ delete
+ .deleted_index_files
+ .push(test_deletion_vector_index_file("old-dv", "data"));
+ let result = commit.commit(vec![update, delete]).await;
+ if action == "THROW_ERROR" {
+ assert!(result
+ .unwrap_err()
+ .to_string()
+ .contains("globally indexed columns"));
+ assert_eq!(latest_snapshot(&io, &path).await.unwrap().id(), 1);
+ } else {
+ result.unwrap();
+ let snapshot = latest_snapshot(&io, &path).await;
+ let entries =
+ TableCommit::read_prev_index_entries(&io,
&format!("{path}/manifest"), &snapshot)
+ .await
+ .unwrap();
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].index_file.file_name, "new-dv");
+ }
+ }
+}
+
+async fn save_schema(io: &FileIO, path: &str, id: i64) {
+ let mut json = serde_json::to_value(test_schema()).unwrap();
+ json["id"] = id.into();
+ io.new_output(&format!("{path}/schema/schema-{id}"))
+ .unwrap()
+ .write(serde_json::to_vec(&json).unwrap().into())
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn snapshot_uses_latest_schema_but_files_keep_writer_schema() {
+ let io = test_file_io();
+ let path = "memory:/latest-schema";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ save_schema(&io, path, 1).await;
+ commit
+ .commit(vec![append_message("old-writer.parquet")])
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.schema_id(), 1);
+ assert_eq!(
+ active_entries(&io, path, &snapshot).await[0]
+ .file()
+ .schema_id,
+ 0
+ );
+}
+
+#[tokio::test]
+async fn index_commit_inherits_watermark_and_schema_compatible_statistics() {
+ let io = test_file_io();
+ let path = "memory:/inherited-metadata";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit(vec![append_message("data.parquet")])
+ .await
+ .unwrap();
+ let mut initial = serde_json::to_value(latest_snapshot(&io,
path).await.unwrap()).unwrap();
+ initial["watermark"] = 1234.into();
+ initial["statistics"] = "stats-1".into();
+ io.new_output(&format!("{path}/snapshot/snapshot-1"))
+ .unwrap()
+ .write(serde_json::to_vec(&initial).unwrap().into())
+ .await
+ .unwrap();
+ io.new_output(&format!("{path}/statistics/stats-1"))
+ .unwrap()
+ .write(bytes::Bytes::from_static(
+ br#"{"schemaId":0,"snapshotId":1,"colStats":{}}"#,
+ ))
+ .await
+ .unwrap();
+ let mut index = CommitMessage::new(vec![], 0, vec![]);
+ index
+ .new_index_files
+ .push(test_global_index_file("global", 0, 0, 9));
+ commit.commit(vec![index]).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.watermark(), Some(1234));
+ assert_eq!(snapshot.statistics(), Some("stats-1"));
+
+ save_schema(&io, path, 1).await;
+ // Older writers may carry statistics across a schema change. Even if the
+ // snapshot already uses the latest schema, the statistics' own schema
must match.
+ let mut stale_statistics = serde_json::to_value(&snapshot).unwrap();
+ stale_statistics["schemaId"] = 1.into();
+ io.new_output(&format!("{path}/snapshot/snapshot-2"))
+ .unwrap()
+ .write(serde_json::to_vec(&stale_statistics).unwrap().into())
+ .await
+ .unwrap();
+ commit
+ .commit(vec![append_message("new.parquet")])
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.watermark(), Some(1234));
+ assert_eq!(snapshot.statistics(), None);
+}
+
+struct LostResponseCommit {
+ manager: SnapshotManager,
+ calls: std::sync::atomic::AtomicUsize,
+ publish_first: bool,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for LostResponseCommit {
+ async fn commit(&self, snapshot: &Snapshot, _: &[PartitionStatistics]) ->
Result<bool> {
+ let attempt = self.calls.fetch_add(1,
std::sync::atomic::Ordering::SeqCst);
+ if attempt == 0 {
+ if self.publish_first {
+ assert!(self.manager.commit_snapshot(snapshot).await?);
+ }
+ return Err(crate::Error::IoUnsupported {
+ message: "lost commit response".into(),
+ });
+ }
+ self.manager.commit_snapshot(snapshot).await
+ }
+}
+
+#[tokio::test]
+async fn publication_errors_retry_without_duplicate_snapshots() {
+ for publish_first in [false, true] {
+ let io = test_file_io();
+ let path = format!("memory:/publication-error-{publish_first}");
+ setup_dirs(&io, &path).await;
+ let mut commit = setup_commit(&io, &path);
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(LostResponseCommit {
+ manager: commit.snapshot_manager.clone(),
+ calls: std::sync::atomic::AtomicUsize::new(0),
+ publish_first,
+ });
+ commit.snapshot_commit = publisher.clone();
+ commit
+ .commit_with_identifier(vec![append_message("data.parquet")], 42)
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(snapshot.id(), 1);
+ assert_eq!(snapshot.total_record_count(), Some(10));
+ assert_eq!(
+ publisher.calls.load(std::sync::atomic::Ordering::SeqCst),
+ if publish_first { 1 } else { 2 }
+ );
+ assert_eq!(active_entries(&io, &path, &snapshot).await.len(), 1);
+ }
+}
+
+#[tokio::test]
+async fn mixed_append_cannot_bypass_stale_dv_check() {
+ let io = test_file_io();
+ let path = "memory:/mixed-stale-dv";
+ setup_dirs(&io, path).await;
+ let commit = setup_data_evolution_commit(&io, path);
+ for name in ["data", "concurrent"] {
+ let mut message = append_message(name);
+ message.new_files[0].file_source = Some(0);
+ commit.commit(vec![message]).await.unwrap();
+ }
+ let mut delete = dv_message("dv", "data");
+ delete.check_from_snapshot = Some(1);
+ let mut append = append_message("another-file");
+ 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}"
+ );
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 2);
+}
+
+async fn manifest_paths(io: &FileIO, path: &str) -> HashSet<String> {
+ io.list_status(&format!("{path}/manifest/"))
+ .await
+ .unwrap()
+ .into_iter()
+ .filter(|status| !status.is_dir)
+ .map(|status| status.path)
+ .collect()
+}
+
+#[tokio::test]
+async fn preparation_failure_cleans_new_metadata_and_preserves_old_snapshots()
{
+ let io = test_file_io();
+ let path = "memory:/prepare-failure";
+ setup_dirs(&io, path).await;
+ let table = test_table_with_options(
+ &io,
+ path,
+ HashMap::from([("manifest.sidecar.enabled".into(), "true".into())]),
+ );
+ let commit = TableCommit::new(table, "test-user".into());
+ for name in ["first", "second"] {
+ commit.commit(vec![append_message(name)]).await.unwrap();
+ }
+ let before = manifest_paths(&io, path).await;
+ let original = latest_snapshot(&io, path).await.unwrap();
+ let mut value = serde_json::to_value(&original).unwrap();
+ value["statistics"] = "invalid-statistics".into();
+ io.new_output(&format!("{path}/snapshot/snapshot-2"))
+ .unwrap()
+ .write(serde_json::to_vec(&value).unwrap().into())
+ .await
+ .unwrap();
+ io.new_output(&format!("{path}/statistics/invalid-statistics"))
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"broken JSON"))
+ .await
+ .unwrap();
+ let mut message = append_message("pending");
+ message
+ .new_changelog_files
+ .push(test_data_file("changelog", 10));
+ message
+ .new_index_files
+ .push(test_global_index_file("pending-index", 0, 0, 9));
+ let error = commit.commit(vec![message]).await.unwrap_err();
+ assert!(
+ error.to_string().contains("Invalid statistics metadata"),
+ "{error}"
+ );
+ assert_eq!(manifest_paths(&io, path).await, before, "failed preparation
must clean lists, delta/changelog manifests, sidecars and index manifests");
+ assert_eq!(active_entries(&io, path, &original).await.len(), 2);
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 2);
+}
+
+#[derive(Debug)]
+struct FailSecondManifest {
+ operator: opendal::Operator,
+ failed: std::sync::atomic::AtomicBool,
+}
+
+#[async_trait::async_trait]
+impl crate::io::FileIOProvider for FailSecondManifest {
+ async fn create(&self, path: &str) -> Result<(opendal::Operator, String)> {
+ let name = path.rsplit('/').next().unwrap();
+ if name.starts_with("manifest-")
+ && !name.starts_with("manifest-list-")
+ && name.ends_with("-1")
+ && !self.failed.swap(true, std::sync::atomic::Ordering::SeqCst)
+ {
+ return Err(crate::Error::IoUnsupported {
+ message: "injected second manifest write failure".into(),
+ });
+ }
+ Ok((
+ self.operator.clone(),
+ path.trim_start_matches("memory:/").to_string(),
+ ))
+ }
+}
+
+#[tokio::test]
+async fn rolling_manifest_failure_cleans_previous_chunks_and_sidecars() {
+ let provider = Arc::new(FailSecondManifest {
+ operator:
opendal::Operator::new(opendal::services::Memory::default()).unwrap(),
+ failed: std::sync::atomic::AtomicBool::new(false),
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let path = "memory:/rolling-failure";
+ setup_dirs(&io, path).await;
+ let table = test_table_with_options(
+ &io,
+ path,
+ HashMap::from([
+ ("manifest.sidecar.enabled".into(), "true".into()),
+ ("manifest.target-file-size".into(), "1 b".into()),
+ ]),
+ );
+ let commit = TableCommit::new(table, "test-user".into());
+ let files = (0..2001)
+ .map(|id| test_data_file(&format!("{id}.parquet"), 1))
+ .collect();
+ let error = commit
+ .commit(vec![CommitMessage::new(vec![], 0, files)])
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("injected second manifest"),
+ "{error}"
+ );
+ assert!(provider.failed.load(std::sync::atomic::Ordering::SeqCst));
+ assert!(manifest_paths(&io, path).await.is_empty());
+ assert!(latest_snapshot(&io, path).await.is_none());
+}
+
+#[tokio::test]
+async fn uncertain_guarded_commit_does_not_abort_published_index_files() {
+ let io = test_file_io();
+ let path = "memory:/uncertain-index";
+ setup_dirs(&io, path).await;
+ let mut commit = setup_commit(&io, path);
+ commit.commit(vec![append_message("data")]).await.unwrap();
+ commit.commit_max_retries = 0;
+ commit.snapshot_commit = Arc::new(LostResponseCommit {
+ manager: commit.snapshot_manager.clone(),
+ calls: std::sync::atomic::AtomicUsize::new(0),
+ publish_first: true,
+ });
+ let index_path = format!("{path}/index/global");
+ io.new_output(&index_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"index"))
+ .await
+ .unwrap();
+ let mut index = CommitMessage::new(vec![], 0, vec![]);
+ index
+ .new_index_files
+ .push(test_global_index_file("global", 0, 0, 9));
+ let error = commit
+ .commit_if_latest_snapshot_with_identifier(vec![index.clone()], 1, 42)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("outcome may be unknown"),
+ "{error}"
+ );
+ assert!(io.exists(&index_path).await.unwrap());
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.id(), 2);
+ assert!(snapshot.index_manifest().is_some());
+ commit
+ .filter_and_commit_with_identifier(vec![index], 42)
+ .await
+ .unwrap();
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), 2);
+}
+
+#[tokio::test]
+async fn dv_publication_response_loss_uses_overwrite_identity() {
+ let io = test_file_io();
+ let path = "memory:/dv-response-loss";
+ setup_dirs(&io, path).await;
+ let mut commit = setup_commit(&io, path);
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(LostResponseCommit {
+ manager: commit.snapshot_manager.clone(),
+ calls: std::sync::atomic::AtomicUsize::new(0),
+ publish_first: true,
+ });
+ commit.snapshot_commit = publisher.clone();
+ commit
+ .commit_with_identifier(vec![dv_message("dv", "data")], 7)
+ .await
+ .unwrap();
+ assert_eq!(
+ latest_snapshot(&io, path).await.unwrap().commit_kind(),
+ &CommitKind::OVERWRITE
+ );
+ assert_eq!(publisher.calls.load(std::sync::atomic::Ordering::SeqCst), 1);
+}
+
+#[tokio::test]
+async fn rest_commit_uses_catalog_snapshot_schema_and_retry_identity() {
+ use crate::api::rest_api::RESTApi;
+ use crate::common::Options;
+ use axum::{
+ body::Bytes,
+ http::{Method, Uri},
+ Json, Router,
+ };
+ use std::sync::Mutex;
+ let io = test_file_io();
+ let path = "memory:/rest-commit-parity";
+ setup_dirs(&io, path).await;
+ let seed = setup_commit(&io, path);
+ seed.commit(vec![append_message("original")]).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 handler_snapshot = snapshot.clone();
+ let handler_posts = posts.clone();
+ let app = Router::new().fallback(move |method: Method, uri: Uri, body:
Bytes| {
+ 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") {
+ serde_json::json!({"snapshot": {"snapshot":
*snapshot.lock().unwrap(), "recordCount": 10}})
+ } else {
+ serde_json::json!({"schemaId": 3})
+ };
+ 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(),
+ test_schema(),
+ Some(env),
+ );
+ let mut commit = TableCommit::new(table, "rest-writer".into());
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let messages = vec![append_message("new")];
+ commit
+ .commit_if_latest_snapshot_with_identifier(messages.clone(), 7, 42)
+ .await
+ .unwrap();
+ commit
+ .filter_and_commit_with_identifier(messages, 42)
+ .await
+ .unwrap();
+ let latest = snapshot.lock().unwrap().clone();
+ assert_eq!(latest.id(), 8);
+ assert_eq!(latest.schema_id(), 3);
+ assert_eq!(latest.total_record_count(), Some(20));
+ assert_eq!(posts.load(std::sync::atomic::Ordering::SeqCst), 1);
+ assert!(!io
+ .exists(&format!("{path}/snapshot/snapshot-8"))
+ .await
+ .unwrap());
+ assert_eq!(active_entries(&io, path, &latest).await.len(), 2);
+ server.abort();
+}
