[Issue #1024] implement S3QS storage for intermediate data passing (#1060)

Author: bianhq

pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/ObjectPath.java

@@ -72,7 +72,7 @@ else if (path.startsWith("/"))
             }
             this.valid = true;
         }
-        else if (path.length() > 0)
+        else if (!path.isEmpty())
         {
             // this is a bucket.
             this.bucket = path;

pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/PhysicalWriter.java

@@ -46,6 +46,17 @@ public interface PhysicalWriter extends Closeable
      */
     long append(ByteBuffer buffer) throws IOException;
 
+    /**
+     * Append content to the file.
+     * @param buffer content buffer
+     * @return start offset of content in the file.
+     * @throws IOException
+     */
+    default long append(byte[] buffer) throws IOException
+    {
+        return append(buffer, 0, buffer.length);
+    }
+
     /**
      * Append content to the file.
      *

pixels-common/src/main/java/io/pixelsdb/pixels/common/physical/Storage.java

@@ -47,7 +47,7 @@ enum Scheme
         redis, // Redis
         gcs,   // google cloud storage
         httpstream,  // http (netty) based stream storage
-        sqsstream,   // AWS SQS+S3 based stream storage
+        s3qs,   // AWS SQS+S3 based storage for intermediate data shuffle
         mock; // mock
 
         /**
@@ -160,7 +160,7 @@ public boolean equals(Scheme other)
      * By default, this method returns false.
      *
      * <p><b>Note:</b> if this method returns true, then getLocations
-     * and getHosts must be override to return the physical locality
+     * and getHosts must be overridden to return the physical locality
      * information of the given path.</p>
      * @return
      */
@@ -228,8 +228,7 @@ default String[] getHosts(String path) throws IOException
      * @return
      * @throws IOException if path is a directory.
      */
-    DataOutputStream create(String path, boolean overwrite,
-                            int bufferSize) throws IOException;
+    DataOutputStream create(String path, boolean overwrite, int bufferSize) throws IOException;
 
     /**
      * For local fs, path is considered as local.
@@ -241,7 +240,7 @@ DataOutputStream create(String path, boolean overwrite,
      * @throws IOException if path is a directory.
      */
     default DataOutputStream create(String path, boolean overwrite,
-                            int bufferSize, short replication) throws IOException
+                                    int bufferSize, short replication) throws IOException
     {
         return create(path, overwrite, bufferSize);
     }

pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/ConfigFactory.java

@@ -19,10 +19,11 @@
  */
 package io.pixelsdb.pixels.common.utils;
 
-import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -63,7 +64,7 @@ public interface UpdateCallback
         void update(String value);
     }
 
-    private Map<String, UpdateCallback> callbacks;
+    private final Map<String, UpdateCallback> callbacks;
 
     private ConfigFactory()
     {
@@ -111,7 +112,7 @@ private ConfigFactory()
                 }
                 else
                 {
-                    in = new FileInputStream(pixelsConfig);
+                    in = Files.newInputStream(Paths.get(pixelsConfig));
                 }
             } catch (IOException e)
             {
@@ -153,7 +154,7 @@ public synchronized void loadProperties(String propFilePath) throws IOException
             }
             else
             {
-                in = new FileInputStream(propFilePath);
+                in = Files.newInputStream(Paths.get(propFilePath));
             }
             this.prop.load(in);
             for (Map.Entry<String, UpdateCallback> entry : this.callbacks.entrySet())

pixels-common/src/main/java/io/pixelsdb/pixels/common/utils/Constants.java

@@ -32,8 +32,8 @@ public final class Constants
     public static final int S3_BUFFER_SIZE = 8 * 1024 * 1024;
     public static final int REDIS_BUFFER_SIZE = 8 * 1024 * 1024;
     public static final int GCS_BUFFER_SIZE = 8 * 1024 * 1024;
+    public static final int S3QS_BUFFER_SIZE = 8 * 1024 * 1024;
     public static final int HTTP_STREAM_BUFFER_SIZE = 8 * 1024 * 1024;
-    public static final int SQS_STREAM_BUFFER_SIZE = 8 * 1024 * 1024;
     public static final int STREAM_READER_RG_BUFFER_SIZE = 1024 * 1024;
     public static final int STREAM_READER_RG_FOOTER_BUFFER_SIZE = 1024;
 

pixels-common/src/main/resources/pixels.properties

@@ -136,6 +136,8 @@ minio.path.prefix=s3://pixels/retina/
 redis.endpoint=localhost:6379
 redis.access.key=redis-user-dummy
 redis.secret.key=redis-password-dummy
+# the number of messages that the receiver tries to poll each time, it should be >= 1 and <= 10
+s3qs.poll.batch.size=3
 
 ###### query engine settings ######
 

pixels-storage/README.md

@@ -11,7 +11,7 @@ Queries will load and call the providers to get access to the underlying storage
 - `pixels-storage-mock` provides a mock storage system for debugging.
 - `pixels-storage-redis` provides redis key-value storage.
 - `pixels-storage-http` provides a streaming storage based on http (netty).
-- `pixels-storage-sqs` provides a streaming storage based on AWS SQS and S3.
+- `pixels-storage-s3qs` provides a storage based on AWS SQS and S3 for intermediate data shuffle.
 
 ## Usage
 Storage provider can be used in either of the following ways:

pixels-storage/pixels-storage-http/src/main/java/io/pixelsdb/pixels/storage/http/PhysicalHttpStreamWriter.java

@@ -25,6 +25,7 @@
 
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.nio.Buffer;
 import java.nio.ByteBuffer;
 
 /**
@@ -71,7 +72,14 @@ public long prepare(int length) throws IOException
     @Override
     public long append(ByteBuffer buffer) throws IOException
     {
-        buffer.flip();
+        /**
+         * Issue #217:
+         * For compatibility reasons if this code is compiled by jdk>=9 but executed in jvm8.
+         *
+         * In jdk8, ByteBuffer.flip() is extended from Buffer.flip(), but in jdk11, different kind of ByteBuffer
+         * has its own flip implementation and may lead to errors.
+         */
+        ((Buffer)buffer).flip();
         int length = buffer.remaining();
         return append(buffer.array(), buffer.arrayOffset() + buffer.position(), length);
     }

pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/AbstractS3.java

@@ -477,7 +477,7 @@ public boolean isDirectory(String path) throws IOException
         return new ObjectPath(path).isFolder;
     }
 
-    public S3Client getClient()
+    public S3Client getS3Client()
     {
         return s3;
     }

pixels-storage/pixels-storage-s3/src/main/java/io/pixelsdb/pixels/storage/s3/AbstractS3Reader.java

@@ -108,7 +108,7 @@ public AbstractS3Reader(Storage storage, String path) throws IOException
         this.length = this.s3.getStatus(path).getLength();
         this.numRequests = 1;
         this.position = 0L;
-        this.client = this.s3.getClient();
+        this.client = this.s3.getS3Client();
         this.enableAsync = Boolean.parseBoolean(ConfigFactory.Instance()
                 .getProperty("s3.enable.async"));
         this.useAsyncClient = Boolean.parseBoolean(ConfigFactory.Instance()
@@ -137,8 +137,7 @@ public void seek(long desired) throws IOException
             position = desired;
             return;
         }
-        throw new IOException("Desired offset " + desired +
-                " is out of bound (" + 0 + "," + length + ")");
+        throw new IOException("Desired offset " + desired + " is out of bound (" + 0 + "," + length + ")");
     }
 
     @Override
@@ -158,7 +157,7 @@ public ByteBuffer readFully(int len) throws IOException
                 client.getObject(request, ResponseTransformer.toBytes());
             this.numRequests++;
             this.position += len;
-            return ByteBuffer.wrap(response.asByteArrayUnsafe());
+            return response.asByteBuffer();
         } catch (Exception e)
         {
             throw new IOException("Failed to read object.", e);
@@ -168,15 +167,15 @@ public ByteBuffer readFully(int len) throws IOException
     @Override
     public void readFully(byte[] buffer) throws IOException
     {
-        ByteBuffer byteBuffer = readFully(buffer.length);
-        System.arraycopy(byteBuffer.array(), 0, buffer, 0, buffer.length);
+        readFully(buffer, 0, buffer.length);
     }
 
     @Override
     public void readFully(byte[] buffer, int off, int len) throws IOException
     {
         ByteBuffer byteBuffer = readFully(len);
-        System.arraycopy(byteBuffer.array(), 0, buffer, off, len);
+        // This is more efficient than byteBuffer.put(buffer, off, len).
+        System.arraycopy(byteBuffer.array(), byteBuffer.arrayOffset() + byteBuffer.position(), buffer, off, len);
     }
 
     /**