[Kernel] Change comparator expression to lazy evaluation

kernel/kernel-api/src/main/java/io/delta/kernel/data/VectorComparator.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.delta.kernel.data;
+
+import io.delta.kernel.annotation.Evolving;
+
+/**
+ * Comparator interface for lazy evaluating comparator
+ */
+@Evolving
+public interface VectorComparator {
+    boolean compare(int compareResult);
+}

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/expressions/DefaultExpressionEvaluator.java

@@ -24,6 +24,7 @@
 import io.delta.kernel.client.ExpressionHandler;
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.VectorComparator;
 import io.delta.kernel.expressions.*;
 import io.delta.kernel.types.*;
 
@@ -35,10 +36,7 @@
 
 import io.delta.kernel.defaults.internal.data.vector.DefaultBooleanVector;
 import io.delta.kernel.defaults.internal.data.vector.DefaultConstantVector;
-import static io.delta.kernel.defaults.internal.expressions.DefaultExpressionUtils.booleanWrapperVector;
-import static io.delta.kernel.defaults.internal.expressions.DefaultExpressionUtils.childAt;
-import static io.delta.kernel.defaults.internal.expressions.DefaultExpressionUtils.compare;
-import static io.delta.kernel.defaults.internal.expressions.DefaultExpressionUtils.evalNullability;
+import static io.delta.kernel.defaults.internal.expressions.DefaultExpressionUtils.*;
 import static io.delta.kernel.defaults.internal.expressions.ImplicitCastExpression.canCastTo;
 
 /**
@@ -405,44 +403,53 @@ ColumnVector visitAlwaysFalse(AlwaysFalse alwaysFalse) {
         @Override
         ColumnVector visitComparator(Predicate predicate) {
             PredicateChildrenEvalResult argResults = evalBinaryExpressionChildren(predicate);
-
-            int numRows = argResults.rowCount;
-            boolean[] result = new boolean[numRows];
-            boolean[] nullability = evalNullability(argResults.leftResult, argResults.rightResult);
-            int[] compareResult = compare(argResults.leftResult, argResults.rightResult);
+            VectorComparator v;
             switch (predicate.getName()) {
                 case "=":
-                    for (int rowId = 0; rowId < numRows; rowId++) {
-                        result[rowId] = compareResult[rowId] == 0;
-                    }
-                    break;
+                    v = new VectorComparator() {
+                        @Override
+                        public boolean compare(int compareResult) {
+                            return compareResult == 0;
+                        }
+                    };
+                    return comparatorVector(argResults.leftResult, argResults.rightResult, v);
                 case ">":
-                    for (int rowId = 0; rowId < numRows; rowId++) {
-                        result[rowId] = compareResult[rowId] > 0;
-                    }
-                    break;
+                    v = new VectorComparator() {
+                        @Override
+                        public boolean compare(int compareResult) {
+                            return compareResult > 0;
+                        }
+                    };
+                    return comparatorVector(argResults.leftResult, argResults.rightResult, v);
                 case ">=":
-                    for (int rowId = 0; rowId < numRows; rowId++) {
-                        result[rowId] = compareResult[rowId] >= 0;
-                    }
-                    break;
+                    v = new VectorComparator() {
+                        @Override
+                        public boolean compare(int compareResult) {
+                            return compareResult >= 0;
+                        }
+                    };
+                    return comparatorVector(argResults.leftResult, argResults.rightResult, v);
                 case "<":
-                    for (int rowId = 0; rowId < numRows; rowId++) {
-                        result[rowId] = compareResult[rowId] < 0;
-                    }
-                    break;
+                    v = new VectorComparator() {
+                        @Override
+                        public boolean compare(int compareResult) {
+                            return compareResult < 0;
+                        }
+                    };
+                    return comparatorVector(argResults.leftResult, argResults.rightResult, v);
                 case "<=":
-                    for (int rowId = 0; rowId < numRows; rowId++) {
-                        result[rowId] = compareResult[rowId] <= 0;
-                    }
-                    break;
+                    v = new VectorComparator() {
+                        @Override
+                        public boolean compare(int compareResult) {
+                            return compareResult <= 0;
+                        }
+                    };
+                    return comparatorVector(argResults.leftResult, argResults.rightResult, v);
                 default:
                     throw DeltaErrors.unsupportedExpression(
                             predicate,
                             Optional.of("unsupported expression encountered"));
             }
-
-            return new DefaultBooleanVector(numRows, Optional.of(nullability), result);
         }
 
         @Override

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/expressions/DefaultExpressionUtils.java

@@ -24,6 +24,7 @@
 import io.delta.kernel.data.ArrayValue;
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.MapValue;
+import io.delta.kernel.data.VectorComparator;
 import io.delta.kernel.expressions.Expression;
 import io.delta.kernel.types.*;
 import io.delta.kernel.internal.util.Utils;
@@ -33,6 +34,20 @@
  * Utility methods used by the default expression evaluator.
  */
 class DefaultExpressionUtils {
+
+    static final Comparator<BigDecimal> BIGDECIMAL_COMPARATOR = Comparator.naturalOrder();
+    static final Comparator<String> STRING_COMPARATOR = Comparator.naturalOrder();
+    static final Comparator<byte[]> BINARY_COMPARTOR = (leftOp, rightOp) -> {
+        int i = 0;
+        while (i < leftOp.length && i < rightOp.length) {
+            if (leftOp[i] != rightOp[i]) {
+                return Byte.compare(leftOp[i], rightOp[i]);
+            }
+            i++;
+        }
+        return Integer.compare(leftOp.length, rightOp.length);
+    };
+
     private DefaultExpressionUtils() {}
 
     /**
@@ -86,6 +101,80 @@ public boolean getBoolean(int rowId) {
         };
     }
 
+    static ColumnVector comparatorVector(
+            ColumnVector left,
+            ColumnVector right, VectorComparator vectorComparator) {
+        checkArgument(
+                left.getSize() == right.getSize(),
+                "Left and right operand have different vector sizes.");
+        return new ColumnVector() {
+
+            @Override
+            public DataType getDataType() {
+                return BooleanType.BOOLEAN;
+            }
+
+            @Override
+            public void close() {
+                Utils.closeCloseables(left, right);
+            }
+
+            @Override
+            public int getSize() {
+                return left.getSize();
+            }
+
+            @Override
+            public boolean isNullAt(int rowId) {
+                return left.isNullAt(rowId) || right.isNullAt(rowId);
+            }
+
+            @Override
+            public boolean getBoolean(int rowId) {
+                if (isNullAt(rowId)) {
+                    return false;
+                }
+                DataType dataType = left.getDataType();
+                if (dataType instanceof BooleanType) {
+                    return vectorComparator.compare(
+                            Boolean.compare(left.getBoolean(rowId), right.getBoolean(rowId)));
+                } else if (dataType instanceof ByteType) {
+                    return vectorComparator.compare(
+                            Byte.compare(left.getByte(rowId), right.getByte(rowId)));
+                } else if (dataType instanceof ShortType) {
+                    return vectorComparator.compare(
+                            Short.compare(left.getShort(rowId), right.getShort(rowId)));
+                } else if (dataType instanceof IntegerType || dataType instanceof DateType) {
+                    return vectorComparator.compare(
+                            Integer.compare(left.getInt(rowId), right.getInt(rowId)));
+                } else if (dataType instanceof LongType || dataType instanceof TimestampType) {
+                    return vectorComparator.compare(
+                            Long.compare(left.getLong(rowId), right.getLong(rowId)));
+                } else if (dataType instanceof FloatType) {
+                    return vectorComparator.compare(
+                            Float.compare(left.getFloat(rowId), right.getFloat(rowId)));
+                } else if (dataType instanceof DoubleType) {
+                    return vectorComparator.compare(
+                            Double.compare(left.getDouble(rowId), right.getDouble(rowId)));
+                } else if (dataType instanceof DecimalType) {
+                    return vectorComparator.compare(
+                            BIGDECIMAL_COMPARATOR.compare(
+                                    left.getDecimal(rowId), right.getDecimal(rowId)));
+                } else if (dataType instanceof StringType) {
+                    return vectorComparator.compare(
+                            STRING_COMPARATOR.compare(
+                                    left.getString(rowId), right.getString(rowId)));
+                } else if (dataType instanceof BinaryType) {
+                    return vectorComparator.compare(
+                            BINARY_COMPARTOR.compare(
+                                    left.getBinary(rowId), right.getBinary(rowId)));
+                } else {
+                    throw new UnsupportedOperationException(dataType + " can not be compared.");
+                }
+            }
+        };
+    }
+
     /**
      * Utility method to compare the left and right according to the natural ordering
      * and return an integer array where each row contains the comparison result (-1, 0, 1) for