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 3ba159ca [core] Align commit failure cleanup with Java (#1045)
3ba159ca is described below
commit 3ba159ca31af0308b18e68b46ebf935e9b7b0d45
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 6 16:19:14 2026 +0800
[core] Align commit failure cleanup with Java (#1045)
---
bindings/go/postpone_fixed_bucket_write.go | 3 +-
bindings/go/write.go | 3 +-
bindings/python/README.md | 9 +-
.../python/python/pypaimon_rust/datafusion.pyi | 6 +-
bindings/python/tests/test_table_commit.py | 20 +-
bindings/python/tests/test_update.py | 20 +-
bindings/python/tests/test_write.py | 5 +-
crates/paimon/src/table/commit_message.rs | 40 +++
crates/paimon/src/table/data_evolution_writer.rs | 2 +
crates/paimon/src/table/format_table_commit.rs | 114 ++++---
.../paimon/src/table/format_table_write_tests.rs | 368 ++++++++++++++++++++-
crates/paimon/src/table/format_table_writer.rs | 14 +-
.../src/table/lumina_index_build_builder/writer.rs | 1 +
.../src/table/sorted_global_index_build_builder.rs | 2 +
crates/paimon/src/table/table_commit.rs | 54 ++-
.../paimon/src/table/table_commit/abort_tests.rs | 121 +++++++
crates/paimon/src/table/table_update.rs | 19 +-
crates/paimon/src/table/table_update_by_row_id.rs | 23 +-
crates/paimon/src/table/table_update_predicate.rs | 8 +-
crates/paimon/src/table/table_upsert.rs | 11 +-
crates/paimon/src/table/table_write.rs | 19 +-
.../paimon/src/table/vindex_index_build_builder.rs | 2 +
crates/paimon/tests/dynamic_bucket_write_test.rs | 87 ++++-
crates/paimon/tests/table_update_test.rs | 75 ++++-
docs/src/getting-started.md | 16 +-
25 files changed, 883 insertions(+), 159 deletions(-)
diff --git a/bindings/go/postpone_fixed_bucket_write.go
b/bindings/go/postpone_fixed_bucket_write.go
index 24ea290b..d3445c16 100644
--- a/bindings/go/postpone_fixed_bucket_write.go
+++ b/bindings/go/postpone_fixed_bucket_write.go
@@ -344,7 +344,8 @@ func (tc *PostponeFixedBucketTableCommit)
TruncateTableWithIdentifier(
)
}
-// Abort performs best-effort cleanup of files in prepared messages.
+// Abort deletes files for an explicitly abandoned, known-uncommitted write.
+// Never call when the commit outcome is unknown.
func (tc *PostponeFixedBucketTableCommit) Abort(
messages *PostponeFixedBucketCommitMessages,
) error {
diff --git a/bindings/go/write.go b/bindings/go/write.go
index c5b793ab..7ae14693 100644
--- a/bindings/go/write.go
+++ b/bindings/go/write.go
@@ -304,7 +304,8 @@ func (tc *TableCommit)
TruncateTableWithIdentifier(commitIdentifier int64) error
return
ffiTableCommitTruncateTableWithIdentifier.symbol(tc.ctx)(tc.inner,
commitIdentifier)
}
-// Abort performs best-effort cleanup of files created for a prepared commit.
+// Abort deletes files for an explicitly abandoned, known-uncommitted write.
+// Never call when the commit outcome is unknown.
func (tc *TableCommit) Abort(messages *CommitMessages) error {
return tc.withMessages(messages, ffiTableCommitAbort.symbol(tc.ctx))
}
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 56613434..ad3230c4 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -197,8 +197,13 @@ mode. `CommitMessage.deserialize(body, version=14)`
decodes it without a table
or builder. Submit decoded messages to their originating table; commit identity
and overwrite mode come from the configured committer. Messages returned
directly
by local writers retain their table and commit-user checks.
-Only v14 is supported. `abort(messages)` deletes newly written files and must
-only be used for messages known not to have committed. Compact increments
remain
+Only v14 is supported. `abort(messages)` deletes newly written files for an
+explicitly abandoned write, matching Java. Call it only for messages known to
be
+uncommitted that will never be submitted. A commit exception does not prove
failure:
+publication may have succeeded before the response failed. Internal error paths
+preserve submitted files instead of calling `abort`. Preparation failures can
clean
+outputs that have not been handed off to the caller.
+Compact increments remain
unsupported by the Rust committer and are rejected.
### Tables resolved outside the Rust catalog
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 2953e1c5..fcfe3c58 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -308,7 +308,7 @@ class BatchTableCommit:
"""Truncate matching partitions. Reject empty input, as Java does."""
...
def abort(self, messages: Sequence[CommitMessage]) -> None:
- """Best-effort deletion of newly written files; only abort uncommitted
messages."""
+ """Delete known-uncommitted files; never call when the commit outcome
is unknown."""
...
class StreamTableCommit:
@@ -321,7 +321,9 @@ class StreamTableCommit:
) -> int:
"""Sort identifiers, skip committed groups, and return the number
committed."""
...
- def abort(self, messages: Sequence[CommitMessage]) -> None: ...
+ def abort(self, messages: Sequence[CommitMessage]) -> None:
+ """Delete known-uncommitted files; never call when the commit outcome
is unknown."""
+ ...
class BatchWriteBuilder:
def _with_commit_user(self, commit_user: str) -> "BatchWriteBuilder":
diff --git a/bindings/python/tests/test_table_commit.py
b/bindings/python/tests/test_table_commit.py
index 2af7a926..00725b0e 100644
--- a/bindings/python/tests/test_table_commit.py
+++ b/bindings/python/tests/test_table_commit.py
@@ -185,15 +185,25 @@ def
test_invalid_batch_bridge_commit_user_preserves_identity(tmp_path, user):
assert _rows(table) == [1]
-def test_abort_serialized_messages_deletes_files(tmp_path):
- table = _table(tmp_path)
- builder = table.new_batch_write_builder()
- messages = _roundtrip(_prepare(builder, [1], [10]))
[email protected]('stream', [False, True])
[email protected]('primary_key', [False, True])
[email protected]('serialized', [False, True])
+def test_explicit_abort_deletes_known_uncommitted_files(tmp_path, stream,
primary_key, serialized):
+ table = _table(tmp_path, primary_key=primary_key, options={'bucket': '1'}
if primary_key else None)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ messages = _prepare(builder, [1], [10], 7 if stream else None)
+ if serialized:
+ messages = _roundtrip(messages)
files = list(tmp_path.rglob("data-*.parquet"))
assert files
- builder.new_commit().abort(messages)
+ assert table.latest_snapshot() is None
+ commit = builder.new_commit()
+ # These messages have never been submitted and will not be committed.
+ commit.abort(messages)
+ commit.abort(messages)
assert not any(file.exists() for file in files)
assert table.latest_snapshot() is None
+ assert _rows(table) == []
def test_batch_writer_and_commit_are_one_shot(tmp_path):
diff --git a/bindings/python/tests/test_update.py
b/bindings/python/tests/test_update.py
index 2e7322ec..33bf8a09 100644
--- a/bindings/python/tests/test_update.py
+++ b/bindings/python/tests/test_update.py
@@ -172,13 +172,16 @@ def
test_grouped_batch_update_checks_input_table_file_overlap(tmp_path):
'grouped_updates.t')
before = set(tmp_path.rglob('*.parquet'))
+ snapshot_id = table.latest_snapshot().id()
overlap = table.new_batch_write_builder().new_update()
with pytest.raises(ValueError, match='overlapping first_row_ids.*0'):
overlap.update_by_arrow_batches_with_row_id(iter([
pa.table({'_ROW_ID': [0], 'name': ['A']}),
pa.table({'_ROW_ID': [1], 'name': ['B']}),
]))
- assert set(tmp_path.rglob('*.parquet')) == before
+ assert set(tmp_path.rglob('*.parquet')) > before
+ assert table.latest_snapshot().id() == snapshot_id
+ before = set(tmp_path.rglob('*.parquet'))
def failing_tables():
yield pa.table({'_ROW_ID': [0], 'name': ['A']})
@@ -186,7 +189,11 @@ def
test_grouped_batch_update_checks_input_table_file_overlap(tmp_path):
with pytest.raises(RuntimeError, match='input failed'):
overlap.update_by_arrow_batches_with_row_id(failing_tables())
- assert set(tmp_path.rglob('*.parquet')) == before
+ assert set(tmp_path.rglob('*.parquet')) > before
+ assert table.latest_snapshot().id() == snapshot_id
+ unchanged = pa.Table.from_batches(context.sql(
+ 'SELECT id, name FROM
paimon.grouped_updates.t')).sort_by('id').to_pydict()
+ assert unchanged == {'id': [1, 2, 3, 4], 'name': ['a', 'b', 'c', 'd']}
builder = table.new_batch_write_builder()
messages = builder.new_update().update_by_arrow_batches_with_row_id(iter([
@@ -271,7 +278,7 @@ def test_stream_row_id_update_and_factory(tmp_path):
assert actual == {'id': [1, 2], 'value': [11, 22]}
-def test_predicate_update_owns_scan_callbacks_and_rollback(tmp_path):
+def
test_predicate_update_owns_scan_callbacks_and_preserves_prepared_files(tmp_path):
assert not hasattr(datafusion, '_MatchedBatchUpdateWriter')
context = SQLContext()
context.register_catalog('paimon', {'warehouse': str(tmp_path)})
@@ -303,6 +310,7 @@ def
test_predicate_update_owns_scan_callbacks_and_rollback(tmp_path):
'score': [100, 999, 999, 999]}
before = set(tmp_path.rglob('*.parquet'))
+ snapshot_id = table.latest_snapshot().id()
seen.clear()
def fail_second(rows):
@@ -314,7 +322,11 @@ def
test_predicate_update_owns_scan_callbacks_and_rollback(tmp_path):
with pytest.raises(RuntimeError, match='callback failure'):
update.update_by_predicate(None, {'value': fail_second},
read_columns=['value'])
assert len(seen) == 2
- assert set(tmp_path.rglob('*.parquet')) == before
+ assert set(tmp_path.rglob('*.parquet')) > before
+ assert table.latest_snapshot().id() == snapshot_id
+ unchanged = pa.Table.from_batches(context.sql(
+ 'SELECT id, value, score FROM
paimon.pred_updates.t')).sort_by('id').to_pydict()
+ assert unchanged == actual
assert update.update_by_predicate(
{'method': 'equal', 'field': 'id', 'literals': [99]},
{'value': 'bad-int'}) == []
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index 657965fc..59e256a3 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -206,9 +206,8 @@ def test_commit_different_builder_same_table_raises():
table.new_batch_write_builder().new_commit().commit(messages)
-def test_abort_cleans_up_written_data():
- # Write data, prepare commit, abort — the written files should be deleted
- # and reading back should return zero rows.
+def test_abort_cleans_up_abandoned_data():
+ # Explicitly discard prepared messages that have never been committed.
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
diff --git a/crates/paimon/src/table/commit_message.rs
b/crates/paimon/src/table/commit_message.rs
index 29c904d1..db24ee9c 100644
--- a/crates/paimon/src/table/commit_message.rs
+++ b/crates/paimon/src/table/commit_message.rs
@@ -276,6 +276,46 @@ pub(crate) struct FormatFileCommit {
pub partition: HashMap<String, String>,
pub record_count: i64,
pub file_size: i64,
+ // Java TwoPhaseCommitMessage keeps this on the message so a later abort
+ // instance makes the same decision. Cloned Rust messages share the flag.
+ preserve_published_target_on_abort:
std::sync::Arc<std::sync::atomic::AtomicBool>,
+}
+
+impl FormatFileCommit {
+ pub(crate) fn new(
+ staged_path: String,
+ target_path: String,
+ partition: HashMap<String, String>,
+ record_count: i64,
+ file_size: i64,
+ ) -> Self {
+ Self {
+ staged_path,
+ target_path,
+ partition,
+ record_count,
+ file_size,
+ // Until publication starts the target is not owned by this write.
+ preserve_published_target_on_abort: std::sync::Arc::new(
+ std::sync::atomic::AtomicBool::new(true),
+ ),
+ }
+ }
+
+ pub(crate) fn preserve_published_target_on_abort(&self) {
+ self.preserve_published_target_on_abort
+ .store(true, std::sync::atomic::Ordering::Relaxed);
+ }
+
+ pub(crate) fn allow_published_target_cleanup(&self) {
+ self.preserve_published_target_on_abort
+ .store(false, std::sync::atomic::Ordering::Relaxed);
+ }
+
+ pub(crate) fn should_preserve_published_target_on_abort(&self) -> bool {
+ self.preserve_published_target_on_abort
+ .load(std::sync::atomic::Ordering::Relaxed)
+ }
}
#[cfg(test)]
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 0e82ade7..2647bff2 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -659,6 +659,8 @@ impl DataEvolutionDeleteWriter {
Ok(Some(message)) => messages.push(message),
Ok(None) => {}
Err(error) => {
+ // Earlier outputs are private to this preparation; nothing
+ // has been returned to a caller or submitted for commit.
let _ = self
.table
.new_write_builder()
diff --git a/crates/paimon/src/table/format_table_commit.rs
b/crates/paimon/src/table/format_table_commit.rs
index ee4a7822..511b3ddc 100644
--- a/crates/paimon/src/table/format_table_commit.rs
+++ b/crates/paimon/src/table/format_table_commit.rs
@@ -16,9 +16,9 @@
// under the License.
//! Publish prepared Format Table files without creating a Paimon snapshot.
-//! Java's `FormatTableCommit` treats published data and catalog partition
-//! registration as separate side effects, with different rollback rules before
-//! and after a partition becomes visible to readers.
+//! Published data and catalog partition registration are separate side
effects.
+//! Follow Java two-phase abort: discard staging, roll back append targets
before
+//! registration completes, and protect overwrite replacements and registered
files.
use std::collections::{HashMap, HashSet};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -74,12 +74,32 @@ impl<'a> FormatTableCommit<'a> {
}
pub(crate) async fn abort(&self, messages: &[CommitMessage]) -> Result<()>
{
+ let mut failure = None;
for message in messages {
- if let Some(file) = &message.format_file {
- self.table.file_io().delete_file(&file.staged_path).await?;
+ let Some(file) = &message.format_file else {
+ failure.get_or_insert_with(|| crate::Error::DataInvalid {
+ message: "Format Table abort requires staged file
messages".into(),
+ source: None,
+ });
+ continue;
+ };
+ // Staging belongs to this write; a preflight failure must never
+ // remove a pre-existing target. Keep processing after cleanup
errors.
+ if file
+ .staged_path
+ .starts_with(&format!("{}/_temporary/", self.table_path))
+ {
+ if let Err(error) =
self.table.file_io().delete_file(&file.staged_path).await {
+ failure.get_or_insert(error);
+ }
+ }
+ if !file.should_preserve_published_target_on_abort() {
+ if let Err(error) =
self.table.file_io().delete_file(&file.target_path).await {
+ failure.get_or_insert(error);
+ }
}
}
- Ok(())
+ failure.map_or(Ok(()), Err)
}
/// `None` is append; `Some(None)` is an overwrite without static
@@ -89,9 +109,21 @@ impl<'a> FormatTableCommit<'a> {
messages: &[CommitMessage],
overwrite: Option<Option<&HashMap<String, Option<Datum>>>>,
) -> Result<()> {
+ // Validation can encounter a target from an earlier attempt or another
+ // writer. Until this attempt starts publishing, only staging is ours.
+ for file in messages
+ .iter()
+ .filter_map(|message| message.format_file.as_ref())
+ {
+ file.preserve_published_target_on_abort();
+ }
let result = self.apply_inner(messages, overwrite).await;
if result.is_err() {
- let _ = self.abort(messages).await;
+ // Java FormatTableCommit rolls back append publication, but
retains
+ // overwrite replacements and successfully registered append files.
+ if let Err(error) = self.abort(messages).await {
+ log::warn!("Failed to abort Format Table files after commit
failure: {error}");
+ }
}
result
}
@@ -159,24 +191,17 @@ impl<'a> FormatTableCommit<'a> {
}
}
- let mut published: Vec<String> = Vec::new();
- for file in &files {
- let result = self.publish(file).await;
- if let Err(error) = result {
- // An overwrite has already removed old files, so replacement
- // files must survive an uncertain partial publish. Append can
- // safely roll back files this attempt uniquely named.
- if overwrite.is_none() {
- for path in
published.iter().chain(std::iter::once(&file.target_path)) {
- let _ = self.table.file_io().delete_file(path).await;
- }
- }
- self.discard_staging(&files).await;
- return Err(error);
+ if overwrite.is_some() {
+ // Old data has been removed. Java protects every replacement
target
+ // before publishing, including a later explicit abort instance.
+ for file in &files {
+ file.preserve_published_target_on_abort();
}
- published.push(file.target_path.clone());
}
- self.discard_staging(&files).await;
+ for file in &files {
+ self.publish(file, overwrite.is_some()).await?;
+ }
+ self.discard_staging(&files).await?;
if managed {
let stats = self.partition_statistics(&files, &selected,
overwrite.is_some());
@@ -212,17 +237,14 @@ impl<'a> FormatTableCommit<'a> {
)
.await?;
} else {
- // Java registers first, then preserves visible files even
- // if the later additive statistics report fails.
- if let Err(error) = env
- .api()
+ // Java registers append partitions before protecting their
+ // targets. Registration failure rolls back this attempt's
+ // files, even if registration succeeded before a lost
response.
+ env.api()
.create_partitions(env.identifier(), specs.clone(),
true)
- .await
- {
- for path in &published {
- let _ =
self.table.file_io().delete_file(path).await;
- }
- return Err(error);
+ .await?;
+ for file in &files {
+ file.preserve_published_target_on_abort();
}
if let Err(error) = env
.api()
@@ -375,7 +397,7 @@ impl<'a> FormatTableCommit<'a> {
})
}
- async fn publish(&self, file: &FormatFileCommit) -> Result<()> {
+ async fn publish(&self, file: &FormatFileCommit, overwrite: bool) ->
Result<()> {
let directory = self.partition_directory(&file.partition)?;
self.table
.file_io()
@@ -390,6 +412,11 @@ impl<'a> FormatTableCommit<'a> {
source: None,
});
}
+ if !overwrite {
+ // Like Java's overwrite=false two-phase writer, the new UUID
target
+ // belongs to this append once publication starts and can be
rolled back.
+ file.allow_published_target_cleanup();
+ }
match self
.table
.file_io()
@@ -400,24 +427,25 @@ impl<'a> FormatTableCommit<'a> {
Err(crate::Error::IoUnexpected { source, .. })
if source.kind() == opendal::ErrorKind::Unsupported =>
{
- let result = self
- .table
+ // Abort uses the message's append/overwrite protection even if
+ // a copy finishes before its response fails.
+ self.table
.file_io()
.copy_file_streaming(&file.staged_path, &file.target_path)
- .await;
- if result.is_err() {
- let _ =
self.table.file_io().delete_file(&file.target_path).await;
- }
- result
+ .await
}
Err(error) => Err(error),
}
}
- async fn discard_staging(&self, files: &[FormatFileCommit]) {
+ async fn discard_staging(&self, files: &[FormatFileCommit]) -> Result<()> {
+ let mut failure = None;
for file in files {
- let _ = self.table.file_io().delete_file(&file.staged_path).await;
+ if let Err(error) =
self.table.file_io().delete_file(&file.staged_path).await {
+ failure.get_or_insert(error);
+ }
}
+ failure.map_or(Ok(()), Err)
}
async fn selected_overwrite_partitions(
diff --git a/crates/paimon/src/table/format_table_write_tests.rs
b/crates/paimon/src/table/format_table_write_tests.rs
index f9bbe50e..94baa8e8 100644
--- a/crates/paimon/src/table/format_table_write_tests.rs
+++ b/crates/paimon/src/table/format_table_write_tests.rs
@@ -357,6 +357,369 @@ async fn
prepared_files_are_hidden_until_commit_and_abort_discards_them() {
assert!(visible_files(&table, "dt=a").await.is_empty());
}
+#[tokio::test]
+async fn invalid_commit_discards_staging_without_deleting_foreign_targets() {
+ let table = memory_table("format_invalid_commit_cleanup", true, &[]);
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch(&[("a", 1)])).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let file = messages[0].format_file.as_ref().unwrap();
+ let outside = "memory:/outside-table/file.parquet";
+ table
+ .file_io()
+ .new_output(outside)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"foreign"))
+ .await
+ .unwrap();
+ let mut invalid = messages.clone();
+ invalid[0].format_file.as_mut().unwrap().target_path = outside.into();
+ assert!(builder
+ .new_commit()
+ .commit(invalid.clone())
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("outside partition"));
+ assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
+ assert!(visible_files(&table, "dt=a").await.is_empty());
+ builder.new_commit().abort(&invalid).await.unwrap();
+ assert_eq!(
+ table
+ .file_io()
+ .new_input(outside)
+ .unwrap()
+ .read()
+ .await
+ .unwrap(),
+ bytes::Bytes::from_static(b"foreign")
+ );
+ append(&table, &batch(&[("a", 2)])).await;
+ assert_eq!(ids(&table).await, [2]);
+}
+
+#[derive(Debug)]
+struct FailStagedAccess {
+ operator: opendal::Operator,
+ staged_path: std::sync::Mutex<String>,
+ accesses: std::sync::atomic::AtomicUsize,
+ fail_at: usize,
+}
+
+#[async_trait::async_trait]
+impl crate::io::FileIOProvider for FailStagedAccess {
+ async fn create(&self, path: &str) -> crate::Result<(opendal::Operator,
String)> {
+ // The first access validates staging; the second starts publication.
+ if path == *self.staged_path.lock().unwrap()
+ && self
+ .accesses
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+ == self.fail_at
+ {
+ return Err(crate::Error::IoUnsupported {
+ message: "staged I/O failed".into(),
+ });
+ }
+ Ok((
+ self.operator.clone(),
+ path.trim_start_matches("memory:/").into(),
+ ))
+ }
+}
+
+#[tokio::test]
+async fn publication_and_staging_cleanup_failures_follow_java_abort() {
+ for overwrite in [false, true] {
+ for fail_at in [1, 2] {
+ // Use a rename-capable backend so access 2 is staging cleanup,
+ // independently of the streaming-copy fallback's extra reads.
+ let temp = tempfile::tempdir().unwrap();
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some(temp.path().display().to_string());
+ let provider = Arc::new(FailStagedAccess {
+ operator: opendal::Operator::from_config(config).unwrap(),
+ staged_path: std::sync::Mutex::new(String::new()),
+ accesses: std::sync::atomic::AtomicUsize::new(0),
+ fail_at,
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let table = table(io, "memory:/format_failed_publish", true, &[]);
+ append(&table, &batch(&[("a", 9), ("b", 10)])).await;
+ let mut builder = table.new_write_builder();
+ if overwrite {
+ builder = builder.with_overwrite();
+ }
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(&[("a", 1), ("b", 2)]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ messages.sort_by_key(|message| {
+ message.format_file.as_ref().unwrap().partition["dt"].clone()
+ });
+ let first = messages[0].format_file.as_ref().unwrap().clone();
+ let second = messages[1].format_file.as_ref().unwrap().clone();
+ *provider.staged_path.lock().unwrap() = second.staged_path.clone();
+ let commit = builder.new_commit();
+ let result = if overwrite {
+ commit.overwrite(messages.clone(), None).await
+ } else {
+ commit.commit(messages.clone()).await
+ };
+ let error = result.unwrap_err();
+ assert!(error.to_string().contains("staged I/O failed"),
"{error}");
+ assert_eq!(
+ table.file_io().exists(&first.target_path).await.unwrap(),
+ overwrite
+ );
+ assert_eq!(
+ table.file_io().exists(&second.target_path).await.unwrap(),
+ overwrite && fail_at == 2
+ );
+ for file in [&first, &second] {
+
assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
+ }
+ // A new committer sees protection through cloned messages, as in
Java.
+ builder.new_commit().abort(&messages).await.unwrap();
+ let expected = if !overwrite {
+ vec![9, 10]
+ } else if fail_at == 1 {
+ vec![1]
+ } else {
+ vec![1, 2]
+ };
+ assert_eq!(ids(&table).await, expected);
+ }
+ }
+}
+
+#[tokio::test]
+async fn
catalog_registration_controls_append_rollback_and_overwrite_protection() {
+ use crate::{Options, RESTApi};
+ use axum::http::{Method, StatusCode, Uri};
+ use axum::response::IntoResponse;
+ use axum::{Json, Router};
+ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+ for overwrite in [false, true] {
+ for (fail_registration, fail_statistics) in [(true, false), (false,
false), (false, true)] {
+ let registered = Arc::new(AtomicBool::new(false));
+ let server_registered = registered.clone();
+ let calls = Arc::new(AtomicUsize::new(0));
+ let app = Router::new().fallback(move |method: Method, uri: Uri| {
+ let registered = server_registered.clone();
+ let calls = calls.clone();
+ async move {
+ if method == Method::POST &&
uri.path().ends_with("/partitions") {
+ let call = calls.fetch_add(1, Ordering::SeqCst);
+ registered.store(true, Ordering::SeqCst);
+ // Registration/statistics took effect but the
response is undecodable.
+ if (call == 0 && fail_registration) || (call > 0 &&
fail_statistics) {
+ return (StatusCode::OK, "lost catalog
response").into_response();
+ }
+ return Json(serde_json::json!({})).into_response();
+ }
+ let partitions = if registered.load(Ordering::SeqCst) {
+ serde_json::json!([{"spec": {"dt": "a"}}])
+ } else {
+ serde_json::json!([])
+ };
+ Json(serde_json::json!({"partitions":
partitions})).into_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("default", "format_abort");
+ let cache =
crate::io::FileIOCacheContext::from_props(options.to_map()).unwrap();
+ let env = super::RESTEnv::new(
+ identifier.clone(),
+ "uuid".into(),
+ api,
+ options,
+ false,
+ None,
+ cache,
+ );
+ let base = memory_table(
+ "format_catalog_abort",
+ true,
+ &[("metastore.partitioned-table", "true")],
+ );
+ let table = Table::new(
+ base.file_io().clone(),
+ identifier,
+ base.location().into(),
+ base.schema().clone(),
+ Some(env),
+ );
+ let mut builder = table.new_write_builder();
+ if overwrite {
+ builder = builder.with_overwrite();
+ }
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch(&[("a", 1)])).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let file = messages[0].format_file.as_ref().unwrap().clone();
+ let commit = builder.new_commit();
+ let result = if overwrite {
+ commit.overwrite(messages.clone(), None).await
+ } else {
+ commit.commit(messages.clone()).await
+ };
+ assert_eq!(result.is_err(), fail_registration, "{result:?}");
+ assert!(registered.load(Ordering::SeqCst));
+ let retain = overwrite || !fail_registration;
+ assert_eq!(
+ table.file_io().exists(&file.target_path).await.unwrap(),
+ retain
+ );
+ assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
+ builder.new_commit().abort(&messages).await.unwrap();
+ assert_eq!(
+ table.file_io().exists(&file.target_path).await.unwrap(),
+ retain
+ );
+ assert_eq!(ids(&table).await, if retain { vec![1] } else { vec![]
});
+ server.abort();
+ }
+ }
+}
+
+#[derive(Debug)]
+struct LosePublicationResponse {
+ operator: opendal::Operator,
+ paths: std::sync::Mutex<(String, String)>,
+ accesses: std::sync::atomic::AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl crate::io::FileIOProvider for LosePublicationResponse {
+ async fn create(&self, path: &str) -> crate::Result<(opendal::Operator,
String)> {
+ let (staged, target) = self.paths.lock().unwrap().clone();
+ if path == staged
+ && self
+ .accesses
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
+ == 1
+ {
+ // The target became visible, but the caller receives an exception.
+ let data = self
+ .operator
+ .read(staged.trim_start_matches("memory:/"))
+ .await
+ .map_err(|error|
crate::Error::from_opendal_with_context(error, "read staging"))?;
+ self.operator
+ .write(target.trim_start_matches("memory:/"), data)
+ .await
+ .map_err(|error| {
+ crate::Error::from_opendal_with_context(error, "publish
target")
+ })?;
+ return Err(crate::Error::IoUnsupported {
+ message: "lost publication response".into(),
+ });
+ }
+ Ok((
+ self.operator.clone(),
+ path.trim_start_matches("memory:/").into(),
+ ))
+ }
+}
+
+#[tokio::test]
+async fn
lost_file_publication_response_rolls_back_append_and_preserves_overwrite() {
+ for overwrite in [false, true] {
+ let provider = Arc::new(LosePublicationResponse {
+ operator:
opendal::Operator::new(opendal::services::Memory::default()).unwrap(),
+ paths: std::sync::Mutex::new((String::new(), String::new())),
+ accesses: std::sync::atomic::AtomicUsize::new(0),
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let table = table(io, "memory:/format_lost_file_response", true, &[]);
+ append(&table, &batch(&[("a", 9)])).await;
+ let mut builder = table.new_write_builder();
+ if overwrite {
+ builder = builder.with_overwrite();
+ }
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch(&[("a", 1)])).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let file = messages[0].format_file.as_ref().unwrap();
+ *provider.paths.lock().unwrap() = (file.staged_path.clone(),
file.target_path.clone());
+ let commit = builder.new_commit();
+ let result = if overwrite {
+ commit.overwrite(messages.clone(), None).await
+ } else {
+ commit.commit(messages.clone()).await
+ };
+ assert!(result
+ .unwrap_err()
+ .to_string()
+ .contains("lost publication response"));
+ assert_eq!(
+ table.file_io().exists(&file.target_path).await.unwrap(),
+ overwrite
+ );
+ assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
+ builder.new_commit().abort(&messages).await.unwrap();
+ assert_eq!(ids(&table).await, if overwrite { vec![1] } else { vec![9]
});
+ }
+}
+
+#[tokio::test]
+async fn abort_attempts_every_staged_file_after_cleanup_failure() {
+ let provider = Arc::new(FailStagedAccess {
+ operator:
opendal::Operator::new(opendal::services::Memory::default()).unwrap(),
+ staged_path: std::sync::Mutex::new(String::new()),
+ accesses: std::sync::atomic::AtomicUsize::new(0),
+ fail_at: 0,
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let table = table(io, "memory:/format_abort_failure", true, &[]);
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(&[("a", 1), ("b", 2)]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let first = messages[0].format_file.as_ref().unwrap();
+ let second = messages[1].format_file.as_ref().unwrap();
+ for file in [first, second] {
+ table
+ .file_io()
+ .copy_file_streaming(&file.staged_path, &file.target_path)
+ .await
+ .unwrap();
+ }
+ *provider.staged_path.lock().unwrap() = first.staged_path.clone();
+ let error = builder.new_commit().abort(&messages).await.unwrap_err();
+ assert!(error.to_string().contains("staged I/O failed"));
+ assert!(table.file_io().exists(&first.staged_path).await.unwrap());
+ assert!(!table.file_io().exists(&second.staged_path).await.unwrap());
+ builder.new_commit().abort(&messages).await.unwrap();
+ for file in [first, second] {
+ assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
+ assert!(table.file_io().exists(&file.target_path).await.unwrap());
+ }
+}
+
#[tokio::test]
async fn closing_a_writer_discards_pending_data_without_publishing() {
let table = memory_table("format_close_pending", true, &[]);
@@ -500,6 +863,7 @@ async fn
target_collision_is_rejected_before_any_overwrite_cleanup() {
.unwrap();
assert!(error.to_string().contains("target already exists"));
assert_eq!(visible_files(&table, "dt=a").await, old_files);
+ assert!(!table.file_io().exists(&file.staged_path).await.unwrap());
builder.new_commit().abort(&messages).await.unwrap();
assert_eq!(visible_files(&table, "dt=a").await, old_files);
}
@@ -536,7 +900,7 @@ async fn
format_file_message_refuses_snapshot_wire_serialization() {
let messages = write.prepare_commit().await.unwrap();
let error = messages[0].serialize().err().unwrap();
assert!(error.to_string().contains("different Java serializer"));
- builder.new_commit().abort(&messages).await.unwrap();
+ // Preserve files after publication errors; explicit abort requires
known-uncommitted messages.
}
#[tokio::test]
@@ -642,7 +1006,7 @@ async fn
output_outside_static_prefix_is_rejected_before_deleting_old_files() {
.unwrap();
assert!(error.to_string().contains("outside the static overwrite"));
assert_eq!(ids(&table).await, [1]);
- builder.new_commit().abort(&messages).await.unwrap();
+ // Preserve files after publication errors; explicit abort requires
known-uncommitted messages.
}
#[tokio::test]
diff --git a/crates/paimon/src/table/format_table_writer.rs
b/crates/paimon/src/table/format_table_writer.rs
index 0d3e10c3..09aada67 100644
--- a/crates/paimon/src/table/format_table_writer.rs
+++ b/crates/paimon/src/table/format_table_writer.rs
@@ -427,13 +427,13 @@ impl FormatTableWriter {
.await?
.size as i64;
let mut message = CommitMessage::new(key.to_vec(), 0, Vec::new());
- message.format_file = Some(FormatFileCommit {
- staged_path: file.staged_path.clone(),
- target_path: file.target_path,
- partition: partition.spec.clone(),
- record_count: file.record_count,
- file_size: size,
- });
+ message.format_file = Some(FormatFileCommit::new(
+ file.staged_path.clone(),
+ file.target_path,
+ partition.spec.clone(),
+ file.record_count,
+ size,
+ ));
self.owned_staging.remove(&file.staged_path);
self.prepared.push(message);
Ok(())
diff --git a/crates/paimon/src/table/lumina_index_build_builder/writer.rs
b/crates/paimon/src/table/lumina_index_build_builder/writer.rs
index 95e8fdda..1cb61468 100644
--- a/crates/paimon/src/table/lumina_index_build_builder/writer.rs
+++ b/crates/paimon/src/table/lumina_index_build_builder/writer.rs
@@ -120,6 +120,7 @@ pub(super) async fn abort_on_build_error<T>(
match result {
Ok(value) => Ok(value),
Err(error) => {
+ // This build has not submitted or exposed any prepared shard.
let _ = commit.abort(messages).await;
Err(error)
}
diff --git a/crates/paimon/src/table/sorted_global_index_build_builder.rs
b/crates/paimon/src/table/sorted_global_index_build_builder.rs
index d8619424..043dd3bf 100644
--- a/crates/paimon/src/table/sorted_global_index_build_builder.rs
+++ b/crates/paimon/src/table/sorted_global_index_build_builder.rs
@@ -231,6 +231,8 @@ impl<'a> SortedGlobalIndexBuildBuilder<'a> {
{
Ok(index_file) => index_file,
Err(error) => {
+ // Earlier outputs are private to this preparation; nothing
+ // has been returned to a caller or submitted for commit.
let _ = commit.abort(&messages).await;
return Err(error);
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 5ad2d51f..2b65486b 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -373,6 +373,8 @@ impl TableCommit {
let changelog_entries =
self.messages_to_changelog_entries(&commit_messages);
let new_index_entries =
self.messages_to_index_entries(&commit_messages);
let check_from_snapshot = Self::check_from_snapshot(&commit_messages)?;
+ // Never delete CommitMessage files on failure. A previous attempt may
+ // have published them even when the caller received an exception.
self.try_commit(
CommitEntriesPlan::Direct {
entries,
@@ -421,28 +423,18 @@ impl TableCommit {
let changelog_entries =
self.messages_to_changelog_entries(&commit_messages);
let new_index_entries =
self.messages_to_index_entries(&commit_messages);
let check_from_snapshot = Self::check_from_snapshot(&commit_messages)?;
- let result = self
- .try_commit(
- CommitEntriesPlan::Direct {
- entries,
- changelog_entries,
- new_index_entries,
- check_from_snapshot,
- },
- Some(expected_snapshot_id),
- commit_identifier,
- false,
- )
- .await;
- if let Err(error) = result {
- // Storage and REST errors can be indeterminate: the snapshot may
- // already reference these files even though the response failed.
- if matches!(&error, crate::Error::DataInvalid { .. }) {
- let _ = self.abort(&commit_messages).await;
- }
- return Err(error);
- }
- Ok(())
+ self.try_commit(
+ CommitEntriesPlan::Direct {
+ entries,
+ changelog_entries,
+ new_index_entries,
+ check_from_snapshot,
+ },
+ Some(expected_snapshot_id),
+ commit_identifier,
+ false,
+ )
+ .await
}
/// Overwrite partitions with new data.
@@ -882,11 +874,16 @@ impl TableCommit {
Ok(())
}
- /// Abort a prepared commit by deleting newly written data, changelog and
index files.
+ /// Delete new files for an explicitly abandoned, known-uncommitted write.
+ ///
+ /// Mirrors Java `FileStoreCommitImpl.abort`; removed inputs are preserved.
+ /// Call only when these messages will never be submitted. Never call after
+ /// a commit whose outcome is unknown: publication may have succeeded
before
+ /// its response failed. File deletion is best-effort.
///
- /// Deletion is best-effort and mirrors Python `FileStoreCommit.abort`:
missing
- /// files or storage errors are ignored so abort cleanup never masks the
- /// original write failure.
+ /// Format Tables follow Java two-phase cleanup instead: discard staging,
+ /// roll back unregistered append targets and preserve overwrite
replacements
+ /// or successfully registered append files.
pub async fn abort(&self, commit_messages: &[CommitMessage]) -> Result<()>
{
if self.table.is_format_table() {
return FormatTableCommit::new(&self.table)
@@ -3617,10 +3614,11 @@ mod tests {
include!("table_commit/parity_tests.rs");
include!("table_commit/recovery_tests.rs");
include!("table_commit/row_tracking_tests.rs");
+ include!("table_commit/abort_tests.rs");
}
#[tokio::test]
- async fn abort_still_cleans_up_for_a_query_auth_table() {
+ async fn abort_is_allowed_for_a_query_auth_table() {
let table = crate::table::query_auth_table();
let commit = crate::table::WriteBuilder::new(&table).new_commit();
commit.abort(&[]).await.unwrap();
@@ -4688,7 +4686,7 @@ mod tests {
let snapshot =
snap_manager.get_latest_snapshot().await.unwrap().unwrap();
assert_eq!(snapshot.id(), 1);
assert!(snapshot.index_manifest().is_none());
- assert!(!file_io.exists(&index_path).await.unwrap());
+ assert!(file_io.exists(&index_path).await.unwrap());
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/abort_tests.rs
b/crates/paimon/src/table/table_commit/abort_tests.rs
new file mode 100644
index 00000000..054785b6
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/abort_tests.rs
@@ -0,0 +1,121 @@
+// 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 explicit_abort_deletes_only_new_message_files() {
+ let io = test_file_io();
+ let path = "memory:/explicit-abort-new-message-files";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ let mut message = append_message("data.parquet");
+ message.new_files[0].extra_files = vec!["data.parquet.index".into()];
+ let mut blob = test_data_file("payload.blob", 10);
+ blob.external_path = Some(format!("{path}/external/payload.blob"));
+ message.new_files.push(blob);
+ message.new_changelog_files = vec![test_data_file("changelog.parquet",
10)];
+ message.compact_before = vec![test_data_file("old.parquet", 10)];
+ message.deleted_files = message.compact_before.clone();
+ message.compact_after = vec![test_data_file("compact.parquet", 10)];
+ message.compact_changelog_files =
vec![test_data_file("compact-changelog.parquet", 10)];
+ let mut hash = test_global_index_file("hash", 0, 0, 9);
+ hash.index_type = "HASH".into();
+ hash.global_index_meta = None;
+ message.new_index_files = vec![hash, test_deletion_vector_index_file("dv",
"data.parquet")];
+ message.compact_new_index_files = vec![test_global_index_file("global", 0,
0, 9)];
+ message.deleted_index_files = vec![test_global_index_file("old-index", 0,
0, 9)];
+ message.compact_deleted_index_files = message.deleted_index_files.clone();
+ let paths = [
+ "bucket-0/data.parquet", "bucket-0/data.parquet.index",
"external/payload.blob",
+ "bucket-0/changelog.parquet", "bucket-0/old.parquet",
"bucket-0/compact.parquet",
+ "bucket-0/compact-changelog.parquet", "index/hash", "index/dv",
"index/global", "index/old-index",
+ ].map(|name| format!("{path}/{name}"));
+ for file in &paths {
+
io.new_output(file).unwrap().write(bytes::Bytes::from_static(b"keep")).await.unwrap();
+ }
+ for _ in 0..2 {
+ commit.abort(std::slice::from_ref(&message)).await.unwrap();
+ for file in &paths {
+ let retained_input = file.ends_with("/old.parquet") ||
file.ends_with("/old-index");
+ assert_eq!(io.exists(file).await.unwrap(), retained_input,
"{file}");
+ }
+ }
+ assert!(latest_snapshot(&io, path).await.is_none());
+}
+
+#[tokio::test]
+async fn failed_commit_preserves_files_after_a_lost_response() {
+ for publish_first in [false, true] {
+ for identifier in [BATCH_COMMIT_IDENTIFIER, 42] {
+ let io = test_file_io();
+ let path =
format!("memory:/abort-response-loss-{publish_first}-{identifier}");
+ setup_dirs(&io, &path).await;
+ let mut commit = setup_commit(&io, &path);
+ 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,
+ });
+ let mut message = append_message("data.parquet");
+ message.new_files[0].extra_files =
vec!["data.parquet.index".into()];
+ message.new_changelog_files =
vec![test_data_file("changelog.parquet", 10)];
+ message.new_index_files = vec![test_global_index_file("global", 0,
0, 9)];
+ let paths = ["bucket-0/data.parquet",
"bucket-0/data.parquet.index", "bucket-0/changelog.parquet", "index/global"]
+ .map(|name| format!("{path}/{name}"));
+ for file in &paths {
+
io.new_output(file).unwrap().write(bytes::Bytes::from_static(b"prepared")).await.unwrap();
+ }
+ let error = commit.commit_with_identifier(vec![message.clone()],
identifier).await.unwrap_err();
+ assert!(error.to_string().contains("outcome may be unknown"),
"{error}");
+ let manifests = manifest_paths(&io, &path).await;
+ // Never explicitly abort messages with an unknown commit outcome.
+ for file in &paths {
+ assert!(io.exists(file).await.unwrap(), "{file}");
+ }
+ assert_eq!(manifest_paths(&io, &path).await, manifests);
+ assert_eq!(latest_snapshot(&io, &path).await.is_some(),
publish_first);
+ commit.filter_and_commit_with_identifier(vec![message],
identifier).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!(active_entries(&io, &path, &snapshot).await.len(), 1);
+ assert!(snapshot.changelog_manifest_list().is_some());
+ assert!(snapshot.index_manifest().is_some());
+ }
+ }
+}
+
+#[tokio::test]
+async fn guarded_retry_preserves_already_published_files() {
+ let io = test_file_io();
+ let path = "memory:/guarded-retry-preserves-files";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit.commit(vec![append_message("original")]).await.unwrap();
+ let index_path = format!("{path}/index/global");
+
io.new_output(&index_path).unwrap().write(bytes::Bytes::from_static(b"index")).await.unwrap();
+ let mut message = CommitMessage::new(vec![], 0, vec![]);
+ message.new_index_files = vec![test_global_index_file("global", 0, 0, 9)];
+ commit.commit_if_latest_snapshot(vec![message.clone()], 1).await.unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ let error = commit.commit_if_latest_snapshot(vec![message.clone()],
1).await.unwrap_err();
+ assert!(error.to_string().contains("Snapshot changed"), "{error}");
+ // A refused retry does not make already published files safe to discard.
+ assert!(io.exists(&index_path).await.unwrap());
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().id(), snapshot.id());
+ assert!(snapshot.index_manifest().is_some());
+}
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index 444c7064..e84c0053 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -125,7 +125,7 @@ impl TableUpdate {
/// Create a row-ID updater sharing one snapshot across per-call columns.
pub async fn new_update_by_row_id(&self) ->
crate::Result<TableUpdateByRowId> {
- TableUpdateByRowId::new(&self.table, self.commit_user.clone()).await
+ TableUpdateByRowId::new(&self.table).await
}
/// Update existing rows from chunks of one Arrow table containing
`_ROW_ID`.
@@ -172,11 +172,8 @@ impl TableUpdate {
.map_or_else(Vec::new, |writer|
writer.commit_messages().to_vec()))
}
.await;
- if result.is_err() {
- if let Some(writer) = &mut writer {
- let _ = writer.abort().await;
- }
- }
+ // Preserve earlier prepared groups on failure. Never delete files
+ // based on CommitMessage.
result
}
@@ -189,14 +186,8 @@ impl TableUpdate {
assignments: Vec<(String, super::UpdateAssignment)>,
read_columns: Vec<String>,
) -> crate::Result<Vec<CommitMessage>> {
- super::table_update_predicate::update(
- &self.table,
- &self.commit_user,
- predicate,
- assignments,
- read_columns,
- )
- .await
+ super::table_update_predicate::update(&self.table, predicate,
assignments, read_columns)
+ .await
}
/// Upsert complete Arrow rows by composite key through the core upsert
diff --git a/crates/paimon/src/table/table_update_by_row_id.rs
b/crates/paimon/src/table/table_update_by_row_id.rs
index 6d79e8de..c320ca60 100644
--- a/crates/paimon/src/table/table_update_by_row_id.rs
+++ b/crates/paimon/src/table/table_update_by_row_id.rs
@@ -18,7 +18,7 @@
//! Row-ID updates sharing one target snapshot and file index.
use super::data_evolution_writer::RowIdFileIndex;
-use crate::table::{CommitMessage, DataEvolutionWriter, Table, TableCommit};
+use crate::table::{CommitMessage, DataEvolutionWriter, Table};
use arrow_array::RecordBatch;
use std::collections::{HashMap, HashSet};
@@ -36,27 +36,21 @@ fn invalid(message: impl Into<String>) -> crate::Error {
/// of the same file group; updating a common column of that group is rejected.
pub struct TableUpdateByRowId {
table: Table,
- commit_user: String,
index: RowIdFileIndex,
updated: HashMap<i32, HashSet<i64>>,
messages: Vec<CommitMessage>,
}
impl TableUpdateByRowId {
- pub(crate) async fn new(table: &Table, commit_user: String) ->
crate::Result<Self> {
- Self::with_index(table, commit_user, RowIdFileIndex::load(table,
None).await?)
+ pub(crate) async fn new(table: &Table) -> crate::Result<Self> {
+ Self::with_index(table, RowIdFileIndex::load(table, None).await?)
}
- pub(super) fn with_index(
- table: &Table,
- commit_user: String,
- index: RowIdFileIndex,
- ) -> crate::Result<Self> {
+ pub(super) fn with_index(table: &Table, index: RowIdFileIndex) ->
crate::Result<Self> {
// Validate table layout without fixing any update columns.
let _ = DataEvolutionWriter::new(table, Vec::new())?;
Ok(Self {
table: table.clone(),
- commit_user,
index,
updated: HashMap::new(),
messages: Vec::new(),
@@ -120,13 +114,12 @@ impl TableUpdateByRowId {
&self.messages
}
- /// Abort staged, uncommitted files after a failed operation.
+ /// Reset update state while preserving prepared files.
pub async fn abort(&mut self) -> crate::Result<()> {
- let result = TableCommit::new(self.table.clone(),
self.commit_user.clone())
- .abort(&self.messages)
- .await;
+ // Never delete CommitMessage files. They may already be published even
+ // when the caller received an exception from commit.
self.messages.clear();
self.updated.clear();
- result
+ Ok(())
}
}
diff --git a/crates/paimon/src/table/table_update_predicate.rs
b/crates/paimon/src/table/table_update_predicate.rs
index f71b5aaa..8bf5ef8b 100644
--- a/crates/paimon/src/table/table_update_predicate.rs
+++ b/crates/paimon/src/table/table_update_predicate.rs
@@ -101,7 +101,6 @@ fn validate(
pub(super) async fn update(
table: &Table,
- commit_user: &str,
predicate: Option<Predicate>,
assignments: Vec<(String, UpdateAssignment)>,
read_columns: Vec<String>,
@@ -135,7 +134,7 @@ pub(super) async fn update(
let plan = read_builder.new_scan().with_scan_all_files().plan().await?;
let index = RowIdFileIndex::from_splits(scan_table.clone(),
plan.splits())?;
let groups = ordered_file_groups(plan.splits())?;
- let mut updater = TableUpdateByRowId::with_index(table,
commit_user.into(), index)?;
+ let mut updater = TableUpdateByRowId::with_index(table, index)?;
let reader = read_builder.new_read()?;
let schema =
crate::arrow::build_target_arrow_schema(table.schema().fields())?;
let columns: Vec<_> = assignments.iter().map(|(name, _)|
name.clone()).collect();
@@ -163,9 +162,8 @@ pub(super) async fn update(
Ok(updater.commit_messages().to_vec())
}
.await;
- if result.is_err() {
- let _ = updater.abort().await;
- }
+ // Preserve earlier prepared groups on failure. Never delete files based
+ // on CommitMessage when a callback or a later update fails.
result
}
diff --git a/crates/paimon/src/table/table_upsert.rs
b/crates/paimon/src/table/table_upsert.rs
index aa8ee45f..7860825a 100644
--- a/crates/paimon/src/table/table_upsert.rs
+++ b/crates/paimon/src/table/table_upsert.rs
@@ -260,15 +260,8 @@ impl TableUpsert {
Ok(())
}
.await;
- if result.is_err() && !messages.is_empty() {
- if let Ok(builder) = self
- .table
- .new_write_builder()
- .with_commit_user(self.commit_user)
- {
- let _ = builder.new_commit().abort(&messages).await;
- }
- }
+ // Preserve earlier prepared groups on failure. Never delete files
+ // based on CommitMessage; a failed response does not prove ownership.
result.map(|()| messages)
}
}
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index b38fed10..25f8e1df 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -1130,9 +1130,7 @@ impl TableWrite {
}
}
if let Some(error) = error {
- self.failed = true;
- let commit = super::TableCommit::new(self.table.clone(),
self.commit_user.clone());
- let _ = commit.abort(&messages).await;
+ self.fail_prepare(&messages).await;
return Err(error);
}
@@ -1142,9 +1140,7 @@ impl TableWrite {
{
Ok(files) => files,
Err(error) => {
- self.failed = true;
- let commit = super::TableCommit::new(self.table.clone(),
self.commit_user.clone());
- let _ = commit.abort(&messages).await;
+ self.fail_prepare(&messages).await;
return Err(error);
}
};
@@ -1165,6 +1161,17 @@ impl TableWrite {
Ok(messages)
}
+ async fn fail_prepare(&mut self, messages: &[CommitMessage]) {
+ self.failed = true;
+ self.close().await;
+ // Only this failed invocation's outputs: no message has been returned
+ // or submitted. Earlier successful prepare calls belong to the caller.
+ let commit = super::TableCommit::new(self.table.clone(),
self.commit_user.clone());
+ if let Err(error) = commit.abort(messages).await {
+ log::warn!("Failed to discard unreturned prepared files: {error}");
+ }
+ }
+
async fn create_writer(&mut self, partition_bytes: Vec<u8>, bucket: i32)
-> Result<()> {
let partition_path = self.resolve_partition_path(&partition_bytes)?;
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index 6b255713..e63f3658 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -182,6 +182,8 @@ impl<'a> VindexIndexBuildBuilder<'a> {
{
Ok(index_file) => index_file,
Err(error) => {
+ // Earlier outputs are private to this preparation; nothing
+ // has been returned to a caller or submitted for commit.
let _ = commit.abort(&messages).await;
return Err(error);
}
diff --git a/crates/paimon/tests/dynamic_bucket_write_test.rs
b/crates/paimon/tests/dynamic_bucket_write_test.rs
index 8e29704a..de409ddd 100644
--- a/crates/paimon/tests/dynamic_bucket_write_test.rs
+++ b/crates/paimon/tests/dynamic_bucket_write_test.rs
@@ -125,7 +125,7 @@ impl paimon::io::FileIOProvider for FailSecondHash {
async fn create(&self, path: &str) -> paimon::Result<(opendal::Operator,
String)> {
let path = path.strip_prefix("file://").unwrap_or(path);
let name = path.rsplit('/').next().unwrap();
- if name.starts_with("index-") {
+ if name.starts_with("index-") && path.contains("/index/") {
let mut indexes = self.indexes.lock().unwrap();
if !indexes.iter().any(|item| item == path) {
indexes.push(path.to_string());
@@ -200,3 +200,88 @@ async fn failed_hash_prepare_cleans_indexes_and_data() {
.to_string()
.contains("cannot be reused"));
}
+
+#[tokio::test]
+async fn failed_later_prepare_preserves_returned_and_committed_files() {
+ use paimon::catalog::Identifier;
+ use paimon::io::FileIOBuilder;
+ use paimon::table::Table;
+ use std::sync::Arc;
+
+ for publish_first in [false, true] {
+ let tmp = tempfile::tempdir().unwrap();
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some("/".into());
+ let provider = Arc::new(FailSecondHash {
+ operator: opendal::Operator::from_config(config).unwrap(),
+ indexes: Default::default(),
+ data: Default::default(),
+ });
+ let io = FileIOBuilder::new("file")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let path = tmp.path().to_str().unwrap();
+ let table = Table::new(
+ io.clone(),
+ Identifier::new("default", "later_failure"),
+ path.into(),
+ pk_schema(&[("bucket", "-1"), ("dynamic-bucket.target-row-num",
"1")]),
+ None,
+ );
+ setup_dirs(&io, path).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1], vec![10]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let returned = provider
+ .data
+ .lock()
+ .unwrap()
+ .iter()
+ .chain(provider.indexes.lock().unwrap().iter())
+ .cloned()
+ .collect::<std::collections::HashSet<_>>();
+ assert!(!returned.is_empty());
+ if publish_first {
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ }
+ writer
+ .write_arrow_batch(&make_batch(vec![2], vec![20]))
+ .await
+ .unwrap();
+ assert!(writer
+ .prepare_commit()
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("injected HASH write failure"));
+ writer.close().await;
+ for file in provider
+ .data
+ .lock()
+ .unwrap()
+ .iter()
+ .chain(provider.indexes.lock().unwrap().iter())
+ {
+ assert_eq!(
+ std::path::Path::new(file).exists(),
+ returned.contains(file),
+ "{file}"
+ );
+ }
+ if !publish_first {
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.total_record_count(), Some(1));
+ }
+}
diff --git a/crates/paimon/tests/table_update_test.rs
b/crates/paimon/tests/table_update_test.rs
index ce85b467..3908a3dc 100644
--- a/crates/paimon/tests/table_update_test.rs
+++ b/crates/paimon/tests/table_update_test.rs
@@ -654,7 +654,7 @@ async fn
grouped_updates_allow_disjoint_columns_on_the_same_file() {
}
#[tokio::test]
-async fn overlapping_groups_abort_previously_prepared_columns() {
+async fn overlapping_groups_preserve_previously_prepared_columns() {
let table = evolution_table().await;
seed(&table).await;
let before = parquet_files(&table).await;
@@ -681,7 +681,9 @@ async fn
overlapping_groups_abort_previously_prepared_columns() {
.contains("overlapping first_row_ids by column"),
"{error}"
);
- assert_eq!(parquet_files(&table).await, before);
+ let after = parquet_files(&table).await;
+ assert!(after.len() > before.len());
+ assert!(before.iter().all(|path| after.contains(path)));
assert_eq!(
read_rows(&table).await,
vec![vec![1, 10, 100], vec![2, 20, 200], vec![3, 30, 300]]
@@ -950,7 +952,61 @@ async fn
row_id_factory_shares_snapshot_and_selects_columns_per_call() {
}
#[tokio::test]
-async fn grouped_input_failure_aborts_files_and_preserves_the_cause() {
+async fn row_id_abort_preserves_prepared_and_published_updates() {
+ for published in [false, true] {
+ let table = evolution_table().await;
+ seed(&table).await;
+ let builder = table.new_write_builder();
+ let update = builder.new_update().unwrap();
+ let mut updater = update.new_update_by_row_id().await.unwrap();
+ let messages = updater
+ .update_columns(
+ vec![matched(vec![0], &[("value", vec![11])])],
+ vec!["value".into()],
+ )
+ .await
+ .unwrap();
+ let prepared_files = parquet_files(&table).await;
+ let snapshot_id = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id();
+ if published {
+ builder.new_commit().commit(messages.clone()).await.unwrap();
+ }
+ updater.abort().await.unwrap();
+ updater.abort().await.unwrap();
+ assert!(updater.commit_messages().is_empty());
+ assert_eq!(parquet_files(&table).await, prepared_files);
+ if !published {
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ snapshot_id
+ );
+ assert_eq!(
+ read_rows(&table).await,
+ vec![vec![1, 10, 100], vec![2, 20, 200], vec![3, 30, 300]]
+ );
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ assert_eq!(
+ read_rows(&table).await,
+ vec![vec![1, 11, 100], vec![2, 20, 200], vec![3, 30, 300]]
+ );
+ }
+}
+
+#[tokio::test]
+async fn grouped_input_failure_preserves_files_and_the_cause() {
let table = evolution_table().await;
seed(&table).await;
let before = parquet_files(&table).await;
@@ -966,7 +1022,9 @@ async fn
grouped_input_failure_aborts_files_and_preserves_the_cause() {
.await
.unwrap_err();
assert!(error.to_string().contains("input generator failed"));
- assert_eq!(parquet_files(&table).await, before);
+ let after = parquet_files(&table).await;
+ assert!(after.len() > before.len());
+ assert!(before.iter().all(|path| after.contains(path)));
assert_eq!(
read_rows(&table).await,
vec![vec![1, 10, 100], vec![2, 20, 200], vec![3, 30, 300]]
@@ -1146,7 +1204,7 @@ async fn predicate_array_assignments_span_file_groups() {
}
#[tokio::test]
-async fn predicate_callback_failure_aborts_earlier_group_files() {
+async fn predicate_callback_failure_preserves_earlier_group_files() {
use paimon::table::UpdateAssignment;
use std::sync::atomic::{AtomicUsize, Ordering};
let table = evolution_table().await;
@@ -1182,7 +1240,9 @@ async fn
predicate_callback_failure_aborts_earlier_group_files() {
.unwrap_err();
assert!(error.to_string().contains("second callback failed"));
assert_eq!(calls.load(Ordering::SeqCst), 2);
- assert_eq!(parquet_files(&table).await, before);
+ let after = parquet_files(&table).await;
+ assert!(after.len() > before.len());
+ assert!(before.iter().all(|path| after.contains(path)));
}
#[tokio::test]
@@ -1461,7 +1521,6 @@ async fn
row_id_nested_overlap_uses_leaf_identity_across_calls() {
.unwrap();
let child = matched(vec![1], &[("profile.a", vec![22])]);
let update = table.new_write_builder().new_update().unwrap();
- let original_files = parquet_files(&table).await;
for reverse in [false, true] {
let mut updater = update.new_update_by_row_id().await.unwrap();
let calls = if reverse {
@@ -1483,7 +1542,7 @@ async fn
row_id_nested_overlap_uses_leaf_identity_across_calls() {
assert_eq!(updater.commit_messages().len(), messages);
assert_eq!(parquet_files(&table).await, staged_files);
updater.abort().await.unwrap();
- assert_eq!(parquet_files(&table).await, original_files);
+ assert_eq!(parquet_files(&table).await, staged_files);
}
// Disjoint sibling leaves of the same file group remain valid.
let mut updater = update.new_update_by_row_id().await.unwrap();
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 85e6487a..8252cbd8 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -186,9 +186,19 @@ When supported indexes are configured, each data file gets
its own index.
The complete serialized index is embedded in the manifest when its size
is at most `file-index.in-manifest-threshold` (default `500 B`); larger indexes
are stored beside the data file as a `.index` sidecar. Indexed write failures
-return an error and clean up newly created files on a best-effort basis.
-Use commit `abort` to clean up files after a successful `prepare_commit`
-when the prepared write will not be committed.
+return an error. Failed preparation discards outputs from that invocation
before
+returning any messages. Files handed off by an earlier successful preparation
+remain available, including after a later preparation or Paimon Table commit
+exception. Explicit commit `abort` deletes newly written files only when the
caller
+knows those messages are uncommitted and will never be submitted. Never call it
+when the Paimon Table commit outcome is unknown: publication can succeed even
when
+its response fails.
+
+Format Tables follow Java's two-phase publication: validation failures discard
+staging only; append publication or partition-registration failures roll back
new
+targets; overwrite failures retain replacement targets after old data is
removed.
+Successfully registered append targets also survive a later abort. Cleanup
attempts
+every file, and a cleanup failure does not replace the original commit error.
`file-index.read.enabled` controls only reading, independently of index
creation.
Existing files are not backfilled. Index generation is not supported for
