Migrate JDBC plugins to page source API

This enables dynamic row filtering and columnar filter evaluation optimizations
on JDBC based sources.
We're also able to use io.trino.plugin.jdbc.JdbcPageSource#isBlocked to yield
when waiting for results from source DB.
RemoteQueryCancellation is removed because all jdbc queries are now executed
on a separate thread pool which allows remote query cancellation.

Author: raunaqmorarka

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

@@ -16,6 +16,7 @@
 import com.google.common.collect.ImmutableSet;
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigDescription;
+import io.airlift.configuration.DefunctConfig;
 import io.airlift.units.Duration;
 import io.airlift.units.MinDuration;
 import jakarta.validation.constraints.AssertTrue;
@@ -29,6 +30,7 @@
 import static jakarta.validation.constraints.Pattern.Flag.CASE_INSENSITIVE;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
+@DefunctConfig("remote-query-async-cancellation.enabled")
 public class BaseJdbcConfig
 {
     private static final String METADATA_CACHE_TTL = "metadata.cache-ttl";

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

@@ -27,10 +27,10 @@
  * Attaches dynamic filter to {@link JdbcSplit} after {@link JdbcDynamicFilteringSplitManager}
  * has waited for the collection of dynamic filters.
  * This allows JDBC based connectors to avoid waiting for dynamic filters again on the worker node
- * in {@link JdbcRecordSetProvider}. The number of splits generated by JDBC based connectors are
+ * in {@link JdbcPageSourceProvider}. The number of splits generated by JDBC based connectors are
  * typically small, therefore attaching dynamic filter here does not add significant overhead.
  * Waiting for dynamic filters in {@link JdbcDynamicFilteringSplitManager} is preferred over waiting
- * for them on the worker node in {@link JdbcRecordSetProvider} to allow connectors to take advantage of
+ * for them on the worker node in {@link JdbcPageSourceProvider} to allow connectors to take advantage of
  * dynamic filters during the splits generation phase.
  */
 public class DynamicFilteringJdbcSplitSource

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForRecordCursor.java renamed to plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForJdbcClient.java

@@ -26,6 +26,6 @@
 @Retention(RUNTIME)
 @Target({FIELD, PARAMETER, METHOD})
 @BindingAnnotation
-public @interface ForRecordCursor
+public @interface ForJdbcClient
 {
 }

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

@@ -13,16 +13,13 @@
  */
 package io.trino.plugin.jdbc;
 
-import com.google.common.util.concurrent.MoreExecutors;
 import com.google.inject.Binder;
-import com.google.inject.Inject;
 import com.google.inject.Key;
 import com.google.inject.Provider;
+import com.google.inject.Provides;
 import com.google.inject.Scopes;
 import com.google.inject.Singleton;
 import com.google.inject.multibindings.Multibinder;
-import com.google.inject.multibindings.ProvidesIntoOptional;
-import dev.failsafe.RetryPolicy;
 import io.airlift.configuration.AbstractConfigurationAwareModule;
 import io.trino.plugin.base.mapping.IdentifierMappingModule;
 import io.trino.plugin.base.session.SessionPropertiesProvider;
@@ -33,7 +30,6 @@
 import io.trino.spi.connector.ConnectorAccessControl;
 import io.trino.spi.connector.ConnectorPageSinkProvider;
 import io.trino.spi.connector.ConnectorPageSourceProvider;
-import io.trino.spi.connector.ConnectorRecordSetProvider;
 import io.trino.spi.connector.ConnectorSplitManager;
 import io.trino.spi.function.table.ConnectorTableFunction;
 import io.trino.spi.procedure.Procedure;
@@ -42,10 +38,12 @@
 
 import static com.google.inject.multibindings.Multibinder.newSetBinder;
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
-import static com.google.inject.multibindings.ProvidesIntoOptional.Type.DEFAULT;
+import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.airlift.configuration.ConditionalModule.conditionalModule;
 import static io.airlift.configuration.ConfigBinder.configBinder;
 import static io.trino.plugin.base.ClosingBinder.closingBinder;
+import static java.lang.String.format;
+import static java.util.concurrent.Executors.newCachedThreadPool;
 import static org.weakref.jmx.guice.ExportBinder.newExporter;
 
 public class JdbcModule
@@ -109,15 +107,10 @@ public void setup(Binder binder)
 
         newOptionalBinder(binder, Key.get(int.class, MaxDomainCompactionThreshold.class));
 
-        newOptionalBinder(binder, Key.get(ExecutorService.class, ForRecordCursor.class))
-                .setDefault()
-                .toProvider(MoreExecutors::newDirectExecutorService)
-                .in(Scopes.SINGLETON);
-
         newSetBinder(binder, JdbcQueryEventListener.class);
 
         closingBinder(binder)
-                .registerExecutor(Key.get(ExecutorService.class, ForRecordCursor.class));
+                .registerExecutor(Key.get(ExecutorService.class, ForJdbcClient.class));
     }
 
     public static Multibinder<SessionPropertiesProvider> sessionPropertiesProviderBinder(Binder binder)
@@ -150,11 +143,11 @@ public static void bindTablePropertiesProvider(Binder binder, Class<? extends Ta
         tablePropertiesProviderBinder(binder).addBinding().to(type).in(Scopes.SINGLETON);
     }
 
-    @ProvidesIntoOptional(DEFAULT)
-    @Inject
+    @Provides
     @Singleton
-    ConnectorRecordSetProvider recordSetProvider(JdbcClient jdbcClient, @ForRecordCursor ExecutorService executor, RetryPolicy<Object> policy)
+    @ForJdbcClient
+    public ExecutorService provideJdbcClientExecutor(CatalogName catalogName)
     {
-        return new JdbcRecordSetProvider(jdbcClient, executor, policy);
+        return newCachedThreadPool(daemonThreadsNamed(format("%s-jdbc-client-%%d", catalogName)));
     }
 }