Skip to content

Commit d97dbbc

Browse files
laughingman7743laughingman7743
committed
[FLINK-37288] Fix SpannerCatalog compatibility with FLINK-38851 changes
1 parent 5f65c13 commit d97dbbc

4 files changed

Lines changed: 150 additions & 46 deletions

File tree

‎flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/database/catalog/AbstractJdbcCatalog.java‎

Lines changed: 29 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,30 @@ public AbstractJdbcCatalog(
130130
String defaultDatabase,
131131
String baseUrl,
132132
Properties connectionProperties) {
133-
super(catalogName, validateJdbcUrl(baseUrl, defaultDatabase));
133+
this(
134+
userClassLoader,
135+
catalogName,
136+
validateJdbcUrl(baseUrl, defaultDatabase),
137+
baseUrl,
138+
connectionProperties,
139+
false);
140+
}
141+
142+
/**
143+
* Protected constructor that allows subclasses to skip URL validation. Subclasses with
144+
* non-standard JDBC URL formats (e.g., Spanner's deep path structure) can use this constructor
145+
* to provide the default database directly without going through {@link #validateJdbcUrl}.
146+
*
147+
* @param skipUrlValidation unused, exists only to differentiate the constructor signature
148+
*/
149+
protected AbstractJdbcCatalog(
150+
ClassLoader userClassLoader,
151+
String catalogName,
152+
String defaultDatabase,
153+
String baseUrl,
154+
Properties connectionProperties,
155+
boolean skipUrlValidation) {
156+
super(catalogName, defaultDatabase);
134157

135158
checkNotNull(userClassLoader);
136159
checkArgument(!StringUtils.isNullOrWhitespaceOnly(baseUrl));
@@ -141,6 +164,10 @@ public AbstractJdbcCatalog(
141164
this.connectionProperties = Preconditions.checkNotNull(connectionProperties);
142165
this.defaultUrl = this.urlFunction.apply(defaultDatabase);
143166

167+
validateConnectionProperties(connectionProperties);
168+
}
169+
170+
protected void validateConnectionProperties(Properties connectionProperties) {
144171
checkArgument(
145172
!StringUtils.isNullOrWhitespaceOnly(connectionProperties.getProperty(USER_KEY)));
146173
checkArgument(
@@ -586,7 +613,7 @@ protected String getSchemaTableName(ObjectPath tablePath) {
586613
throw new UnsupportedOperationException();
587614
}
588615

589-
private Function<String, String> calculateUrlFunction(String url) {
616+
protected Function<String, String> calculateUrlFunction(String url) {
590617
final String[] parts;
591618
final int questionMarkIndex = url.indexOf('?');
592619
if (questionMarkIndex == -1) {

‎flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/database/catalog/AbstractJdbcCatalogTest.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.flink.connector.jdbc.core.database.catalog;
20+
21+
import org.junit.jupiter.api.Test;
22+
23+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
24+
25+
/** Test for {@link AbstractJdbcCatalog}. */
26+
class AbstractJdbcCatalogTest {
27+
28+
@Test
29+
void testValidJdbcUrl() {
30+
AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234/db", "db");
31+
AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234/db", null);
32+
AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234/", "db");
33+
AbstractJdbcCatalog.validateJdbcUrl("jdbc:dialect://localhost:1234", "db");
34+
}
35+
36+
@Test
37+
void testInvalidJdbcUrl() {
38+
assertThatThrownBy(
39+
() ->
40+
AbstractJdbcCatalog.validateJdbcUrl(
41+
"jdbc:dialect://localhost:1234", null))
42+
.isInstanceOf(IllegalArgumentException.class)
43+
.hasMessage(AbstractJdbcCatalog.NO_DATABASES_HINT);
44+
assertThatThrownBy(
45+
() ->
46+
AbstractJdbcCatalog.validateJdbcUrl(
47+
"jdbc:dialect://localhost:1234/", null))
48+
.isInstanceOf(IllegalArgumentException.class)
49+
.hasMessage(AbstractJdbcCatalog.NO_DATABASES_HINT);
50+
assertThatThrownBy(
51+
() ->
52+
AbstractJdbcCatalog.validateJdbcUrl(
53+
"jdbc:dialect://localhost:1234/db", ""))
54+
.isInstanceOf(IllegalArgumentException.class)
55+
.hasMessage(AbstractJdbcCatalog.DATABASE_NOT_UNIQUE_HINT);
56+
assertThatThrownBy(
57+
() ->
58+
AbstractJdbcCatalog.validateJdbcUrl(
59+
"jdbc:dialect://localhost:1234/db", "not_db"))
60+
.isInstanceOf(IllegalArgumentException.class)
61+
.hasMessage(AbstractJdbcCatalog.DATABASE_NOT_UNIQUE_HINT);
62+
}
63+
}

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

Lines changed: 52 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
import java.util.Optional;
6161
import java.util.Properties;
6262
import java.util.Set;
63+
import java.util.function.Function;
6364

6465
import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
6566

@@ -101,33 +102,35 @@ public SpannerCatalog(
101102
String defaultDatabase,
102103
String baseUrl,
103104
Properties connectProperties) {
104-
super(userClassLoader, catalogName, defaultDatabase, baseUrl, connectProperties);
105+
// Use the protected constructor to skip validateJdbcUrl, which does not support
106+
// Spanner's deep path URL structure (e.g., /projects/.../databases/).
107+
super(userClassLoader, catalogName, defaultDatabase, baseUrl, connectProperties, true);
105108
this.dialectTypeMapper = new SpannerTypeMapper();
106109
}
107110

108111
@Override
109-
protected String getDatabaseUrl(String databaseName) {
110-
// Spanner JDBC URL uses semicolon for parameters instead of question mark
111-
// Handle semicolon params the same way AbstractJdbcCatalog handles '?' params
112-
// Example: "jdbc:cloudspanner://host/projects/.../databases/;autoConfigEmulator=true"
113-
// -> "jdbc:cloudspanner://host/projects/.../databases/mydb;autoConfigEmulator=true"
114-
int semiColonIndex = baseUrl.indexOf(';');
115-
112+
protected Function<String, String> calculateUrlFunction(String url) {
113+
// Spanner JDBC URLs use semicolons for parameters instead of question marks.
114+
// Example: "jdbc:cloudspanner://host/.../databases/;autoConfigEmulator=true"
115+
// -> urlFunction("mydb") -> "jdbc:cloudspanner://host/.../databases/mydb;autoConfigEmulator=true"
116+
int semiColonIndex = url.indexOf(';');
116117
if (semiColonIndex == -1) {
117-
// No parameters: traditional baseUrl + databaseName
118-
return baseUrl + databaseName;
119-
}
120-
121-
// Parameters present: insert database name before ';'
122-
String urlWithoutParams = baseUrl.substring(0, semiColonIndex);
123-
String params = baseUrl.substring(semiColonIndex);
124-
125-
// Remove trailing '/' from params if AbstractJdbcCatalog added it
126-
if (params.endsWith("/")) {
127-
params = params.substring(0, params.length() - 1);
118+
// No semicolon params
119+
String trimmed = url.trim();
120+
String prefix = trimmed.endsWith("/") ? trimmed : trimmed + "/";
121+
return dbName -> prefix + dbName;
128122
}
123+
// Has semicolon params: split into prefix and params
124+
String urlWithoutParams = url.substring(0, semiColonIndex);
125+
String params = url.substring(semiColonIndex);
126+
String prefix = urlWithoutParams.endsWith("/") ? urlWithoutParams : urlWithoutParams + "/";
127+
return dbName -> prefix + dbName + params;
128+
}
129129

130-
return urlWithoutParams + databaseName + params;
130+
@Override
131+
protected void validateConnectionProperties(Properties connectionProperties) {
132+
// Spanner uses Google Cloud credentials, not traditional username/password.
133+
// Skip the parent's USER_KEY/PASSWORD_KEY validation.
131134
}
132135

133136
private ConnectionOptions getConnectionOptions() {
@@ -137,7 +140,9 @@ private ConnectionOptions getConnectionOptions() {
137140
private SpannerOptions getSpannerOptions(ConnectionOptions options) {
138141
SpannerOptions.Builder builder =
139142
SpannerOptions.newBuilder().setProjectId(options.getProjectId());
140-
if (options.getHost().contains("localhost")) {
143+
// Detect the emulator via the driver's own flag (set by ";autoConfigEmulator=true") rather
144+
// than string-matching the host, so emulator hosts other than "localhost" also work.
145+
if (options.isAutoConfigEmulator()) {
141146
builder.setEmulatorHost(options.getHost());
142147
} else {
143148
builder.setHost(options.getHost());
@@ -206,7 +211,8 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
206211
String databaseName = tablePath.getDatabaseName();
207212

208213
try (Connection conn =
209-
DriverManager.getConnection(getDatabaseUrl(databaseName), connectionProperties)) {
214+
DriverManager.getConnection(
215+
this.urlFunction.apply(databaseName), connectionProperties)) {
210216
DatabaseMetaData metaData = conn.getMetaData();
211217
Optional<UniqueConstraint> primaryKey =
212218
getPrimaryKey(
@@ -220,30 +226,33 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
220226
// The INFORMATION_SCHEMA.COLUMNS table is used to retrieve nullability.
221227
Map<String, Boolean> nullables = getColumnNullables(conn, tablePath);
222228

223-
PreparedStatement ps =
229+
try (PreparedStatement ps =
224230
conn.prepareStatement(
225-
String.format("SELECT * FROM %s;", getSchemaTableName(tablePath)));
226-
227-
ResultSetMetaData resultSetMetaData = ps.getMetaData();
228-
229-
String[] columnNames = new String[resultSetMetaData.getColumnCount()];
230-
DataType[] types = new DataType[resultSetMetaData.getColumnCount()];
231-
232-
for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
233-
final String columnName = resultSetMetaData.getColumnName(i);
234-
columnNames[i - 1] = columnName;
235-
types[i - 1] = fromJDBCType(tablePath, resultSetMetaData, i);
236-
if (!nullables.getOrDefault(columnName, true)) {
237-
types[i - 1] = types[i - 1].notNull();
231+
String.format("SELECT * FROM %s;", getSchemaTableName(tablePath)))) {
232+
ResultSetMetaData resultSetMetaData = ps.getMetaData();
233+
234+
String[] columnNames = new String[resultSetMetaData.getColumnCount()];
235+
DataType[] types = new DataType[resultSetMetaData.getColumnCount()];
236+
237+
for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
238+
final String columnName = resultSetMetaData.getColumnName(i);
239+
columnNames[i - 1] = columnName;
240+
types[i - 1] = fromJDBCType(tablePath, resultSetMetaData, i);
241+
if (!nullables.getOrDefault(columnName, true)) {
242+
types[i - 1] = types[i - 1].notNull();
243+
}
238244
}
239-
}
240245

241-
Schema.Builder schemaBuilder = Schema.newBuilder().fromFields(columnNames, types);
242-
primaryKey.ifPresent(
243-
pk -> schemaBuilder.primaryKeyNamed(pk.getName(), pk.getColumns()));
244-
Schema tableSchema = schemaBuilder.build();
246+
Schema.Builder schemaBuilder = Schema.newBuilder().fromFields(columnNames, types);
247+
primaryKey.ifPresent(
248+
pk -> schemaBuilder.primaryKeyNamed(pk.getName(), pk.getColumns()));
249+
Schema tableSchema = schemaBuilder.build();
245250

246-
return CatalogTable.of(tableSchema, null, Lists.newArrayList(), getOptions(tablePath));
251+
return CatalogTable.newBuilder()
252+
.schema(tableSchema)
253+
.options(getOptions(tablePath))
254+
.build();
255+
}
247256
} catch (Exception e) {
248257
throw new CatalogException(
249258
String.format("Failed getting table %s", tablePath.getFullName()), e);
@@ -281,7 +290,7 @@ public List<String> listTables(String databaseName)
281290
throw new DatabaseNotExistException(getName(), databaseName);
282291
}
283292

284-
final String url = getDatabaseUrl(databaseName);
293+
final String url = this.urlFunction.apply(databaseName);
285294
try (Connection conn = DriverManager.getConnection(url, connectionProperties)) {
286295
// get all schemas
287296
List<String> schemas;

‎flink-connector-jdbc-spanner/src/test/java/org/apache/flink/connector/jdbc/spanner/database/SpannerFactoryTest.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,12 @@ class SpannerFactoryTest implements SpannerTestBase {
4848
@BeforeEach
4949
void setup() {
5050
String jdbcUrl = getMetadata().getJdbcUrl();
51-
baseUrl = jdbcUrl.substring(0, jdbcUrl.lastIndexOf("/"));
51+
// Preserve semicolon parameters in baseUrl (e.g., ;autoConfigEmulator=true)
52+
int semiColonIndex = jdbcUrl.indexOf(';');
53+
String params = semiColonIndex != -1 ? jdbcUrl.substring(semiColonIndex) : "";
54+
String urlWithoutParams =
55+
semiColonIndex != -1 ? jdbcUrl.substring(0, semiColonIndex) : jdbcUrl;
56+
baseUrl = urlWithoutParams.substring(0, urlWithoutParams.lastIndexOf("/") + 1) + params;
5257
catalog =
5358
JdbcFactoryLoader.loadCatalog(
5459
Thread.currentThread().getContextClassLoader(),

0 commit comments

Comments
 (0)