feat: Add CLP_GET_* UDFs with rewrites for schemaless querying.

presto-clp/pom.xml

@@ -135,6 +135,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.facebook.presto</groupId>
+            <artifactId>presto-main-base</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-math3</artifactId>

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

@@ -15,6 +15,7 @@
 
 import com.facebook.airlift.bootstrap.LifeCycleManager;
 import com.facebook.airlift.log.Logger;
+import com.facebook.presto.plugin.clp.optimization.ClpPlanOptimizerProvider;
 import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider;
 import com.facebook.presto.spi.connector.Connector;
 import com.facebook.presto.spi.connector.ConnectorMetadata;

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

@@ -0,0 +1,68 @@
+/*
+ * 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 com.facebook.presto.plugin.clp;
+
+import com.facebook.presto.common.block.Block;
+import com.facebook.presto.common.type.StandardTypes;
+import com.facebook.presto.spi.function.Description;
+import com.facebook.presto.spi.function.ScalarFunction;
+import com.facebook.presto.spi.function.SqlType;
+import io.airlift.slice.Slice;
+
+public final class ClpFunctions
+{
+    private ClpFunctions()
+    {
+    }
+
+    @ScalarFunction(value = "CLP_GET_INT", deterministic = false)
+    @Description("Retrieves an integer value corresponding to the given JSON path.")
+    @SqlType(StandardTypes.BIGINT)
+    public static long clpGetInt(@SqlType(StandardTypes.VARCHAR) Slice jsonPath)
+    {
+        throw new UnsupportedOperationException("CLP_GET_INT is a placeholder function without implementation.");
+    }
+
+    @ScalarFunction(value = "CLP_GET_FLOAT", deterministic = false)
+    @Description("Retrieves a floating point value corresponding to the given JSON path.")
+    @SqlType(StandardTypes.DOUBLE)
+    public static double clpGetFloat(@SqlType(StandardTypes.VARCHAR) Slice jsonPath)
+    {
+        throw new UnsupportedOperationException("CLP_GET_FLOAT is a placeholder function without implementation.");
+    }
+
+    @ScalarFunction(value = "CLP_GET_BOOL", deterministic = false)
+    @Description("Retrieves a boolean value corresponding to the given JSON path.")
+    @SqlType(StandardTypes.BOOLEAN)
+    public static boolean clpGetBool(@SqlType(StandardTypes.VARCHAR) Slice jsonPath)
+    {
+        throw new UnsupportedOperationException("CLP_GET_BOOL is a placeholder function without implementation.");
+    }
+
+    @ScalarFunction(value = "CLP_GET_STRING", deterministic = false)
+    @Description("Retrieves a string value corresponding to the given JSON path.")
+    @SqlType(StandardTypes.VARCHAR)
+    public static Slice clpGetString(@SqlType(StandardTypes.VARCHAR) Slice jsonPath)
+    {
+        throw new UnsupportedOperationException("CLP_GET_STRING is a placeholder function without implementation.");
+    }
+
+    @ScalarFunction(value = "CLP_GET_STRING_ARRAY", deterministic = false)
+    @Description("Retrieves an array value corresponding to the given JSON path and converts each element into a string.")
+    @SqlType("ARRAY(VARCHAR)")
+    public static Block clpGetStringArray(@SqlType(StandardTypes.VARCHAR) Slice jsonPath)
+    {
+        throw new UnsupportedOperationException("CLP_GET_STRING_ARRAY is a placeholder function without implementation.");
+    }
+}

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

@@ -16,6 +16,9 @@
 import com.facebook.presto.spi.Plugin;
 import com.facebook.presto.spi.connector.ConnectorFactory;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+import java.util.Set;
 
 public class ClpPlugin
         implements Plugin
@@ -25,4 +28,10 @@ public Iterable<ConnectorFactory> getConnectorFactories()
     {
         return ImmutableList.of(new ClpConnectorFactory());
     }
+
+    @Override
+    public Set<Class<?>> getFunctions()
+    {
+        return ImmutableSet.of(ClpFunctions.class);
+    }
 }

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

@@ -11,9 +11,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.facebook.presto.plugin.clp;
+package com.facebook.presto.plugin.clp.optimization;
 
 import com.facebook.airlift.log.Logger;
+import com.facebook.presto.plugin.clp.ClpExpression;
+import com.facebook.presto.plugin.clp.ClpTableHandle;
+import com.facebook.presto.plugin.clp.ClpTableLayoutHandle;
 import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPlanOptimizer;
@@ -31,7 +34,6 @@
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
 import com.google.common.collect.ImmutableSet;
 
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
@@ -42,15 +44,15 @@
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
-public class ClpPlanOptimizer
+public class ClpComputePushDown
         implements ConnectorPlanOptimizer
 {
-    private static final Logger log = Logger.get(ClpPlanOptimizer.class);
+    private static final Logger log = Logger.get(ClpComputePushDown.class);
     private final FunctionMetadataManager functionManager;
     private final StandardFunctionResolution functionResolution;
     private final ClpSplitFilterProvider splitFilterProvider;
 
-    public ClpPlanOptimizer(FunctionMetadataManager functionManager, StandardFunctionResolution functionResolution, ClpSplitFilterProvider splitFilterProvider)
+    public ClpComputePushDown(FunctionMetadataManager functionManager, StandardFunctionResolution functionResolution, ClpSplitFilterProvider splitFilterProvider)
     {
         this.functionManager = requireNonNull(functionManager, "functionManager is null");
         this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
@@ -99,59 +101,75 @@ public PlanNode visitFilter(FilterNode node, RewriteContext<Void> context)
             if (!(node.getSource() instanceof TableScanNode)) {
                 return node;
             }
+
+            return processFilter(node, (TableScanNode) node.getSource());
+        }
+
+        private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNode)
+        {
             hasVisitedFilter = true;
 
-            TableScanNode tableScanNode = (TableScanNode) node.getSource();
-            Map<VariableReferenceExpression, ColumnHandle> assignments = new HashMap<>(tableScanNode.getAssignments());
             TableHandle tableHandle = tableScanNode.getTable();
             ClpTableHandle clpTableHandle = (ClpTableHandle) tableHandle.getConnectorHandle();
+
             String tableScope = CONNECTOR_NAME + "." + clpTableHandle.getSchemaTableName().toString();
-            ClpExpression clpExpression = node.getPredicate().accept(
+            Map<VariableReferenceExpression, ColumnHandle> assignments = tableScanNode.getAssignments();
+
+            ClpExpression clpExpression = filterNode.getPredicate().accept(
                     new ClpFilterToKqlConverter(
                             functionResolution,
                             functionManager,
+                            assignments,
                             splitFilterProvider.getColumnNames(tableScope)),
-                    assignments);
+                    null);
             Optional<String> kqlQuery = clpExpression.getPushDownExpression();
             Optional<String> metadataSqlQuery = clpExpression.getMetadataSqlQuery();
             Optional<RowExpression> remainingPredicate = clpExpression.getRemainingExpression();
 
             // Perform required metadata filter checks before handling the KQL query (if kqlQuery
             // isn't present, we'll return early, skipping subsequent checks).
             splitFilterProvider.checkContainsRequiredFilters(ImmutableSet.of(tableScope), metadataSqlQuery.orElse(""));
-            if (metadataSqlQuery.isPresent()) {
+            boolean hasMetadataFilter = metadataSqlQuery.isPresent() && !metadataSqlQuery.get().isEmpty();
+            if (hasMetadataFilter) {
                 metadataSqlQuery = Optional.of(splitFilterProvider.remapSplitFilterPushDownExpression(tableScope, metadataSqlQuery.get()));
                 log.debug("Metadata SQL query: %s", metadataSqlQuery);
             }
 
-            if (!kqlQuery.isPresent()) {
-                return node;
-            }
-            log.debug("KQL query: %s", kqlQuery);
-
-            ClpTableLayoutHandle clpTableLayoutHandle = new ClpTableLayoutHandle(clpTableHandle, kqlQuery, metadataSqlQuery);
-            TableScanNode newTableScanNode = new TableScanNode(
-                    tableScanNode.getSourceLocation(),
-                    idAllocator.getNextId(),
-                    new TableHandle(
-                            tableHandle.getConnectorId(),
-                            clpTableHandle,
-                            tableHandle.getTransaction(),
-                            Optional.of(clpTableLayoutHandle)),
-                    tableScanNode.getOutputVariables(),
-                    tableScanNode.getAssignments(),
-                    tableScanNode.getTableConstraints(),
-                    tableScanNode.getCurrentConstraint(),
-                    tableScanNode.getEnforcedConstraint(),
-                    tableScanNode.getCteMaterializationInfo());
-            if (!remainingPredicate.isPresent()) {
-                return newTableScanNode;
+            if (kqlQuery.isPresent() || hasMetadataFilter) {
+                if (kqlQuery.isPresent()) {
+                    log.debug("KQL query: %s", kqlQuery);
+                }
+
+                ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle(clpTableHandle, kqlQuery, metadataSqlQuery);
+                TableHandle newTableHandle = new TableHandle(
+                        tableHandle.getConnectorId(),
+                        clpTableHandle,
+                        tableHandle.getTransaction(),
+                        Optional.of(layoutHandle));
+
+                tableScanNode = new TableScanNode(
+                        tableScanNode.getSourceLocation(),
+                        idAllocator.getNextId(),
+                        newTableHandle,
+                        tableScanNode.getOutputVariables(),
+                        tableScanNode.getAssignments(),
+                        tableScanNode.getTableConstraints(),
+                        tableScanNode.getCurrentConstraint(),
+                        tableScanNode.getEnforcedConstraint(),
+                        tableScanNode.getCteMaterializationInfo());
             }
 
-            return new FilterNode(node.getSourceLocation(),
-                    idAllocator.getNextId(),
-                    newTableScanNode,
-                    remainingPredicate.get());
+            if (remainingPredicate.isPresent()) {
+                // Not all predicate pushed down, need new FilterNode
+                return new FilterNode(
+                        filterNode.getSourceLocation(),
+                        idAllocator.getNextId(),
+                        tableScanNode,
+                        remainingPredicate.get());
+            }
+            else {
+                return tableScanNode;
+            }
         }
     }
 }