fix: logic short-circuit in OrNode

Author: aqni

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

@@ -27,7 +27,7 @@ public enum JoinOption {
   LEFT(true, false),
   RIGHT(false, true),
   FULL(true, true),
-  MARK(false, true, false, true, false, true),
+  MARK(false, true, true, true, false, true),
   SINGLE(false, true, true, false, false, false);
 
   private final boolean toOutputProbeSideUnmatched;

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

@@ -99,15 +99,24 @@ public BaseIntVector filter(
       IntStream.range(0, input.getRowCount()).forEach(remainIndex::add);
     } else {
       for (int i = 0; i < selection.getValueCount(); i++) {
+        if (selection.getField().isNullable() && selection.isNull(i)) {
+          continue;
+        }
         remainIndex.add((int) selection.getValueAsLong(i));
       }
     }
 
     ResultBuilder resultBuilder = new ResultBuilder(allocator, selection);
-    return filter(
-        allocator, dictionaryProvider, input, selection, children, resultBuilder, remainIndex);
+    BaseIntVector result =
+        filter(
+            allocator, dictionaryProvider, input, selection, children, resultBuilder, remainIndex);
+    if (result != null) {
+      return result;
+    }
+    return ValueVectors.slice(allocator, selection, "or");
   }
 
+  @Nullable
   private static BaseIntVector filter(
       BufferAllocator allocator,
       DictionaryProvider dictionaryProvider,
@@ -120,15 +129,15 @@ private static BaseIntVector filter(
     try (BaseIntVector subSelection =
         children.get(0).filter(allocator, dictionaryProvider, input, selection)) {
       if (subSelection == null) {
-        return ValueVectors.slice(allocator, selection, "or");
+        return null;
       }
       for (int i = 0; i < subSelection.getValueCount(); i++) {
         int index = (int) subSelection.getValueAsLong(i);
         resultBuilder.append(index);
         remainIndex.remove(index);
       }
       if (remainIndex.isEmpty()) {
-        return ValueVectors.slice(allocator, selection, "or");
+        return null;
       }
       List<PredicateExpression> remainChildren = children.subList(1, children.size());
       if (remainChildren.isEmpty()) {
@@ -175,6 +184,9 @@ public ResultBuilder(BufferAllocator allocator, BaseIntVector selection) {
       } else {
         invertedIndex = new IntObjectHashMap<>();
         for (int i = 0; i < selection.getValueCount(); i++) {
+          if (selection.getField().isNullable() && selection.isNull(i)) {
+            continue;
+          }
           int value = (int) selection.getValueAsLong(i);
           invertedIndex.put(value, Integer.valueOf(i));
         }

optimizer/src/main/java/cn/edu/tsinghua/iginx/physical/optimizer/naive/NaivePhysicalPlanner.java

@@ -315,16 +315,17 @@ public PhysicalTask<?> construct(RowTransform operator, RequestContext context)
   public PhysicalTask<?> construct(Select operator, RequestContext context) {
     Source source = operator.getSource();
     PhysicalTask<?> sourceTask = fetch(operator.getSource(), context);
-    if (sourceTask.getResultClass() == RowStream.class) {
-      return new UnaryRowMemoryPhysicalTask(
-          convert(sourceTask, context, RowStream.class), operator, context);
-    }
 
     StoragePhysicalTask storageTask = tryPushDownAloneWithProject(sourceTask, context, operator);
     if (storageTask != null) {
       return storageTask;
     }
 
+    if (sourceTask.getResultClass() == RowStream.class) {
+      return new UnaryRowMemoryPhysicalTask(
+          convert(sourceTask, context, RowStream.class), operator, context);
+    }
+
     PhysicalTask<BatchStream> batchTask = convert(sourceTask, context, BatchStream.class);
 
     if (operator.getTagFilter() != null) {
@@ -371,7 +372,7 @@ private StoragePhysicalTask tryPushDownAloneWithProject(
     }
     Project project = (Project) sourceOperator;
 
-    if (project.getTagFilter() == null) {
+    if (project.getTagFilter() != null) {
       return null;
     }
     return reConstruct(storageTask, context, false, operator);