[Issue #468] improve operator name setting. (#470)

bianhq · web-flow · commit 0bf317bba9ae · 2023-06-08T04:05:19.000+02:00
Ensure that the operator name is set for serverless worker inputs when the inputs are added to an operator.
diff --git a/docs/TPC-H.md b/docs/TPC-H.md
@@ -68,17 +68,25 @@ Connect to trino-cli:
 cd ~/opt/trino-server
 ./bin/trino --server localhost:8080 --catalog pixels --schema tpch
 ```
-Execute the TPC-H queries in trino-cli.
+In trino-cli, select the ordered data layout by setting the two session properties:
+```sql
+set session pixels.ordered_path_enabled=true
+set session pixels.compact_path_enabled=false
+```
+By default, both paths are enabled. You can also enable the compact path and disable the ordered path when [data compaction](#data-compaction) is done.
+After selecting the data layout, execute the TPC-H queries in trino-cli.
 
 ## Data Compaction*
 This is optional. It is only needed if we want to test the query performance on the compact layout.
 In pixels-cli, use the following commands to compact the files in the ordered path of each table:
 ```bash
 COMPACT -s tpch -t customer -n no -c 2
 COMPACT -s tpch -t lineitem -n no -c 16
+COMPACT -s tpch -t nation -n no -c 1
 COMPACT -s tpch -t orders -n no -c 8
 COMPACT -s tpch -t part -n no -c 1
 COMPACT -s tpch -t partsupp -n no -c 8
+COMPACT -s tpch -t region -n no -c 1
 COMPACT -s tpch -t supplier -n no -c 1
 ```
 The tables `nation` and `region` are too small, no need to compact them.
diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/AggregationOperator.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/AggregationOperator.java
@@ -70,13 +70,22 @@ public AggregationOperator(String name, List<AggregationInput> finalAggrInputs,
         requireNonNull(finalAggrInputs, "finalAggrInputs is null");
         checkArgument(!finalAggrInputs.isEmpty(), "finalAggrInputs is empty");
         this.finalAggrInputs = ImmutableList.copyOf(finalAggrInputs);
+        for (AggregationInput aggrInput : this.finalAggrInputs)
+        {
+            aggrInput.setOperatorName(name);
+        }
+
         if (scanInputs == null || scanInputs.isEmpty())
         {
             this.scanInputs = ImmutableList.of();
         }
         else
         {
             this.scanInputs = ImmutableList.copyOf(scanInputs);
+            for (ScanInput scanInput : this.scanInputs)
+            {
+                scanInput.setOperatorName(name);
+            }
         }
     }
 
@@ -122,7 +131,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()
                 int i = 0;
                 for (AggregationInput preAggrInput : this.finalAggrInputs)
                 {
-                    preAggrInput.setOperatorName(this.getName());
                     this.finalAggrOutputs[i++] = InvokerFactory.Instance()
                             .getInvoker(WorkerType.AGGREGATION).invoke(preAggrInput);
                 }
@@ -157,7 +165,6 @@ public CompletableFuture<Void> executePrev()
                     int i = 0;
                     for (ScanInput scanInput : this.scanInputs)
                     {
-                        scanInput.setOperatorName(this.getName());
                         this.scanOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.SCAN).invoke(scanInput);
                     }
diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/PartitionedJoinOperator.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/PartitionedJoinOperator.java
@@ -66,6 +66,10 @@ public PartitionedJoinOperator(String name, List<PartitionInput> smallPartitionI
         else
         {
             this.smallPartitionInputs = ImmutableList.copyOf(smallPartitionInputs);
+            for (PartitionInput partitionInput : this.smallPartitionInputs)
+            {
+                partitionInput.setOperatorName(name);
+            }
         }
         if (largePartitionInputs == null || largePartitionInputs.isEmpty())
         {
@@ -74,6 +78,10 @@ public PartitionedJoinOperator(String name, List<PartitionInput> smallPartitionI
         else
         {
             this.largePartitionInputs = ImmutableList.copyOf(largePartitionInputs);
+            for (PartitionInput partitionInput : this.largePartitionInputs)
+            {
+                partitionInput.setOperatorName(name);
+            }
         }
     }
 
@@ -139,7 +147,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()
             for (int i = 0; i < joinInputs.size(); ++i)
             {
                 JoinInput joinInput = joinInputs.get(i);
-                joinInput.setOperatorName(this.getName());
                 if (joinAlgo == JoinAlgorithm.PARTITIONED)
                 {
                     joinOutputs[i] = InvokerFactory.Instance()
@@ -191,7 +198,6 @@ public CompletableFuture<Void> executePrev()
                     int i = 0;
                     for (PartitionInput partitionInput : largePartitionInputs)
                     {
-                        partitionInput.setOperatorName(this.getName());
                         largePartitionOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.PARTITION).invoke((partitionInput));
                     }
@@ -210,7 +216,6 @@ public CompletableFuture<Void> executePrev()
                     int i = 0;
                     for (PartitionInput partitionInput : smallPartitionInputs)
                     {
-                        partitionInput.setOperatorName(this.getName());
                         smallPartitionOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.PARTITION).invoke((partitionInput));
                     }
@@ -228,7 +233,6 @@ public CompletableFuture<Void> executePrev()
                     int i = 0;
                     for (PartitionInput partitionInput : smallPartitionInputs)
                     {
-                        partitionInput.setOperatorName(this.getName());
                         smallPartitionOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.PARTITION).invoke((partitionInput));
                     }
@@ -239,7 +243,6 @@ public CompletableFuture<Void> executePrev()
                     i = 0;
                     for (PartitionInput partitionInput : largePartitionInputs)
                     {
-                        partitionInput.setOperatorName(this.getName());
                         largePartitionOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.PARTITION).invoke((partitionInput));
                     }
diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/SingleStageJoinOperator.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/SingleStageJoinOperator.java
@@ -51,6 +51,8 @@ public class SingleStageJoinOperator extends JoinOperator
     public SingleStageJoinOperator(String name, JoinInput joinInput, JoinAlgorithm joinAlgo)
     {
         super(name);
+        // ImmutableList.of() add the reference of joinInput into the returned list
+        joinInput.setOperatorName(name);
         this.joinInputs = ImmutableList.of(joinInput);
         this.joinAlgo = joinAlgo;
     }
@@ -59,6 +61,10 @@ public SingleStageJoinOperator(String name, List<JoinInput> joinInputs, JoinAlgo
     {
         super(name);
         this.joinInputs = ImmutableList.copyOf(joinInputs);
+        for (JoinInput joinInput : this.joinInputs)
+        {
+            joinInput.setOperatorName(name);
+        }
         this.joinAlgo = joinAlgo;
     }
 
@@ -116,7 +122,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()
             for (int i = 0; i < joinInputs.size(); ++i)
             {
                 JoinInput joinInput = joinInputs.get(i);
-                joinInput.setOperatorName(this.getName());
                 if (joinAlgo == JoinAlgorithm.BROADCAST)
                 {
                     joinOutputs[i] = InvokerFactory.Instance()
diff --git a/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/StarlingAggregationOperator.java b/pixels-planner/src/main/java/io/pixelsdb/pixels/planner/plan/physical/StarlingAggregationOperator.java
@@ -71,13 +71,21 @@ public StarlingAggregationOperator(String name, List<AggregationInput> finalAggr
         requireNonNull(finalAggrInputs, "finalAggrInputs is null");
         checkArgument(!finalAggrInputs.isEmpty(), "finalAggrInputs is empty");
         this.finalAggrInputs = ImmutableList.copyOf(finalAggrInputs);
+        for (AggregationInput aggrInput : this.finalAggrInputs)
+        {
+            aggrInput.setOperatorName(name);
+        }
         if (partitionInputs == null || partitionInputs.isEmpty())
         {
             this.partitionInputs = ImmutableList.of();
         }
         else
         {
             this.partitionInputs = ImmutableList.copyOf(partitionInputs);
+            for (PartitionInput partitionInput : this.partitionInputs)
+            {
+                partitionInput.setOperatorName(name);
+            }
         }
     }
 
@@ -123,7 +131,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()
                 int i = 0;
                 for (AggregationInput preAggrInput : this.finalAggrInputs)
                 {
-                    preAggrInput.setOperatorName(this.getName());
                     this.finalAggrOutputs[i++] = InvokerFactory.Instance()
                             .getInvoker(WorkerType.AGGREGATION).invoke(preAggrInput);
                 }
@@ -158,7 +165,6 @@ public CompletableFuture<Void> executePrev()
                     int i = 0;
                     for (PartitionInput partitionInput : this.partitionInputs)
                     {
-                        partitionInput.setOperatorName(this.getName());
                         this.partitionOutputs[i++] = InvokerFactory.Instance()
                                 .getInvoker(WorkerType.PARTITION).invoke(partitionInput);
                     }
diff --git a/pixels-planner/src/test/java/io/pixelsdb/pixels/planner/TestOutput.java b/pixels-planner/src/test/java/io/pixelsdb/pixels/planner/TestOutput.java
@@ -19,7 +19,9 @@
  */
 package io.pixelsdb.pixels.planner;
 
+import com.alibaba.fastjson.JSON;
 import com.google.gson.Gson;
+import io.pixelsdb.pixels.planner.plan.physical.input.ScanInput;
 import io.pixelsdb.pixels.planner.plan.physical.output.ScanOutput;
 import org.junit.Test;
 
@@ -42,4 +44,19 @@ public void testEncodeScanOutput()
         assert json != null && !json.isEmpty();
         System.out.println(json);
     }
+
+    @Test
+    public void testEncodeScanInput()
+    {
+        ScanInput scanInput = new ScanInput();
+        scanInput.setTransId(1);
+        scanInput.setOperatorName("scan");
+        System.out.println(JSON.toJSONString(scanInput));
+        Gson gson = new Gson();
+        System.out.println(gson.toJson(scanInput));
+        ScanInput scanInput1 = JSON.parseObject(JSON.toJSONString(scanInput), ScanInput.class);
+        ScanInput scanInput2 = gson.fromJson(gson.toJson(scanInput), ScanInput.class);
+        System.out.println(scanInput1.getOperatorName());
+        System.out.println(scanInput2.getOperatorName());
+    }
 }
diff --git a/pixels-turbo/pixels-worker-vhive/src/main/java/io/pixelsdb/pixels/worker/vhive/utils/ServiceImpl.java b/pixels-turbo/pixels-worker-vhive/src/main/java/io/pixelsdb/pixels/worker/vhive/utils/ServiceImpl.java
@@ -72,9 +72,9 @@ public void execute(TurboProto.WorkerRequest request,
                 output = handler.handleRequest(input);
                 Utils.stopProfile(JFRFilename);
 
-                Utils.upload(JFRFilename, String.format("%s_%s/%s",
+                Utils.upload(JFRFilename, String.format("%s/%s/%s",
                         input.getTransId(), notNullOrElse(input.getOperatorName(), "default"), JFRFilename));
-                log.info(String.format("upload JFR file to experiments/%s_%s/%s successfully",
+                log.info(String.format("upload JFR file to experiments/%s/%s/%s successfully",
                         input.getTransId(),  notNullOrElse(input.getOperatorName(), "default"), JFRFilename));
             } else
             {
@@ -83,10 +83,10 @@ public void execute(TurboProto.WorkerRequest request,
                 output = handler.handleRequest(input);
             }
             Utils.dump(JSONFilename, input, output);
-            Utils.upload(JSONFilename, String.format("%s_%s/%s",
-                    input.getTransId(),  notNullOrElse(input.getOperatorName(), "none"), JSONFilename));
-            log.info(String.format("upload JSON file to experiments/%s_%s/%s successfully",
-                    input.getTransId(),  notNullOrElse(input.getOperatorName(), "none"), JSONFilename));
+            Utils.upload(JSONFilename, String.format("%s/%s/%s",
+                    input.getTransId(),  notNullOrElse(input.getOperatorName(), "default"), JSONFilename));
+            log.info(String.format("upload JSON file to experiments/%s/%s/%s successfully",
+                    input.getTransId(),  notNullOrElse(input.getOperatorName(), "default"), JSONFilename));
 
             log.info(String.format("get output successfully: %s", JSON.toJSONString(output)));
         } catch (Exception e)

Original file line number	Diff line number	Diff line change
`@@ -70,13 +70,22 @@ public AggregationOperator(String name, List<AggregationInput> finalAggrInputs,`
`70`	`70`	`requireNonNull(finalAggrInputs, "finalAggrInputs is null");`
`71`	`71`	`checkArgument(!finalAggrInputs.isEmpty(), "finalAggrInputs is empty");`
`72`	`72`	`this.finalAggrInputs = ImmutableList.copyOf(finalAggrInputs);`
	`73`	`+ for (AggregationInput aggrInput : this.finalAggrInputs)`
	`74`	`+ {`
	`75`	`+ aggrInput.setOperatorName(name);`
	`76`	`+ }`
	`77`	`+`
`73`	`78`	`if (scanInputs == null \|\| scanInputs.isEmpty())`
`74`	`79`	`{`
`75`	`80`	`this.scanInputs = ImmutableList.of();`
`76`	`81`	`}`
`77`	`82`	`else`
`78`	`83`	`{`
`79`	`84`	`this.scanInputs = ImmutableList.copyOf(scanInputs);`
	`85`	`+ for (ScanInput scanInput : this.scanInputs)`
	`86`	`+ {`
	`87`	`+ scanInput.setOperatorName(name);`
	`88`	`+ }`
`80`	`89`	`}`
`81`	`90`	`}`
`82`	`91`
`@@ -122,7 +131,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()`
`122`	`131`	`int i = 0;`
`123`	`132`	`for (AggregationInput preAggrInput : this.finalAggrInputs)`
`124`	`133`	`{`
`125`		`- preAggrInput.setOperatorName(this.getName());`
`126`	`134`	`this.finalAggrOutputs[i++] = InvokerFactory.Instance()`
`127`	`135`	`.getInvoker(WorkerType.AGGREGATION).invoke(preAggrInput);`
`128`	`136`	`}`
`@@ -157,7 +165,6 @@ public CompletableFuture<Void> executePrev()`
`157`	`165`	`int i = 0;`
`158`	`166`	`for (ScanInput scanInput : this.scanInputs)`
`159`	`167`	`{`
`160`		`- scanInput.setOperatorName(this.getName());`
`161`	`168`	`this.scanOutputs[i++] = InvokerFactory.Instance()`
`162`	`169`	`.getInvoker(WorkerType.SCAN).invoke(scanInput);`
`163`	`170`	`}`
Original file line number	Diff line number	Diff line change
`@@ -66,6 +66,10 @@ public PartitionedJoinOperator(String name, List<PartitionInput> smallPartitionI`
`66`	`66`	`else`
`67`	`67`	`{`
`68`	`68`	`this.smallPartitionInputs = ImmutableList.copyOf(smallPartitionInputs);`
	`69`	`+ for (PartitionInput partitionInput : this.smallPartitionInputs)`
	`70`	`+ {`
	`71`	`+ partitionInput.setOperatorName(name);`
	`72`	`+ }`
`69`	`73`	`}`
`70`	`74`	`if (largePartitionInputs == null \|\| largePartitionInputs.isEmpty())`
`71`	`75`	`{`
`@@ -74,6 +78,10 @@ public PartitionedJoinOperator(String name, List<PartitionInput> smallPartitionI`
`74`	`78`	`else`
`75`	`79`	`{`
`76`	`80`	`this.largePartitionInputs = ImmutableList.copyOf(largePartitionInputs);`
	`81`	`+ for (PartitionInput partitionInput : this.largePartitionInputs)`
	`82`	`+ {`
	`83`	`+ partitionInput.setOperatorName(name);`
	`84`	`+ }`
`77`	`85`	`}`
`78`	`86`	`}`
`79`	`87`
`@@ -139,7 +147,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()`
`139`	`147`	`for (int i = 0; i < joinInputs.size(); ++i)`
`140`	`148`	`{`
`141`	`149`	`JoinInput joinInput = joinInputs.get(i);`
`142`		`- joinInput.setOperatorName(this.getName());`
`143`	`150`	`if (joinAlgo == JoinAlgorithm.PARTITIONED)`
`144`	`151`	`{`
`145`	`152`	`joinOutputs[i] = InvokerFactory.Instance()`
`@@ -191,7 +198,6 @@ public CompletableFuture<Void> executePrev()`
`191`	`198`	`int i = 0;`
`192`	`199`	`for (PartitionInput partitionInput : largePartitionInputs)`
`193`	`200`	`{`
`194`		`- partitionInput.setOperatorName(this.getName());`
`195`	`201`	`largePartitionOutputs[i++] = InvokerFactory.Instance()`
`196`	`202`	`.getInvoker(WorkerType.PARTITION).invoke((partitionInput));`
`197`	`203`	`}`
`@@ -210,7 +216,6 @@ public CompletableFuture<Void> executePrev()`
`210`	`216`	`int i = 0;`
`211`	`217`	`for (PartitionInput partitionInput : smallPartitionInputs)`
`212`	`218`	`{`
`213`		`- partitionInput.setOperatorName(this.getName());`
`214`	`219`	`smallPartitionOutputs[i++] = InvokerFactory.Instance()`
`215`	`220`	`.getInvoker(WorkerType.PARTITION).invoke((partitionInput));`
`216`	`221`	`}`
`@@ -228,7 +233,6 @@ public CompletableFuture<Void> executePrev()`
`228`	`233`	`int i = 0;`
`229`	`234`	`for (PartitionInput partitionInput : smallPartitionInputs)`
`230`	`235`	`{`
`231`		`- partitionInput.setOperatorName(this.getName());`
`232`	`236`	`smallPartitionOutputs[i++] = InvokerFactory.Instance()`
`233`	`237`	`.getInvoker(WorkerType.PARTITION).invoke((partitionInput));`
`234`	`238`	`}`
`@@ -239,7 +243,6 @@ public CompletableFuture<Void> executePrev()`
`239`	`243`	`i = 0;`
`240`	`244`	`for (PartitionInput partitionInput : largePartitionInputs)`
`241`	`245`	`{`
`242`		`- partitionInput.setOperatorName(this.getName());`
`243`	`246`	`largePartitionOutputs[i++] = InvokerFactory.Instance()`
`244`	`247`	`.getInvoker(WorkerType.PARTITION).invoke((partitionInput));`
`245`	`248`	`}`
Original file line number	Diff line number	Diff line change
`@@ -51,6 +51,8 @@ public class SingleStageJoinOperator extends JoinOperator`
`51`	`51`	`public SingleStageJoinOperator(String name, JoinInput joinInput, JoinAlgorithm joinAlgo)`
`52`	`52`	`{`
`53`	`53`	`super(name);`
	`54`	`+ // ImmutableList.of() add the reference of joinInput into the returned list`
	`55`	`+ joinInput.setOperatorName(name);`
`54`	`56`	`this.joinInputs = ImmutableList.of(joinInput);`
`55`	`57`	`this.joinAlgo = joinAlgo;`
`56`	`58`	`}`
`@@ -59,6 +61,10 @@ public SingleStageJoinOperator(String name, List<JoinInput> joinInputs, JoinAlgo`
`59`	`61`	`{`
`60`	`62`	`super(name);`
`61`	`63`	`this.joinInputs = ImmutableList.copyOf(joinInputs);`
	`64`	`+ for (JoinInput joinInput : this.joinInputs)`
	`65`	`+ {`
	`66`	`+ joinInput.setOperatorName(name);`
	`67`	`+ }`
`62`	`68`	`this.joinAlgo = joinAlgo;`
`63`	`69`	`}`
`64`	`70`
`@@ -116,7 +122,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()`
`116`	`122`	`for (int i = 0; i < joinInputs.size(); ++i)`
`117`	`123`	`{`
`118`	`124`	`JoinInput joinInput = joinInputs.get(i);`
`119`		`- joinInput.setOperatorName(this.getName());`
`120`	`125`	`if (joinAlgo == JoinAlgorithm.BROADCAST)`
`121`	`126`	`{`
`122`	`127`	`joinOutputs[i] = InvokerFactory.Instance()`
Original file line number	Diff line number	Diff line change
`@@ -71,13 +71,21 @@ public StarlingAggregationOperator(String name, List<AggregationInput> finalAggr`
`71`	`71`	`requireNonNull(finalAggrInputs, "finalAggrInputs is null");`
`72`	`72`	`checkArgument(!finalAggrInputs.isEmpty(), "finalAggrInputs is empty");`
`73`	`73`	`this.finalAggrInputs = ImmutableList.copyOf(finalAggrInputs);`
	`74`	`+ for (AggregationInput aggrInput : this.finalAggrInputs)`
	`75`	`+ {`
	`76`	`+ aggrInput.setOperatorName(name);`
	`77`	`+ }`
`74`	`78`	`if (partitionInputs == null \|\| partitionInputs.isEmpty())`
`75`	`79`	`{`
`76`	`80`	`this.partitionInputs = ImmutableList.of();`
`77`	`81`	`}`
`78`	`82`	`else`
`79`	`83`	`{`
`80`	`84`	`this.partitionInputs = ImmutableList.copyOf(partitionInputs);`
	`85`	`+ for (PartitionInput partitionInput : this.partitionInputs)`
	`86`	`+ {`
	`87`	`+ partitionInput.setOperatorName(name);`
	`88`	`+ }`
`81`	`89`	`}`
`82`	`90`	`}`
`83`	`91`
`@@ -123,7 +131,6 @@ public CompletableFuture<CompletableFuture<?>[]> execute()`
`123`	`131`	`int i = 0;`
`124`	`132`	`for (AggregationInput preAggrInput : this.finalAggrInputs)`
`125`	`133`	`{`
`126`		`- preAggrInput.setOperatorName(this.getName());`
`127`	`134`	`this.finalAggrOutputs[i++] = InvokerFactory.Instance()`
`128`	`135`	`.getInvoker(WorkerType.AGGREGATION).invoke(preAggrInput);`
`129`	`136`	`}`
`@@ -158,7 +165,6 @@ public CompletableFuture<Void> executePrev()`
`158`	`165`	`int i = 0;`
`159`	`166`	`for (PartitionInput partitionInput : this.partitionInputs)`
`160`	`167`	`{`
`161`		`- partitionInput.setOperatorName(this.getName());`
`162`	`168`	`this.partitionOutputs[i++] = InvokerFactory.Instance()`
`163`	`169`	`.getInvoker(WorkerType.PARTITION).invoke(partitionInput);`
`164`	`170`	`}`