Merge branch 'release-0.293-clp-connector' into xwei/add-k8s-manual-doc

Author: anlowee

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/ClpConnector.java

@@ -17,9 +17,12 @@
 import com.facebook.airlift.log.Logger;
 import com.facebook.presto.spi.connector.Connector;
 import com.facebook.presto.spi.connector.ConnectorMetadata;
+import com.facebook.presto.spi.connector.ConnectorPlanOptimizerProvider;
 import com.facebook.presto.spi.connector.ConnectorRecordSetProvider;
 import com.facebook.presto.spi.connector.ConnectorSplitManager;
 import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
+import com.facebook.presto.spi.function.FunctionMetadataManager;
+import com.facebook.presto.spi.function.StandardFunctionResolution;
 import com.facebook.presto.spi.transaction.IsolationLevel;
 
 import javax.inject.Inject;
@@ -35,14 +38,33 @@ public class ClpConnector
     private final ClpMetadata metadata;
     private final ClpRecordSetProvider recordSetProvider;
     private final ClpSplitManager splitManager;
+    private final FunctionMetadataManager functionManager;
+    private final StandardFunctionResolution functionResolution;
+    private final ClpMetadataFilterProvider metadataFilterProvider;
 
     @Inject
-    public ClpConnector(LifeCycleManager lifeCycleManager, ClpMetadata metadata, ClpRecordSetProvider recordSetProvider, ClpSplitManager splitManager)
+    public ClpConnector(
+            LifeCycleManager lifeCycleManager,
+            ClpMetadata metadata,
+            ClpRecordSetProvider recordSetProvider,
+            ClpSplitManager splitManager,
+            FunctionMetadataManager functionManager,
+            StandardFunctionResolution functionResolution,
+            ClpMetadataFilterProvider metadataFilterProvider)
     {
         this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
         this.metadata = requireNonNull(metadata, "metadata is null");
         this.recordSetProvider = requireNonNull(recordSetProvider, "recordSetProvider is null");
         this.splitManager = requireNonNull(splitManager, "splitManager is null");
+        this.functionManager = requireNonNull(functionManager, "functionManager is null");
+        this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
+        this.metadataFilterProvider = requireNonNull(metadataFilterProvider, "metadataFilterProvider is null");
+    }
+
+    @Override
+    public ConnectorPlanOptimizerProvider getConnectorPlanOptimizerProvider()
+    {
+        return new ClpPlanOptimizerProvider(functionManager, functionResolution, metadataFilterProvider);
     }
 
     @Override

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

@@ -23,7 +23,6 @@
 import com.facebook.presto.spi.connector.ConnectorFactory;
 import com.facebook.presto.spi.function.FunctionMetadataManager;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
-import com.facebook.presto.spi.relation.RowExpressionService;
 import com.google.inject.Injector;
 
 import java.util.Map;
@@ -33,10 +32,12 @@
 public class ClpConnectorFactory
         implements ConnectorFactory
 {
+    public static final String CONNECTOR_NAME = "clp";
+
     @Override
     public String getName()
     {
-        return "clp";
+        return CONNECTOR_NAME;
     }
 
     @Override
@@ -54,7 +55,6 @@ public Connector create(String catalogName, Map<String, String> config, Connecto
             Bootstrap app = new Bootstrap(new JsonModule(), new ClpModule(), binder -> {
                 binder.bind(FunctionMetadataManager.class).toInstance(context.getFunctionMetadataManager());
                 binder.bind(NodeManager.class).toInstance(context.getNodeManager());
-                binder.bind(RowExpressionService.class).toInstance(context.getRowExpressionService());
                 binder.bind(StandardFunctionResolution.class).toInstance(context.getStandardFunctionResolution());
                 binder.bind(TypeManager.class).toInstance(context.getTypeManager());
             });

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

@@ -18,22 +18,30 @@
 import java.util.Optional;
 
 /**
- * Represents the result of converting a Presto RowExpression into a CLP-compatible KQL query. In
- * every case, `pushDownExpression` represents the part of the RowExpression that could be
- * converted to a KQL expression, and `remainingExpression` represents the part that could not be
- * converted.
+ * Represents the result of:
+ * <ul>
+ *     <li>splitting a {@link RowExpression} into an expression that can be pushed down to CLP
+ *     (`pushDownExpression`) and any remaining expression that cannot be
+ *     (`remainingExpression`).</li>
+ *     <li>a SQL query for filtering splits using CLP's metadata database.</li>
+ * </ul>
  */
 public class ClpExpression
 {
     // Optional KQL query or column name representing the fully or partially translatable part of the expression.
     private final Optional<String> pushDownExpression;
 
+    // Optional SQL string extracted from the query plan, which is only made of up of columns in
+    // CLP's metadata database.
+    private final Optional<String> metadataSqlQuery;
+
     // 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 metadataSqlQuery, RowExpression remainingExpression)
     {
         this.pushDownExpression = Optional.ofNullable(pushDownExpression);
+        this.metadataSqlQuery = Optional.ofNullable(metadataSqlQuery);
         this.remainingExpression = Optional.ofNullable(remainingExpression);
     }
 
@@ -42,7 +50,7 @@ public ClpExpression(String pushDownExpression, RowExpression remainingExpressio
      */
     public ClpExpression()
     {
-        this(null, null);
+        this(null, null, null);
     }
 
     /**
@@ -52,7 +60,19 @@ public ClpExpression()
      */
     public ClpExpression(String pushDownExpression)
     {
-        this(pushDownExpression, null);
+        this(pushDownExpression, null, null);
+    }
+
+    /**
+     * Creates a ClpExpression from a fully translatable KQL string or column name, as well as a
+     * metadata SQL string.
+     *
+     * @param pushDownExpression
+     * @param metadataSqlQuery
+     */
+    public ClpExpression(String pushDownExpression, String metadataSqlQuery)
+    {
+        this(pushDownExpression, metadataSqlQuery, null);
     }
 
     /**
@@ -62,14 +82,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> getMetadataSqlQuery()
+    {
+        return metadataSqlQuery;
+    }
+
     public Optional<RowExpression> getRemainingExpression()
     {
         return remainingExpression;