[JingsongLi]

Purpose

Two concurrent writers can both observe that a snapshot is absent, overwrite the same snapshot path, and both report success. Replace check-then-rename/write publication with atomic creation that never replaces an existing snapshot.

Brief change log

• Add a FileIO atomic publication primitive and an optional FileIOProvider override. Use staged no-replace publication for the built-in local filesystem and FUSE, conditional rename where supported, and conditional creation for supported whole-object stores.
• Treat an existing target as a conflict and preserve errors without falling back to an unconditional write. Backends without a reliable atomic primitive reject filesystem snapshot publication and require a catalog or provider implementation.
• Keep snapshot serialization and LATEST hints in SnapshotManager, and retain the existing TableCommit conflict checks, retries, and uncertain-outcome recovery.
• Add deterministic concurrent-writer, row-ID rebase, provider-error, unsupported-backend, rename-error, and OSS conditional PUT regressions; adapt filesystem fixtures to use safe publication.
• Restore an expired tagged snapshot before deleting newer history during DataFusion rollback, so rejected publication preserves the existing snapshots, tags, and LATEST hint.

Tests

• Paimon tests with fulltext,vortex: 4,705 unit tests and 702 integration tests passed across full-suite and focused runs; 7 existing tests ignored. The unchanged cache subprocess test failed once under load and passed an isolated rerun.
• DataFusion: all 15 procedure unit tests passed, including the new rollback failure/success regressions. The failure regression reproduced deletion of snapshots 2/3 before the fix and passed afterward. All 4 SQL rollback tests passed with TZ=UTC.
• Broader DataFusion unit run: 355 passed; 7 tests could not load the locally absent Spark-provisioned partitioned tables.
• OSS HTTP transport regression verifies that two conditional PUTs produce exactly one winner and preserve its complete bytes under default, never-enabled versioning.
• cargo fmt --all -- --check, explicit rustfmt validation for the included recovery tests, and git diff --check passed.
• Strict workspace Clippy is blocked by four existing clippy::nonminimal_bool findings in unchanged de_vector_read.rs, table_merge_into.rs, and vector_search_common.rs. Workspace Clippy passed with -D warnings -A clippy::nonminimal_bool.
• Real cloud endpoints and JindoSDK were not exercised.

API and Format

• Add a default optional FileIOProvider::write_atomic_if_not_exists method. Existing providers retain their path resolution; providers can explicitly supply atomic publication when their operator cannot.
• Snapshot serialization and the on-disk format are unchanged. Custom filesystem operators without atomic complete-file publication now return an unsupported I/O error. OSS uses conditional writes under the default bucket configuration with versioning never enabled. Jindo remains read-only.

Documentation

Document atomic snapshot publication, provider overrides, FUSE requirements, and unsupported backend behavior in the getting-started guide and Rust API comments.

[leaves12138]

LGTM