[pull] master from trinodb:master

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

@@ -26,6 +26,7 @@
 import io.trino.spi.type.TypeManager;
 
 import java.util.Map;
+import java.util.function.Supplier;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Strings.isNullOrEmpty;
@@ -36,13 +37,13 @@ public class JdbcConnectorFactory
         implements ConnectorFactory
 {
     private final String name;
-    private final Module module;
+    private final Supplier<Module> module;
 
-    public JdbcConnectorFactory(String name, Module module)
+    public JdbcConnectorFactory(String name, Supplier<Module> module)
     {
         checkArgument(!isNullOrEmpty(name), "name is null or empty");
         this.name = name;
-        this.module = module;
+        this.module = requireNonNull(module, "module is null");
     }
 
     @Override
@@ -55,7 +56,6 @@ public String getName()
     public Connector create(String catalogName, Map<String, String> requiredConfig, ConnectorContext context)
     {
         requireNonNull(requiredConfig, "requiredConfig is null");
-        requireNonNull(module, "module is null");
         checkStrictSpiVersionMatch(context, this);
 
         Bootstrap app = new Bootstrap(
@@ -65,7 +65,7 @@ public Connector create(String catalogName, Map<String, String> requiredConfig,
                 binder -> binder.bind(OpenTelemetry.class).toInstance(context.getOpenTelemetry()),
                 binder -> binder.bind(CatalogName.class).toInstance(new CatalogName(catalogName)),
                 new JdbcModule(),
-                module);
+                module.get());
 
         Injector injector = app
                 .doNotInitializeLogging()

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

@@ -19,6 +19,8 @@
 import io.trino.spi.Plugin;
 import io.trino.spi.connector.ConnectorFactory;
 
+import java.util.function.Supplier;
+
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Strings.isNullOrEmpty;
 import static io.airlift.configuration.ConfigurationAwareModule.combine;
@@ -28,9 +30,9 @@ public class JdbcPlugin
         implements Plugin
 {
     private final String name;
-    private final Module module;
+    private final Supplier<Module> module;
 
-    public JdbcPlugin(String name, Module module)
+    public JdbcPlugin(String name, Supplier<Module> module)
     {
         checkArgument(!isNullOrEmpty(name), "name is null or empty");
         this.name = name;
@@ -42,9 +44,9 @@ public Iterable<ConnectorFactory> getConnectorFactories()
     {
         return ImmutableList.of(new JdbcConnectorFactory(
                 name,
-                combine(
+                () -> combine(
                         new CredentialProviderModule(),
                         new ExtraCredentialsBasedIdentityCacheMappingModule(),
-                        module)));
+                        module.get())));
     }
 }

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/H2QueryRunner.java

@@ -81,7 +81,7 @@ public static QueryRunner createH2QueryRunner(
 
             createSchema(properties, "tpch");
 
-            queryRunner.installPlugin(new JdbcPlugin("base_jdbc", module));
+            queryRunner.installPlugin(new JdbcPlugin("base_jdbc", () -> module));
             queryRunner.createCatalog("jdbc", "base_jdbc", properties);
 
             copyTpchTables(queryRunner, "tpch", TINY_SCHEMA_NAME, tables);

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcPlugin.java

@@ -13,15 +13,32 @@
  */
 package io.trino.plugin.jdbc;
 
+import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableMap;
+import com.google.inject.Binder;
+import com.google.inject.Inject;
+import com.google.inject.Module;
+import com.google.inject.Scopes;
+import io.airlift.configuration.AbstractConfigurationAwareModule;
 import io.trino.spi.Plugin;
+import io.trino.spi.catalog.CatalogName;
+import io.trino.spi.connector.Connector;
 import io.trino.spi.connector.ConnectorFactory;
 import io.trino.testing.TestingConnectorContext;
+import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
+
 import static com.google.common.collect.Iterables.getOnlyElement;
+import static io.airlift.configuration.ConditionalModule.conditionalModule;
 import static io.trino.plugin.base.mapping.MappingConfig.CASE_INSENSITIVE_NAME_MATCHING;
 import static io.trino.plugin.base.mapping.RuleBasedIdentifierMappingUtils.createRuleBasedIdentifierMappingFile;
+import static io.trino.plugin.jdbc.TestingH2JdbcModule.createH2ConnectionUrl;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 public class TestJdbcPlugin
 {
@@ -47,9 +64,84 @@ public void testRuleBasedIdentifierCanBeUsedTogetherWithCacheBased()
                 .shutdown();
     }
 
+    @RepeatedTest(100)
+    void testConfigurationDoesNotLeakBetweenCatalogs()
+    {
+        TestingJdbcPlugin plugin = new TestingJdbcPlugin("test_jdbc", TestingJdbcModule::new);
+        ConnectorFactory connectorFactory = getOnlyElement(plugin.getConnectorFactories());
+
+        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
+            Future<Connector> pushDownEnabledFuture = executor.submit(() -> connectorFactory.create(
+                    TestingJdbcModule.CATALOG_WITH_PUSH_DOWN_ENABLED,
+                    ImmutableMap.of("connection-url", createH2ConnectionUrl(), "join-pushdown.enabled", "true"),
+                    new TestingConnectorContext()));
+            Future<Connector> pushDownDisabledFuture = executor.submit(() -> connectorFactory.create(
+                    TestingJdbcModule.CATALOG_WITH_PUSH_DOWN_DISABLED,
+                    ImmutableMap.of("connection-url", createH2ConnectionUrl(), "join-pushdown.enabled", "false"),
+                    new TestingConnectorContext()));
+
+            AtomicReference<Connector> catalogWithPushDownEnabled = new AtomicReference<>();
+            AtomicReference<Connector> catalogWithPushDownDisabled = new AtomicReference<>();
+            assertThatCode(() -> {
+                catalogWithPushDownEnabled.set(pushDownEnabledFuture.get());
+                catalogWithPushDownDisabled.set(pushDownDisabledFuture.get());
+            }).doesNotThrowAnyException();
+
+            catalogWithPushDownEnabled.get().shutdown();
+            catalogWithPushDownDisabled.get().shutdown();
+        }
+    }
+
+    private static class TestingJdbcPlugin
+            extends JdbcPlugin
+    {
+        public TestingJdbcPlugin(String name, Supplier<Module> module)
+        {
+            super(name, module);
+        }
+    }
+
+    private static class TestingJdbcModule
+            extends AbstractConfigurationAwareModule
+    {
+        public static final String CATALOG_WITH_PUSH_DOWN_ENABLED = "catalogWithPushDownEnabled";
+        public static final String CATALOG_WITH_PUSH_DOWN_DISABLED = "catalogWithPushDownDisabled";
+
+        @Override
+        protected void setup(Binder binder)
+        {
+            install(conditionalModule(
+                    JdbcMetadataConfig.class,
+                    JdbcMetadataConfig::isJoinPushdownEnabled,
+                    new ModuleCheckingThatPushDownCanBeEnabled()));
+            install(new TestingH2JdbcModule());
+        }
+    }
+
+    private static class ModuleCheckingThatPushDownCanBeEnabled
+            implements Module
+    {
+        @Override
+        public void configure(Binder binder)
+        {
+            binder.bind(PushDownCanBeEnabledChecker.class).in(Scopes.SINGLETON);
+        }
+    }
+
+    private static class PushDownCanBeEnabledChecker
+    {
+        @Inject
+        public PushDownCanBeEnabledChecker(CatalogName catalogName)
+        {
+            if (!TestingJdbcModule.CATALOG_WITH_PUSH_DOWN_ENABLED.equals(catalogName.toString())) {
+                throw new RuntimeException("Catalog '%s' should not have push-down enabled".formatted(catalogName));
+            }
+        }
+    }
+
     private static ConnectorFactory getConnectorFactory()
     {
-        Plugin plugin = new JdbcPlugin("jdbc", new TestingH2JdbcModule());
+        Plugin plugin = new JdbcPlugin("jdbc", TestingH2JdbcModule::new);
         return getOnlyElement(plugin.getConnectorFactories());
     }
 }

plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJmxStats.java

@@ -37,7 +37,7 @@ public class TestJmxStats
     public void testJmxStatsExposure()
             throws Exception
     {
-        Plugin plugin = new JdbcPlugin("base_jdbc", new TestingH2JdbcModule());
+        Plugin plugin = new JdbcPlugin("base_jdbc", TestingH2JdbcModule::new);
         ConnectorFactory factory = getOnlyElement(plugin.getConnectorFactories());
         factory.create(
                 "test",

plugin/trino-clickhouse/src/main/java/io/trino/plugin/clickhouse/ClickHousePlugin.java

@@ -20,6 +20,6 @@ public class ClickHousePlugin
 {
     public ClickHousePlugin()
     {
-        super("clickhouse", new ClickHouseClientModule());
+        super("clickhouse", ClickHouseClientModule::new);
     }
 }

plugin/trino-druid/src/main/java/io/trino/plugin/druid/DruidJdbcPlugin.java

@@ -20,6 +20,6 @@ public class DruidJdbcPlugin
 {
     public DruidJdbcPlugin()
     {
-        super("druid", new DruidJdbcClientModule());
+        super("druid", DruidJdbcClientModule::new);
     }
 }

plugin/trino-example-jdbc/src/main/java/io/trino/plugin/example/ExamplePlugin.java

@@ -20,6 +20,6 @@ public class ExamplePlugin
 {
     public ExamplePlugin()
     {
-        super("example_jdbc", new ExampleClientModule());
+        super("example_jdbc", ExampleClientModule::new);
     }
 }

plugin/trino-exasol/src/main/java/io/trino/plugin/exasol/ExasolPlugin.java

@@ -20,6 +20,6 @@ public class ExasolPlugin
 {
     public ExasolPlugin()
     {
-        super("exasol", new ExasolClientModule());
+        super("exasol", ExasolClientModule::new);
     }
 }

plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgnitePlugin.java

@@ -20,6 +20,6 @@ public class IgnitePlugin
 {
     public IgnitePlugin()
     {
-        super("ignite", new IgniteClientModule());
+        super("ignite", IgniteClientModule::new);
     }
 }

plugin/trino-mariadb/src/main/java/io/trino/plugin/mariadb/MariaDbPlugin.java

@@ -20,6 +20,6 @@ public class MariaDbPlugin
 {
     public MariaDbPlugin()
     {
-        super("mariadb", new MariaDbClientModule());
+        super("mariadb", MariaDbClientModule::new);
     }
 }

plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlPlugin.java

@@ -20,6 +20,6 @@ public class MySqlPlugin
 {
     public MySqlPlugin()
     {
-        super("mysql", new MySqlClientModule());
+        super("mysql", MySqlClientModule::new);
     }
 }

plugin/trino-oracle/src/main/java/io/trino/plugin/oracle/OraclePlugin.java

@@ -20,6 +20,6 @@ public class OraclePlugin
 {
     public OraclePlugin()
     {
-        super("oracle", new OracleClientModule());
+        super("oracle", OracleClientModule::new);
     }
 }

plugin/trino-postgresql/src/main/java/io/trino/plugin/postgresql/PostgreSqlPlugin.java

@@ -22,6 +22,6 @@ public class PostgreSqlPlugin
 {
     public PostgreSqlPlugin()
     {
-        super("postgresql", combine(new PostgreSqlClientModule(), new PostgreSqlConnectionFactoryModule()));
+        super("postgresql", () -> combine(new PostgreSqlClientModule(), new PostgreSqlConnectionFactoryModule()));
     }
 }

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlJdbcConnectionAccesses.java

@@ -60,7 +60,7 @@ protected QueryRunner createQueryRunner()
                 .setAdditionalSetup(runner -> {
                     runner.installPlugin(new JdbcPlugin(
                             "counting_postgresql",
-                            combine(new PostgreSqlClientModule(), new TestingPostgreSqlModule(connectionFactory))));
+                            () -> combine(new PostgreSqlClientModule(), new TestingPostgreSqlModule(connectionFactory))));
                     runner.createCatalog("counting_postgresql", "counting_postgresql", ImmutableMap.of(
                             "connection-url", postgreSqlServer.getJdbcUrl(),
                             "connection-user", postgreSqlServer.getUser(),

plugin/trino-postgresql/src/test/java/io/trino/plugin/postgresql/TestPostgreSqlJdbcConnectionCreation.java

@@ -58,7 +58,7 @@ protected QueryRunner createQueryRunner()
                 .setAdditionalSetup(runner -> {
                     runner.installPlugin(new JdbcPlugin(
                             "counting_postgresql",
-                            combine(new PostgreSqlClientModule(), new TestingPostgreSqlModule(connectionFactory))));
+                            () -> combine(new PostgreSqlClientModule(), new TestingPostgreSqlModule(connectionFactory))));
                     runner.createCatalog("counting_postgresql", "counting_postgresql", ImmutableMap.of(
                             "connection-url", postgreSqlServer.getJdbcUrl(),
                             "connection-user", postgreSqlServer.getUser(),

plugin/trino-redshift/src/main/java/io/trino/plugin/redshift/RedshiftPlugin.java

@@ -20,6 +20,6 @@ public class RedshiftPlugin
 {
     public RedshiftPlugin()
     {
-        super("redshift", new RedshiftClientModule());
+        super("redshift", RedshiftClientModule::new);
     }
 }

plugin/trino-singlestore/src/main/java/io/trino/plugin/singlestore/SingleStorePlugin.java

@@ -21,6 +21,6 @@ public class SingleStorePlugin
 {
     public SingleStorePlugin()
     {
-        super("singlestore", new SingleStoreClientModule());
+        super("singlestore", SingleStoreClientModule::new);
     }
 }

plugin/trino-snowflake/src/main/java/io/trino/plugin/snowflake/SnowflakePlugin.java

@@ -20,6 +20,6 @@ public class SnowflakePlugin
 {
     public SnowflakePlugin()
     {
-        super("snowflake", new SnowflakeClientModule());
+        super("snowflake", SnowflakeClientModule::new);
     }
 }

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerPlugin.java

@@ -22,6 +22,6 @@ public class SqlServerPlugin
 {
     public SqlServerPlugin()
     {
-        super("sqlserver", combine(new SqlServerClientModule(), new SqlServerConnectionFactoryModule()));
+        super("sqlserver", () -> combine(new SqlServerClientModule(), new SqlServerConnectionFactoryModule()));
     }
 }

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/TestSqlServerJdbcConnectionAccesses.java

@@ -61,7 +61,7 @@ protected QueryRunner createQueryRunner()
                 .setAdditionalSetup(runner -> {
                     runner.installPlugin(new JdbcPlugin(
                             "counting_sqlserver",
-                            combine(new SqlServerClientModule(), new TestingSqlServerModule(connectionFactory))));
+                            () -> combine(new SqlServerClientModule(), new TestingSqlServerModule(connectionFactory))));
                     runner.createCatalog("counting_sqlserver", "counting_sqlserver", ImmutableMap.of(
                             "connection-url", sqlServer.getJdbcUrl(),
                             "connection-user", sqlServer.getUsername(),

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/TestSqlServerJdbcConnectionCreation.java

@@ -59,7 +59,7 @@ protected QueryRunner createQueryRunner()
                 .setAdditionalSetup(runner -> {
                     runner.installPlugin(new JdbcPlugin(
                             "counting_sqlserver",
-                            combine(new SqlServerClientModule(), new TestingSqlServerModule(connectionFactory))));
+                            () -> combine(new SqlServerClientModule(), new TestingSqlServerModule(connectionFactory))));
                     runner.createCatalog("counting_sqlserver", "counting_sqlserver", ImmutableMap.of(
                             "connection-url", sqlServer.getJdbcUrl(),
                             "connection-user", sqlServer.getUsername(),

plugin/trino-vertica/src/main/java/io/trino/plugin/vertica/VerticaPlugin.java

@@ -20,6 +20,6 @@ public class VerticaPlugin
 {
     public VerticaPlugin()
     {
-        super("vertica", new VerticaClientModule());
+        super("vertica", VerticaClientModule::new);
     }
 }

testing/trino-tests/src/test/java/io/trino/security/TestAccessControl.java

@@ -272,7 +272,7 @@ public ScalarFunctionImplementation getScalarFunctionImplementation(FunctionId f
                 }))
                 .build()));
         queryRunner.createCatalog("mock", "mock");
-        queryRunner.installPlugin(new JdbcPlugin("base_jdbc", new TestingH2JdbcModule()));
+        queryRunner.installPlugin(new JdbcPlugin("base_jdbc", TestingH2JdbcModule::new));
         queryRunner.createCatalog("jdbc", "base_jdbc", TestingH2JdbcModule.createProperties());
         for (String tableName : ImmutableList.of("orders", "nation", "region", "lineitem")) {
             queryRunner.execute(format("CREATE TABLE %1$s AS SELECT * FROM tpch.tiny.%1$s WITH NO DATA", tableName));

testing/trino-tests/src/test/java/io/trino/tests/TestQueryAssertions.java

@@ -75,7 +75,7 @@ protected void configureCatalog(QueryRunner queryRunner)
         queryRunner.installPlugin(new TpchPlugin());
         queryRunner.createCatalog("tpch", "tpch", ImmutableMap.of());
 
-        queryRunner.installPlugin(new JdbcPlugin("base_jdbc", new TestingH2JdbcModule()));
+        queryRunner.installPlugin(new JdbcPlugin("base_jdbc", TestingH2JdbcModule::new));
         Map<String, String> jdbcConfigurationProperties = TestingH2JdbcModule.createProperties();
         queryRunner.createCatalog("jdbc", "base_jdbc", jdbcConfigurationProperties);