refactor: Extract split filter provider interface to improve flexibility for user customization.

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.metadata.filter.ClpMetadataFilterProvider;
 import com.facebook.presto.spi.connector.Connector;
 import com.facebook.presto.spi.connector.ConnectorMetadata;
 import com.facebook.presto.spi.connector.ConnectorPlanOptimizerProvider;

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

@@ -16,6 +16,8 @@
 import com.facebook.airlift.configuration.AbstractConfigurationAwareModule;
 import com.facebook.presto.plugin.clp.metadata.ClpMetadataProvider;
 import com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider;
+import com.facebook.presto.plugin.clp.metadata.filter.ClpMetadataFilterProvider;
+import com.facebook.presto.plugin.clp.metadata.filter.ClpMySqlMetadataFilterProvider;
 import com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider;
 import com.facebook.presto.plugin.clp.split.ClpSplitProvider;
 import com.facebook.presto.spi.PrestoException;
@@ -37,13 +39,13 @@ protected void setup(Binder binder)
         binder.bind(ClpConnector.class).in(Scopes.SINGLETON);
         binder.bind(ClpMetadata.class).in(Scopes.SINGLETON);
         binder.bind(ClpRecordSetProvider.class).in(Scopes.SINGLETON);
-        binder.bind(ClpMetadataFilterProvider.class).in(Scopes.SINGLETON);
         binder.bind(ClpSplitManager.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(ClpConfig.class);
 
         ClpConfig config = buildConfigObject(ClpConfig.class);
         if (config.getMetadataProviderType() == MetadataProviderType.MYSQL) {
             binder.bind(ClpMetadataProvider.class).to(ClpMySqlMetadataProvider.class).in(Scopes.SINGLETON);
+            binder.bind(ClpMetadataFilterProvider.class).to(ClpMySqlMetadataFilterProvider.class).in(Scopes.SINGLETON);
         }
         else {
             throw new PrestoException(CLP_UNSUPPORTED_METADATA_SOURCE, "Unsupported metadata provider type: " + config.getMetadataProviderType());

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

@@ -14,6 +14,7 @@
 package com.facebook.presto.plugin.clp;
 
 import com.facebook.airlift.log.Logger;
+import com.facebook.presto.plugin.clp.metadata.filter.ClpMetadataFilterProvider;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorPlanOptimizer;
 import com.facebook.presto.spi.ConnectorPlanRewriter;
@@ -94,7 +95,7 @@ public PlanNode visitFilter(FilterNode node, RewriteContext<Void> context)
             // isn't present, we'll return early, skipping subsequent checks).
             metadataFilterProvider.checkContainsRequiredFilters(clpTableHandle.getSchemaTableName(), metadataSqlQuery.orElse(""));
             if (metadataSqlQuery.isPresent()) {
-                metadataSqlQuery = Optional.of(metadataFilterProvider.remapFilterSql(scope, metadataSqlQuery.get()));
+                metadataSqlQuery = Optional.of(metadataFilterProvider.remapMetadataFilterPushDown(scope, metadataSqlQuery.get()));
                 log.debug("Metadata SQL query: %s", metadataSqlQuery);
             }
 

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

@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.plugin.clp;
 
+import com.facebook.presto.plugin.clp.metadata.filter.ClpMetadataFilterProvider;
 import com.facebook.presto.spi.ConnectorPlanOptimizer;
 import com.facebook.presto.spi.connector.ConnectorPlanOptimizerProvider;
 import com.facebook.presto.spi.function.FunctionMetadataManager;

presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/filter/ClpMetadataDatabaseSpecificDeserializer.java

@@ -0,0 +1,49 @@
+/*
+ * 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.metadata.filter;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+import java.io.IOException;
+
+import static com.facebook.presto.plugin.clp.metadata.filter.ClpMetadataFilter.MetadataDatabaseSpecific;
+
+/**
+ * Uses the given implementation of {@link MetadataDatabaseSpecific} to deserialize the
+ * {@code "metadataDatabaseSpecific"} field in the filter. The implementation is originally chosen
+ * by the {@code clp.metadata-provider-type} config option.
+ */
+public class ClpMetadataDatabaseSpecificDeserializer
+        extends JsonDeserializer<MetadataDatabaseSpecific>
+{
+    private final Class<? extends MetadataDatabaseSpecific> actualMetadataDatabaseSpecificClass;
+
+    public ClpMetadataDatabaseSpecificDeserializer(Class<? extends MetadataDatabaseSpecific> actualMetadataDatabaseSpecificClass)
+    {
+        this.actualMetadataDatabaseSpecificClass = actualMetadataDatabaseSpecificClass;
+    }
+
+    @Override
+    public MetadataDatabaseSpecific deserialize(JsonParser p, DeserializationContext ctxt) throws IOException
+    {
+        ObjectNode node = p.getCodec().readTree(p);
+        ObjectMapper mapper = (ObjectMapper) p.getCodec();
+
+        return mapper.treeToValue(node, actualMetadataDatabaseSpecificClass);
+    }
+}