[Issue #1277] Integrate S3QS storage backend with coordinator-managed partitioned shuffle execution

pixels-common/src/main/java/io/pixelsdb/pixels/common/task/TaskQueue.java

@@ -60,6 +60,10 @@ public TaskQueue(Collection<E> tasks)
     public boolean offerAllPending(Collection<E> tasks)
     {
         checkPendingTasks(tasks);
+        synchronized (this.allTasks)
+        {
+            this.allTasks.addAll(tasks);
+        }
         return this.pendingQueue.addAll(tasks);
     }
 
@@ -80,6 +84,11 @@ private void checkPendingTasks(Collection<E> tasks)
      */
     public boolean offerPending(E task)
     {
+        checkPendingTasks(Collections.singletonList(task));
+        synchronized (this.allTasks)
+        {
+            this.allTasks.add(task);
+        }
         return this.pendingQueue.offer(task);
     }
 
@@ -150,6 +159,53 @@ public E removeNextExpired()
 
     public List<E> getAllTasks()
     {
-        return ImmutableList.copyOf(this.allTasks);
+        synchronized (this.allTasks)
+        {
+            return ImmutableList.copyOf(this.allTasks);
+        }
+    }
+
+    public int getTotalTaskCount()
+    {
+        synchronized (this.allTasks)
+        {
+            return this.allTasks.size();
+        }
+    }
+
+    public int getPendingTaskCount()
+    {
+        return this.pendingQueue.size();
+    }
+
+    public int getRunningTaskCount()
+    {
+        return this.runningTasks.size();
+    }
+
+    public int getCompletedTaskCount()
+    {
+        return getTaskStatusCount(Task.Status.COMPLETE);
+    }
+
+    public int getFailedTaskCount()
+    {
+        return getTaskStatusCount(Task.Status.FAILED);
+    }
+
+    public int getTaskStatusCount(Task.Status status)
+    {
+        synchronized (this.allTasks)
+        {
+            int count = 0;
+            for (E task : this.allTasks)
+            {
+                if (task.getStatus() == status)
+                {
+                    count++;
+                }
+            }
+            return count;
+        }
     }
 }

pixels-common/src/main/java/io/pixelsdb/pixels/common/task/Worker.java

@@ -88,6 +88,21 @@ public void terminate()
         }
     }
 
+    /**
+     * Whether this worker has explicitly finished or been terminated.
+     *
+     * This is different from {@link #isAlive()}: a lease can expire while a
+     * physical invocation is still running and must still be considered by
+     * coordinator admission or drain control.
+     */
+    public boolean isTerminated()
+    {
+        synchronized (this.lease)
+        {
+            return this.terminated;
+        }
+    }
+
     /**
      * Extent the lease of this worker if the worker is alive. This method should only be called on the coordinator.
      * @return the new start time (milliseconds since the Unix epoch) of the extended lease

pixels-common/src/main/java/io/pixelsdb/pixels/common/turbo/WorkerType.java

@@ -30,10 +30,13 @@ public enum WorkerType
     UNKNOWN, // The first enum value is the default value.
     SCAN, SCAN_STREAMING,
     PARTITION, PARTITION_STREAMING,
+    PARTITION_S3QS,
     BROADCAST_JOIN, BROADCAST_JOIN_STREAMING,
     BROADCAST_CHAIN_JOIN, BROADCAST_CHAIN_JOIN_STREAMING,
     PARTITIONED_JOIN, PARTITIONED_JOIN_STREAMING,
+    PARTITIONED_JOIN_S3QS,
     PARTITIONED_CHAIN_JOIN, PARTITIONED_CHAIN_JOIN_STREAMING,
+    PARTITIONED_CHAIN_JOIN_S3QS,
     SORT,
     SORTED_JOIN,
     AGGREGATION;

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

@@ -197,6 +197,8 @@ aggr.partition.size.rows=1280000
 ### pixels-turbo - query execution ###
 executor.input.storage.scheme=s3
 executor.intermediate.storage.scheme=s3
+# the storage scheme dedicated to the shuffle transport path, empty means disabled
+executor.shuffle.storage.scheme=
 executor.intermediate.folder=/pixels-turbo/intermediate/
 executor.output.storage.scheme=output-storage-scheme-dummy
 executor.output.folder=output-folder-dummy

pixels-planner/pom.xml

@@ -31,6 +31,10 @@
             <groupId>io.pixelsdb</groupId>
             <artifactId>pixels-executor</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.pixelsdb</groupId>
+            <artifactId>pixels-storage-s3qs</artifactId>
+        </dependency>
 
         <!-- grpc -->
         <dependency>
@@ -57,7 +61,6 @@
         <dependency>
             <groupId>io.pixelsdb</groupId>
             <artifactId>pixels-storage-s3</artifactId>
-            <scope>test</scope>
             <optional>true</optional>
         </dependency>
         <dependency>
@@ -96,4 +99,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
+</project>