feat: Add support for defining and using specific columns for metadata filtering.

presto-clp/pom.xml

@@ -65,6 +65,16 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.facebook.presto</groupId>
             <artifactId>presto-spi</artifactId>

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java

@@ -29,6 +29,7 @@ public class ClpConfig
     private String metadataDbName;
     private String metadataDbUser;
     private String metadataDbPassword;
+    private String metadataFilterConfig;
     private String metadataTablePrefix;
     private long metadataRefreshInterval = 60;
     private long metadataExpireInterval = 600;
@@ -107,6 +108,18 @@ public ClpConfig setMetadataDbPassword(String metadataDbPassword)
         return this;
     }
 
+    public String getMetadataFilterConfig()
+    {
+        return metadataFilterConfig;
+    }
+
+    @Config("clp.metadata-filter-config")
+    public ClpConfig setMetadataFilterConfig(String metadataFilterConfig)
+    {
+        this.metadataFilterConfig = metadataFilterConfig;
+        return this;
+    }
+
     public String getMetadataTablePrefix()
     {
         return metadataTablePrefix;

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConnectorFactory.java

@@ -33,10 +33,12 @@
 public class ClpConnectorFactory
         implements ConnectorFactory
 {
+    public static final String CONNECTOR_NAME = "clp";
+
     @Override
     public String getName()
     {
-        return "clp";
+        return CONNECTOR_NAME;
     }
 
     @Override

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java

@@ -18,6 +18,7 @@
 import com.facebook.presto.spi.ErrorCodeSupplier;
 
 import static com.facebook.presto.common.ErrorType.EXTERNAL;
+import static com.facebook.presto.common.ErrorType.USER_ERROR;
 
 public enum ClpErrorCode
         implements ErrorCodeSupplier
@@ -26,7 +27,10 @@ public enum ClpErrorCode
     CLP_UNSUPPORTED_METADATA_SOURCE(1, EXTERNAL),
     CLP_UNSUPPORTED_SPLIT_SOURCE(2, EXTERNAL),
     CLP_UNSUPPORTED_TYPE(3, EXTERNAL),
-    CLP_UNSUPPORTED_CONFIG_OPTION(4, EXTERNAL);
+    CLP_UNSUPPORTED_CONFIG_OPTION(4, EXTERNAL),
+
+    CLP_METADATA_FILTER_CONFIG_NOT_FOUND(10, USER_ERROR),
+    CLP_MANDATORY_METADATA_FILTER_NOT_VALID(11, USER_ERROR);
 
     private final ErrorCode errorCode;
 

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java

@@ -27,13 +27,15 @@ public class ClpExpression
 {
     // Optional KQL query or column name representing the fully or partially translatable part of the expression.
     private final Optional<String> pushDownExpression;
-
+    // Optinal SQL string extracted from the definition, which is only made of given metadata columns.
+    private final Optional<String> metadataSql;
     // The remaining (non-translatable) portion of the RowExpression, if any.
     private final Optional<RowExpression> remainingExpression;
 
-    public ClpExpression(String pushDownExpression, RowExpression remainingExpression)
+    public ClpExpression(String pushDownExpression, String metadataSql, RowExpression remainingExpression)
     {
         this.pushDownExpression = Optional.ofNullable(pushDownExpression);
+        this.metadataSql = Optional.ofNullable(metadataSql);
         this.remainingExpression = Optional.ofNullable(remainingExpression);
     }
 
@@ -42,7 +44,7 @@ public ClpExpression(String pushDownExpression, RowExpression remainingExpressio
      */
     public ClpExpression()
     {
-        this(null, null);
+        this(null, null, null);
     }
 
     /**
@@ -52,7 +54,18 @@ public ClpExpression()
      */
     public ClpExpression(String pushDownExpression)
     {
-        this(pushDownExpression, null);
+        this(pushDownExpression, null, null);
+    }
+
+    /**
+     * Creates a ClpExpression from a fully translatable KQL string and a metadata SQL string.
+     *
+     * @param pushDownExpression
+     * @param metadataSql
+     */
+    public ClpExpression(String pushDownExpression, String metadataSql)
+    {
+        this(pushDownExpression, metadataSql, null);
     }
 
     /**
@@ -62,14 +75,19 @@ public ClpExpression(String pushDownExpression)
      */
     public ClpExpression(RowExpression remainingExpression)
     {
-        this(null, remainingExpression);
+        this(null, null, remainingExpression);
     }
 
     public Optional<String> getPushDownExpression()
     {
         return pushDownExpression;
     }
 
+    public Optional<String> getMetadataSql()
+    {
+        return metadataSql;
+    }
+
     public Optional<RowExpression> getRemainingExpression()
     {
         return remainingExpression;

presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpFilterToKqlConverter.java

@@ -38,6 +38,7 @@
 import java.util.Optional;
 import java.util.Set;
 
+import static com.facebook.presto.common.function.OperatorType.BETWEEN;
 import static com.facebook.presto.common.function.OperatorType.EQUAL;
 import static com.facebook.presto.common.function.OperatorType.GREATER_THAN;
 import static com.facebook.presto.common.function.OperatorType.GREATER_THAN_OR_EQUAL;
@@ -49,6 +50,8 @@
 import static com.facebook.presto.common.function.OperatorType.flip;
 import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
 import static com.facebook.presto.plugin.clp.ClpErrorCode.CLP_PUSHDOWN_UNSUPPORTED_EXPRESSION;
+import static com.facebook.presto.plugin.clp.ClpUtils.escapeKqlSpecialCharsForStringValue;
+import static com.facebook.presto.plugin.clp.ClpUtils.isNumericType;
 import static com.facebook.presto.spi.relation.SpecialFormExpression.Form.AND;
 import static java.lang.Integer.parseInt;
 import static java.lang.String.format;
@@ -84,15 +87,18 @@ public class ClpFilterToKqlConverter
     private final StandardFunctionResolution standardFunctionResolution;
     private final FunctionMetadataManager functionMetadataManager;
     private final Map<VariableReferenceExpression, ColumnHandle> assignments;
+    private final Set<String> metadataFilterColumns;
 
     public ClpFilterToKqlConverter(
             StandardFunctionResolution standardFunctionResolution,
             FunctionMetadataManager functionMetadataManager,
-            Map<VariableReferenceExpression, ColumnHandle> assignments)
+            Map<VariableReferenceExpression, ColumnHandle> assignments,
+            Set<String> metadataFilterColumns)
     {
         this.standardFunctionResolution = requireNonNull(standardFunctionResolution, "standardFunctionResolution is null");
         this.functionMetadataManager = requireNonNull(functionMetadataManager, "function metadata manager is null");
         this.assignments = requireNonNull(assignments, "assignments is null");
+        this.metadataFilterColumns = requireNonNull(metadataFilterColumns, "metadataFilterColumns is null");
     }
 
     @Override
@@ -114,11 +120,68 @@ public ClpExpression visitCall(CallExpression node, Void context)
             if (operatorType.isComparisonOperator() && operatorType != IS_DISTINCT_FROM) {
                 return handleLogicalBinary(operatorType, node);
             }
+            if (BETWEEN == operatorType) {
+                return handleBetween(node);
+            }
         }
 
         return new ClpExpression(node);
     }
 
+    /**
+     * Handles the BETWEEN expression for numeric range, all arguments must be numeric. If the first
+     * argument is not a variable reference expression or either the second or the third is not
+     * constant expressions it won't translate the expression.
+     *
+     * <p></p>
+     * Example: <code>col1 BETWEEN 0 AND 5</code> → <code>col1 >= 0 AND col1 <= 5</code>
+     *
+     * @param node the BETWEEN call expression
+     * @return a ClpExpression containing either the equivalent KQL query, or the original
+     * expression if it couldn't be translated
+     */
+    private ClpExpression handleBetween(CallExpression node)
+    {
+        if (node.getArguments().size() != 3) {
+            throw new PrestoException(CLP_PUSHDOWN_UNSUPPORTED_EXPRESSION,
+                    "BETWEEN operator must have exactly three arguments. Received: " + node);
+        }
+        if (!(node.getArguments().get(0) instanceof VariableReferenceExpression)
+                || !(node.getArguments().get(1) instanceof ConstantExpression)
+                || !(node.getArguments().get(2) instanceof ConstantExpression)) {
+            return new ClpExpression(node);
+        }
+        if (!isNumericType(node.getArguments().get(0).getType())
+            || !isNumericType(node.getArguments().get(1).getType())
+            || !isNumericType(node.getArguments().get(2).getType())) {
+            // Let the Presto SQL analyzer throw exception
+            return new ClpExpression(node);
+        }
+        Optional<String> variableReferencePushDownExpression = node.getArguments().get(0).accept(this, null).getPushDownExpression();
+        if (!variableReferencePushDownExpression.isPresent()) {
+            return new ClpExpression(node);
+        }
+
+        String lowerBoundConstantPushDownExpression = getLiteralString((ConstantExpression) node.getArguments().get(1));
+        String upperBoundConstantPushDownExpression = getLiteralString((ConstantExpression) node.getArguments().get(2));
+        String metadataSql = null;
+        String kql = format(
+                "%s >= %s AND %s <= %s",
+                variableReferencePushDownExpression.get(),
+                lowerBoundConstantPushDownExpression,
+                variableReferencePushDownExpression.get(),
+                upperBoundConstantPushDownExpression);
+        if (metadataFilterColumns.contains(variableReferencePushDownExpression.get())) {
+            metadataSql = format(
+                    "\"%s\" >= %s AND \"%s\" <= %s",
+                    variableReferencePushDownExpression.get(),
+                    lowerBoundConstantPushDownExpression,
+                    variableReferencePushDownExpression.get(),
+                    upperBoundConstantPushDownExpression);
+        }
+        return new ClpExpression(kql, metadataSql);
+    }
+
     @Override
     public ClpExpression visitConstant(ConstantExpression node, Void context)
     {
@@ -203,7 +266,13 @@ private ClpExpression handleNot(CallExpression node)
         if (expression.getRemainingExpression().isPresent() || !expression.getPushDownExpression().isPresent()) {
             return new ClpExpression(node);
         }
-        return new ClpExpression("NOT " + expression.getPushDownExpression().get());
+        String notPushDownExpression = "NOT " + expression.getPushDownExpression().get();
+        if (expression.getMetadataSql().isPresent()) {
+            return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSql());
+        }
+        else {
+            return new ClpExpression(notPushDownExpression);
+        }
     }
 
     /**
@@ -350,24 +419,40 @@ private ClpExpression buildClpExpression(
             Type literalType,
             RowExpression originalNode)
     {
+        String metadataSql = null;
         if (operator.equals(EQUAL)) {
             if (literalType instanceof VarcharType) {
-                return new ClpExpression(format("%s: \"%s\"", variableName, literalString));
+                if (metadataFilterColumns.contains(variableName)) {
+                    metadataSql = format("\"%s\" = '%s'", variableName, literalString);
+                }
+                return new ClpExpression(format("%s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), metadataSql);
             }
             else {
-                return new ClpExpression(format("%s: %s", variableName, literalString));
+                if (metadataFilterColumns.contains(variableName)) {
+                    metadataSql = format("\"%s\" = %s", variableName, literalString);
+                }
+                return new ClpExpression(format("%s: %s", variableName, literalString), metadataSql);
             }
         }
         else if (operator.equals(NOT_EQUAL)) {
             if (literalType instanceof VarcharType) {
-                return new ClpExpression(format("NOT %s: \"%s\"", variableName, literalString));
+                if (metadataFilterColumns.contains(variableName)) {
+                    metadataSql = format("NOT \"%s\" = '%s'", variableName, literalString);
+                }
+                return new ClpExpression(format("NOT %s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), metadataSql);
             }
             else {
-                return new ClpExpression(format("NOT %s: %s", variableName, literalString));
+                if (metadataFilterColumns.contains(variableName)) {
+                    metadataSql = format("NOT \"%s\" = %s", variableName, literalString);
+                }
+                return new ClpExpression(format("NOT %s: %s", variableName, literalString), metadataSql);
             }
         }
         else if (LOGICAL_BINARY_OPS_FILTER.contains(operator) && !(literalType instanceof VarcharType)) {
-            return new ClpExpression(format("%s %s %s", variableName, operator.getOperator(), literalString));
+            if (metadataFilterColumns.contains(variableName)) {
+                metadataSql = format("\"%s\" %s %s", variableName, operator.getOperator(), literalString);
+            }
+            return new ClpExpression(format("%s %s %s", variableName, operator.getOperator(), literalString), metadataSql);
         }
         return new ClpExpression(originalNode);
     }
@@ -568,16 +653,24 @@ private Optional<Integer> parseLengthLiteral(RowExpression lengthExpression, Str
      */
     private ClpExpression handleAnd(SpecialFormExpression node)
     {
+        StringBuilder metadataQueryBuilder = new StringBuilder();
+        metadataQueryBuilder.append("(");
         StringBuilder queryBuilder = new StringBuilder();
         queryBuilder.append("(");
         List<RowExpression> remainingExpressions = new ArrayList<>();
+        boolean hasMetadataSql = false;
         boolean hasPushDownExpression = false;
         for (RowExpression argument : node.getArguments()) {
             ClpExpression expression = argument.accept(this, null);
             if (expression.getPushDownExpression().isPresent()) {
                 hasPushDownExpression = true;
                 queryBuilder.append(expression.getPushDownExpression().get());
                 queryBuilder.append(" AND ");
+                if (expression.getMetadataSql().isPresent()) {
+                    hasMetadataSql = true;
+                    metadataQueryBuilder.append(expression.getMetadataSql().get());
+                    metadataQueryBuilder.append(" AND ");
+                }
             }
             if (expression.getRemainingExpression().isPresent()) {
                 remainingExpressions.add(expression.getRemainingExpression().get());
@@ -588,16 +681,21 @@ private ClpExpression handleAnd(SpecialFormExpression node)
         }
         else if (!remainingExpressions.isEmpty()) {
             if (remainingExpressions.size() == 1) {
-                return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 5) + ")", remainingExpressions.get(0));
+                return new ClpExpression(
+                        queryBuilder.substring(0, queryBuilder.length() - 5) + ")",
+                        hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null,
+                        remainingExpressions.get(0));
             }
             else {
                 return new ClpExpression(
                         queryBuilder.substring(0, queryBuilder.length() - 5) + ")",
+                        hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null,
                         new SpecialFormExpression(node.getSourceLocation(), AND, BOOLEAN, remainingExpressions));
             }
         }
         // Remove the last " AND " from the query
-        return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 5) + ")");
+        return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 5) + ")",
+                hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null);
     }
 
     /**
@@ -614,18 +712,30 @@ else if (!remainingExpressions.isEmpty()) {
      */
     private ClpExpression handleOr(SpecialFormExpression node)
     {
+        StringBuilder metadataQueryBuilder = new StringBuilder();
+        metadataQueryBuilder.append("(");
         StringBuilder queryBuilder = new StringBuilder();
         queryBuilder.append("(");
+        boolean hasAllMetadataSql = true;
         for (RowExpression argument : node.getArguments()) {
             ClpExpression expression = argument.accept(this, null);
             if (expression.getRemainingExpression().isPresent() || !expression.getPushDownExpression().isPresent()) {
                 return new ClpExpression(node);
             }
             queryBuilder.append(expression.getPushDownExpression().get());
             queryBuilder.append(" OR ");
+            if (hasAllMetadataSql && expression.getMetadataSql().isPresent()) {
+                metadataQueryBuilder.append(expression.getMetadataSql().get());
+                metadataQueryBuilder.append(" OR ");
+            }
+            else {
+                hasAllMetadataSql = false;
+            }
         }
         // Remove the last " OR " from the query
-        return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 4) + ")");
+        return new ClpExpression(
+                queryBuilder.substring(0, queryBuilder.length() - 4) + ")",
+                hasAllMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 4) + ")" : null);
     }
 
     /**