Enable dynamic filtering in JDBC connector

core/trino-main/src/main/java/io/trino/split/RecordPageSourceProvider.java

@@ -47,6 +47,6 @@ public ConnectorPageSource createPageSource(
             List<ColumnHandle> columns,
             DynamicFilter dynamicFilter)
     {
-        return new RecordPageSource(recordSetProvider.getRecordSet(transaction, session, split, table, columns));
+        return new RecordPageSource(recordSetProvider.getRecordSet(transaction, session, split, table, columns, dynamicFilter));
     }
 }

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorRecordSetProvider.java

@@ -17,6 +17,19 @@
 
 public interface ConnectorRecordSetProvider
 {
+    default RecordSet getRecordSet(
+            ConnectorTransactionHandle transaction,
+            ConnectorSession session,
+            ConnectorSplit split,
+            ConnectorTableHandle table,
+            List<? extends ColumnHandle> columns,
+            DynamicFilter dynamicFilter)
+    {
+        // By default, ignore dynamic filter (as it is an optimization and doesn't affect correctness).
+        return getRecordSet(transaction, session, split, table, columns);
+    }
+
+    @Deprecated
     default RecordSet getRecordSet(
             ConnectorTransactionHandle transaction,
             ConnectorSession session,

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorRecordSetProvider.java

@@ -21,6 +21,7 @@
 import io.trino.spi.connector.ConnectorSplit;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.ConnectorTransactionHandle;
+import io.trino.spi.connector.DynamicFilter;
 import io.trino.spi.connector.RecordSet;
 
 import javax.inject.Inject;
@@ -42,6 +43,12 @@ public ClassLoaderSafeConnectorRecordSetProvider(@ForClassLoaderSafe ConnectorRe
         this.classLoader = requireNonNull(classLoader, "classLoader is null");
     }
 
+    @Override
+    public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, ConnectorTableHandle table, List<? extends ColumnHandle> columns, DynamicFilter dynamicFilter)
+    {
+        return getRecordSet(transaction, session, split, table, columns);
+    }
+
     @Override
     public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, ConnectorTableHandle table, List<? extends ColumnHandle> columns)
     {

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DynamicFilteringJdbcConfig.java

@@ -0,0 +1,57 @@
+/*
+ * 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 io.trino.plugin.jdbc;
+
+import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
+import io.airlift.units.Duration;
+import io.airlift.units.MinDuration;
+
+import javax.validation.constraints.NotNull;
+
+import java.util.concurrent.TimeUnit;
+
+public class DynamicFilteringJdbcConfig
+{
+    private boolean enableDynamicFiltering;
+    private Duration dynamicFilteringWaitTimeout = new Duration(0, TimeUnit.MINUTES);
+
+    public boolean isEnableDynamicFiltering()
+    {
+        return enableDynamicFiltering;
+    }
+
+    @Config("dynamic-filtering.enabled")
+    public DynamicFilteringJdbcConfig setEnableDynamicFiltering(boolean enableDynamicFiltering)
+    {
+        this.enableDynamicFiltering = enableDynamicFiltering;
+        return this;
+    }
+
+    @NotNull
+    @MinDuration("0ms")
+    public Duration getDynamicFilteringWaitTimeout()
+    {
+        return dynamicFilteringWaitTimeout;
+    }
+
+    @Config("dynamic-filtering.wait-timeout")
+    @ConfigDescription("Duration to wait for completion of dynamic filters")
+    public DynamicFilteringJdbcConfig setDynamicFilteringWaitTimeout(Duration timeout)
+    {
+        this.dynamicFilteringWaitTimeout = timeout;
+        return this;
+    }
+}

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DynamicFilteringJdbcSessionProperties.java

@@ -0,0 +1,69 @@
+/*
+ * 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 io.trino.plugin.jdbc;
+
+import com.google.common.collect.ImmutableList;
+import io.airlift.units.Duration;
+import io.trino.plugin.base.session.SessionPropertiesProvider;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.session.PropertyMetadata;
+
+import javax.inject.Inject;
+
+import java.util.List;
+
+import static io.trino.plugin.base.session.PropertyMetadataUtil.durationProperty;
+import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+
+public class DynamicFilteringJdbcSessionProperties
+        implements SessionPropertiesProvider
+{
+    private static final String DYNAMIC_FILTERING_ENABLED = "dynamic_filtering_enabled";
+    private static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
+
+    private final List<PropertyMetadata<?>> properties;
+
+    @Inject
+    public DynamicFilteringJdbcSessionProperties(DynamicFilteringJdbcConfig config)
+    {
+        properties = ImmutableList.of(
+                booleanProperty(
+                        DYNAMIC_FILTERING_ENABLED,
+                        "If dynamic filtering is enabled",
+                        config.isEnableDynamicFiltering(),
+                        false),
+                durationProperty(
+                        DYNAMIC_FILTERING_WAIT_TIMEOUT,
+                        "Duration to wait for completion of dynamic filters",
+                        config.getDynamicFilteringWaitTimeout(),
+                        false));
+    }
+
+    public static boolean isEnableDynamicFiltering(ConnectorSession session)
+    {
+        return session.getProperty(DYNAMIC_FILTERING_ENABLED, Boolean.class);
+    }
+
+    public static Duration getDynamicFilteringWaitTimeout(ConnectorSession session)
+    {
+        return session.getProperty(DYNAMIC_FILTERING_WAIT_TIMEOUT, Duration.class);
+    }
+
+    @Override
+    public List<PropertyMetadata<?>> getSessionProperties()
+    {
+        return properties;
+    }
+}

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcModule.java

@@ -75,9 +75,11 @@ public void setup(Binder binder)
         configBinder(binder).bindConfig(BaseJdbcConfig.class);
 
         configBinder(binder).bindConfig(TypeHandlingJdbcConfig.class);
+        configBinder(binder).bindConfig(DynamicFilteringJdbcConfig.class);
         bindSessionPropertiesProvider(binder, TypeHandlingJdbcSessionProperties.class);
         bindSessionPropertiesProvider(binder, JdbcMetadataSessionProperties.class);
         bindSessionPropertiesProvider(binder, JdbcWriteSessionProperties.class);
+        bindSessionPropertiesProvider(binder, DynamicFilteringJdbcSessionProperties.class);
 
         binder.bind(CachingJdbcClient.class).in(Scopes.SINGLETON);
         binder.bind(JdbcClient.class).to(Key.get(CachingJdbcClient.class)).in(Scopes.SINGLETON);

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcRecordSetProvider.java

@@ -21,6 +21,7 @@
 import io.trino.spi.connector.ConnectorSplit;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.ConnectorTransactionHandle;
+import io.trino.spi.connector.DynamicFilter;
 import io.trino.spi.connector.RecordSet;
 
 import javax.inject.Inject;
@@ -29,7 +30,10 @@
 import java.util.concurrent.ExecutorService;
 
 import static com.google.common.base.Verify.verify;
+import static io.trino.plugin.jdbc.DynamicFilteringJdbcSessionProperties.getDynamicFilteringWaitTimeout;
+import static io.trino.plugin.jdbc.DynamicFilteringJdbcSessionProperties.isEnableDynamicFiltering;
 import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 public class JdbcRecordSetProvider
         implements ConnectorRecordSetProvider
@@ -44,6 +48,31 @@ public JdbcRecordSetProvider(JdbcClient jdbcClient, @ForRecordCursor ExecutorSer
         this.executor = requireNonNull(executor, "executor is null");
     }
 
+    @Override
+    public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, ConnectorTableHandle table, List<? extends ColumnHandle> columns, DynamicFilter dynamicFilter)
+    {
+        if (isEnableDynamicFiltering(session)) {
+            long timeoutMillis = getDynamicFilteringWaitTimeout(session).toMillis();
+
+            // no time for further narrow down or done already
+            if (timeoutMillis == 0 || !dynamicFilter.isAwaitable()) {
+                JdbcTableHandle handle = ((JdbcTableHandle) table).withConstraint(dynamicFilter.getCurrentPredicate());
+                return getRecordSet(transaction, session, split, handle, columns);
+            }
+
+            try {
+                dynamicFilter.isBlocked().get(timeoutMillis, MILLISECONDS);
+                JdbcTableHandle handle = ((JdbcTableHandle) table).withConstraint(dynamicFilter.getCurrentPredicate());
+                return getRecordSet(transaction, session, split, handle, columns);
+            }
+            catch (Exception e) {
+                return getRecordSet(transaction, session, split, table, columns);
+            }
+        }
+
+        return getRecordSet(transaction, session, split, table, columns);
+    }
+
     @Override
     public RecordSet getRecordSet(ConnectorTransactionHandle transaction, ConnectorSession session, ConnectorSplit split, ConnectorTableHandle table, List<? extends ColumnHandle> columns)
     {

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcTableHandle.java

@@ -103,6 +103,16 @@ public JdbcTableHandle(
         this.nextSyntheticColumnId = nextSyntheticColumnId;
     }
 
+    public JdbcTableHandle withConstraint(TupleDomain<ColumnHandle> newConstraint)
+    {
+        TupleDomain<ColumnHandle> newDomain = constraint.intersect(newConstraint);
+        if (newDomain == constraint) {
+            return this;
+        }
+
+        return new JdbcTableHandle(relationHandle, newDomain, constraintExpressions, sortOrder, limit, columns, otherReferencedTables, nextSyntheticColumnId);
+    }
+
     /**
      * @deprecated Use {@code asPlainTable().getSchemaTableName()} instead, but see those methods for more information, as this is not a drop-in replacement.
      */