MartijnVisser commented on code in PR #156: URL: https://github.com/apache/flink-connector-jdbc/pull/156#discussion_r4081196493
########## flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/dialect/SpannerDialectConverter.java: ########## @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.connector.jdbc.spanner.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.DecimalType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; + +import java.lang.reflect.Array; +import java.sql.Timestamp; +import java.sql.Types; +import java.time.LocalDateTime; + +/** + * Runtime converter that responsible to convert between JDBC object and Flink internal object for + * Spanner. + */ +@Internal +public class SpannerDialectConverter extends AbstractDialectConverter { + + private static final long serialVersionUID = 1L; + + @Override + public String converterName() { + return "Spanner"; + } + + public SpannerDialectConverter(RowType rowType) { + super(rowType); + } + + @Override + protected JdbcDeserializationConverter createInternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + ArrayType arrayType = (ArrayType) type; + return createSpannerArrayConverter(arrayType); + } else if (root == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + // Spanner has a single TIMESTAMP type (an absolute instant); treat TIMESTAMP_LTZ as a + // plain timestamp value, mirroring how TIMESTAMP_WITHOUT_TIME_ZONE is handled. + return val -> + val instanceof LocalDateTime + ? TimestampData.fromLocalDateTime((LocalDateTime) val) + : TimestampData.fromTimestamp((Timestamp) val); + } else { + return super.createInternalConverter(type); + } + } + + @Override + protected JdbcSerializationConverter createNullableExternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + if (root == LogicalTypeRoot.ARRAY) { Review Comment: Overriding `createNullableExternalConverter` drops the null check, so a NULL array fails with an NPE on `arrayData.size()`. Overriding `createExternalConverter` keeps the parent wrapper. ########## flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/dialect/SpannerDialectConverter.java: ########## @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.connector.jdbc.spanner.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.DecimalType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; + +import java.lang.reflect.Array; +import java.sql.Timestamp; +import java.sql.Types; +import java.time.LocalDateTime; + +/** + * Runtime converter that responsible to convert between JDBC object and Flink internal object for + * Spanner. + */ +@Internal +public class SpannerDialectConverter extends AbstractDialectConverter { + + private static final long serialVersionUID = 1L; + + @Override + public String converterName() { + return "Spanner"; + } + + public SpannerDialectConverter(RowType rowType) { + super(rowType); + } + + @Override + protected JdbcDeserializationConverter createInternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + ArrayType arrayType = (ArrayType) type; + return createSpannerArrayConverter(arrayType); + } else if (root == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + // Spanner has a single TIMESTAMP type (an absolute instant); treat TIMESTAMP_LTZ as a + // plain timestamp value, mirroring how TIMESTAMP_WITHOUT_TIME_ZONE is handled. + return val -> + val instanceof LocalDateTime + ? TimestampData.fromLocalDateTime((LocalDateTime) val) + : TimestampData.fromTimestamp((Timestamp) val); + } else { + return super.createInternalConverter(type); + } + } + + @Override + protected JdbcSerializationConverter createNullableExternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + if (root == LogicalTypeRoot.ARRAY) { + ArrayType arrayType = (ArrayType) type; + LogicalType elementType = arrayType.getElementType(); + String typeName = getSpannerArrayTypeName(elementType); + + return (val, index, statement) -> { + ArrayData arrayData = val.getArray(index); + int size = arrayData.size(); + Object[] elements = new Object[size]; + + // Convert each element + for (int i = 0; i < size; i++) { + elements[i] = extractArrayElement(arrayData, i, elementType); + } + + // Create JDBC Array and set it + java.sql.Array jdbcArray = statement.createArrayOf(typeName, elements); + statement.setArray(index, jdbcArray); + }; + } else if (root == LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE) { + // Spanner stores TIMESTAMP as an absolute instant; write TIMESTAMP_LTZ like a plain + // timestamp. JdbcTypeUtil has no mapping for TIMESTAMP_LTZ, so the null branch is + // handled here with Types.TIMESTAMP instead of the parent's nullable wrapper. + final int precision = LogicalTypeChecks.getPrecision(type); + return (val, index, statement) -> { + if (val == null || val.isNullAt(index)) { + statement.setNull(index, Types.TIMESTAMP); + } else { + statement.setTimestamp(index, val.getTimestamp(index, precision).toTimestamp()); Review Comment: With `-Duser.timezone=Asia/Tokyo`, `TO_TIMESTAMP_LTZ(0, 3)` is stored as 1969-12-31T15:00:00Z; reading shifts it back. I think this wants `toInstant()` here and `fromInstant` on read. ########## flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/lineage/SpannerLocationExtractorFactory.java: ########## @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.connector.jdbc.spanner.database.lineage; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.lineage.JdbcLocationExtractorFactory; + +import io.openlineage.client.utils.jdbc.JdbcExtractor; +import io.openlineage.client.utils.jdbc.OverridingJdbcExtractor; + +/** Implementation of {@link JdbcLocationExtractorFactory} for Spanner. */ +@Internal +public class SpannerLocationExtractorFactory implements JdbcLocationExtractorFactory { + + @Override + public JdbcExtractor createExtractor() { + // TODO: Replace with SpannerJdbcExtractor when OpenLineage adds native Spanner support + return new OverridingJdbcExtractor("cloudspanner", "9010"); Review Comment: 9010 is the emulator port, and every `spanner.googleapis.com` database gets the namespace `cloudspanner://spanner.googleapis.com:9010`. Could this include project, instance and database, with a test? ########## docs/content/docs/connectors/table/jdbc.md: ########## @@ -887,6 +902,7 @@ Flink supports connect to several databases which uses dialect like MySQL, Oracl <code>TINYINT(1)</code></td> <td></td> <td><code>BOOLEAN</code></td> + <td><code>BOOLEAN</code></td> Review Comment: Spanner calls this `BOOL`, and the cell lands after the Flink SQL column. Same in the DATE row. ########## docs/content/docs/connectors/table/jdbc.md: ########## @@ -481,7 +486,8 @@ The JDBC catalog supports the following options: - `base-url`: required (should not contain the database name) - for Postgres Catalog this should be `"jdbc:postgresql://<ip>:<port>"` - for MySQL Catalog this should be `"jdbc:mysql://<ip>:<port>"` - - for OceanBase Catalog this should be `jdbc:oceanbase://<ip>:<port>` + - for OceanBase Catalog this should be `"jdbc:oceanbase://<ip>:<port>"` + - for Spanner Catalog this should be `"jdbc:cloudspanner:/projects/<project>/instances/<instance>/databases/<database>"` Review Comment: The catalog appends the database here, giving `.../databases/<database>/<db>`; the tests end base-url in `/databases/`. Same in zh. ########## flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/dialect/SpannerDialect.java: ########## @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.connector.jdbc.spanner.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialect; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; + +import java.util.Arrays; +import java.util.EnumSet; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** JDBC dialect for Spanner. */ +@Internal +public class SpannerDialect extends AbstractDialect { + + private static final long serialVersionUID = 1L; + + // Define MAX/MIN precision of TIMESTAMP type according to Spanner docs: + // https://cloud.google.com/spanner/docs/reference/standard-sql/data-types#timestamp_type + private static final int MAX_TIMESTAMP_PRECISION = 9; + private static final int MIN_TIMESTAMP_PRECISION = 0; + + // Define MAX/MIN precision of DECIMAL type according to Spanner docs: + // https://cloud.google.com/spanner/docs/reference/standard-sql/data-types#decimal_types + private static final int MAX_DECIMAL_PRECISION = 38; + private static final int MIN_DECIMAL_PRECISION = 0; + + @Override + public SpannerDialectConverter getRowConverter(RowType rowType) { + return new SpannerDialectConverter(rowType); + } + + @Override + public Optional<String> defaultDriverName() { + return Optional.of("com.google.cloud.spanner.jdbc.JdbcDriver"); + } + + @Override + public String dialectName() { + return "Spanner"; + } + + @Override + public String getLimitClause(long limit) { + return "LIMIT " + limit; + } + + /** Spanner upsert query using INSERT OR UPDATE INTO syntax. */ + @Override + public Optional<String> getUpsertStatement( + String tableName, String[] fieldNames, String[] uniqueKeyFields) { + String columns = + Arrays.stream(fieldNames) + .map(this::quoteIdentifier) + .collect(Collectors.joining(", ")); + String placeholders = + Arrays.stream(fieldNames).map(f -> ":" + f).collect(Collectors.joining(", ")); + return Optional.of( + "INSERT OR UPDATE INTO " + + quoteIdentifier(tableName) + + "(" + + columns + + ")" + + " VALUES (" + + placeholders + + ")"); + } + + @Override + public String quoteIdentifier(String identifier) { + // Spanner (GoogleSQL) quotes identifiers with backticks. Quote each dot-separated component + // so schema-qualified names (e.g. "my_schema.my_table") stay valid. + return Arrays.stream(identifier.split("\\.")) + .map(part -> "`" + part + "`") + .collect(Collectors.joining(".")); + } + + @Override + public Optional<Range> decimalPrecisionRange() { + return Optional.of(Range.of(MIN_DECIMAL_PRECISION, MAX_DECIMAL_PRECISION)); + } + + @Override + public Optional<Range> timestampPrecisionRange() { + return Optional.of(Range.of(MIN_TIMESTAMP_PRECISION, MAX_TIMESTAMP_PRECISION)); + } + + @Override + public Set<LogicalTypeRoot> supportedTypes() { + // The data types used in Spanner are list at: + // https://cloud.google.com/spanner/docs/reference/standard-sql/data-types + + // TODO: We can't convert BINARY data type to + // PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO in + // LegacyTypeInfoDataTypeConverter. + + return EnumSet.of( + LogicalTypeRoot.CHAR, + LogicalTypeRoot.VARCHAR, + LogicalTypeRoot.BOOLEAN, + LogicalTypeRoot.VARBINARY, + LogicalTypeRoot.DECIMAL, + LogicalTypeRoot.TINYINT, Review Comment: Spanner only returns INT64, so an INT column fails on read with a ClassCastException. Narrow in `createInternalConverter` or drop the smaller integer types here. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/statement/FieldNamedPreparedStatement.java: ########## @@ -254,6 +255,21 @@ static FieldNamedPreparedStatement prepareStatement( */ void setObject(int fieldIndex, Object x) throws SQLException; + /** + * Sets the designated parameter to the given <code>java.sql.Array</code> object. The driver + * converts this to an SQL <code>ARRAY</code> value when it sends it to the database. + * + * @see PreparedStatement#setArray(int, Array) + */ + void setArray(int fieldIndex, Array x) throws SQLException; Review Comment: This interface is @PublicEvolving, so new abstract methods break external implementors. Could these be default methods, with `createArrayOf` going through `statement.getConnection()`? ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/database/catalog/AbstractJdbcCatalog.java: ########## @@ -130,7 +130,30 @@ public AbstractJdbcCatalog( String defaultDatabase, String baseUrl, Properties connectionProperties) { - super(catalogName, validateJdbcUrl(baseUrl, defaultDatabase)); + this( + userClassLoader, + catalogName, + validateJdbcUrl(baseUrl, defaultDatabase), + baseUrl, + connectionProperties, + false); + } + + /** + * Protected constructor that allows subclasses to skip URL validation. Subclasses with + * non-standard JDBC URL formats (e.g., Spanner's deep path structure) can use this constructor + * to provide the default database directly without going through {@link #validateJdbcUrl}. + * + * @param skipUrlValidation unused, exists only to differentiate the constructor signature + */ + protected AbstractJdbcCatalog( + ClassLoader userClassLoader, + String catalogName, + String defaultDatabase, + String baseUrl, + Properties connectionProperties, + boolean skipUrlValidation) { Review Comment: An ignored flag that only changes the signature is hard to keep on a @PublicEvolving class. I think a protected hook resolving the default database reads better. ########## flink-connector-jdbc-spanner/pom.xml: ########## @@ -0,0 +1,98 @@ +<?xml version="1.0" encoding="UTF-8"?> Review Comment: Needs the ASF header, as does `SpannerTableRow.java` (`apache-rat:check`). On main the testcontainers artifacts are `testcontainers-gcloud` and `testcontainers-jdbc`. ########## flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/catalog/SpannerCatalog.java: ########## @@ -0,0 +1,345 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.connector.jdbc.spanner.database.catalog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog; +import org.apache.flink.connector.jdbc.core.database.catalog.JdbcCatalogTypeMapper; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.types.DataType; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.StringUtils; + +import com.google.cloud.spanner.Spanner; +import com.google.cloud.spanner.SpannerOptions; +import com.google.cloud.spanner.admin.database.v1.DatabaseAdminClient; +import com.google.cloud.spanner.admin.database.v1.DatabaseAdminClient.ListDatabasesPage; +import com.google.cloud.spanner.admin.database.v1.DatabaseAdminClient.ListDatabasesPagedResponse; +import com.google.cloud.spanner.connection.ConnectionOptions; +import com.google.spanner.admin.database.v1.Database; +import com.google.spanner.admin.database.v1.InstanceName; +import org.apache.commons.compress.utils.Lists; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Properties; +import java.util.Set; +import java.util.function.Function; + +import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties; + +/** Catalog for Spanner. */ +@Internal +public class SpannerCatalog extends AbstractJdbcCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(SpannerCatalog.class); + + private static final Set<String> builtinSchemas = + new HashSet<String>() { + { + add("INFORMATION_SCHEMA"); + add("SPANNER_SYS"); + } + }; + + private final JdbcCatalogTypeMapper dialectTypeMapper; + + @VisibleForTesting + public SpannerCatalog( + ClassLoader userClassLoader, + String catalogName, + String defaultDatabase, + String username, + String pwd, + String baseUrl) { + this( + userClassLoader, + catalogName, + defaultDatabase, + baseUrl, + getBriefAuthProperties(username, pwd)); + } + + public SpannerCatalog( + ClassLoader userClassLoader, + String catalogName, + String defaultDatabase, + String baseUrl, + Properties connectProperties) { + // Use the protected constructor to skip validateJdbcUrl, which does not support + // Spanner's deep path URL structure (e.g., /projects/.../databases/). + super(userClassLoader, catalogName, defaultDatabase, baseUrl, connectProperties, true); + this.dialectTypeMapper = new SpannerTypeMapper(); + } + + @Override + protected Function<String, String> calculateUrlFunction(String url) { + // Spanner JDBC URLs use semicolons for parameters instead of question marks. + // Example: "jdbc:cloudspanner://host/.../databases/;autoConfigEmulator=true" + // -> urlFunction("mydb") -> + // "jdbc:cloudspanner://host/.../databases/mydb;autoConfigEmulator=true" + int semiColonIndex = url.indexOf(';'); + if (semiColonIndex == -1) { + // No semicolon params + String trimmed = url.trim(); + String prefix = trimmed.endsWith("/") ? trimmed : trimmed + "/"; + return dbName -> prefix + dbName; + } + // Has semicolon params: split into prefix and params + String urlWithoutParams = url.substring(0, semiColonIndex); + String params = url.substring(semiColonIndex); + String prefix = urlWithoutParams.endsWith("/") ? urlWithoutParams : urlWithoutParams + "/"; + return dbName -> prefix + dbName + params; + } + + @Override + protected void validateConnectionProperties(Properties connectionProperties) { + // Spanner uses Google Cloud credentials, not traditional username/password. + // Skip the parent's USER_KEY/PASSWORD_KEY validation. + } + + private ConnectionOptions getConnectionOptions() { + return ConnectionOptions.newBuilder().setUri(defaultUrl.replace("jdbc:", "")).build(); + } + + private SpannerOptions getSpannerOptions(ConnectionOptions options) { + SpannerOptions.Builder builder = + SpannerOptions.newBuilder().setProjectId(options.getProjectId()); Review Comment: This drops the base-url credentials, so `listDatabases` (and thus `databaseExists`, `tableExists`) uses default credentials. Passing `options.getCredentials()` should keep them aligned. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
