feat(s3stream): copy write based on max part size (#883)

Signed-off-by: Robin Han <[email protected]>
AutoMQ · Jan 4, 2024 · 1dab15a · 1dab15a
1 parent d3d9a52
commit 1dab15a
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 0 deletions.
diff --git a/s3stream/src/main/java/com/automq/stream/s3/StreamObjectCompactor.java b/s3stream/src/main/java/com/automq/stream/s3/StreamObjectCompactor.java
@@ -175,7 +175,12 @@ static List<List<S3ObjectMetadata>> group0(List<S3ObjectMetadata> objects, long
         long groupSize = 0;
         long groupNextOffset = -1L;
         List<S3ObjectMetadata> group = new LinkedList<>();
+        int partCount = 0;
         for (S3ObjectMetadata object : objects) {
+            int objectPartCount = (int) ((object.objectSize() + Writer.MAX_PART_SIZE - 1) / Writer.MAX_PART_SIZE);
+            if (objectPartCount >= Writer.MAX_PART_COUNT) {
+                continue;
+            }
             if (groupNextOffset == -1L) {
                 groupNextOffset = object.startOffset();
             }
@@ -185,6 +190,7 @@ static List<List<S3ObjectMetadata>> group0(List<S3ObjectMetadata> objects, long
                 || (groupSize + object.objectSize() > maxStreamObjectSize && !group.isEmpty())
                 // object count in group is larger than MAX_OBJECT_GROUP_COUNT
                 || group.size() >= MAX_OBJECT_GROUP_COUNT
+                || partCount + objectPartCount > Writer.MAX_PART_COUNT
             ) {
                 objectGroups.add(group);
                 group = new LinkedList<>();
@@ -193,6 +199,7 @@ static List<List<S3ObjectMetadata>> group0(List<S3ObjectMetadata> objects, long
             group.add(object);
             groupSize += object.objectSize();
             groupNextOffset = object.endOffset();
+            partCount += objectPartCount;
         }
         if (!group.isEmpty()) {
             objectGroups.add(group);

diff --git a/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java b/s3stream/src/main/java/com/automq/stream/s3/operator/MultiPartWriter.java
@@ -103,6 +103,18 @@ public boolean hasBatchingPart() {
 
     @Override
     public void copyWrite(String sourcePath, long start, long end) {
+        long nextStart = start;
+        for (; ; ) {
+            long currentEnd = Math.min(nextStart + Writer.MAX_PART_SIZE, end);
+            copyWrite0(sourcePath, nextStart, currentEnd);
+            nextStart = currentEnd;
+            if (currentEnd == end) {
+                break;
+            }
+        }
+    }
+
+    public void copyWrite0(String sourcePath, long start, long end) {
         long targetSize = end - start;
         if (objectPart == null) {
             if (targetSize < minPartSize) {