[FLINK-37288] Handle Spanner semicolon params in getDatabaseUrl

Author: laughingman7743

flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/catalog/SpannerCatalog.java

@@ -105,6 +105,31 @@ public SpannerCatalog(
         this.dialectTypeMapper = new SpannerTypeMapper();
     }
 
+    @Override
+    protected String getDatabaseUrl(String databaseName) {
+        // Spanner JDBC URL uses semicolon for parameters instead of question mark
+        // Handle semicolon params the same way AbstractJdbcCatalog handles '?' params
+        // Example: "jdbc:cloudspanner://host/projects/.../databases/;autoConfigEmulator=true"
+        //       -> "jdbc:cloudspanner://host/projects/.../databases/mydb;autoConfigEmulator=true"
+        int semiColonIndex = baseUrl.indexOf(';');
+
+        if (semiColonIndex == -1) {
+            // No parameters: traditional baseUrl + databaseName
+            return baseUrl + databaseName;
+        }
+
+        // Parameters present: insert database name before ';'
+        String urlWithoutParams = baseUrl.substring(0, semiColonIndex);
+        String params = baseUrl.substring(semiColonIndex);
+
+        // Remove trailing '/' from params if AbstractJdbcCatalog added it
+        if (params.endsWith("/")) {
+            params = params.substring(0, params.length() - 1);
+        }
+
+        return urlWithoutParams + databaseName + params;
+    }
+
     private ConnectionOptions getConnectionOptions() {
         return ConnectionOptions.newBuilder().setUri(defaultUrl.replace("jdbc:", "")).build();
     }
@@ -218,10 +243,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
                     pk -> schemaBuilder.primaryKeyNamed(pk.getName(), pk.getColumns()));
             Schema tableSchema = schemaBuilder.build();
 
-            return CatalogTable.newBuilder()
-                    .schema(tableSchema)
-                    .options(getOptions(tablePath))
-                    .build();
+            return CatalogTable.of(tableSchema, null, Lists.newArrayList(), getOptions(tablePath));
         } catch (Exception e) {
             throw new CatalogException(
                     String.format("Failed getting table %s", tablePath.getFullName()), e);

flink-connector-jdbc-spanner/src/test/java/org/apache/flink/connector/jdbc/spanner/database/catalog/SpannerCatalogTestBase.java

@@ -152,13 +152,20 @@ protected static TableRow createTableGroupedBy(String name) {
     @BeforeAll
     static void beforeAll() throws SQLException {
         String jdbcUrl = SpannerDatabase.getMetadata().getJdbcUrl();
-        baseUrl = jdbcUrl.substring(0, jdbcUrl.lastIndexOf("/"));
+        // Preserve connection parameters in baseUrl with '/' before ';'
+        // Example: jdbc:cloudspanner://host/.../databases/testdb;autoConfigEmulator=true
+        //       -> jdbc:cloudspanner://host/.../databases/;autoConfigEmulator=true
+        int semiColonIndex = jdbcUrl.indexOf(';');
+        String params = semiColonIndex != -1 ? jdbcUrl.substring(semiColonIndex) : "";
+        String urlWithoutParams =
+                semiColonIndex != -1 ? jdbcUrl.substring(0, semiColonIndex) : jdbcUrl;
+        // Include trailing '/' before params so database name can be inserted
+        baseUrl = urlWithoutParams.substring(0, urlWithoutParams.lastIndexOf("/") + 1) + params;
         metadata = SpannerDatabase.getMetadata();
 
         Properties props =
                 JdbcConnectionOptions.getBriefAuthProperties(
                         metadata.getUsername(), metadata.getPassword());
-        props.put("autoConfigEmulator", "true");
         catalog =
                 new SpannerCatalog(
                         Thread.currentThread().getContextClassLoader(),