Implement pushdown of dynamic filters in phoenix connector

trinodb · Aug 4, 2022 · 4b2c5b7 · 4b2c5b7
1 parent 29bff60
commit 4b2c5b7
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 3 deletions.
diff --git a/plugin/trino-phoenix5/pom.xml b/plugin/trino-phoenix5/pom.xml
@@ -107,6 +107,11 @@
             <artifactId>modernizer-maven-annotations</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.weakref</groupId>
+            <artifactId>jmxutils</artifactId>
+        </dependency>
+
         <!-- used by tests but also needed transitively -->
         <dependency>
             <groupId>io.airlift</groupId>

diff --git a/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixClientModule.java b/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixClientModule.java
@@ -29,11 +29,16 @@
 import io.trino.plugin.jdbc.DecimalModule;
 import io.trino.plugin.jdbc.DefaultQueryBuilder;
 import io.trino.plugin.jdbc.DriverConnectionFactory;
+import io.trino.plugin.jdbc.DynamicFilteringStats;
 import io.trino.plugin.jdbc.ForBaseJdbc;
+import io.trino.plugin.jdbc.ForJdbcDynamicFiltering;
 import io.trino.plugin.jdbc.ForLazyConnectionFactory;
 import io.trino.plugin.jdbc.ForRecordCursor;
 import io.trino.plugin.jdbc.JdbcClient;
 import io.trino.plugin.jdbc.JdbcDiagnosticModule;
+import io.trino.plugin.jdbc.JdbcDynamicFilteringConfig;
+import io.trino.plugin.jdbc.JdbcDynamicFilteringSessionProperties;
+import io.trino.plugin.jdbc.JdbcDynamicFilteringSplitManager;
 import io.trino.plugin.jdbc.JdbcMetadataConfig;
 import io.trino.plugin.jdbc.JdbcMetadataSessionProperties;
 import io.trino.plugin.jdbc.JdbcPageSinkProvider;
@@ -73,14 +78,24 @@
 import static io.trino.plugin.jdbc.JdbcModule.bindTablePropertiesProvider;
 import static io.trino.plugin.phoenix5.ConfigurationInstantiator.newEmptyConfiguration;
 import static io.trino.plugin.phoenix5.PhoenixErrorCode.PHOENIX_CONFIG_ERROR;
+import static java.util.Objects.requireNonNull;
+import static org.weakref.jmx.guice.ExportBinder.newExporter;
 
 public class PhoenixClientModule
         extends AbstractConfigurationAwareModule
 {
+    private final String catalogName;
+
+    public PhoenixClientModule(String catalogName)
+    {
+        this.catalogName = requireNonNull(catalogName, "catalogName is null");
+    }
+
     @Override
     protected void setup(Binder binder)
     {
-        binder.bind(ConnectorSplitManager.class).annotatedWith(ForClassLoaderSafe.class).to(PhoenixSplitManager.class).in(Scopes.SINGLETON);
+        binder.bind(ConnectorSplitManager.class).annotatedWith(ForJdbcDynamicFiltering.class).to(PhoenixSplitManager.class).in(Scopes.SINGLETON);
+        binder.bind(ConnectorSplitManager.class).annotatedWith(ForClassLoaderSafe.class).to(JdbcDynamicFilteringSplitManager.class).in(Scopes.SINGLETON);
         binder.bind(ConnectorSplitManager.class).to(ClassLoaderSafeConnectorSplitManager.class).in(Scopes.SINGLETON);
         binder.bind(ConnectorRecordSetProvider.class).annotatedWith(ForClassLoaderSafe.class).to(JdbcRecordSetProvider.class).in(Scopes.SINGLETON);
         binder.bind(ConnectorRecordSetProvider.class).to(ClassLoaderSafeConnectorRecordSetProvider.class).in(Scopes.SINGLETON);
@@ -94,9 +109,15 @@ protected void setup(Binder binder)
         bindSessionPropertiesProvider(binder, JdbcMetadataSessionProperties.class);
         bindSessionPropertiesProvider(binder, JdbcWriteSessionProperties.class);
         bindSessionPropertiesProvider(binder, PhoenixSessionProperties.class);
+        bindSessionPropertiesProvider(binder, JdbcDynamicFilteringSessionProperties.class);
+
+        binder.bind(DynamicFilteringStats.class).in(Scopes.SINGLETON);
+        newExporter(binder).export(DynamicFilteringStats.class)
+                .as(generator -> generator.generatedNameOf(DynamicFilteringStats.class, catalogName));
 
         configBinder(binder).bindConfig(JdbcMetadataConfig.class);
         configBinder(binder).bindConfig(JdbcWriteConfig.class);
+        configBinder(binder).bindConfig(JdbcDynamicFilteringConfig.class);
 
         binder.bind(PhoenixClient.class).in(Scopes.SINGLETON);
         binder.bind(JdbcClient.class).annotatedWith(ForBaseJdbc.class).to(Key.get(PhoenixClient.class)).in(Scopes.SINGLETON);

diff --git a/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixConnectorFactory.java b/plugin/trino-phoenix5/src/main/java/io/trino/plugin/phoenix5/PhoenixConnectorFactory.java
@@ -53,7 +53,7 @@ public Connector create(String catalogName, Map<String, String> requiredConfig,
         try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
             Bootstrap app = new Bootstrap(
                     new JsonModule(),
-                    new PhoenixClientModule(),
+                    new PhoenixClientModule(catalogName),
                     binder -> {
                         binder.bind(CatalogName.class).toInstance(new CatalogName(catalogName));
                         binder.bind(ClassLoader.class).toInstance(PhoenixConnectorFactory.class.getClassLoader());

diff --git a/plugin/trino-phoenix5/src/test/java/io/trino/plugin/phoenix5/TestPhoenixConnectorTest.java b/plugin/trino-phoenix5/src/test/java/io/trino/plugin/phoenix5/TestPhoenixConnectorTest.java
@@ -89,7 +89,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_LIMIT_PUSHDOWN:
             case SUPPORTS_TOPN_PUSHDOWN:
             case SUPPORTS_AGGREGATION_PUSHDOWN:
-            case SUPPORTS_DYNAMIC_FILTER_PUSHDOWN:
                 return false;
 
             case SUPPORTS_CREATE_TABLE_WITH_TABLE_COMMENT: