only one worker will pass schema to down stream

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;

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);

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

@@ -74,6 +74,7 @@ public Worker<CFWorkerInfo> registerWorker(CFWorkerInfo workerInfo) throws Worke
         {
             throw new WorkerCoordinateException("failed to register worker, error code=" + response.getErrorCode());
         }
+        workerInfo.setPassSchema(response.getPassSchema());
         return new Worker<>(response.getWorkerId(),
                 new Lease(response.getLeasePeriodMs(), response.getLeaseStartTimeMs()), workerInfo);
     }

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

@@ -68,7 +68,7 @@ public void registerWorker(TurboProto.RegisterWorkerRequest request,
         stageCoordinator.addWorker(worker);
         TurboProto.RegisterWorkerResponse response = TurboProto.RegisterWorkerResponse.newBuilder()
                 .setErrorCode(SUCCESS).setWorkerId(workerId).setLeasePeriodMs(lease.getPeriodMs())
-                .setLeaseStartTimeMs(lease.getStartTimeMs()).build();
+                .setLeaseStartTimeMs(lease.getStartTimeMs()).setPassSchema(worker.getWorkerInfo().getPassSchema()).build();
         responseObserver.onNext(response);
         responseObserver.onCompleted();
     }

proto/turbo.proto

@@ -109,6 +109,7 @@ message RegisterWorkerResponse {
   int64 workerId = 2; // the unique id assigned by the coordinator for this worker
   int64 leaseStartTimeMs = 3; // the time since the epoch in milliseconds of the lease
   int64 leasePeriodMs = 4; // the valid period in milliseconds of the lease
+  bool passSchema = 5; // if this worker send schema to down stream worker
 }
 
 message GetDownstreamWorkersRequest {