Address review comments

Author: anlowee

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

@@ -58,7 +58,7 @@ public ClpExpression(String pushDownExpression)
     }
 
     /**
-     * Creates a ClpExpression from a fully translatable KQL string and give it metadata SQL.
+     * Creates a ClpExpression from a fully translatable KQL string and a metadata SQL string.
      *
      * @param pushDownExpression
      * @param metadataSql

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;
@@ -118,35 +119,58 @@ public ClpExpression visitCall(CallExpression node, Void context)
             if (operatorType.isComparisonOperator() && operatorType != IS_DISTINCT_FROM) {
                 return handleLogicalBinary(operatorType, node);
             }
-            else if (OperatorType.BETWEEN == operatorType) {
-                return handleBetween(operatorType, node);
+            if (BETWEEN == operatorType) {
+                return handleBetween(node);
             }
         }
 
         return new ClpExpression(node);
     }
 
-    private ClpExpression handleBetween(OperatorType operator, CallExpression node)
+    /**
+     * Handles the BETWEEN expression for numeric range, the range must be numeric. If the first
+     * argument is not a variable reference expression or either range boundaries 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,
-                    "Logical binary operator must have exactly two arguments. Received: " + node);
+                    "Between operator must have exactly three arguments. Received: " + node);
         }
-        Optional<String> variableReferenceDefinition = node.getArguments().get(0).accept(this, null).getPushDownExpression();
-        Optional<String> lowerBoundConstantDefinition = node.getArguments().get(1).accept(this, null).getPushDownExpression();
-        Optional<String> upperBoundConstantDefinition = node.getArguments().get(2).accept(this, null).getPushDownExpression();
-        if (!variableReferenceDefinition.isPresent() || !lowerBoundConstantDefinition.isPresent() || !upperBoundConstantDefinition.isPresent()) {
+        if (!(node.getArguments().get(0) instanceof VariableReferenceExpression)
+                || !(node.getArguments().get(1) instanceof ConstantExpression)
+                || !(node.getArguments().get(2) instanceof ConstantExpression)) {
             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 = String.format(
-                "\"%s\" >= %s AND \"%s\" <= %s",
-                variableReferenceDefinition.get(),
-                lowerBoundConstantDefinition.get(),
-                variableReferenceDefinition.get(),
-                upperBoundConstantDefinition.get());
-        if (metadataFilterColumns.contains(variableReferenceDefinition.get())) {
-            metadataSql = kql;
+        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);
     }
@@ -235,11 +259,12 @@ private ClpExpression handleNot(CallExpression node)
         if (expression.getRemainingExpression().isPresent() || !expression.getPushDownExpression().isPresent()) {
             return new ClpExpression(node);
         }
+        String notPushDownExpression = "NOT " + expression.getPushDownExpression().get();
         if (expression.getMetadataSql().isPresent()) {
-            return new ClpExpression("NOT " + expression.getPushDownExpression().get(), "NOT " + expression.getMetadataSql());
+            return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSql());
         }
         else {
-            return new ClpExpression("NOT " + expression.getPushDownExpression().get());
+            return new ClpExpression(notPushDownExpression);
         }
     }
 

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

@@ -24,9 +24,7 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -37,8 +35,37 @@
 import static com.facebook.presto.plugin.clp.ClpErrorCode.CLP_METADATA_FILTER_CONFIG_NOT_FOUND;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
+/**
+ * Provides metadata filter configuration and utilities for remapping and validating metadata
+ * filters based on a JSON configuration file.
+ * <p>
+ * The configuration file locates at {@code clp.metadata-filter-config} and defines metadata filters
+ * for different scopes:
+ * <ul>
+ *   <li><b>Catalog-level</b>: applies to all schemas and tables under the catalog.</li>
+ *   <li><b>Schema-level</b>: applies to all tables under the specified catalog and schema.</li>
+ *   <li><b>Table-level</b>: applies to the fully qualified catalog.schema.table.</li>
+ * </ul>
+ * <p>
+ * Each scope maps to a list of filter definitions. Each filter includes:
+ * <ul>
+ *   <li>{@code filterName}: must match a column name in the table's schema. Note that only numeric
+ *   type column can be used as metadata filter now.</li>
+ *   <li>{@code rangeMapping} (optional): specifies how the filter should be remapped when it
+ *   targets metadata-only columns. Note that this option is valid only if the column is numeric
+ *   type.
+ *       For example, a condition like {@code "msg.timestamp" > 1234} will be rewritten as
+ *       {@code end_timestamp > 1234} to ensure metadata-based filtering produces a superset of the
+ *       actual result.</li>
+ * </ul>
+ * <p>
+ * This provider is used by {@code ClpFilterToKqlConverter} to determine which columns are eligible
+ * for metadata filter push down, and by {@code ClpMySqlSplitProvider} to construct metadata filter
+ * queries that determine which splits to read.
+ */
 public class ClpMetadataFilterProvider
 {
     private final Map<String, TableConfig> filterMap;
@@ -68,8 +95,7 @@ public void checkContainsAllFilters(SchemaTableName schemaTableName, String meta
     {
         boolean hasAllMetadataFilterColumns = true;
         String notFoundFilterColumnName = "";
-        for (String columnName : getFilterNames(
-                CONNECTOR_NAME + "." + schemaTableName.toString())) {
+        for (String columnName : getFilterNames(format("%s.%s", CONNECTOR_NAME, schemaTableName))) {
             if (!metadataFilterKqlQuery.contains(columnName)) {
                 hasAllMetadataFilterColumns = false;
                 notFoundFilterColumnName = columnName;
@@ -101,12 +127,14 @@ public String remapFilterSql(String scope, String sql)
         String remappedSql = sql;
         for (Map.Entry<String, RangeMapping> entry : mappings.entrySet()) {
             remappedSql = remappedSql.replaceAll(
-                    "\"(" + entry.getKey() + ")\"\\s(>=?)\\s([0-9]*)", entry.getValue().upperBound + " $2 $3");
+                    format("\"(%s)\"\\s(>=?)\\s([0-9]*)", entry.getKey()),
+                    format("%s $2 $3", entry.getValue().upperBound));
             remappedSql = remappedSql.replaceAll(
-                    "\"(" + entry.getKey() + ")\"\\s(<=?)\\s([0-9]*)", entry.getValue().lowerBound + " $2 $3");
+                    format("\"(%s)\"\\s(<=?)\\s([0-9]*)", entry.getKey()),
+                    format("%s $2 $3", entry.getValue().lowerBound));
             remappedSql = remappedSql.replaceAll(
-                    "\"(" + entry.getKey() + ")\"\\s(=)\\s([0-9]*)",
-                    "(" + entry.getValue().lowerBound + " <= $3 AND " + entry.getValue().upperBound + " >= $3)");
+                    format("\"(%s)\"\\s(=)\\s([0-9]*)", entry.getKey()),
+                    format("(%s <= $3 AND %s >= $3)", entry.getValue().lowerBound, entry.getValue().upperBound));
         }
         return remappedSql;
     }
@@ -115,18 +143,19 @@ public Set<String> getFilterNames(String scope)
     {
         String[] splitScope = scope.split("\\.");
         if (0 == splitScope.length) {
-            return Collections.emptySet();
+            return ImmutableSet.of();
         }
-        Set<String> filterNames = new HashSet<>(getAllFilerNamesFromTableConfig(filterMap.get(splitScope[0])));
+        ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+        builder.addAll(getAllFilerNamesFromTableConfig(filterMap.get(splitScope[0])));
 
         if (1 < splitScope.length) {
-            filterNames.addAll(getAllFilerNamesFromTableConfig(filterMap.get(splitScope[0] + "." + splitScope[1])));
+            builder.addAll(getAllFilerNamesFromTableConfig(filterMap.get(splitScope[0] + "." + splitScope[1])));
         }
 
         if (3 == splitScope.length) {
-            filterNames.addAll(getAllFilerNamesFromTableConfig(filterMap.get(scope)));
+            builder.addAll(getAllFilerNamesFromTableConfig(filterMap.get(scope)));
         }
-        return ImmutableSet.copyOf(filterNames);
+        return builder.build();
     }
 
     private Set<String> getAllFilerNamesFromTableConfig(TableConfig tableConfig)

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

@@ -89,8 +89,7 @@ public PlanNode visitFilter(FilterNode node, RewriteContext<Void> context)
             Optional<String> metadataSql = clpExpression.getMetadataSql();
             Optional<RowExpression> remainingPredicate = clpExpression.getRemainingExpression();
 
-            // This must be checked before check if the KQL is present, otherwise KQL can be emptry then this function
-            // directly exits.
+            // Check metadata filters before the KQL query
             metadataFilterProvider.checkContainsAllFilters(clpTableHandle.getSchemaTableName(), metadataSql.orElse(""));
             if (metadataSql.isPresent()) {
                 metadataSql = Optional.of(metadataFilterProvider.remapFilterSql(scope, metadataSql.get()));

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

@@ -15,84 +15,12 @@
 
 import com.facebook.airlift.log.Logger;
 
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
 public class ClpUtils
 {
     public static class KqlUtils
     {
         private static final Logger log = Logger.get(KqlUtils.class);
 
-        public static String toSql(String kqlQuery)
-        {
-            String sqlQuery = kqlQuery;
-
-            // Translate string filters
-            Pattern stringFilterPattern = Pattern.compile("([a-zA-Z_][a-zA-Z0-9_]*):\\s*\"([^\"]+)\"");
-            Matcher stringFilterMatcher = stringFilterPattern.matcher(sqlQuery);
-            StringBuilder translateStringFiltersSb = new StringBuilder();
-
-            int lastStringFilterEnd = 0;
-            while (stringFilterMatcher.find()) {
-                String columnName = stringFilterMatcher.group(1);
-                String value = stringFilterMatcher.group(2);
-
-                // Append everything between last match and this one
-                translateStringFiltersSb.append(sqlQuery, lastStringFilterEnd, stringFilterMatcher.start());
-
-                if (value.contains("*")) {
-                    translateStringFiltersSb.append("\"")
-                            .append(columnName)
-                            .append("\" LIKE '")
-                            .append(value.replace('*', '%'))
-                            .append("'");
-                }
-                else {
-                    translateStringFiltersSb.append("\"")
-                            .append(columnName)
-                            .append("\" = '")
-                            .append(value)
-                            .append("'");
-                }
-
-                lastStringFilterEnd = stringFilterMatcher.end();
-            }
-            translateStringFiltersSb.append(sqlQuery.substring(lastStringFilterEnd));
-            sqlQuery = translateStringFiltersSb.toString();
-
-            // Translate numeric filters
-            Pattern numericFilterPattern = Pattern.compile("([a-zA-Z_][a-zA-Z0-9_]*)\\s*([><]=?|:)\\s*(-?\\d+(?:\\.\\d+)?)");
-            Matcher numericFilterMatcher = numericFilterPattern.matcher(sqlQuery);
-            StringBuilder translateNumericFilterSb = new StringBuilder();
-
-            int lastNumericFilterEnd = 0;
-            while (numericFilterMatcher.find()) {
-                String columnName = numericFilterMatcher.group(1);
-                String operator = numericFilterMatcher.group(2);
-                if (":".equals(operator)) {
-                    operator = "=";
-                }
-                String value = numericFilterMatcher.group(3);
-
-                // Append everything between last match and this one
-                translateNumericFilterSb.append(sqlQuery, lastNumericFilterEnd, numericFilterMatcher.start());
-
-                translateNumericFilterSb.append("\"")
-                        .append(columnName)
-                        .append("\" ")
-                        .append(operator)
-                        .append(" ")
-                        .append(value);
-
-                lastNumericFilterEnd = numericFilterMatcher.end();
-            }
-            translateNumericFilterSb.append(sqlQuery.substring(lastNumericFilterEnd));
-            sqlQuery = translateNumericFilterSb.toString();
-
-            return sqlQuery;
-        }
-
         public static String escapeKqlSpecialCharsForStringValue(String literalString)
         {
             String escaped = literalString;

presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java

@@ -73,7 +73,7 @@ public List<ClpSplit> listSplits(ClpTableLayoutHandle clpTableLayoutHandle)
             String metadataFilterQuery = clpTableLayoutHandle.getMetadataSql().get();
             archivePathQuery += " AND (" + metadataFilterQuery + ")";
         }
-        log.info("Query for archive: %s", archivePathQuery);
+        log.debug("Query for archive: %s", archivePathQuery);
 
         try (Connection connection = getConnection()) {
             // Fetch archive IDs and create splits

presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java

@@ -48,13 +48,13 @@ public final class ClpMetadataDbSetUp
     public static final String METADATA_DB_TABLE_PREFIX = "clp_";
     public static final String METADATA_DB_URL_TEMPLATE = "jdbc:h2:file:%s;MODE=MySQL;DATABASE_TO_UPPER=FALSE";
     public static final String METADATA_DB_USER = "sa";
-    public static final String ARCHIVE_STORAGE_DIRECTORY_BASE = "/tmp/archives/";
+    public static final String ARCHIVES_STORAGE_DIRECTORY_BASE = "/tmp/archives/";
 
     private static final Logger log = Logger.get(ClpMetadataDbSetUp.class);
     private static final String DATASETS_TABLE_NAME = METADATA_DB_TABLE_PREFIX + DATASETS_TABLE_SUFFIX;
-    private static final String ARCHIVE_TABLE_COLUMN_BEGIN_TIMESTAMP = "begin_timestamp";
+    private static final String ARCHIVES_TABLE_COLUMN_BEGIN_TIMESTAMP = "begin_timestamp";
     private static final String ARCHIVES_TABLE_COLUMN_PAGINATION_ID = "pagination_id";
-    private static final String ARCHIVE_TABLE_COLUMN_END_TIMESTAMP = "end_timestamp";
+    private static final String ARCHIVES_TABLE_COLUMN_END_TIMESTAMP = "end_timestamp";
 
     private ClpMetadataDbSetUp()
     {
@@ -143,17 +143,17 @@ public static ClpMySqlSplitProvider setupSplit(DbHandle dbHandle, Map<String, Li
                         archiveTableName,
                         ARCHIVES_TABLE_COLUMN_PAGINATION_ID,
                         ARCHIVES_TABLE_COLUMN_ID,
-                        ARCHIVE_TABLE_COLUMN_BEGIN_TIMESTAMP,
-                        ARCHIVE_TABLE_COLUMN_END_TIMESTAMP);
+                        ARCHIVES_TABLE_COLUMN_BEGIN_TIMESTAMP,
+                        ARCHIVES_TABLE_COLUMN_END_TIMESTAMP);
 
                 stmt.execute(createArchiveTableSQL);
 
                 String insertArchiveTableSQL = format(
                         "INSERT INTO %s (%s, %s, %s) VALUES (?, ?, ?)",
                         archiveTableName,
                         ARCHIVES_TABLE_COLUMN_ID,
-                        ARCHIVE_TABLE_COLUMN_BEGIN_TIMESTAMP,
-                        ARCHIVE_TABLE_COLUMN_END_TIMESTAMP);
+                        ARCHIVES_TABLE_COLUMN_BEGIN_TIMESTAMP,
+                        ARCHIVES_TABLE_COLUMN_END_TIMESTAMP);
                 try (PreparedStatement pstmt = conn.prepareStatement(insertArchiveTableSQL)) {
                     for (ArchiveTableRow split : tableSplits.getValue()) {
                         pstmt.setString(1, split.id);
@@ -213,7 +213,7 @@ private static void updateDatasetsTable(Connection conn, String tableName)
                 DATASETS_TABLE_COLUMN_ARCHIVE_STORAGE_DIRECTORY);
         try (PreparedStatement pstmt = conn.prepareStatement(insertDatasetsTableSql)) {
             pstmt.setString(1, tableName);
-            pstmt.setString(2, ARCHIVE_STORAGE_DIRECTORY_BASE + tableName);
+            pstmt.setString(2, ARCHIVES_STORAGE_DIRECTORY_BASE + tableName);
             pstmt.executeUpdate();
         }
     }