[Issue #1082] support message grouping and terminating in S3QS

pixels-core/src/main/java/io/pixelsdb/pixels/core/PixelsWriterImpl.java

@@ -219,6 +219,7 @@ public static class Builder
         private boolean builderPartitioned = false;
         private boolean builderNullsPadding = false;
         private Optional<List<Integer>> builderPartKeyColumnIds = Optional.empty();
+        private PhysicalWriter builderPhysicalWriter = null;
 
         private Builder()
         {
@@ -230,6 +231,12 @@ public Builder setSchema(TypeDescription schema)
             return this;
         }
 
+        public Builder setPhysicalWriter(PhysicalWriter writer)
+        {
+            this.builderPhysicalWriter = requireNonNull(writer);
+            return this;
+        }
+
         public Builder setHasHiddenColumn(boolean hasHiddenColumn)
         {
             this.builderHasHiddenColumn = hasHiddenColumn;
@@ -336,24 +343,27 @@ public PixelsWriter build() throws PixelsWriterException
         {
             requireNonNull(this.builderStorage, "storage is not set");
             requireNonNull(this.builderFilePath, "file path is not set");
-            PhysicalWriter fsWriter = null;
-            try
-            {
-                fsWriter = PhysicalWriterUtil.newPhysicalWriter(
-                        this.builderStorage, this.builderFilePath, this.builderBlockSize, this.builderReplication,
-                        this.builderBlockPadding, this.builderOverwrite);
-            } catch (IOException e)
-            {
-                LOGGER.error("Failed to create PhysicalWriter");
-                throw new PixelsWriterException(
-                        "Failed to create PixelsWriter due to error of creating PhysicalWriter", e);
-            }
 
-            if (fsWriter == null)
+            if(this.builderPhysicalWriter == null)
             {
-                LOGGER.error("Failed to create PhysicalWriter");
-                throw new PixelsWriterException(
-                        "Failed to create PixelsWriter due to error of creating PhysicalWriter");
+                try
+                {
+                    this.builderPhysicalWriter = PhysicalWriterUtil.newPhysicalWriter(
+                            this.builderStorage, this.builderFilePath, this.builderBlockSize, this.builderReplication,
+                            this.builderBlockPadding, this.builderOverwrite);
+                }
+                catch (IOException e)
+                {
+                    LOGGER.error("Failed to create PhysicalWriter");
+                    throw new PixelsWriterException(
+                            "Failed to create PixelsWriter due to error of creating PhysicalWriter", e);
+                }
+                if (this.builderPhysicalWriter == null)
+                {
+                    LOGGER.error("Failed to create PhysicalWriter");
+                    throw new PixelsWriterException(
+                            "Failed to create PixelsWriter due to error of creating PhysicalWriter");
+                }
             }
 
             return new PixelsWriterImpl(
@@ -364,7 +374,7 @@ public PixelsWriter build() throws PixelsWriterException
                     builderCompressionKind,
                     builderCompressionBlockSize,
                     builderTimeZone,
-                    fsWriter,
+                    builderPhysicalWriter,
                     builderEncodingLevel,
                     builderNullsPadding,
                     builderPartitioned,

pixels-core/src/main/java/io/pixelsdb/pixels/core/PixelsWriterStreamImpl.java

@@ -186,6 +186,7 @@ public static class Builder
         private EncodingLevel builderEncodingLevel = EncodingLevel.EL0;
         private boolean builderPartitioned = false;
         private boolean builderNullsPadding = false;
+        private PhysicalWriter fsWriter = null;
         private Optional<List<Integer>> builderPartKeyColumnIds = Optional.empty();
 
         // added compared to PixelsWriterImpl
@@ -256,6 +257,11 @@ public Builder setNullsPadding(boolean nullsPadding)
             this.builderNullsPadding = nullsPadding;
             return this;
         }
+        public Builder setPhysicalWriter(PhysicalWriter fsWriter)
+        {
+            this.fsWriter = fsWriter;
+            return this;
+        }
 
         public Builder setEncodingLevel(EncodingLevel encodingLevel)
         {
@@ -305,16 +311,20 @@ public PixelsWriter build() throws PixelsWriterException
                             (this.builderPartKeyColumnIds.isPresent() && !this.builderPartKeyColumnIds.get().isEmpty()),
                     "partition column ids are present while partitioned is false, or vice versa");
 
-            PhysicalWriter fsWriter;
-            try
-            {
-                fsWriter = PhysicalWriterUtil.newPhysicalWriter(
-                        this.builderStorage, this.builderFilePath, null);
-            } catch (IOException e)
+            PhysicalWriter fsWriter = this.fsWriter;
+            if (fsWriter == null)
             {
-                LOGGER.error("Failed to create PhysicalWriter");
-                throw new PixelsWriterException(
-                        "Failed to create PixelsWriter due to error of creating PhysicalWriter", e);
+                try
+                {
+                    fsWriter = PhysicalWriterUtil.newPhysicalWriter(
+                            this.builderStorage, this.builderFilePath, null);
+                }
+                catch (IOException e)
+                {
+                    LOGGER.error("Failed to create PhysicalWriter");
+                    throw new PixelsWriterException(
+                            "Failed to create PixelsWriter due to error of creating PhysicalWriter", e);
+                }
             }
 
             if (fsWriter == null)
@@ -597,7 +607,8 @@ private void writeRowGroup() throws IOException
                 columnWriters[i] = newColumnWriter(children.get(i), columnWriterOption);
             }
             physicalWriter.flush();
-        } catch (IOException e)
+        }
+        catch (IOException e)
         {
             LOGGER.error(e.getMessage());
             throw e;

pixels-storage/pixels-storage-s3qs/src/main/java/io/pixelsdb/pixels/storage/s3qs/PhysicalS3QSReader.java

@@ -74,7 +74,8 @@ public PhysicalS3QSReader(Storage storage, String path) throws IOException
             this.buffer = responseBytes.asByteArrayUnsafe();
             this.length = this.buffer.length;
             this.position = 0;
-        } catch (Exception e)
+        }
+        catch (Exception e)
         {
             this.position = 0;
             throw new IOException("Failed to read object.", e);

pixels-storage/pixels-storage-s3qs/src/main/java/io/pixelsdb/pixels/storage/s3qs/PhysicalS3QSWriter.java

@@ -116,10 +116,17 @@ public long append(byte[] buffer, int offset, int length) throws IOException
     @Override
     public void close() throws IOException
     {
-        this.out.close();
-        if (this.queue != null && !this.queue.isClosed())
+        try
         {
-            this.queue.push(this.pathStr);
+            this.out.close();
+            if (this.queue != null && !this.queue.isClosed())
+            {
+                this.queue.push(this.pathStr);
+            }
+        }
+        catch (IOException e)
+        {
+            throw e;
         }
     }