right child stage should know left child stage worker number

pixels-planner/src/main/java/io/pixelsdb/pixels/planner/coordinate/StageCoordinator.java

@@ -62,6 +62,8 @@ public class StageCoordinator
     private final List<Worker<CFWorkerInfo>> workers = new ArrayList<>();
     private final Map<Long, List<Integer>> workerIdToWorkerIndex = new ConcurrentHashMap<>();
     private int workerIndexAssigner;
+    private int leftChildWorkerNum;
+    private int rightChildWorkerNum;
     private final Object lock = new Object();
 
     /**
@@ -82,6 +84,20 @@ public StageCoordinator(int stageId, int workerNum)
         this.taskQueue = null;
         this.downStreamWorkerNum = 0;
         this.workerIndexAssigner = 0;
+        this.leftChildWorkerNum = 0;
+        this.rightChildWorkerNum = 0;
+    }
+
+    public StageCoordinator(int stageId, int workerNum, int workerIndexAssigner)
+    {
+        this.stageId = stageId;
+        this.isQueued = false;
+        this.fixedWorkerNum = workerNum;
+        this.taskQueue = null;
+        this.downStreamWorkerNum = 0;
+        this.workerIndexAssigner = workerIndexAssigner;
+        this.leftChildWorkerNum = 0;
+        this.rightChildWorkerNum = 0;
     }
 
     /**
@@ -100,6 +116,8 @@ public StageCoordinator(int stageId, List<Task> pendingTasks, int workerIndex)
         this.taskQueue = new TaskQueue<>(pendingTasks);
         this.downStreamWorkerNum = 0;
         this.workerIndexAssigner = workerIndex;
+        this.leftChildWorkerNum = 0;
+        this.rightChildWorkerNum = 0;
     }
 
     /**
@@ -284,4 +302,34 @@ public int getFixedWorkerNum()
     {
         return this.fixedWorkerNum;
     }
+
+    public void setLeftChildWorkerNum(int num)
+    {
+        leftChildWorkerNum = num;
+    }
+
+    public int getLeftChildWorkerNum()
+    {
+        return leftChildWorkerNum;
+    }
+
+    public boolean leftChildWorkerIsEmpty()
+    {
+        return leftChildWorkerNum == 0;
+    }
+
+    public void setRightChildWorkerNum(int num)
+    {
+        rightChildWorkerNum = num;
+    }
+
+    public int getRightChildWorkerNum()
+    {
+        return rightChildWorkerNum;
+    }
+
+    public boolean rightChildWorkerIsEmpty()
+    {
+        return rightChildWorkerNum == 0;
+    }
 }

pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/PartitionedJoinOperator.java

@@ -143,7 +143,24 @@ public void initPlanCoordinator(PlanCoordinator planCoordinator, int parentStage
             joinInput.setStageId(this.joinStageId);
         }
         StageDependency joinStageDependency = new StageDependency(joinStageId, parentStageId, wideDependOnParent);
-        StageCoordinator joinStageCoordinator = new StageCoordinator(joinStageId, this.joinInputs.size());
+        StageCoordinator parentStageCoordinator = planCoordinator.getStageCoordinator(parentStageId);
+        StageCoordinator joinStageCoordinator;
+        if (parentStageCoordinator != null)
+        {
+            if (parentStageCoordinator.leftChildWorkerIsEmpty())
+            {
+                joinStageCoordinator = new StageCoordinator(joinStageId, this.joinInputs.size());
+                parentStageCoordinator.setLeftChildWorkerNum(this.joinInputs.size());
+            } else
+            {
+                joinStageCoordinator = new StageCoordinator(joinStageId, this.joinInputs.size(),
+                        parentStageCoordinator.getLeftChildWorkerNum());
+                parentStageCoordinator.setRightChildWorkerNum(this.joinInputs.size());
+            }
+        } else
+        {
+            joinStageCoordinator = new StageCoordinator(joinStageId, this.joinInputs.size());
+        }
         planCoordinator.addStageCoordinator(joinStageCoordinator, joinStageDependency);
         if (this.joinAlgo == JoinAlgorithm.PARTITIONED || this.joinAlgo == JoinAlgorithm.PARTITIONED_CHAIN)
         {