fix: tpc-h (IGinX-THU#519)

* test: remove temp tables and udf in tpc-h test

* fix: tpc-h

Author: aqni

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/compute/PhysicalFunctions.java

@@ -83,25 +83,31 @@ public static IntVector filter(BufferAllocator allocator, BitVector bitmap) {
 
   public static <OUTPUT extends FieldVector> void takeTo(
       BaseIntVector selection, OUTPUT output, OUTPUT input) {
-    if (selection.getField().isNullable()) {
-      throw new IllegalArgumentException("Selection vector must be not nullable");
-    }
-
     int outputOffset = output.getValueCount();
     int outputRowCount = outputOffset + selection.getValueCount();
+
+    boolean useSetSafe;
     if (output instanceof BaseFixedWidthVector) {
-      output.setValueCount(outputRowCount);
-      for (int selectionIndex = 0; selectionIndex < selection.getValueCount(); selectionIndex++) {
-        int outputIndex = outputOffset + selectionIndex;
-        output.copyFrom((int) selection.getValueAsLong(selectionIndex), outputIndex, input);
-      }
+      ((BaseFixedWidthVector) output).allocateNew(outputRowCount);
+      useSetSafe = false;
     } else {
-      for (int selectionIndex = 0; selectionIndex < selection.getValueCount(); selectionIndex++) {
-        int outputIndex = outputOffset + selectionIndex;
-        output.copyFromSafe((int) selection.getValueAsLong(selectionIndex), outputIndex, input);
+      output.setInitialCapacity(outputRowCount);
+      useSetSafe = true;
+    }
+    boolean isNullable = input.getField().isNullable();
+    for (int selectionIndex = 0; selectionIndex < selection.getValueCount(); selectionIndex++) {
+      int outputIndex = outputOffset + selectionIndex;
+      if (isNullable && selection.isNull(selectionIndex)) {
+        output.setNull(outputIndex);
+      } else {
+        if (useSetSafe) {
+          output.copyFromSafe((int) selection.getValueAsLong(selectionIndex), outputIndex, input);
+        } else {
+          output.copyFrom((int) selection.getValueAsLong(selectionIndex), outputIndex, input);
+        }
       }
-      output.setValueCount(outputRowCount);
     }
+    output.setValueCount(outputRowCount);
   }
 
   public static void takeTo(
@@ -119,9 +125,6 @@ public static void takeTo(
   @SuppressWarnings("unchecked")
   public static <OUTPUT extends FieldVector> OUTPUT take(
       BufferAllocator allocator, BaseIntVector selection, OUTPUT input) {
-    if (selection.getField().isNullable()) {
-      throw new IllegalArgumentException("Selection vector must be not nullable");
-    }
     TransferPair transferPair = input.getTransferPair(allocator);
     OUTPUT result = (OUTPUT) transferPair.getTo();
     result.setInitialCapacity(selection.getValueCount());

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/compute/join/JoinHashMap.java

@@ -32,9 +32,6 @@
 import javax.annotation.Nullable;
 import javax.annotation.WillClose;
 import javax.annotation.WillCloseWhenClosed;
-import org.apache.arrow.algorithm.sort.DefaultVectorComparators;
-import org.apache.arrow.algorithm.sort.IndexSorter;
-import org.apache.arrow.algorithm.sort.StableVectorComparator;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.util.Preconditions;
 import org.apache.arrow.vector.*;
@@ -106,7 +103,7 @@ public void probe(VectorSchemaRoot probeSideBatch) throws ComputeException {
           IntVector buildSideCandidateIndices = buildSideCandidateIndicesBuilder.build();
           IntVector proSideCandidateIndices = probeSideCandidateIndicesBuilder.build()) {
         boolean[] probeSideMatched = new boolean[probeSideBatch.getRowCount()];
-        output(
+        outputMatchedAndUnmatched(
             outputDictionaryProvider,
             buildSideCandidateIndices,
             proSideCandidateIndices,
@@ -143,12 +140,12 @@ public void flush() throws ComputeException {
           IntVector buildSideIndices = buildSideIndicesBuilder.build(buildSideUnmatchedCount);
           IntVector probeSideIndices = probeSideIndicesBuilder.build(buildSideUnmatchedCount);
           BitVector mark = markBuilder.build(buildSideUnmatchedCount)) {
-        output(dictionary, buildSideIndices, probeSideIndices, mark);
+        output(dictionary, buildSideIndices, probeSideIndices, mark, 0);
       }
     }
   }
 
-  private void output(
+  private void outputMatchedAndUnmatched(
       ArrayDictionaryProvider dictionary,
       IntVector buildSideCandidateIndices,
       IntVector proSideCandidateIndices,
@@ -160,32 +157,37 @@ private void output(
                 allocator, "buildSideIndices", buildSideCandidateIndices.getValueCount());
         SelectionBuilder probeSideIndicesBuilder =
             new SelectionBuilder(
-                allocator, "probeSideIndices", proSideCandidateIndices.getValueCount());
-        MarkBuilder markBuilder = getMarkBuilder(probeSideMatched.length)) {
+                allocator, "probeSideIndices", proSideCandidateIndices.getValueCount())) {
 
+      int matchedCount = 0;
       try (VectorSchemaRoot candidate =
               getDictionaryEncodedBatch(
                   buildSideCandidateIndices, proSideCandidateIndices, dictionary);
           BaseIntVector indicesSelection = matcher.filter(allocator, dictionary, candidate, null)) {
-        outputMatched(
-            buildSideIndicesBuilder,
-            probeSideIndicesBuilder,
-            markBuilder,
-            buildSideCandidateIndices,
-            proSideCandidateIndices,
-            indicesSelection,
-            probeSideMatched);
+        matchedCount =
+            outputMatched(
+                buildSideIndicesBuilder,
+                probeSideIndicesBuilder,
+                buildSideCandidateIndices,
+                proSideCandidateIndices,
+                indicesSelection,
+                probeSideMatched);
       }
 
+      int unmatchedCount = 0;
       if (joinOption.isToOutputProbeSideUnmatched()) {
-        outputProbeSideUnmatched(probeSideIndicesBuilder, markBuilder, probeSideMatched);
+        unmatchedCount = outputProbeSideUnmatched(probeSideIndicesBuilder, probeSideMatched);
       }
 
-      try (IntVector probeSideIndices = probeSideIndicesBuilder.build();
-          IntVector buildSideIndices =
-              buildSideIndicesBuilder.build(probeSideIndices.getValueCount());
-          BitVector mark = markBuilder.build(probeSideIndices.getValueCount())) {
-        output(dictionary, buildSideIndices, probeSideIndices, mark);
+      try (MarkBuilder markBuilder = getMarkBuilder(matchedCount + unmatchedCount)) {
+        markBuilder.appendTrue(matchedCount);
+        markBuilder.appendFalse(unmatchedCount);
+        try (IntVector probeSideIndices = probeSideIndicesBuilder.build();
+            IntVector buildSideIndices =
+                buildSideIndicesBuilder.build(probeSideIndices.getValueCount());
+            BitVector mark = markBuilder.build(probeSideIndices.getValueCount())) {
+          output(dictionary, buildSideIndices, probeSideIndices, mark, unmatchedCount);
+        }
       }
     }
   }
@@ -198,10 +200,9 @@ private MarkBuilder getMarkBuilder(int capacity) {
     }
   }
 
-  private void outputMatched(
+  private int outputMatched(
       SelectionBuilder buildSideIndicesBuilder,
       SelectionBuilder probeSideIndicesBuilder,
-      MarkBuilder markBuilder,
       IntVector buildSideCandidateIndices,
       IntVector proSideCandidateIndices,
       @Nullable BaseIntVector indicesSelection,
@@ -210,20 +211,18 @@ private void outputMatched(
     Preconditions.checkState(
         buildSideCandidateIndices.getValueCount() == proSideCandidateIndices.getValueCount());
     if (indicesSelection == null) {
-      outputMatched(
+      return outputMatched(
           buildSideIndicesBuilder,
           probeSideIndicesBuilder,
-          markBuilder,
           buildSideCandidateIndices,
           proSideCandidateIndices,
           probeSideMatched,
           proSideCandidateIndices.getValueCount(),
           i -> i);
     } else {
-      outputMatched(
+      return outputMatched(
           buildSideIndicesBuilder,
           probeSideIndicesBuilder,
-          markBuilder,
           buildSideCandidateIndices,
           proSideCandidateIndices,
           probeSideMatched,
@@ -232,10 +231,9 @@ private void outputMatched(
     }
   }
 
-  private void outputMatched(
+  private int outputMatched(
       SelectionBuilder buildSideIndicesBuilder,
       SelectionBuilder probeSideIndicesBuilder,
-      MarkBuilder markBuilder,
       IntVector buildSideCandidateIndices,
       IntVector proSideCandidateIndices,
       boolean[] probeSideMatched,
@@ -266,21 +264,19 @@ private void outputMatched(
       buildSideIndicesBuilder.append(buildSideMatchedIndex);
       probeSideIndicesBuilder.append(probeSideMatchedIndex);
     }
-    markBuilder.appendTrue(probeSideMatchedCount);
+    return probeSideMatchedCount;
   }
 
-  private void outputProbeSideUnmatched(
-      SelectionBuilder probeSideIndicesBuilder,
-      MarkBuilder markBuilder,
-      boolean[] probeSideMatched) {
+  private int outputProbeSideUnmatched(
+      SelectionBuilder probeSideIndicesBuilder, boolean[] probeSideMatched) {
     int probeSideUnmatchedCount = 0;
     for (int probeSideIndex = 0; probeSideIndex < probeSideMatched.length; probeSideIndex++) {
       if (!probeSideMatched[probeSideIndex]) {
         probeSideUnmatchedCount++;
         probeSideIndicesBuilder.append(probeSideIndex);
       }
     }
-    markBuilder.appendFalse(probeSideUnmatchedCount);
+    return probeSideUnmatchedCount;
   }
 
   private VectorSchemaRoot getDictionaryEncodedBatch(
@@ -306,7 +302,8 @@ private void output(
       ArrayDictionaryProvider dictionaryProvider,
       BaseIntVector buildSideIndices,
       BaseIntVector probeSideIndices,
-      @Nullable BitVector mark)
+      @Nullable BitVector mark,
+      int unmatchedCount)
       throws ComputeException {
     Preconditions.checkArgument(
         buildSideIndices.getValueCount() == probeSideIndices.getValueCount());
@@ -335,7 +332,7 @@ private void output(
         VectorSchemaRoot output =
             ScalarExpressions.evaluate(
                 allocator, dictionaryProvider, result, null, outputExpressions);
-        BaseIntVector selection = getSelection(probeSideIndices)) {
+        BaseIntVector selection = getSelection(probeSideIndices, unmatchedCount)) {
       resultConsumer.consume(
           dictionaryProvider.slice(allocator),
           VectorSchemaRoots.transfer(allocator, output),
@@ -344,33 +341,55 @@ private void output(
   }
 
   @Nullable
-  private BaseIntVector getSelection(BaseIntVector probeSideIndices) {
-    if (!joinOption.isToOutputProbeSideUnmatched() || !joinOption.isOrderByProbeSideOrdinal()) {
+  private BaseIntVector getSelection(BaseIntVector probeSideIndices, int unmatchedCount) {
+    int total = probeSideIndices.getValueCount();
+    int matchedCount = total - unmatchedCount;
+    if (matchedCount == 0 || unmatchedCount == 0 || !joinOption.isOrderByProbeSideOrdinal()) {
       return null;
     }
-    if (isInOrder(probeSideIndices)) {
+
+    if (isInOrder(probeSideIndices, matchedCount - 1, matchedCount)) {
       return null;
     }
 
-    IntVector selection = new IntVector("selection", allocator);
-    selection.allocateNew(probeSideIndices.getValueCount());
-    selection.setValueCount(probeSideIndices.getValueCount());
-    new IndexSorter<>()
-        .sort(
-            probeSideIndices,
-            selection,
-            new StableVectorComparator<>(
-                DefaultVectorComparators.createDefaultComparator(probeSideIndices)));
-    return selection;
+    try (SelectionBuilder selectionBuilder = new SelectionBuilder(allocator, "selection", total)) {
+      // merge from [0,matchedCount) and from [matchedCount,probeSideIndices.getValueCount())
+      int leftCursor = 0;
+      int rightCursor = matchedCount;
+      while (true) {
+        if (leftCursor < matchedCount) {
+          if (rightCursor < total) {
+            if (isInOrder(probeSideIndices, leftCursor, rightCursor)) {
+              selectionBuilder.append(leftCursor++);
+            } else {
+              selectionBuilder.append(rightCursor++);
+            }
+          } else {
+            selectionBuilder.append(leftCursor++);
+          }
+        } else {
+          if (rightCursor < total) {
+            selectionBuilder.append(rightCursor++);
+          } else {
+            break;
+          }
+        }
+      }
+      return selectionBuilder.build();
+    }
   }
 
-  private static boolean isInOrder(BaseIntVector vector) {
-    for (int i = 1; i < vector.getValueCount(); i++) {
-      if (vector.getValueAsLong(i) < vector.getValueAsLong((i - 1))) {
+  private static boolean isInOrder(BaseIntVector vector, int leftIndex, int rightIndex) {
+    // null is biggest
+    if (vector.getField().isNullable()) {
+      if (vector.isNull(rightIndex)) {
+        return true;
+      }
+      if (vector.isNull(leftIndex)) {
         return false;
       }
     }
-    return true;
+    return vector.getValueAsLong(leftIndex) <= vector.getValueAsLong(rightIndex);
   }
 
   public static class Builder implements AutoCloseable {

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/compute/join/JoinOption.java

@@ -27,8 +27,8 @@ public enum JoinOption {
   LEFT(true, false),
   RIGHT(false, true),
   FULL(true, true),
-  MARK(true, false, true),
-  SINGLE(false, false, false);
+  MARK(false, true, false, true, false, true),
+  SINGLE(false, true, true, false, false, false);
 
   private final boolean toOutputProbeSideUnmatched;
   private final boolean toOutputBuildSideUnmatched;
@@ -41,10 +41,6 @@ public enum JoinOption {
     this(toOutputBuildSideUnmatched, toOutputProbeSideUnmatched, false, true, true, false);
   }
 
-  JoinOption(boolean allowedToMatchMultiple, boolean toOutputAllMatched, boolean toOutputMark) {
-    this(false, true, true, allowedToMatchMultiple, toOutputAllMatched, toOutputMark);
-  }
-
   JoinOption(
       boolean toOutputBuildSideUnmatched,
       boolean toOutputProbeSideUnmatched,

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/compute/scalar/ScalarFunction.java

@@ -70,7 +70,7 @@ default OUTPUT invoke(
     }
     try (VectorSchemaRoot flattened =
         VectorSchemaRoots.flatten(allocator, dictionaryProvider, input, selection)) {
-      return invoke(allocator, selection, flattened);
+      return invoke(allocator, flattened);
     }
   }
 }

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/compute/scalar/convert/cast/AbstractScalarCast.java

@@ -62,7 +62,7 @@ public OUTPUT evaluate(
       return evaluate(allocator, input);
     }
     try (FieldVector selected = PhysicalFunctions.take(allocator, selection, input)) {
-      return evaluate(allocator, selection);
+      return evaluate(allocator, selected);
     }
   }