one worker write to one port of down stream worker

Author: huasiy

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

@@ -31,13 +31,15 @@
 public class Worker<WI extends WorkerInfo>
 {
     private final long workerId;
+    private int workerPortIndex;
     private final Lease lease;
     private final WI workerInfo;
     private boolean terminated;
 
-    public Worker(long workerId, Lease lease, WI workerInfo)
+    public Worker(long workerId, Lease lease, int workerPortIndex, WI workerInfo)
     {
         this.workerId = workerId;
+        this.workerPortIndex = workerPortIndex;
         this.lease = requireNonNull(lease, "lease is null");
         this.workerInfo = requireNonNull(workerInfo, "worker info is null");
         this.terminated = false;
@@ -48,6 +50,10 @@ public long getWorkerId()
         return workerId;
     }
 
+    public void setWorkerPortIndex(int index) { this.workerPortIndex = index; }
+
+    public int getWorkerPortIndex() { return workerPortIndex; }
+
     public WI getWorkerInfo()
     {
         return workerInfo;

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

@@ -115,6 +115,7 @@ public void addWorker(Worker<CFWorkerInfo> worker)
                 if (downStreamWorkerNum > fixedWorkerNum)
                 {
                     // one-to-multiple stream
+                    // TODO: find a query to test
                     List<Integer> workerIndexs = new ArrayList<>();
                     int num = downStreamWorkerNum / fixedWorkerNum;
                     if (downStreamWorkerNum > fixedWorkerNum*num)
@@ -129,10 +130,11 @@ public void addWorker(Worker<CFWorkerInfo> worker)
                 } else
                 {
                     // multiple-to-one stream
-                    if (workerIndexAssigner < downStreamWorkerNum)
-                    {
-                        worker.getWorkerInfo().setPassSchema(true);
-                    }
+//                    if (workerIndexAssigner < downStreamWorkerNum)
+//                    {
+//                        worker.getWorkerInfo().setPassSchema(true);
+//                    }
+                    worker.setWorkerPortIndex(this.workerIndexAssigner / this.downStreamWorkerNum);
                     List<Integer> workerIndexes = new ArrayList<>();
                     workerIndexes.add(this.workerIndexAssigner % this.downStreamWorkerNum);
                     this.workerIndexAssigner++;
@@ -141,7 +143,7 @@ public void addWorker(Worker<CFWorkerInfo> worker)
             } else
             {
                 // assume one-to-one stream
-                worker.getWorkerInfo().setPassSchema(true);
+                worker.setWorkerPortIndex(0);
                 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,9 +74,8 @@ 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);
+                new Lease(response.getLeasePeriodMs(), response.getLeaseStartTimeMs()), response.getWorkerPortIndex(), workerInfo);
     }
 
     /**

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

@@ -57,7 +57,7 @@ public void registerWorker(TurboProto.RegisterWorkerRequest request,
         CFWorkerInfo workerInfo = new CFWorkerInfo(request.getWorkerInfo());
         Lease lease = new Lease(WorkerLeasePeriodMs, System.currentTimeMillis());
         long workerId = CFWorkerManager.Instance().createWorkerId();
-        Worker<CFWorkerInfo> worker = new Worker<>(workerId, lease, workerInfo);
+        Worker<CFWorkerInfo> worker = new Worker<>(workerId, lease, 0, workerInfo);
         CFWorkerManager.Instance().registerCFWorker(worker);
         log.debug("register worker, local address: " + workerInfo.getIp() + ", transId: " + workerInfo.getTransId()
                 + ", stageId: " + workerInfo.getStageId() + ", workerId: " + workerId);
@@ -67,8 +67,8 @@ public void registerWorker(TurboProto.RegisterWorkerRequest request,
         requireNonNull(stageCoordinator, "stage coordinator is not found");
         stageCoordinator.addWorker(worker);
         TurboProto.RegisterWorkerResponse response = TurboProto.RegisterWorkerResponse.newBuilder()
-                .setErrorCode(SUCCESS).setWorkerId(workerId).setLeasePeriodMs(lease.getPeriodMs())
-                .setLeaseStartTimeMs(lease.getStartTimeMs()).setPassSchema(worker.getWorkerInfo().getPassSchema()).build();
+                .setErrorCode(SUCCESS).setWorkerId(workerId).setWorkerPortIndex(worker.getWorkerPortIndex()).setLeasePeriodMs(lease.getPeriodMs())
+                .setLeaseStartTimeMs(lease.getStartTimeMs()).build();
         responseObserver.onNext(response);
         responseObserver.onCompleted();
     }