Add pluggable custom filter predicates

Author: xiangfu0

pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/MultiPartitionColumnsSegmentPruner.java

@@ -119,7 +119,10 @@ public Set<String> getPartitionColumns() {
   private boolean isPartitionMatch(Expression filterExpression,
       Map<String, SegmentPartitionInfo> columnPartitionInfoMap) {
     Function function = filterExpression.getFunctionCall();
-    FilterKind filterKind = FilterKind.valueOf(function.getOperator());
+    FilterKind filterKind = FilterKind.fromOperator(function.getOperator());
+    if (filterKind == null) {
+      return true;
+    }
     List<Expression> operands = function.getOperands();
     switch (filterKind) {
       case AND:

pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/SinglePartitionColumnSegmentPruner.java

@@ -109,7 +109,10 @@ public Set<String> prune(BrokerRequest brokerRequest, Set<String> segments) {
 
   private boolean isPartitionMatch(Expression filterExpression, SegmentPartitionInfo partitionInfo) {
     Function function = filterExpression.getFunctionCall();
-    FilterKind filterKind = FilterKind.valueOf(function.getOperator());
+    FilterKind filterKind = FilterKind.fromOperator(function.getOperator());
+    if (filterKind == null) {
+      return true;
+    }
     List<Expression> operands = function.getOperands();
     switch (filterKind) {
       case AND:

pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpruner/TimeSegmentPruner.java

@@ -173,7 +173,10 @@ public Set<String> prune(BrokerRequest brokerRequest, Set<String> segments) {
   @Nullable
   private List<Interval> getFilterTimeIntervals(Expression filterExpression) {
     Function function = filterExpression.getFunctionCall();
-    FilterKind filterKind = FilterKind.valueOf(function.getOperator());
+    FilterKind filterKind = FilterKind.fromOperator(function.getOperator());
+    if (filterKind == null) {
+      return null;
+    }
     List<Expression> operands = function.getOperands();
     switch (filterKind) {
       case AND:

pinot-common/src/main/java/org/apache/pinot/common/filter/FilterPredicatePlugin.java

@@ -44,7 +44,7 @@ public interface FilterPredicatePlugin {
   /**
    * Returns the canonical name of this filter predicate (e.g., "SEMANTIC_MATCH").
    * Must be unique across all registered plugins and built-in filter kinds.
-   * The name is case-insensitive and will be uppercased for matching.
+   * The name is case-insensitive and will be canonicalized for matching.
    */
   String name();
 
@@ -94,6 +94,19 @@ default List<Integer> getOptionalOperandIndices() {
     return List.of();
   }
 
+  /**
+   * Returns whether this predicate accepts additional trailing operands beyond those declared in
+   * {@link #getOperandTypes()}.
+   *
+   * <p>Variadic custom predicates are currently registered as homogeneous variadic functions in Calcite,
+   * so every declared operand type must be identical. For example,
+   * {@code List.of(OperandType.STRING, OperandType.STRING)} with this flag enabled accepts
+   * {@code (STRING, STRING, STRING, ...)}.
+   */
+  default boolean acceptsVariadicArguments() {
+    return false;
+  }
+
   // --- Predicate Creation Layer (RequestContextUtils) ---
 
   /**

pinot-common/src/main/java/org/apache/pinot/common/filter/FilterPredicateRegistry.java

@@ -18,21 +18,25 @@
  */
 package org.apache.pinot.common.filter;
 
+import com.google.common.base.Preconditions;
 import java.util.Collection;
-import java.util.Locale;
 import java.util.Map;
-import java.util.ServiceLoader;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nullable;
+import org.apache.pinot.common.function.FunctionRegistry;
+import org.apache.pinot.common.function.TransformFunctionType;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.plugin.PluginManager;
+import org.apache.pinot.sql.FilterKind;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 
 /**
  * Registry for custom filter predicate plugins.
  *
- * <p>Plugins are discovered automatically via {@link ServiceLoader} during {@link #init()},
+ * <p>Plugins are discovered automatically via {@link PluginManager} during {@link #init()},
  * and can also be registered programmatically via {@link #register(FilterPredicatePlugin)}.
  *
  * <p>This registry is consulted at multiple points in the query pipeline when the filter name
@@ -53,15 +57,14 @@ private FilterPredicateRegistry() {
   }
 
   /**
-   * Discovers and registers all {@link FilterPredicatePlugin} implementations found via ServiceLoader.
-   * Safe to call multiple times; subsequent calls after the first are no-ops.
+   * Discovers and registers all {@link FilterPredicatePlugin} implementations visible from the application classpath
+   * and loaded Pinot plugins. Safe to call multiple times; subsequent calls after the first are no-ops.
    */
   public static void init() {
     if (!INITIALIZED.compareAndSet(false, true)) {
       return;
     }
-    ServiceLoader<FilterPredicatePlugin> loader = ServiceLoader.load(FilterPredicatePlugin.class);
-    for (FilterPredicatePlugin plugin : loader) {
+    for (FilterPredicatePlugin plugin : PluginManager.get().loadServices(FilterPredicatePlugin.class)) {
       register(plugin);
     }
   }
@@ -73,29 +76,32 @@ public static void init() {
    * @throws IllegalStateException if a different plugin is already registered with the same name
    */
   public static void register(FilterPredicatePlugin plugin) {
-    String name = plugin.name().toUpperCase(Locale.ROOT);
-    FilterPredicatePlugin existing = REGISTRY.putIfAbsent(name, plugin);
+    String canonicalName = canonicalize(plugin.name());
+    Preconditions.checkState(!isBuiltInName(canonicalName),
+        "Cannot register FilterPredicatePlugin '%s' (%s): name collides with a built-in Pinot predicate or function",
+        plugin.name(), plugin.getClass().getName());
+    FilterPredicatePlugin existing = REGISTRY.putIfAbsent(canonicalName, plugin);
     if (existing != null && existing != plugin) {
       throw new IllegalStateException(
           String.format("Cannot register FilterPredicatePlugin '%s' (%s): already registered by %s",
-              name, plugin.getClass().getName(), existing.getClass().getName()));
+              plugin.name(), plugin.getClass().getName(), existing.getClass().getName()));
     }
-    LOGGER.info("Registered custom filter predicate plugin: {} ({})", name, plugin.getClass().getName());
+    LOGGER.info("Registered custom filter predicate plugin: {} ({})", plugin.name(), plugin.getClass().getName());
   }
 
   /**
    * Returns the plugin registered for the given filter name, or null if none is registered.
    */
   @Nullable
   public static FilterPredicatePlugin get(String name) {
-    return REGISTRY.get(name.toUpperCase(Locale.ROOT));
+    return REGISTRY.get(canonicalize(name));
   }
 
   /**
    * Returns true if a custom filter predicate plugin is registered for the given name.
    */
   public static boolean isRegistered(String name) {
-    return REGISTRY.containsKey(name.toUpperCase(Locale.ROOT));
+    return REGISTRY.containsKey(canonicalize(name));
   }
 
   /**
@@ -112,4 +118,32 @@ public static void clear() {
     REGISTRY.clear();
     INITIALIZED.set(false);
   }
+
+  private static String canonicalize(String name) {
+    return FunctionRegistry.canonicalize(name);
+  }
+
+  private static boolean isBuiltInName(String canonicalName) {
+    if (FunctionRegistry.contains(canonicalName)) {
+      return true;
+    }
+    for (FilterKind filterKind : FilterKind.values()) {
+      if (canonicalize(filterKind.name()).equals(canonicalName)) {
+        return true;
+      }
+    }
+    for (TransformFunctionType functionType : TransformFunctionType.values()) {
+      for (String name : functionType.getNames()) {
+        if (canonicalize(name).equals(canonicalName)) {
+          return true;
+        }
+      }
+    }
+    for (AggregationFunctionType functionType : AggregationFunctionType.values()) {
+      if (canonicalize(functionType.getName()).equals(canonicalName)) {
+        return true;
+      }
+    }
+    return false;
+  }
 }

pinot-common/src/main/java/org/apache/pinot/common/request/context/RequestContextUtils.java

@@ -22,7 +22,6 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import org.apache.commons.lang3.EnumUtils;
 import org.apache.pinot.common.filter.FilterPredicatePlugin;
 import org.apache.pinot.common.filter.FilterPredicateRegistry;
 import org.apache.pinot.common.request.Expression;
@@ -136,13 +135,22 @@ private static FilterContext getFilterInner(Expression thriftExpression) {
   private static FilterContext getFilterInner(Function thriftFunction) {
     String functionOperator = thriftFunction.getOperator();
 
+    FilterPredicatePlugin customPlugin = FilterPredicateRegistry.get(functionOperator);
+    if (customPlugin != null) {
+      List<ExpressionContext> operands = new ArrayList<>(thriftFunction.getOperandsSize());
+      for (Expression operand : thriftFunction.getOperands()) {
+        operands.add(getExpression(operand));
+      }
+      return FilterContext.forPredicate(customPlugin.createPredicate(operands));
+    }
+
     // convert "WHERE startsWith(col, 'str')" to "WHERE startsWith(col, 'str') = true"
-    if (!EnumUtils.isValidEnum(FilterKind.class, functionOperator)) {
+    FilterKind filterKind = FilterKind.fromOperator(functionOperator);
+    if (filterKind == null) {
       return FilterContext.forPredicate(
           new EqPredicate(ExpressionContext.forFunction(getFunction(thriftFunction)), "true"));
     }
 
-    FilterKind filterKind = FilterKind.valueOf(thriftFunction.getOperator().toUpperCase());
     List<Expression> operands = thriftFunction.getOperands();
     int numOperands = operands.size();
     switch (filterKind) {
@@ -331,7 +339,7 @@ private static FilterContext getFilterInner(ExpressionContext filterExpression)
   }
 
   private static FilterContext getFilterInner(FunctionContext filterFunction) {
-    String functionOperator = filterFunction.getFunctionName().toUpperCase();
+    String functionOperator = filterFunction.getFunctionName();
 
     // Check if this is a custom filter predicate registered via plugin
     FilterPredicatePlugin customPlugin = FilterPredicateRegistry.get(functionOperator);
@@ -341,11 +349,11 @@ private static FilterContext getFilterInner(FunctionContext filterFunction) {
     }
 
     // convert "WHERE startsWith(col, 'str')" to "WHERE startsWith(col, 'str') = true"
-    if (!EnumUtils.isValidEnum(FilterKind.class, functionOperator)) {
+    FilterKind filterKind = FilterKind.fromOperator(functionOperator);
+    if (filterKind == null) {
       return FilterContext.forPredicate(new EqPredicate(ExpressionContext.forFunction(filterFunction), "true"));
     }
 
-    FilterKind filterKind = FilterKind.valueOf(filterFunction.getFunctionName().toUpperCase());
     List<ExpressionContext> operands = filterFunction.getArguments();
     int numOperands = operands.size();
     switch (filterKind) {

pinot-common/src/main/java/org/apache/pinot/sql/FilterKind.java

@@ -18,6 +18,8 @@
  */
 package org.apache.pinot.sql;
 
+import javax.annotation.Nullable;
+
 public enum FilterKind {
   AND,
   OR,
@@ -50,4 +52,19 @@ public boolean isRange() {
     return this == GREATER_THAN || this == GREATER_THAN_OR_EQUAL || this == LESS_THAN || this == LESS_THAN_OR_EQUAL
         || this == BETWEEN || this == RANGE;
   }
+
+  /**
+   * Resolves the given operator name to a built-in filter kind.
+   *
+   * <p>Returns {@code null} for custom predicates registered through the plugin system.
+   */
+  @Nullable
+  public static FilterKind fromOperator(String operator) {
+    for (FilterKind filterKind : values()) {
+      if (filterKind.name().equalsIgnoreCase(operator)) {
+        return filterKind;
+      }
+    }
+    return null;
+  }
 }

pinot-common/src/main/java/org/apache/pinot/sql/parsers/rewriter/PredicateComparisonRewriter.java

@@ -20,7 +20,6 @@
 
 import com.google.common.base.Preconditions;
 import java.util.List;
-import org.apache.commons.lang3.EnumUtils;
 import org.apache.pinot.common.filter.FilterPredicatePlugin;
 import org.apache.pinot.common.filter.FilterPredicateRegistry;
 import org.apache.pinot.common.request.Expression;
@@ -82,8 +81,9 @@ private static Expression updatePredicate(Expression expression) {
   private static Expression updateFunctionExpression(Expression expression) {
     Function function = expression.getFunctionCall();
     String functionOperator = function.getOperator();
+    FilterKind filterKind = FilterKind.fromOperator(functionOperator);
 
-    if (!EnumUtils.isValidEnum(FilterKind.class, functionOperator)) {
+    if (filterKind == null) {
       // Check if this is a custom filter predicate registered via plugin
       FilterPredicatePlugin customPlugin = FilterPredicateRegistry.get(functionOperator);
       if (customPlugin != null) {
@@ -96,7 +96,6 @@ private static Expression updateFunctionExpression(Expression expression) {
       expression = convertPredicateToEqualsBooleanExpression(expression);
       return expression;
     } else {
-      FilterKind filterKind = FilterKind.valueOf(function.getOperator());
       List<Expression> operands = function.getOperands();
       switch (filterKind) {
         case AND:

pinot-common/src/test/java/org/apache/pinot/common/filter/FilterPredicateRegistryTest.java

@@ -0,0 +1,91 @@
+/**
+ * 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.filter;
+
+import java.util.List;
+import org.apache.pinot.common.request.Expression;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.predicate.EqPredicate;
+import org.apache.pinot.common.request.context.predicate.Predicate;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+
+public class FilterPredicateRegistryTest {
+
+  @AfterMethod
+  public void tearDown() {
+    FilterPredicateRegistry.clear();
+  }
+
+  @Test
+  public void testCanonicalizedLookup() {
+    FilterPredicatePlugin plugin = new TestFilterPredicatePlugin("LIKE_ANY");
+
+    FilterPredicateRegistry.register(plugin);
+
+    Assert.assertSame(FilterPredicateRegistry.get("LIKE_ANY"), plugin);
+    Assert.assertSame(FilterPredicateRegistry.get("like_any"), plugin);
+    Assert.assertSame(FilterPredicateRegistry.get("likeany"), plugin);
+  }
+
+  @Test
+  public void testRejectsBuiltInFilterName() {
+    IllegalStateException exception = Assert.expectThrows(IllegalStateException.class,
+        () -> FilterPredicateRegistry.register(new TestFilterPredicatePlugin("TEXT_MATCH")));
+
+    Assert.assertTrue(exception.getMessage().contains("collides with a built-in Pinot predicate or function"));
+  }
+
+  @Test
+  public void testRejectsBuiltInFunctionName() {
+    IllegalStateException exception = Assert.expectThrows(IllegalStateException.class,
+        () -> FilterPredicateRegistry.register(new TestFilterPredicatePlugin("startsWith")));
+
+    Assert.assertTrue(exception.getMessage().contains("collides with a built-in Pinot predicate or function"));
+  }
+
+  private static final class TestFilterPredicatePlugin implements FilterPredicatePlugin {
+    private final String _name;
+
+    private TestFilterPredicatePlugin(String name) {
+      _name = name;
+    }
+
+    @Override
+    public String name() {
+      return _name;
+    }
+
+    @Override
+    public void validateFilterExpression(List<Expression> operands) {
+    }
+
+    @Override
+    public List<OperandType> getOperandTypes() {
+      return List.of(OperandType.STRING, OperandType.STRING);
+    }
+
+    @Override
+    public Predicate createPredicate(List<ExpressionContext> operands) {
+      return new EqPredicate(ExpressionContext.forIdentifier("col"), "value");
+    }
+  }
+}