add broadcast join stream invoker

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

@@ -28,7 +28,7 @@ public enum WorkerType
     UNKNOWN, // The first enum value is the default value.
     SCAN, SCAN_STREAM,
     PARTITION, PARTITION_STREAMING,
-    BROADCAST_JOIN,
+    BROADCAST_JOIN, BROADCAST_JOIN_STREAMING,
     BROADCAST_CHAIN_JOIN,
     PARTITIONED_JOIN, PARTITIONED_JOIN_STREAMING,
     PARTITIONED_CHAIN_JOIN,

pixels-turbo/pixels-invoker-vhive/src/main/java/io/pixelsdb/pixels/invoker/vhive/BroadcastJoinStreamInvoker.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2023 PixelsDB.
+ *
+ * This file is part of Pixels.
+ *
+ * Pixels is free software: you can redistribute it and/or modify
+ * it under the terms of the Affero GNU General Public License as
+ * published by the Free Software Foundation, either version 3 of
+ * the License, or (at your option) any later version.
+ *
+ * Pixels is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * Affero GNU General Public License for more details.
+ *
+ * You should have received a copy of the Affero GNU General Public
+ * License along with Pixels.  If not, see
+ * <https://www.gnu.org/licenses/>.
+ */
+package io.pixelsdb.pixels.invoker.vhive;
+
+import com.alibaba.fastjson.JSON;
+import com.google.common.util.concurrent.ListenableFuture;
+import io.pixelsdb.pixels.common.turbo.Input;
+import io.pixelsdb.pixels.common.turbo.Output;
+import io.pixelsdb.pixels.planner.plan.physical.input.BroadcastJoinInput;
+import io.pixelsdb.pixels.planner.plan.physical.output.JoinOutput;
+import io.pixelsdb.pixels.turbo.TurboProto;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.concurrent.CompletableFuture;
+
+public class BroadcastJoinStreamInvoker extends VhiveInvoker
+{
+    private final Logger log = LogManager.getLogger(BroadcastJoinInvoker.class);
+
+    protected BroadcastJoinStreamInvoker(String functionName)
+    {
+        super(functionName);
+    }
+
+    @Override
+    public Output parseOutput(String outputJson)
+    {
+        return JSON.parseObject(outputJson, JoinOutput.class);
+    }
+
+    @Override
+    public CompletableFuture<Output> invoke(Input input)
+    {
+//        log.info(String.format("invoke BroadcastJoinStreamInput: %s", JSON.toJSONString(input, SerializerFeature.PrettyFormat, SerializerFeature.DisableCircularReferenceDetect)));
+        ListenableFuture<TurboProto.WorkerResponse> future = Vhive.Instance().getAsyncClient().broadcastJoinStream((BroadcastJoinInput) input);
+        return genCompletableFuture(future);
+    }
+}

pixels-turbo/pixels-invoker-vhive/src/main/java/io/pixelsdb/pixels/invoker/vhive/BroadcastJoinStreamInvokerProvider.java

@@ -0,0 +1,31 @@
+package io.pixelsdb.pixels.invoker.vhive;
+
+import io.pixelsdb.pixels.common.turbo.FunctionService;
+import io.pixelsdb.pixels.common.turbo.Invoker;
+import io.pixelsdb.pixels.common.turbo.InvokerProvider;
+import io.pixelsdb.pixels.common.turbo.WorkerType;
+import io.pixelsdb.pixels.common.utils.ConfigFactory;
+
+public class BroadcastJoinStreamInvokerProvider implements InvokerProvider
+{
+    private static final ConfigFactory config = ConfigFactory.Instance();
+
+    @Override
+    public Invoker createInvoker()
+    {
+        String broadcastJoinWorker = config.getProperty("broadcast.join.worker.name");
+        return new BroadcastJoinStreamInvoker(broadcastJoinWorker);
+    }
+
+    @Override
+    public WorkerType workerType()
+    {
+        return WorkerType.BROADCAST_JOIN_STREAMING;
+    }
+
+    @Override
+    public boolean compatibleWith(FunctionService functionService)
+    {
+        return functionService.equals(FunctionService.vhive);
+    }
+}

pixels-turbo/pixels-invoker-vhive/src/main/java/io/pixelsdb/pixels/invoker/vhive/WorkerAsyncClient.java

@@ -93,6 +93,15 @@ public ListenableFuture<TurboProto.WorkerResponse> broadcastJoin(BroadcastJoinIn
         return this.stub.process(request);
     }
 
+    public ListenableFuture<TurboProto.WorkerResponse> broadcastJoinStream(BroadcastJoinInput input)
+    {
+        TurboProto.WorkerRequest request = TurboProto.WorkerRequest.newBuilder()
+                .setWorkerType(String.valueOf(WorkerType.BROADCAST_JOIN_STREAMING))
+                .setJson(JSON.toJSONString(input, SerializerFeature.DisableCircularReferenceDetect))
+                .build();
+        return this.stub.process(request);
+    }
+
     public ListenableFuture<TurboProto.WorkerResponse> partitionChainJoin(PartitionedChainJoinInput input)
     {
         TurboProto.WorkerRequest request = TurboProto.WorkerRequest.newBuilder()

pixels-turbo/pixels-invoker-vhive/src/main/resources/META-INF/services/io.pixelsdb.pixels.common.turbo.InvokerProvider

@@ -5,6 +5,7 @@ io.pixelsdb.pixels.invoker.vhive.PartitionStreamInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.PartitionJoinInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.PartitionedJoinStreamInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.PartitionChainJoinInvokerProvider
+io.pixelsdb.pixels.invoker.vhive.BroadcastJoinStreamInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.BroadcastJoinInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.BroadcastChainJoinInvokerProvider
 io.pixelsdb.pixels.invoker.vhive.AggregationInvokerProvider