[Bug][python] Concurrent OSS snapshot commits can overwrite successful writes

[wangzhigang1999]

Search before asking

• I searched the existing issues and pull requests. Java's corresponding OSS write issue was fixed in [core] Rewrite oss tryToWriteAtomic using atomic putObject API (#8226) #8228; the Python path still uses temporary-file-and-rename writes.

Paimon version

Master at d77efe0d2e3d15e33df0eb8211f833035e0959b8. The Python source is unchanged in master 8f5ce6b84.

Compute Engine

PyPaimon FileSystem Catalog, Python 3.12.6, PyArrow 19.0.1, Alibaba Cloud OSS with fs.oss.impl=legacy. The test bucket has never enabled versioning.

Minimal reproduce step

Configure an OSS bucket and run concurrent atomic creations against the same fresh object. For example, run this against a disposable prefix using credentials from environment variables:

import os
import threading
import uuid
from concurrent.futures import ThreadPoolExecutor

from pypaimon.common.file_io import FileIO
from pypaimon.common.options import Options

root = os.environ['OSS_TEST_PREFIX'].rstrip('/') + '/' + uuid.uuid4().hex
options = Options({
    'fs.oss.impl': 'legacy',
    'fs.oss.endpoint': os.environ['OSS_ENDPOINT'],
    'fs.oss.accessKeyId': os.environ['OSS_ACCESS_KEY_ID'],
    'fs.oss.accessKeySecret': os.environ['OSS_ACCESS_KEY_SECRET'],
    'fs.oss.securityToken': os.environ.get('OSS_SECURITY_TOKEN'),
})
io = FileIO.get(root, options)
print('test prefix:', root)
failures = 0
for round_id in range(20):
    path = root + '/snapshot-' + str(round_id)
    barrier = threading.Barrier(8)

    def write(writer_id):
        content = 'writer-' + str(writer_id)
        barrier.wait(timeout=30)
        return content, io.try_to_write_atomic(path, content)

    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(write, range(8)))
    winners = [content for content, success in results if success]
    if len(winners) != 1 or io.read_file_utf8(path) != winners[0]:
        failures += 1
        print('round', round_id, 'successful writers:', winners)
print('rounds violating atomic creation:', failures)
assert failures == 0

OSS_TEST_PREFIX should be an oss://bucket/disposable-prefix URI. The script leaves the objects under the printed/configured prefix for inspection. The race is timing-dependent, so repeat if necessary.

What doesn't meet your expectations?

Exactly one creator should return True; subsequent contenders should return False without replacing the successful content. Instead, several contenders can return True for the same object, and previously successful content is overwritten.

PyArrowFileIO.try_to_write_atomic() delegates to the temporary-file-and-rename implementation. Its destination existence check and the OSS copy/move are separate operations. Two writers can both observe an absent snapshot and then overwrite the same destination. The snapshot commit retry loop consequently treats conflicting publications as successful.

A real OSS comparison using separate processes reproduced the impact:

Check	Current master	Conditional PUT implementation
Atomic creation: 64 processes, 300 rounds, 64 KiB payloads
Rounds with multiple successful creators	300/300	0/300
Successful writes later overwritten	7,545	0
Table append: 16 processes, 10 commits/process, 2,000 rows/commit
Completed commit calls	160	160
Rows read back, out of 320,000 expected	232,000	320,000
Snapshot objects	116	160

Both table runs used commit.max-retries=64. The old calls completed; the validation assertions failed because 88,000 expected rows were absent from the visible table state. These are correctness results from one controlled comparison, not throughput benchmarks.

Anything else?

Java OSSFileIO uses PutObject with x-oss-forbid-overwrite=true after #8228. Python needs the corresponding OSS-specific conditional creation, with conflicts distinguished from permission/network failures and with configured SSE headers preserved.

OSS ignores the overwrite prohibition for buckets with versioning enabled or suspended. This boundary must be handled explicitly; the same guarantee cannot be assumed for every S3-compatible service.

Are you willing to submit a PR?

• I'm willing to submit a PR!