Support conjugates for scalar functions, add more scalar functions

pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ComparisonFunctions.java

@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pinot.common.function.scalar;
+
+import org.apache.pinot.spi.annotations.ScalarFunction;
+
+public class ComparisonFunctions {
+
+  private static final double DOUBLE_COMPARISON_TOLERANCE = 1e-7d;
+
+  private ComparisonFunctions() {
+  }
+
+  @ScalarFunction
+  public static boolean greaterThan(double a, double b) {
+    return a > b;
+  }
+
+  @ScalarFunction
+  public static boolean greaterThanOrEqual(double a, double b) {
+    return a >= b;
+  }
+
+  @ScalarFunction
+  public static boolean lessThan(double a, double b) {
+    return a < b;
+  }
+
+  @ScalarFunction
+  public static boolean lessThanOrEqual(double a, double b) {
+    return a <= b;
+  }
+
+  @ScalarFunction
+  public static boolean notEquals(double a, double b) {
+    return Math.abs(a - b) >= DOUBLE_COMPARISON_TOLERANCE;
+  }
+
+  @ScalarFunction
+  public static boolean equals(double a, double b) {
+    // To avoid approximation errors
+    return Math.abs(a - b) < DOUBLE_COMPARISON_TOLERANCE;
+  }
+
+  @ScalarFunction
+  public static boolean between(double val, double a, double b) {
+    return val > a && val < b;
+  }
+}

pinot-common/src/main/java/org/apache/pinot/common/function/scalar/ObjectFunctions.java

@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pinot.common.function.scalar;
+
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.annotations.ScalarFunction;
+
+public class ObjectFunctions {
+  private ObjectFunctions() {
+  }
+
+  @ScalarFunction(nullableParameters = true)
+  public static boolean isNull(@Nullable Object obj) {
+    return obj == null;
+  }
+
+  @ScalarFunction(nullableParameters = true)
+  public static boolean isNotNull(@Nullable Object obj) {
+    return !isNull(obj);
+  }
+}

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/InbuiltFunctionEvaluator.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pinot.segment.local.function;
 
+import com.google.common.base.Preconditions;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.commons.lang3.StringUtils;
@@ -68,17 +69,27 @@ private ExecutableNode planExecution(ExpressionContext expression) {
           childNodes[i] = planExecution(arguments.get(i));
         }
         String functionName = function.getFunctionName();
-        FunctionInfo functionInfo = FunctionRegistry.getFunctionInfo(functionName, numArguments);
-        if (functionInfo == null) {
-          if (FunctionRegistry.containsFunction(functionName)) {
-            throw new IllegalStateException(
-              String.format("Unsupported function: %s with %d parameters", functionName, numArguments));
-          } else {
-            throw new IllegalStateException(
-              String.format("Unsupported function: %s not found", functionName));
-          }
+        switch (functionName) {
+          case "and":
+            return new AndExecutionNode(childNodes);
+          case "or":
+            return new OrExecutionNode(childNodes);
+          case "not":
+            Preconditions.checkState(numArguments == 1, "NOT function expects 1 argument, got: %s", numArguments);
+            return new NotExecutionNode(childNodes[0]);
+          default:
+            FunctionInfo functionInfo = FunctionRegistry.getFunctionInfo(functionName, numArguments);
+            if (functionInfo == null) {
+              if (FunctionRegistry.containsFunction(functionName)) {
+                throw new IllegalStateException(
+                    String.format("Unsupported function: %s with %d parameters", functionName, numArguments));
+              } else {
+                throw new IllegalStateException(
+                    String.format("Unsupported function: %s not found", functionName));
+              }
+            }
+            return new FunctionExecutionNode(functionInfo, childNodes);
         }
-        return new FunctionExecutionNode(functionInfo, childNodes);
       default:
         throw new IllegalStateException();
     }
@@ -106,6 +117,84 @@ private interface ExecutableNode {
     Object execute(Object[] values);
   }
 
+  private static class NotExecutionNode implements ExecutableNode {
+    private final ExecutableNode _argumentNode;
+
+    NotExecutionNode(ExecutableNode argumentNode) {
+      _argumentNode = argumentNode;
+    }
+
+    @Override
+    public Object execute(GenericRow row) {
+      return !((Boolean) _argumentNode.execute(row));
+    }
+
+    @Override
+    public Object execute(Object[] values) {
+      return !((Boolean) _argumentNode.execute(values));
+    }
+  }
+
+  private static class OrExecutionNode implements ExecutableNode {
+    private final ExecutableNode[] _argumentNodes;
+
+    OrExecutionNode(ExecutableNode[] argumentNodes) {
+      _argumentNodes = argumentNodes;
+    }
+
+    @Override
+    public Object execute(GenericRow row) {
+      for (ExecutableNode executableNode :_argumentNodes) {
+        Boolean res = (Boolean) executableNode.execute(row);
+        if (res) {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    @Override
+    public Object execute(Object[] values) {
+      for (ExecutableNode executableNode :_argumentNodes) {
+        Boolean res = (Boolean) executableNode.execute(values);
+        if (res) {
+          return true;
+        }
+      }
+      return false;
+    }
+  }
+
+  private static class AndExecutionNode implements ExecutableNode {
+    private final ExecutableNode[] _argumentNodes;
+
+    AndExecutionNode(ExecutableNode[] argumentNodes) {
+      _argumentNodes = argumentNodes;
+    }
+
+    @Override
+    public Object execute(GenericRow row) {
+      for (ExecutableNode executableNode :_argumentNodes) {
+        Boolean res = (Boolean) executableNode.execute(row);
+        if (!res) {
+          return false;
+        }
+      }
+      return true;
+    }
+
+    @Override
+    public Object execute(Object[] values) {
+      for (ExecutableNode executableNode :_argumentNodes) {
+        Boolean res = (Boolean) executableNode.execute(values);
+        if (!res) {
+          return false;
+        }
+      }
+      return true;
+    }
+  }
+
   private static class FunctionExecutionNode implements ExecutableNode {
     final FunctionInvoker _functionInvoker;
     final FunctionInfo _functionInfo;

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/RecordTransformerTest.java

@@ -81,6 +81,7 @@ private static GenericRow getRecord() {
     record.putValue("svStringWithLengthLimit", "123");
     record.putValue("mvString1", new Object[]{"123", 123, 123L, 123f, 123.0});
     record.putValue("mvString2", new Object[]{123, 123L, 123f, 123.0, "123"});
+    record.putValue("svNullString", null);
     return record;
   }
 
@@ -178,6 +179,150 @@ record = transformer.transform(record);
     }
   }
 
+  @Test
+  public void testScalarOps() {
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build();
+
+    // expression true, filtered
+    GenericRow genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svInt = 123"), null, null));
+    RecordTransformer transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svDouble > 120"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svDouble >= 123"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svDouble < 200"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svDouble <= 123"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svLong != 125"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svLong = 123"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null, new FilterConfig("between(svLong, 100, 125)"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+  }
+
+  private GenericRow getNullColumnsRecord() {
+    GenericRow record = new GenericRow();
+    record.putValue("svNullString", null);
+    record.putValue("svInt", (byte) 123);
+
+    record.putValue("mvLong", Collections.singletonList(123f));
+    record.putValue("mvNullFloat", null);
+    return record;
+  }
+
+  @Test
+  public void testObjectOps() {
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build();
+
+    // expression true, filtered
+    GenericRow genericRow = getNullColumnsRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null, new FilterConfig("svNullString is null"), null, null));
+    RecordTransformer transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getNullColumnsRecord();
+    tableConfig.setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("svInt is not null"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getNullColumnsRecord();
+    tableConfig.setIngestionConfig(new IngestionConfig(null, null, new FilterConfig("mvLong is not null"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getNullColumnsRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null, new FilterConfig("mvNullFloat is null"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+  }
+
+  @Test
+  public void testLogicalScalarOps() {
+    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("testTable").build();
+
+    // expression true, filtered
+    GenericRow genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svInt = 123 AND svDouble <= 200"), null, null));
+    RecordTransformer transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+
+    // expression true, filtered
+    genericRow = getRecord();
+    tableConfig.setIngestionConfig(
+        new IngestionConfig(null, null,
+            new FilterConfig("svInt = 125 OR svLong <= 200"), null, null));
+    transformer = new FilterTransformer(tableConfig);
+    transformer.transform(genericRow);
+    Assert.assertTrue(genericRow.getFieldToValueMap().containsKey(GenericRow.SKIP_RECORD_KEY));
+  }
+
   @Test
   public void testNullValueTransformer() {
     RecordTransformer transformer = new NullValueTransformer(TABLE_CONFIG, SCHEMA);