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 fb5e8f7e fix: prevent concurrent snapshot overwrites (#1231)
fb5e8f7e is described below
commit fb5e8f7efeb592fab916b8ee8873405a947e2f17
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 27 16:57:07 2026 +0800
fix: prevent concurrent snapshot overwrites (#1231)
---
crates/integrations/datafusion/src/procedures.rs | 140 +++++++++---
crates/paimon/src/io/file_io.rs | 212 +++++++++++++++++
crates/paimon/src/io/file_io/fuse.rs | 19 ++
crates/paimon/src/table/branch_write_tests.rs | 11 +-
crates/paimon/src/table/snapshot_commit.rs | 4 +-
crates/paimon/src/table/snapshot_manager.rs | 252 ++++++++++++++++++---
.../src/table/table_commit/recovery_tests.rs | 113 +++++++++
crates/paimon/tests/dynamic_bucket_write_test.rs | 18 +-
docs/src/getting-started.md | 18 ++
9 files changed, 713 insertions(+), 74 deletions(-)
diff --git a/crates/integrations/datafusion/src/procedures.rs
b/crates/integrations/datafusion/src/procedures.rs
index 6b1d0b28..99439c35 100644
--- a/crates/integrations/datafusion/src/procedures.rs
+++ b/crates/integrations/datafusion/src/procedures.rs
@@ -823,6 +823,30 @@ async fn clean_larger_than(
Ok(())
}
+async fn rollback_to_snapshot(
+ sm: &SnapshotManager,
+ tm: &TagManager,
+ snapshot: &Snapshot,
+) -> DFResult<()> {
+ let id = snapshot.id();
+ // Restore expired tagged snapshots before removing any newer history. A
+ // backend may reject publication, in which case rollback must not delete
it.
+ if !sm
+ .file_io()
+ .exists(&sm.snapshot_path(id))
+ .await
+ .map_err(to_datafusion_error)?
+ {
+ sm.commit_snapshot(snapshot)
+ .await
+ .map_err(to_datafusion_error)?;
+ sm.write_earliest_hint(id)
+ .await
+ .map_err(to_datafusion_error)?;
+ }
+ clean_larger_than(sm, tm, id).await
+}
+
async fn proc_rollback_to(
ctx: &SessionContext,
catalog: &Arc<dyn Catalog>,
@@ -869,41 +893,14 @@ async fn proc_rollback_to(
.parse()
.map_err(|_| DataFusionError::Plan(format!("Invalid
snapshot_id: '{id_str}'")))?;
let snapshot = resolve_snapshot_by_id(&sm, &tm, id).await?;
- clean_larger_than(&sm, &tm, id).await?;
- if !sm
- .file_io()
- .exists(&sm.snapshot_path(id))
- .await
- .map_err(to_datafusion_error)?
- {
- sm.commit_snapshot(&snapshot)
- .await
- .map_err(to_datafusion_error)?;
- sm.write_earliest_hint(id)
- .await
- .map_err(to_datafusion_error)?;
- }
+ rollback_to_snapshot(&sm, &tm, &snapshot).await?;
} else if let Some(tag_name) = args.get("tag") {
let snapshot = tm
.get(tag_name)
.await
.map_err(to_datafusion_error)?
.ok_or_else(|| DataFusionError::Plan(format!("Tag '{tag_name}'
does not exist")))?;
- let snapshot_id = snapshot.id();
- clean_larger_than(&sm, &tm, snapshot_id).await?;
- if !sm
- .file_io()
- .exists(&sm.snapshot_path(snapshot_id))
- .await
- .map_err(to_datafusion_error)?
- {
- sm.commit_snapshot(&snapshot)
- .await
- .map_err(to_datafusion_error)?;
- sm.write_earliest_hint(snapshot_id)
- .await
- .map_err(to_datafusion_error)?;
- }
+ rollback_to_snapshot(&sm, &tm, &snapshot).await?;
} else {
return Err(DataFusionError::Plan(
"rollback_to requires either 'snapshot_id' or 'tag'
argument".to_string(),
@@ -1955,6 +1952,91 @@ mod tests {
assert!(result.is_err());
}
+ #[derive(Debug)]
+ struct RejectSnapshotPublication(opendal::Operator);
+
+ #[async_trait::async_trait]
+ impl paimon::io::FileIOProvider for RejectSnapshotPublication {
+ async fn create(&self, path: &str) ->
paimon::Result<(opendal::Operator, String)> {
+ Ok((
+ self.0.clone(),
+ path.strip_prefix("memory:/").unwrap().into(),
+ ))
+ }
+
+ async fn write_atomic_if_not_exists(
+ &self,
+ _path: &str,
+ _bytes: bytes::Bytes,
+ ) -> paimon::Result<Option<bool>> {
+ Err(paimon::Error::IoUnsupported {
+ message: "snapshot publication unavailable".into(),
+ })
+ }
+ }
+
+ #[tokio::test]
+ async fn test_rollback_publication_failure_preserves_history() {
+ let op =
opendal::Operator::new(opendal::services::Memory::default()).unwrap();
+ let file_io = FileIOBuilder::new("memory")
+ .with_provider(Arc::new(RejectSnapshotPublication(op)))
+ .build()
+ .unwrap();
+ let path = "memory:/rollback_failure";
+ let sm = SnapshotManager::new(file_io.clone(), path.into());
+ let tm = TagManager::new(file_io.clone(), path.into());
+ // Snapshot 1 has expired, but a tag still retains it. Seed the newer
+ // history through ordinary writes on a backend rejecting publication.
+ for id in [2, 3] {
+ file_io
+ .new_output(&sm.snapshot_path(id))
+ .unwrap()
+ .write(
+ serde_json::to_vec(&test_snapshot(id, id as u64 * 1000))
+ .unwrap()
+ .into(),
+ )
+ .await
+ .unwrap();
+ }
+ sm.write_latest_hint(3).await.unwrap();
+ tm.create("v1", &test_snapshot(1, 1000)).await.unwrap();
+ tm.create("v3", &test_snapshot(3, 3000)).await.unwrap();
+ let target = resolve_snapshot_by_id(&sm, &tm, 1).await.unwrap();
+
+ let error = rollback_to_snapshot(&sm, &tm, &target).await.unwrap_err();
+ assert!(error
+ .to_string()
+ .contains("snapshot publication unavailable"));
+ assert_eq!(sm.list_all_ids().await.unwrap(), vec![2, 3]);
+ assert_eq!(sm.get_latest_snapshot_id().await.unwrap(), Some(3));
+ assert!(tm.tag_exists("v1").await.unwrap());
+ assert!(tm.tag_exists("v3").await.unwrap());
+ assert!(!file_io.exists(&sm.snapshot_path(1)).await.unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_rollback_restores_expired_snapshot_before_cleaning_history()
{
+ let (_, sm, tm) = setup("memory:/rollback_expired").await;
+ for id in [2, 3] {
+ sm.commit_snapshot(&test_snapshot(id, id as u64 * 1000))
+ .await
+ .unwrap();
+ }
+ let target = test_snapshot(1, 1000);
+ tm.create("v1", &target).await.unwrap();
+ tm.create("v3", &test_snapshot(3, 3000)).await.unwrap();
+
+ rollback_to_snapshot(&sm, &tm, &target).await.unwrap();
+
+ assert_eq!(sm.get_snapshot(1).await.unwrap(), target);
+ assert_eq!(sm.list_all_ids().await.unwrap(), vec![1]);
+ assert_eq!(sm.get_latest_snapshot_id().await.unwrap(), Some(1));
+ assert_eq!(sm.earliest_snapshot_id().await.unwrap(), Some(1));
+ assert!(tm.tag_exists("v1").await.unwrap());
+ assert!(!tm.tag_exists("v3").await.unwrap());
+ }
+
#[tokio::test]
async fn test_later_or_equal_exact_live() {
let (_, sm, tm) = setup("memory:/test_later_exact").await;
diff --git a/crates/paimon/src/io/file_io.rs b/crates/paimon/src/io/file_io.rs
index acf9a2f4..79779c3a 100644
--- a/crates/paimon/src/io/file_io.rs
+++ b/crates/paimon/src/io/file_io.rs
@@ -87,6 +87,20 @@ pub trait FileIOProvider: std::fmt::Debug + Send + Sync +
'static {
/// Rename requires both paths to resolve to the same shared service
instance.
async fn create(&self, path: &str) -> crate::Result<(Operator, String)>;
+ /// Optionally publish a complete immutable file without replacing an
existing file.
+ ///
+ /// Return `Some(true)` on success or `Some(false)` for a definite
conflict.
+ /// Return `None` only if no publication was attempted; FileIO then uses
the
+ /// resolved operator's safe publication primitives. An error may have an
+ /// unknown outcome and is propagated without falling back to another
write.
+ async fn write_atomic_if_not_exists(
+ &self,
+ _path: &str,
+ _bytes: Bytes,
+ ) -> crate::Result<Option<bool>> {
+ Ok(None)
+ }
+
/// Return a stable namespace for the resolved storage target. `None`
/// disables caching; distinct endpoints must use distinct namespaces.
async fn create_with_cache_namespace(
@@ -923,6 +937,137 @@ impl FileIO {
Ok(())
}
+ /// Publish a complete immutable file only if the target does not exist.
+ /// Returns `false` only for a definite publication conflict. Other
failures
+ /// may have an unknown outcome and must never trigger an unconditional
write.
+ pub(crate) async fn write_atomic_if_not_exists(
+ &self,
+ path: &str,
+ bytes: Bytes,
+ ) -> Result<bool> {
+ if let FileIOBackend::Provider(provider) = &self.backend {
+ if let Some(published) = provider
+ .write_atomic_if_not_exists(path, bytes.clone())
+ .await?
+ {
+ return Ok(published);
+ }
+ }
+ let source = self.file_source(path)?;
+ let (op, relative_path, _) = source.resolve(path).await?;
+
+ #[cfg(feature = "storage-fs")]
+ if matches!(source, FileSource::Static { local_fs: true, .. }) {
+ return Self::write_local_atomic_if_not_exists(path, &op,
&relative_path, bytes).await;
+ }
+
+ let info = op.info();
+ let capability = info.capability();
+ let result = if capability.rename_with_if_not_exists {
+ let staging_path = format!("{relative_path}.tmp-{}",
uuid::Uuid::new_v4());
+ op.write(&staging_path, bytes).await?;
+ let result = op
+ .rename_with(&staging_path, &relative_path)
+ .if_not_exists(true)
+ .await;
+ if result.as_ref().is_err_and(|error| {
+ matches!(
+ error.kind(),
+ opendal::ErrorKind::AlreadyExists |
opendal::ErrorKind::ConditionNotMatch
+ )
+ }) {
+ let _ = op.delete(&staging_path).await;
+ }
+ result
+ } else if capability.write_with_if_not_exists
+ // Only known whole-object publishers are safe here: position
+ // writers can expose incomplete files, and COS ignores conditional
+ // puts on buckets that have ever enabled versioning. OSS requires
+ // the bucket's default, never-enabled versioning configuration.
+ && matches!(info.scheme(), "memory" | "s3" | "oss" | "gcs")
+ {
+ op.write_with(&relative_path, bytes)
+ .if_not_exists(true)
+ .await
+ .map(|_| ())
+ } else {
+ return Err(Error::IoUnsupported {
+ message: format!(
+ "Storage '{}' cannot atomically publish a complete
snapshot without overwriting '{path}'; use a catalog SnapshotCommit with
concurrency control",
+ info.scheme()
+ ),
+ });
+ };
+ match result {
+ Ok(()) => Ok(true),
+ Err(error)
+ if matches!(
+ error.kind(),
+ opendal::ErrorKind::AlreadyExists |
opendal::ErrorKind::ConditionNotMatch
+ ) =>
+ {
+ Ok(false)
+ }
+ Err(error) => Err(Error::from_opendal_with_context(
+ error,
+ format!("Failed to publish '{path}'"),
+ )),
+ }
+ }
+
+ #[cfg(feature = "storage-fs")]
+ async fn write_local_atomic_if_not_exists(
+ path: &str,
+ op: &Operator,
+ relative_path: &str,
+ bytes: Bytes,
+ ) -> Result<bool> {
+ // OpenDAL's filesystem conditional write exposes the target before
+ // its contents are complete. Stage locally, then publish without
+ // replacing another process's snapshot.
+ let relative = normalize_path(relative_path);
+ if std::path::Path::new(&relative)
+ .components()
+ .any(|component| matches!(component,
std::path::Component::ParentDir))
+ {
+ return Err(Error::IoUnsupported {
+ message: format!("Cannot publish a snapshot through a parent
path: '{path}'"),
+ });
+ }
+ let target = std::path::Path::new(&op.info().root()).join(relative);
+ tokio::task::spawn_blocking(move || -> opendal::Result<bool> {
+ use std::io::Write;
+
+ let parent = target.parent().ok_or_else(|| {
+ opendal::Error::new(opendal::ErrorKind::IsADirectory, "target
has no parent")
+ })?;
+ std::fs::create_dir_all(parent).map_err(new_std_io_error)?;
+ let mut builder = tempfile::Builder::new();
+ builder.prefix(".snapshot-");
+ #[cfg(unix)]
+ {
+ use std::os::unix::fs::PermissionsExt;
+ builder.permissions(std::fs::Permissions::from_mode(0o666));
+ }
+ let mut staged =
builder.tempfile_in(parent).map_err(new_std_io_error)?;
+ staged.write_all(&bytes).map_err(new_std_io_error)?;
+ staged.as_file().sync_all().map_err(new_std_io_error)?;
+ match staged.persist_noclobber(target) {
+ Ok(_) => Ok(true),
+ Err(error) if error.error.kind() ==
std::io::ErrorKind::AlreadyExists => Ok(false),
+ Err(error) => Err(new_std_io_error(error.error)),
+ }
+ })
+ .await
+ .map_err(|error| Error::UnexpectedError {
+ message: format!("Failed to publish '{path}'"),
+ source: Some(Box::new(error)),
+ })?
+ .map_err(|error| {
+ Error::from_opendal_with_context(error, format!("Failed to publish
'{path}'"))
+ })
+ }
+
/// Copy a large file without materializing its entire contents in memory.
/// Format Table publication uses this when a backend does not support
/// rename, as is common for object stores and the in-memory test backend.
@@ -1158,6 +1303,9 @@ impl FileIOBuilder {
/// form, so the operator's root decides what they resolve against.
Scheme'd paths
/// (`s3://…`, `oss://…`) are rejected rather than misresolved: an
object-store operator
/// needs bucket/scheme resolution this hook deliberately does not provide.
+ /// Snapshot publication requires atomic no-replace publication from the
operator.
+ /// Filesystem conditional writes alone expose incomplete files; use the
built-in
+ /// `file` backend or a catalog SnapshotCommit when conditional rename is
unavailable.
pub fn with_fs_operator(mut self, operator: Operator) -> Self {
self.operator = Some(operator);
self
@@ -4185,4 +4333,68 @@ mod input_output_test {
Bytes::from_static(b"new metadata")
);
}
+
+ #[cfg(feature = "storage-oss")]
+ #[tokio::test]
+ async fn test_atomic_publication_oss_has_one_winner() {
+ use axum::body::Body;
+ use axum::http::{HeaderMap, Response, StatusCode, Uri};
+ use axum::Router;
+ use opendal::Configurator;
+
+ let objects = Arc::new(Mutex::new(HashMap::<String, Bytes>::new()));
+ let stored = objects.clone();
+ // An OSS bucket with default versioning atomically enforces this
header.
+ let app = Router::new().fallback(move |uri: Uri, headers: HeaderMap,
body: Bytes| {
+ let stored = stored.clone();
+ async move {
+ let mut objects = stored.lock().unwrap();
+ if headers.get("x-oss-forbid-overwrite").is_some_and(|value|
value == "true")
+ && objects.contains_key(uri.path())
+ {
+ return Response::builder()
+ .status(StatusCode::CONFLICT)
+
.body(Body::from("<Error><Code>FileAlreadyExists</Code><Message>exists</Message></Error>"))
+ .unwrap();
+ }
+ objects.insert(uri.path().into(), body);
+ Response::builder()
+ .header("etag", "\"mock-etag\"")
+ .body(Body::empty())
+ .unwrap()
+ }
+ });
+ let listener =
tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let endpoint = format!("http://{}";, listener.local_addr().unwrap());
+ let server = tokio::spawn(async move { axum::serve(listener,
app).await.unwrap() });
+ let mut config = opendal_service_oss::OssConfig::default();
+ config.endpoint = Some(endpoint);
+ config.addressing_style = Some("path".into());
+ config.skip_signature = true;
+ let op = crate::io::with_http_transport(
+ Operator::new(config.into_builder().bucket("bucket")).unwrap(),
+ "paimon-atomic-publication-test",
+ );
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(op)
+ .build()
+ .unwrap();
+ let path = "/table/snapshot/snapshot-1";
+ let first = Bytes::from_static(b"first snapshot");
+ let second = Bytes::from_static(b"second snapshot");
+ let (a, b) = tokio::join!(
+ io.write_atomic_if_not_exists(path, first.clone()),
+ io.write_atomic_if_not_exists(path, second.clone())
+ );
+ server.abort();
+ let (a, b) = (a.unwrap(), b.unwrap());
+ assert_ne!(a, b, "exactly one conditional OSS PUT must win");
+ assert_eq!(
+ objects
+ .lock()
+ .unwrap()
+ .get("/bucket/table/snapshot/snapshot-1"),
+ Some(if a { &first } else { &second })
+ );
+ }
}
diff --git a/crates/paimon/src/io/file_io/fuse.rs
b/crates/paimon/src/io/file_io/fuse.rs
index d0a287db..27c03963 100644
--- a/crates/paimon/src/io/file_io/fuse.rs
+++ b/crates/paimon/src/io/file_io/fuse.rs
@@ -35,6 +35,25 @@ impl FileIOProvider for FuseFileIO {
self.fallback.create(path).await
}
+ async fn write_atomic_if_not_exists(
+ &self,
+ path: &str,
+ bytes: Bytes,
+ ) -> crate::Result<Option<bool>> {
+ if let Some(relative) = self.relative_path(path) {
+ // Require the mount to implement no-replace publication. An
+ // unsupported hard link must never fall back to copy or rename.
+ FileIO::write_local_atomic_if_not_exists(path, &self.mounted,
relative, bytes)
+ .await
+ .map(Some)
+ } else {
+ self.fallback
+ .write_atomic_if_not_exists(path, bytes)
+ .await
+ .map(Some)
+ }
+ }
+
async fn create_with_cache_namespace(
&self,
path: &str,
diff --git a/crates/paimon/src/table/branch_write_tests.rs
b/crates/paimon/src/table/branch_write_tests.rs
index 115a3557..2ce70d95 100644
--- a/crates/paimon/src/table/branch_write_tests.rs
+++ b/crates/paimon/src/table/branch_write_tests.rs
@@ -24,14 +24,7 @@ use std::sync::Arc;
async fn table(primary_key: bool, options: &[(&str, &str)]) ->
(tempfile::TempDir, Table) {
let dir = tempfile::tempdir().unwrap();
- // Scope the real filesystem operator to this temporary directory.
OpenDAL's
- // default root "/" cannot list an absolute path on another Windows drive.
- let mut config = opendal_service_fs::FsConfig::default();
- config.root = Some(dir.path().to_str().unwrap().to_string());
- let file_io = FileIOBuilder::new("file")
- .with_fs_operator(opendal::Operator::from_config(config).unwrap())
- .build()
- .unwrap();
+ let file_io = FileIOBuilder::new("file").build().unwrap();
let mut builder = Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("value", DataType::Int(IntType::new()));
@@ -44,7 +37,7 @@ async fn table(primary_key: bool, options: &[(&str, &str)])
-> (tempfile::TempDi
let table = Table::new(
file_io.clone(),
Identifier::new("db", "t"),
- "file:/db.db/t".into(),
+ dir.path().join("db.db/t").to_str().unwrap().into(),
TableSchema::new(0, &builder.build().unwrap()),
None,
);
diff --git a/crates/paimon/src/table/snapshot_commit.rs
b/crates/paimon/src/table/snapshot_commit.rs
index 7f646090..7d49a62b 100644
--- a/crates/paimon/src/table/snapshot_commit.rs
+++ b/crates/paimon/src/table/snapshot_commit.rs
@@ -30,7 +30,7 @@ use std::sync::Arc;
/// Interface to commit a snapshot atomically.
///
/// Two implementations:
-/// - `RenamingSnapshotCommit` — file system atomic rename
+/// - `RenamingSnapshotCommit` — atomic no-replace file publication
/// - `RESTSnapshotCommit` — via Catalog API (e.g. REST)
#[async_trait]
pub trait SnapshotCommit: Send + Sync {
@@ -46,7 +46,7 @@ pub trait SnapshotCommit: Send + Sync {
) -> Result<bool>;
}
-/// A SnapshotCommit using file renaming to commit.
+/// A SnapshotCommit using atomic file publication without replacing an
existing snapshot.
///
/// Reference: [Java
RenamingSnapshotCommit](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java)
pub struct RenamingSnapshotCommit {
diff --git a/crates/paimon/src/table/snapshot_manager.rs
b/crates/paimon/src/table/snapshot_manager.rs
index 1ca5344d..0c536874 100644
--- a/crates/paimon/src/table/snapshot_manager.rs
+++ b/crates/paimon/src/table/snapshot_manager.rs
@@ -268,10 +268,9 @@ impl SnapshotManager {
/// Writes the snapshot JSON to the target path. Returns `false` if the
/// target already exists (another writer won the race).
///
- /// On file systems that support atomic rename, we write to a temp file
- /// first then rename. On backends where rename is not supported (e.g.
- /// memory, object stores), we fall back to a direct write after an
- /// existence check.
+ /// Uses atomic no-replace publication: a staged file on the local
filesystem,
+ /// conditional rename, or a conditional object write. Backends without a
+ /// safe primitive must use a catalog-provided SnapshotCommit instead.
pub async fn commit_snapshot(&self, snapshot: &Snapshot) ->
crate::Result<bool> {
let target_path = self.snapshot_path(snapshot.id());
@@ -280,37 +279,14 @@ impl SnapshotManager {
source: Some(Box::new(e)),
})?;
- // Try rename-based atomic commit first, fall back to check-and-write.
- //
- // TODO: opendal's rename uses POSIX semantics which silently
overwrites the target.
- // The exists() check below narrows the race window but does not
eliminate it.
- // Java Paimon uses `lock.runWithLock(() -> !fileIO.exists(newPath)
&& callable.call())`
- // for full mutual exclusion. We need an external lock mechanism
(like Java's Lock
- // interface) for backends without atomic rename-no-replace support.
- let tmp_path = format!("{}.tmp-{}", target_path, uuid::Uuid::new_v4());
- let output = self.file_io.new_output(&tmp_path)?;
- output.write(bytes::Bytes::from(json.clone())).await?;
-
- // Check before rename to avoid silent overwrite (opendal uses POSIX
rename semantics)
- if self.file_io.exists(&target_path).await? {
- let _ = self.file_io.delete_file(&tmp_path).await;
+ if !self
+ .file_io
+ .write_atomic_if_not_exists(&target_path, bytes::Bytes::from(json))
+ .await?
+ {
return Ok(false);
}
- match self.file_io.rename(&tmp_path, &target_path).await {
- Ok(()) => {}
- Err(_) => {
- // Rename not supported (e.g. memory/object store).
- // Clean up temp file, then check-and-write.
- let _ = self.file_io.delete_file(&tmp_path).await;
- if self.file_io.exists(&target_path).await? {
- return Ok(false);
- }
- let output = self.file_io.new_output(&target_path)?;
- output.write(bytes::Bytes::from(json)).await?;
- }
- }
-
// Update LATEST hint (best-effort)
let _ = self.write_latest_hint(snapshot.id()).await;
Ok(true)
@@ -829,6 +805,218 @@ mod tests {
assert!(!result);
}
+ #[derive(Debug)]
+ struct SnapshotPublicationProvider {
+ op: opendal::Operator,
+ barrier: tokio::sync::Barrier,
+ synchronize: std::sync::atomic::AtomicBool,
+ reject: bool,
+ }
+
+ #[async_trait::async_trait]
+ impl crate::io::FileIOProvider for SnapshotPublicationProvider {
+ async fn create(&self, path: &str) ->
crate::Result<(opendal::Operator, String)> {
+ let relative = path.strip_prefix("memory:/").unwrap();
+ // Synchronize every resolution of the publication path. In the old
+ // check-then-write implementation both writers pass each check
+ // before either reaches its unconditional write.
+ if relative.ends_with("/snapshot-1")
+ && self.synchronize.load(std::sync::atomic::Ordering::SeqCst)
+ {
+ self.barrier.wait().await;
+ }
+ Ok((self.op.clone(), relative.to_string()))
+ }
+
+ async fn write_atomic_if_not_exists(
+ &self,
+ _: &str,
+ _: bytes::Bytes,
+ ) -> crate::Result<Option<bool>> {
+ if self.reject {
+ return Err(crate::Error::IoUnsupported {
+ message: "provider rejected snapshot publication".into(),
+ });
+ }
+ Ok(None)
+ }
+ }
+
+ #[tokio::test]
+ async fn test_commit_snapshot_concurrent_publication_has_one_winner() {
+ use std::sync::atomic::{AtomicBool, Ordering};
+ use std::sync::Arc;
+
+ let provider = Arc::new(SnapshotPublicationProvider {
+ op:
opendal::Operator::from_config(opendal::services::MemoryConfig::default()).unwrap(),
+ barrier: tokio::sync::Barrier::new(2),
+ synchronize: AtomicBool::new(true),
+ reject: false,
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let manager = SnapshotManager::new(io,
"memory:/racing-snapshot".into());
+ let first = test_snapshot(1);
+ let mut json = serde_json::to_value(&first).unwrap();
+ json["commitUser"] = "other-writer".into();
+ let second: Snapshot = serde_json::from_value(json).unwrap();
+ let (a, b) = tokio::time::timeout(std::time::Duration::from_secs(5),
async {
+ tokio::join!(
+ manager.commit_snapshot(&first),
+ manager.commit_snapshot(&second)
+ )
+ })
+ .await
+ .expect("publication must not deadlock");
+ provider.synchronize.store(false, Ordering::SeqCst);
+ let (a, b) = (a.unwrap(), b.unwrap());
+ assert_ne!(a, b, "exactly one writer must win the same snapshot ID");
+ assert_eq!(
+ manager.get_snapshot(1).await.unwrap(),
+ if a { first } else { second }
+ );
+ assert_eq!(manager.get_latest_snapshot_id().await.unwrap(), Some(1));
+ let files = manager
+ .file_io
+ .list_status(&manager.snapshot_dir())
+ .await
+ .unwrap();
+ assert!(files.iter().all(|file| !file.path.contains(".tmp-")));
+ }
+
+ #[tokio::test]
+ async fn
test_commit_snapshot_provider_error_never_falls_back_to_operator() {
+ let provider = std::sync::Arc::new(SnapshotPublicationProvider {
+ op:
opendal::Operator::from_config(opendal::services::MemoryConfig::default()).unwrap(),
+ barrier: tokio::sync::Barrier::new(2),
+ synchronize: std::sync::atomic::AtomicBool::new(false),
+ reject: true,
+ });
+ let io = FileIOBuilder::new("memory")
+ .with_provider(provider.clone())
+ .build()
+ .unwrap();
+ let manager = SnapshotManager::new(io,
"memory:/provider-error".into());
+ let error = manager
+ .commit_snapshot(&test_snapshot(1))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, crate::Error::IoUnsupported { ref message } if
message == "provider rejected snapshot publication")
+ );
+ assert!(provider.op.list("/").await.unwrap().is_empty());
+ }
+
+ #[cfg(feature = "storage-fs")]
+ #[tokio::test]
+ async fn test_commit_snapshot_local_concurrent_independent_writers() {
+ let dir = tempfile::tempdir().unwrap();
+ let table_path = dir.path().to_str().unwrap().to_string();
+ let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(16));
+ let results = futures::future::join_all((0..16).map(|writer| {
+ let table_path = table_path.clone();
+ let barrier = barrier.clone();
+ async move {
+ // Separate FileIO instances also cover writers in other
processes:
+ // publication must not rely on a mutex owned by a committer.
+ let io = FileIOBuilder::new("file").build().unwrap();
+ let manager = SnapshotManager::new(io, table_path);
+ let mut json = serde_json::to_value(test_snapshot(1)).unwrap();
+ json["commitUser"] = format!("writer-{writer}").into();
+ let snapshot: Snapshot = serde_json::from_value(json).unwrap();
+ barrier.wait().await;
+ let won = manager.commit_snapshot(&snapshot).await.unwrap();
+ (snapshot, won)
+ }
+ }))
+ .await;
+ let winners = results.iter().filter(|(_, won)|
*won).collect::<Vec<_>>();
+ assert_eq!(winners.len(), 1);
+ let io = FileIOBuilder::new("file").build().unwrap();
+ let manager = SnapshotManager::new(io.clone(), table_path);
+ assert_eq!(manager.get_snapshot(1).await.unwrap(), winners[0].0);
+ assert_eq!(manager.get_latest_snapshot_id().await.unwrap(), Some(1));
+ assert_eq!(
+ io.list_status(&manager.snapshot_dir()).await.unwrap().len(),
+ 2
+ );
+ assert!(!manager.commit_snapshot(&test_snapshot(1)).await.unwrap());
+ assert_eq!(manager.get_snapshot(1).await.unwrap(), winners[0].0);
+ }
+
+ #[tokio::test]
+ async fn test_commit_snapshot_refuses_non_atomic_backend_without_writing()
{
+ let op =
opendal::Operator::from_config(opendal::services::MemoryConfig::default())
+ .unwrap()
+ .layer(opendal::layers::CapabilityOverrideLayer::new(|mut cap| {
+ cap.write_with_if_not_exists = false;
+ cap
+ }));
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(op.clone())
+ .build()
+ .unwrap();
+ let manager = SnapshotManager::new(io, "/unsafe-snapshot".into());
+ let error = manager
+ .commit_snapshot(&test_snapshot(1))
+ .await
+ .unwrap_err();
+ assert!(matches!(error, crate::Error::IoUnsupported { .. }));
+ assert!(error.to_string().contains("SnapshotCommit"), "{error}");
+ assert!(
+ op.list("/").await.unwrap().is_empty(),
+ "unsafe backends must not receive even a staging write"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_commit_snapshot_rename_error_never_falls_back_to_overwrite()
{
+ let op =
opendal::Operator::from_config(opendal::services::MemoryConfig::default())
+ .unwrap()
+ .layer(opendal::layers::CapabilityOverrideLayer::new(|mut cap| {
+ // Advertise rename but inject an Unsupported error from the
+ // underlying memory service. Conditional put remains
available.
+ cap.rename = true;
+ cap.rename_with_if_not_exists = true;
+ cap
+ }));
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(op.clone())
+ .build()
+ .unwrap();
+ let manager = SnapshotManager::new(io.clone(),
"/failed-rename".into());
+ let error = manager
+ .commit_snapshot(&test_snapshot(1))
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, crate::Error::IoUnexpected { ref source, .. } if
source.kind() == opendal::ErrorKind::Unsupported)
+ );
+ assert!(!io.exists(&manager.snapshot_path(1)).await.unwrap());
+ assert!(!io.exists(&manager.latest_hint_path()).await.unwrap());
+ }
+
+ #[cfg(feature = "storage-fs")]
+ #[tokio::test]
+ async fn
test_commit_snapshot_refuses_custom_fs_without_conditional_rename() {
+ let dir = tempfile::tempdir().unwrap();
+ let mut config = opendal_service_fs::FsConfig::default();
+ config.root = Some(dir.path().to_str().unwrap().into());
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(opendal::Operator::from_config(config).unwrap())
+ .build()
+ .unwrap();
+ let manager = SnapshotManager::new(io, "file:/custom-table".into());
+ let error = manager
+ .commit_snapshot(&test_snapshot(1))
+ .await
+ .unwrap_err();
+ assert!(matches!(error, crate::Error::IoUnsupported { .. }));
+ assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);
+ }
+
#[tokio::test]
async fn test_commit_updates_latest_hint() {
let (_, sm) = setup("memory:/test_commit_hint").await;
diff --git a/crates/paimon/src/table/table_commit/recovery_tests.rs
b/crates/paimon/src/table/table_commit/recovery_tests.rs
index fbe1d67f..8210166d 100644
--- a/crates/paimon/src/table/table_commit/recovery_tests.rs
+++ b/crates/paimon/src/table/table_commit/recovery_tests.rs
@@ -36,6 +36,119 @@ async fn empty_unpartitioned_overwrite_clears_old_rows() {
);
}
+struct RacingTablePublisher {
+ manager: SnapshotManager,
+ barrier: tokio::sync::Barrier,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for RacingTablePublisher {
+ async fn commit(
+ &self,
+ _: Option<&str>,
+ snapshot: &Snapshot,
+ _: &str,
+ _: &[PartitionStatistics],
+ ) -> Result<bool> {
+ if snapshot.id() == 1 {
+ self.barrier.wait().await;
+ }
+ self.manager.commit_snapshot(snapshot).await
+ }
+}
+
+#[tokio::test]
+async fn concurrent_appends_rebase_without_losing_files_or_reusing_row_ids() {
+ let io = test_file_io();
+ let path = "memory:/concurrent-append-publication";
+ setup_dirs(&io, path).await;
+ let table = test_table_with_options(
+ &io,
+ path,
+ HashMap::from([("row-tracking.enabled".into(), "true".into())]),
+ );
+ let publisher = Arc::new(RacingTablePublisher {
+ manager: table.snapshot_manager(),
+ barrier: tokio::sync::Barrier::new(2),
+ });
+ let mut first = TableCommit::new(table.clone(), "writer-a".into());
+ first.snapshot_commit = publisher.clone();
+ first.commit_min_retry_wait_ms = 0;
+ first.commit_max_retry_wait_ms = 0;
+ let mut second = TableCommit::new(table, "writer-b".into());
+ second.snapshot_commit = publisher;
+ second.commit_min_retry_wait_ms = 0;
+ second.commit_max_retry_wait_ms = 0;
+ let mut a = append_message("a.parquet");
+ let mut b = append_message("b.parquet");
+ a.new_files[0].file_source = Some(0);
+ b.new_files[0].file_source = Some(0);
+ let (a, b) = tokio::time::timeout(std::time::Duration::from_secs(5), async
{
+ tokio::join!(
+ first.commit_with_identifier(vec![a], 7),
+ second.commit_with_identifier(vec![b], 7)
+ )
+ })
+ .await
+ .expect("concurrent commits must complete");
+ a.unwrap();
+ b.unwrap();
+ let latest = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(latest.id(), 2);
+ assert_eq!(latest.total_record_count(), Some(20));
+ assert_eq!(latest.next_row_id(), Some(20));
+ let mut entries = active_entries(&io, path, &latest).await;
+ entries.sort_by_key(|entry| entry.file().first_row_id);
+ assert_eq!(entries.len(), 2);
+ assert_eq!(entries[0].file().first_row_id, Some(0));
+ assert_eq!(entries[1].file().first_row_id, Some(10));
+ assert_eq!(entries[0].file().row_count, 10);
+ assert_eq!(entries[1].file().row_count, 10);
+ let mut names = entries
+ .iter()
+ .map(|entry| entry.file().file_name.as_str())
+ .collect::<Vec<_>>();
+ names.sort_unstable();
+ assert_eq!(names, ["a.parquet", "b.parquet"]);
+}
+
+#[tokio::test]
+async fn unsupported_publication_preserves_prepared_data() {
+ let op =
opendal::Operator::from_config(opendal::services::MemoryConfig::default())
+ .unwrap()
+ .layer(opendal::layers::CapabilityOverrideLayer::new(|mut cap| {
+ cap.write_with_if_not_exists = false;
+ cap
+ }));
+ let io = FileIOBuilder::new("file")
+ .with_fs_operator(op)
+ .build()
+ .unwrap();
+ let path = "/unsupported-publication";
+ setup_dirs(&io, path).await;
+ let data_path = format!("{path}/bucket-0/data.parquet");
+ io.new_output(&data_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ let mut commit = setup_commit(&io, path);
+ commit.commit_max_retries = 0;
+ let error = commit
+ .commit(vec![append_message("data.parquet")])
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("cannot atomically publish"),
+ "{error}"
+ );
+ assert!(latest_snapshot(&io, path).await.is_none());
+ assert_eq!(
+ io.new_input(&data_path).unwrap().read().await.unwrap(),
+ "prepared"
+ );
+}
+
#[tokio::test]
async fn recovery_rejects_duplicate_file_after_identity_expiry() {
let io = test_file_io();
diff --git a/crates/paimon/tests/dynamic_bucket_write_test.rs
b/crates/paimon/tests/dynamic_bucket_write_test.rs
index de409ddd..ffeb2d92 100644
--- a/crates/paimon/tests/dynamic_bucket_write_test.rs
+++ b/crates/paimon/tests/dynamic_bucket_write_test.rs
@@ -231,6 +231,20 @@ async fn
failed_later_prepare_preserves_returned_and_committed_files() {
);
setup_dirs(&io, path).await;
let builder = table.new_write_builder();
+ // Inject HASH failures only during preparation. Publish the resulting
+ // messages through the built-in filesystem's atomic snapshot
primitive.
+ let commit_table = Table::new(
+ FileIOBuilder::new("file").build().unwrap(),
+ table.identifier().clone(),
+ path.into(),
+ table.schema().clone(),
+ None,
+ );
+ let committer = commit_table
+ .new_write_builder()
+ .with_commit_user(builder.commit_user())
+ .unwrap()
+ .new_commit();
let mut writer = builder.new_write().unwrap();
writer
.write_arrow_batch(&make_batch(vec![1], vec![10]))
@@ -247,7 +261,7 @@ async fn
failed_later_prepare_preserves_returned_and_committed_files() {
.collect::<std::collections::HashSet<_>>();
assert!(!returned.is_empty());
if publish_first {
- builder.new_commit().commit(messages.clone()).await.unwrap();
+ committer.commit(messages.clone()).await.unwrap();
}
writer
.write_arrow_batch(&make_batch(vec![2], vec![20]))
@@ -274,7 +288,7 @@ async fn
failed_later_prepare_preserves_returned_and_committed_files() {
);
}
if !publish_first {
- builder.new_commit().commit(messages).await.unwrap();
+ committer.commit(messages).await.unwrap();
}
let snapshot = table
.snapshot_manager()
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 811fe07b..26cb18f8 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -218,6 +218,24 @@ 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.
+Snapshot publication never replaces an existing snapshot. The built-in local
+filesystem stages a complete file before publishing it without replacement;
+memory and supported object stores use conditional creation. A publication
+conflict rebases the commit on the latest snapshot and retries. Backends
without
+atomic publication of a complete file must use a catalog-provided
`SnapshotCommit`
+with concurrency control; filesystem commits on those backends return an
+unsupported I/O error. This includes COS conditional writes, which do not
reliably
+prevent overwrites on buckets that have ever enabled versioning, and position
+writers such as ADLS and custom filesystem operators without conditional
rename.
+OSS conditional publication assumes the bucket uses its default versioning
+configuration and has never enabled versioning.
+Application storage providers can implement
`FileIOProvider::write_atomic_if_not_exists`
+to supply their own atomic publication primitive. FUSE routes use local staged
+publication and require the mount to support publishing without replacement.
+Storage failures never fall back to an unconditional snapshot write.
+All concurrent writers, including writers from other engines, must use
compatible
+concurrency control.
+
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.
