Refactor JdbcMetadata to take in TableLocationProvider (#25582)

Summary:
**Issue**: JDBC connectors were tightly coupled to `BaseJdbcConfig.connectionUrl` for table location information, making it difficult for connectors like XDB to use alternative connection management (e.g., SMC tier).

**Solution**: Introduced `TableLocationProvider` interface to decouple table location logic from JDBC configuration:
- Created `TableLocationProvider` interface for flexible table location resolution
- Added `DefaultTableLocationProvider` that uses connection-url from `BaseJdbcConfig` (maintains backward compatibility)
- Refactored `JdbcMetadata` and `JdbcMetadataFactory` to accept `TableLocationProvider` instead of `BaseJdbcConfig`

This enables connectors to provide custom table location logic while maintaining full backward compatibility for existing JDBC connectors.

Added test coverage.

Pull Request resolved: #25582

Differential Revision: D78571326

Author: pradeepvaka

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/BaseJdbcConfig.java

@@ -33,6 +33,17 @@
 import static java.util.Locale.ENGLISH;
 import static java.util.concurrent.TimeUnit.MINUTES;
 
+/**
+ * Base configuration class for JDBC connectors.
+ *
+ * This class is provided for convenience and contains common configuration properties
+ * that many JDBC connectors may need. However, core JDBC functionality should not
+ * depend on this class, as JDBC connectors may choose to use their own mechanisms
+ * for connection management, authentication, and other configuration needs.
+ *
+ * Connectors are free to implement their own configuration classes and connection
+ * strategies without extending or using this base configuration.
+ */
 public class BaseJdbcConfig
 {
     private String connectionUrl;
@@ -176,5 +187,9 @@ public void validateConfig()
             throw new ConfigurationException(ImmutableList.of(new Message("Only one of 'case-insensitive-name-matching=true' or 'case-sensitive-name-matching=true' can be set. " +
                     "These options are mutually exclusive.")));
         }
+
+        if (connectionUrl == null) {
+            throw new ConfigurationException(ImmutableList.of(new Message("connection-url is required but was not provided")));
+        }
     }
 }

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/DefaultTableLocationProvider.java

@@ -0,0 +1,41 @@
+/*
+ * 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.jdbc;
+
+import javax.inject.Inject;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Default implementation of TableLocationProvider that uses the connection URL
+ * from BaseJdbcConfig as the table location.
+ */
+public class DefaultTableLocationProvider
+        implements TableLocationProvider
+{
+    private final String connectionUrl;
+
+    @Inject
+    public DefaultTableLocationProvider(BaseJdbcConfig baseJdbcConfig)
+    {
+        requireNonNull(baseJdbcConfig, "baseJdbcConfig is null");
+        this.connectionUrl = requireNonNull(baseJdbcConfig.getConnectionUrl(), "connection-url is null");
+    }
+
+    @Override
+    public String getTableLocation()
+    {
+        return connectionUrl;
+    }
+}

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcMetadata.java

@@ -58,12 +58,12 @@ public class JdbcMetadata
     private final String url;
     private final AtomicReference<Runnable> rollbackAction = new AtomicReference<>();
 
-    public JdbcMetadata(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, boolean allowDropTable, BaseJdbcConfig baseJdbcConfig)
+    public JdbcMetadata(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, boolean allowDropTable, TableLocationProvider tableLocationProvider)
     {
         this.jdbcMetadataCache = requireNonNull(jdbcMetadataCache, "jdbcMetadataCache is null");
         this.jdbcClient = requireNonNull(jdbcClient, "client is null");
         this.allowDropTable = allowDropTable;
-        this.url = baseJdbcConfig.getConnectionUrl();
+        this.url = requireNonNull(tableLocationProvider, "tableLocationProvider is null").getTableLocation();
     }
 
     @Override

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcMetadataFactory.java

@@ -22,20 +22,20 @@ public class JdbcMetadataFactory
     private final JdbcMetadataCache jdbcMetadataCache;
     private final JdbcClient jdbcClient;
     private final boolean allowDropTable;
-    private final BaseJdbcConfig baseJdbcConfig;
+    private final TableLocationProvider tableLocationProvider;
 
     @Inject
-    public JdbcMetadataFactory(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, JdbcMetadataConfig config, BaseJdbcConfig baseJdbcConfig)
+    public JdbcMetadataFactory(JdbcMetadataCache jdbcMetadataCache, JdbcClient jdbcClient, JdbcMetadataConfig config, TableLocationProvider tableLocationProvider)
     {
         this.jdbcMetadataCache = requireNonNull(jdbcMetadataCache, "jdbcMetadataCache is null");
         this.jdbcClient = requireNonNull(jdbcClient, "jdbcClient is null");
         requireNonNull(config, "config is null");
         this.allowDropTable = config.isAllowDropTable();
-        this.baseJdbcConfig = requireNonNull(baseJdbcConfig, "baseJdbcConfig is null");
+        this.tableLocationProvider = requireNonNull(tableLocationProvider, "tableLocationProvider is null");
     }
 
     public JdbcMetadata create()
     {
-        return new JdbcMetadata(jdbcMetadataCache, jdbcClient, allowDropTable, baseJdbcConfig);
+        return new JdbcMetadata(jdbcMetadataCache, jdbcClient, allowDropTable, tableLocationProvider);
     }
 }

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcModule.java

@@ -54,5 +54,7 @@ public void configure(Binder binder)
         newOptionalBinder(binder, JdbcSessionPropertiesProvider.class);
         binder.bind(JdbcConnector.class).in(Scopes.SINGLETON);
         configBinder(binder).bindConfig(JdbcMetadataConfig.class);
+        configBinder(binder).bindConfig(BaseJdbcConfig.class);
+        binder.bind(TableLocationProvider.class).to(DefaultTableLocationProvider.class).in(Scopes.SINGLETON);
     }
 }

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/TableLocationProvider.java

@@ -0,0 +1,29 @@
+/*
+ * 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.jdbc;
+
+/**
+ * Provides table location information for JDBC connectors.
+ * Different implementations can provide location information based on
+ * different sources (e.g., connection URL, service discovery, etc.)
+ */
+public interface TableLocationProvider
+{
+    /**
+     * Returns the location/URL for the table.
+     * This could be a connection URL, service endpoint, or any other
+     * location identifier relevant to the specific connector.
+     */
+    String getTableLocation();
+}

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/SimpleTestJdbcConfig.java

@@ -0,0 +1,67 @@
+/*
+ * 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.jdbc;
+
+import com.facebook.airlift.configuration.Config;
+
+import javax.validation.constraints.NotNull;
+
+/**
+ * Simple JDBC configuration that doesn't extend BaseJdbcConfig.
+ * This demonstrates that JDBC connectors can implement their own
+ * configuration mechanisms without depending on BaseJdbcConfig.
+ */
+public class SimpleTestJdbcConfig
+{
+    private String jdbcUrl;
+    private String username;
+    private String password;
+
+    @NotNull
+    public String getJdbcUrl()
+    {
+        return jdbcUrl;
+    }
+
+    @Config("jdbc-url")
+    public SimpleTestJdbcConfig setJdbcUrl(String jdbcUrl)
+    {
+        this.jdbcUrl = jdbcUrl;
+        return this;
+    }
+
+    public String getUsername()
+    {
+        return username;
+    }
+
+    @Config("username")
+    public SimpleTestJdbcConfig setUsername(String username)
+    {
+        this.username = username;
+        return this;
+    }
+
+    public String getPassword()
+    {
+        return password;
+    }
+
+    @Config("password")
+    public SimpleTestJdbcConfig setPassword(String password)
+    {
+        this.password = password;
+        return this;
+    }
+}

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/SimpleTestTableLocationProvider.java

@@ -0,0 +1,42 @@
+/*
+ * 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.jdbc;
+
+import javax.inject.Inject;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Simple TableLocationProvider implementation that doesn't depend on BaseJdbcConfig.
+ * This demonstrates that TableLocationProvider implementations can use their own
+ * configuration mechanisms without depending on BaseJdbcConfig.
+ */
+public class SimpleTestTableLocationProvider
+        implements TableLocationProvider
+{
+    private final String jdbcUrl;
+
+    @Inject
+    public SimpleTestTableLocationProvider(SimpleTestJdbcConfig config)
+    {
+        requireNonNull(config, "config is null");
+        this.jdbcUrl = config.getJdbcUrl();
+    }
+
+    @Override
+    public String getTableLocation()
+    {
+        return jdbcUrl;
+    }
+}

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/TestBaseJdbcConfig.java

@@ -15,13 +15,16 @@
 
 import com.facebook.airlift.configuration.testing.ConfigAssertions;
 import com.google.common.collect.ImmutableMap;
+import com.google.inject.ConfigurationException;
 import io.airlift.units.Duration;
 import org.testng.annotations.Test;
 
 import java.util.Map;
 
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.expectThrows;
 
 public class TestBaseJdbcConfig
 {
@@ -68,4 +71,69 @@ public void testExplicitPropertyMappings()
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }
+
+    @Test
+    public void testValidConfigValidation()
+    {
+        BaseJdbcConfig config = new BaseJdbcConfig();
+        config.setConnectionUrl("jdbc:mysql://localhost:3306/test");
+
+        // Should not throw any exception
+        config.validateConfig();
+
+        assertEquals(config.getConnectionUrl(), "jdbc:mysql://localhost:3306/test");
+    }
+
+    @Test
+    public void testNullConnectionUrlValidation()
+    {
+        BaseJdbcConfig config = new BaseJdbcConfig();
+        // connectionUrl is null by default
+
+        ConfigurationException exception = expectThrows(
+                ConfigurationException.class,
+                config::validateConfig);
+        assertEquals(exception.getErrorMessages().iterator().next().getMessage(),
+                "connection-url is required but was not provided");
+    }
+
+    @Test
+    public void testMutuallyExclusiveNameMatchingOptions()
+    {
+        BaseJdbcConfig config = new BaseJdbcConfig();
+        config.setConnectionUrl("jdbc:mysql://localhost:3306/test");
+        config.setCaseInsensitiveNameMatching(true);
+        config.setCaseSensitiveNameMatching(true);
+
+        ConfigurationException exception = expectThrows(
+                ConfigurationException.class,
+                config::validateConfig);
+        assertEquals(exception.getErrorMessages().iterator().next().getMessage(),
+                "Only one of 'case-insensitive-name-matching=true' or 'case-sensitive-name-matching=true' can be set. " +
+                "These options are mutually exclusive.");
+    }
+
+    @Test
+    public void testCaseInsensitiveNameMatchingOnly()
+    {
+        BaseJdbcConfig config = new BaseJdbcConfig();
+        config.setConnectionUrl("jdbc:mysql://localhost:3306/test");
+        config.setCaseInsensitiveNameMatching(true);
+        config.setCaseSensitiveNameMatching(false);
+
+        // Should not throw any exception
+        config.validateConfig();
+    }
+
+    @Test
+    public void testCaseSensitiveNameMatchingOnly()
+    {
+        BaseJdbcConfig config = new BaseJdbcConfig();
+        config.setConnectionUrl("jdbc:mysql://localhost:3306/test");
+        config.setCaseInsensitiveNameMatching(false);
+        config.setCaseSensitiveNameMatching(true);
+
+        // Should not throw any exception
+        config.validateConfig();
+    }
 }