laughingman7743 commented on code in PR #156: URL: https://github.com/apache/flink-connector-jdbc/pull/156#discussion_r4083013616
########## 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: Good catch. Arrays are now handled in `createExternalConverter`, so the parent's nullable wrapper takes care of NULL arrays. Added a row with NULL arrays to `SpannerDynamicTableSinkITCase#testArray`. ########## 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: Thanks, reproduced exactly as described (stored as `1969-12-31T15:00:00Z` under `Asia/Tokyo`). TIMESTAMP_LTZ is now written with `Timestamp.from(toInstant())` and read with `TimestampData.fromInstant`, including array elements. Added `testTimestampLtzWithNonUtcDefaultTimeZone`, which fails without the fix. ########## 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: Went with narrowing in `createInternalConverter`: TINYINT/SMALLINT/INTEGER values are narrowed from the INT64 `Long`, and out-of-range values fail with an `ArithmeticException` instead of being silently truncated. The source ITCase now reads INT/SMALLINT/TINYINT columns with their min/max values. ########## 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: Both are now default methods throwing `SQLFeatureNotSupportedException`, so existing implementors are unaffected. `FieldNamedPreparedStatementImpl` implements them, and `createArrayOf` goes through `statement.getConnection()`; the extra connection field is gone. ########## 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: Replaced the flag with a protected constructor that takes a `BiFunction<String, String, String>` default database resolver. Since `AbstractCatalog` validates the default database in its constructor, the resolution has to happen before `super(...)`, so an overridable instance method was not an option. The public constructors pass `AbstractJdbcCatalog::validateJdbcUrl`, and `SpannerCatalog` passes its own static resolver. ########## 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: `getSpannerOptions` now sets `options.getCredentials()`, so the admin client uses the same credentials as the JDBC connections (the application default credentials when none are configured, `NoCredentials` for the emulator). Added unit tests for `;credentials=<key file>` and `;oauthToken=...`. ########## 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: Replaced `OverridingJdbcExtractor` with a `SpannerJdbcExtractor`. It follows the OpenLineage naming convention `spanner://{projectId}:{instanceId}` and appends the database, e.g. `spanner://my-project:my-instance/my-db`, since the connector only uses the table name as the dataset name. Added `SpannerJdbcExtractorTest` covering the default host, `spanner.googleapis.com`, the emulator, and a base-url without a database. ########## 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: Fixed in both en and zh: the base-url now ends with `/databases/`. -- 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]
