This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 98afc10300 [core] Rewrite oss tryToWriteAtomic using atomic putObject
API (#8226) (#8228)
98afc10300 is described below
commit 98afc1030036e716ea13ca594cd2efd68646ec77
Author: MaxLinyun <[email protected]>
AuthorDate: Sat Jun 20 13:10:43 2026 +0800
[core] Rewrite oss tryToWriteAtomic using atomic putObject API (#8226)
(#8228)
---
.../java/org/apache/paimon/fs/PluginFileIO.java | 5 +++
.../main/java/org/apache/paimon/oss/OSSFileIO.java | 39 ++++++++++++++++++++++
2 files changed, 44 insertions(+)
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
index 2af4919f08..d5b62a2944 100644
--- a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
+++ b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java
@@ -91,6 +91,11 @@ public abstract class PluginFileIO implements FileIO,
HadoopOptionsProvider {
return wrap(() -> fileIO(src).rename(src, dst));
}
+ @Override
+ public boolean tryToWriteAtomic(Path path, String content) throws
IOException {
+ return wrap(() -> fileIO(path).tryToWriteAtomic(path, content));
+ }
+
private FileIO fileIO(Path path) throws IOException {
if (lazyFileIO == null) {
synchronized (this) {
diff --git
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
index 85d7e8cf9e..b9c18bf994 100644
---
a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
+++
b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java
@@ -26,9 +26,12 @@ import org.apache.paimon.fs.TwoPhaseOutputStream;
import org.apache.paimon.options.Options;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.ReflectionUtils;
+import org.apache.paimon.utils.StringUtils;
import com.aliyun.oss.OSSClient;
+import com.aliyun.oss.OSSException;
import com.aliyun.oss.common.comm.ServiceClient;
+import com.aliyun.oss.model.ObjectMetadata;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem;
@@ -36,9 +39,11 @@ import
org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystemStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.URI;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -180,6 +185,40 @@ public class OSSFileIO extends HadoopCompliantFileIO
implements HadoopOptionsPro
}
}
+ @Override
+ public boolean tryToWriteAtomic(Path path, String content) throws
IOException {
+ URI uri = path.toUri();
+ String bucket = uri.getHost();
+ String objectKey = uri.getPath().substring(1);
+ byte[] bytes = content.getBytes(StandardCharsets.UTF_8);
+
+ ObjectMetadata metadata = new ObjectMetadata();
+ metadata.setContentLength(bytes.length);
+ metadata.setHeader("x-oss-forbid-overwrite", "true");
+
+ String sseAlgorithm =
+
hadoopOptions.getString("fs.oss.server-side-encryption-algorithm", "");
+ if (StringUtils.isNotEmpty(sseAlgorithm)) {
+ metadata.setServerSideEncryption(sseAlgorithm);
+ }
+
+ AliyunOSSFileSystem fs = (AliyunOSSFileSystem)
getFileSystem(path(path));
+ AliyunOSSFileSystemStore store = fs.getStore();
+ try {
+ OSSClient ossClient = ReflectionUtils.getPrivateFieldValue(store,
"ossClient");
+ ossClient.putObject(bucket, objectKey, new
ByteArrayInputStream(bytes), metadata);
+ return true;
+ } catch (OSSException e) {
+ if ("FileAlreadyExists".equals(e.getErrorCode())) {
+ LOG.warn("Failed to atomic write {}: object already exists",
path);
+ return false;
+ }
+ throw new IOException("Failed to atomic write " + path, e);
+ } catch (Exception e) {
+ throw new IOException("Failed to atomic write " + path, e);
+ }
+ }
+
@Override
public void close() {
if (!allowCache) {
