only one worker will pass schema to down stream

pixelsdb · Oct 28, 2024 · e0245f6 · e0245f6
1 parent 119fe31
commit e0245f6
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/coordinate/CFWorkerInfo.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/coordinate/CFWorkerInfo.java
@@ -36,6 +36,7 @@ public class CFWorkerInfo implements WorkerInfo
     private final int stageId;
     private final String operatorName;
     private final List<Integer> hashValues;
+    private boolean passSchema;
 
     public CFWorkerInfo(String ip, int port, long transId, int stageId,
                         String operatorName, List<Integer> hashValues)
@@ -46,6 +47,7 @@ public CFWorkerInfo(String ip, int port, long transId, int stageId,
         this.stageId = stageId;
         this.operatorName = operatorName;
         this.hashValues = hashValues;
+        this.passSchema = false;
     }
 
     public CFWorkerInfo(TurboProto.WorkerInfo workerInfo)
@@ -56,8 +58,13 @@ public CFWorkerInfo(TurboProto.WorkerInfo workerInfo)
         this.stageId = workerInfo.getStageId();
         this.operatorName = workerInfo.getOperatorName();
         this.hashValues = workerInfo.getHashValuesList();
+        this.passSchema = false;
     }
 
+    public void setPassSchema(boolean passSchema) { this.passSchema = passSchema; }
+
+    public boolean getPassSchema() { return passSchema; }
+
     public String getIp()
     {
         return ip;

diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/coordinate/StageCoordinator.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/coordinate/StageCoordinator.java
@@ -129,6 +129,10 @@ public void addWorker(Worker<CFWorkerInfo> worker)
                 } else
                 {
                     // multiple-to-one stream
+                    if (workerIndexAssigner <= downStreamWorkerNum)
+                    {
+                        worker.getWorkerInfo().setPassSchema(true);
+                    }
                     List<Integer> workerIndexes = new ArrayList<>();
                     workerIndexes.add(this.workerIndexAssigner % this.downStreamWorkerNum);
                     this.workerIndexAssigner++;
@@ -137,6 +141,7 @@ public void addWorker(Worker<CFWorkerInfo> worker)
             } else
             {
                 // assume one-to-one stream
+                worker.getWorkerInfo().setPassSchema(true);
                 List<Integer> workerIndexs = new ArrayList<>(this.workerIndexAssigner);
                 this.workerIndexAssigner++;
                 this.workerIdToWorkerIndex.put(worker.getWorkerId(), workerIndexs);