update partSizeThreshold and add doc

Author: jerry-024

paimon-common/src/main/java/org/apache/paimon/fs/MultiPartUploadTwoPhaseOutputStream.java

@@ -57,7 +57,12 @@ public MultiPartUploadTwoPhaseOutputStream(
         this.position = 0;
     }
 
-    public abstract int partSizeThreshold();
+    // OSS limit:  100KB ~ 5GB
+    // S3 limit:  5MiB ~ 5GiB
+    // Considering memory usage, and referencing Flink's setting of 10MiB.
+    public int partSizeThreshold() {
+        return 10 << 20;
+    }
 
     public abstract Committer committer(
             String uploadId, List<T> uploadedParts, String objectName, long position);
@@ -102,6 +107,10 @@ public void write(byte[] b, int off, int len) throws IOException {
             offset += count;
             remaining -= count;
             position += count;
+            // consume buffer if it is full
+            if (buffer.size() >= partSizeThreshold()) {
+                uploadPart();
+            }
         }
     }
 
@@ -144,15 +153,18 @@ private void uploadPart() throws IOException {
         File tempFile = null;
         int partNumber = uploadedParts.size() + 1;
         try {
-            byte[] data = buffer.toByteArray();
             tempFile = Files.createTempFile("multi-part-" + UUID.randomUUID(), ".tmp").toFile();
             try (FileOutputStream fos = new FileOutputStream(tempFile)) {
-                fos.write(data);
+                buffer.writeTo(fos);
                 fos.flush();
             }
             T partETag =
                     multiPartUploadStore.uploadPart(
-                            objectName, uploadId, uploadedParts.size() + 1, tempFile, data.length);
+                            objectName,
+                            uploadId,
+                            uploadedParts.size() + 1,
+                            tempFile,
+                            checkedDownCast(tempFile.length()));
             uploadedParts.add(partETag);
             buffer.reset();
         } catch (Exception e) {
@@ -166,4 +178,13 @@ private void uploadPart() throws IOException {
             }
         }
     }
+
+    private static int checkedDownCast(long value) {
+        int downCast = (int) value;
+        if (downCast != value) {
+            throw new IllegalArgumentException(
+                    "Cannot downcast long value " + value + " to integer.");
+        }
+        return downCast;
+    }
 }

paimon-common/src/test/java/org/apache/paimon/fs/MultiPartUploadTwoPhaseOutputStreamTest.java

@@ -55,8 +55,6 @@ void testWriteFlushAndCommit() throws IOException {
                 new TestMultiPartUploadTwoPhaseOutputStream(store, objectPath, 5);
 
         stream.write("hello".getBytes(StandardCharsets.UTF_8));
-        assertThat(store.getUploadedParts()).hasSize(0);
-        stream.flush();
         assertThat(store.getUploadedParts()).hasSize(1);
         assertThat(store.getUploadedParts()).extracting(TestPart::getPartNumber).containsExactly(1);
         assertThat(store.getUploadedParts())

paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoTwoPhaseOutputStream.java

@@ -37,11 +37,6 @@ public JindoTwoPhaseOutputStream(
         super(multiPartUploadStore, hadoopPath);
     }
 
-    @Override
-    public int partSizeThreshold() {
-        return 8 << 20;
-    }
-
     @Override
     public Committer committer(
             String uploadId, List<JdoObjectPart> uploadedParts, String objectName, long position) {

paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OssTwoPhaseOutputStream.java

@@ -38,11 +38,6 @@ public OssTwoPhaseOutputStream(
         super(multiPartUploadStore, hadoopPath);
     }
 
-    @Override
-    public int partSizeThreshold() {
-        return 8 << 20;
-    }
-
     @Override
     public Committer committer(
             String uploadId, List<PartETag> uploadedParts, String objectName, long position) {

paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3MultiPartUpload.java

@@ -82,13 +82,7 @@ public PartETag uploadPart(
             throws IOException {
         final UploadPartRequest uploadRequest =
                 s3accessHelper.newUploadPartRequest(
-                        objectName,
-                        uploadId,
-                        partNumber,
-                        checkedDownCast(byteLength),
-                        null,
-                        file,
-                        0L);
+                        objectName, uploadId, partNumber, byteLength, null, file, 0L);
         return s3accessHelper.uploadPart(uploadRequest).getPartETag();
     }
 
@@ -108,13 +102,4 @@ private static final class InternalWriteOperationHelper extends WriteOperationHe
             super(owner, conf, statisticsContext, auditSpanSource, auditSpan);
         }
     }
-
-    private static int checkedDownCast(long value) {
-        int downCast = (int) value;
-        if (downCast != value) {
-            throw new IllegalArgumentException(
-                    "Cannot downcast long value " + value + " to integer.");
-        }
-        return downCast;
-    }
 }

paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/S3TwoPhaseOutputStream.java

@@ -38,11 +38,6 @@ public S3TwoPhaseOutputStream(
         super(multiPartUploadStore, hadoopPath);
     }
 
-    @Override
-    public int partSizeThreshold() {
-        return 5 << 20;
-    }
-
     @Override
     public Committer committer(
             String uploadId, List<PartETag> uploadedParts, String objectName, long position) {