laughingman7743 commented on code in PR #156:
URL:
https://github.com/apache/flink-connector-jdbc/pull/156#discussion_r4087099737
##########
docs/content/docs/connectors/table/jdbc.md:
##########
@@ -45,17 +45,18 @@ See how to link with it for cluster execution [here]({{<
ref "docs/dev/configura
A driver dependency is also required to connect to a specified database. Here
are drivers currently supported:
-| Driver | Group Id | Artifact Id | JAR
|
-|:-----------|:---------------------------|:-----------------------|:----------------------------------------------------------------------------------------------------------------------------------|
-| MySQL | `mysql` | `mysql-connector-java` |
[Download](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/)
|
-| Oracle | `com.oracle.database.jdbc` | `ojdbc8` |
[Download](https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8)
|
-| PostgreSQL | `org.postgresql` | `postgresql` |
[Download](https://jdbc.postgresql.org/download/)
|
-| Derby | `org.apache.derby` | `derby` |
[Download](http://db.apache.org/derby/derby_downloads.html)
|
-| SQL Server | `com.microsoft.sqlserver` | `mssql-jdbc` |
[Download](https://docs.microsoft.com/en-us/sql/connect/jdbc/download-microsoft-jdbc-driver-for-sql-server?view=sql-server-ver16)
|
-| CrateDB | `io.crate` | `crate-jdbc` |
[Download](https://repo1.maven.org/maven2/io/crate/crate-jdbc/)
|
-| Db2 | `com.ibm.db2.jcc` | `db2jcc` |
[Download](https://www.ibm.com/support/pages/download-db2-fix-packs-version-db2-linux-unix-and-windows)
|
-| Trino | `io.trino` | `trino-jdbc` |
[Download](https://repo1.maven.org/maven2/io/trino/trino-jdbc/)
|
-| OceanBase | `com.oceanbase` | `oceanbase-client` |
[Download](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/)
|
+| Driver | Group Id | Artifact Id | JAR
|
+|:-----------|:---------------------------|:----------------------------|:----------------------------------------------------------------------------------------------------------------------------------------|
+| MySQL | `mysql` | `mysql-connector-java` |
[Download](https://repo.maven.apache.org/maven2/mysql/mysql-connector-java/)
|
+| Oracle | `com.oracle.database.jdbc` | `ojdbc8` |
[Download](https://mvnrepository.com/artifact/com.oracle.database.jdbc/ojdbc8)
|
+| PostgreSQL | `org.postgresql` | `postgresql` |
[Download](https://jdbc.postgresql.org/download/)
|
+| Derby | `org.apache.derby` | `derby` |
[Download](http://db.apache.org/derby/derby_downloads.html)
|
+| SQL Server | `com.microsoft.sqlserver` | `mssql-jdbc` |
[Download](https://docs.microsoft.com/en-us/sql/connect/jdbc/download-microsoft-jdbc-driver-for-sql-server?view=sql-server-ver16)
|
+| CrateDB | `io.crate` | `crate-jdbc` |
[Download](https://repo1.maven.org/maven2/io/crate/crate-jdbc/)
|
+| Db2 | `com.ibm.db2.jcc` | `db2jcc` |
[Download](https://www.ibm.com/support/pages/download-db2-fix-packs-version-db2-linux-unix-and-windows)
|
+| Trino | `io.trino` | `trino-jdbc` |
[Download](https://repo1.maven.org/maven2/io/trino/trino-jdbc/)
|
+| OceanBase | `com.oceanbase` | `oceanbase-client` |
[Download](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/)
|
+| Spanner | `com.google.cloud` | `google-cloud-spanner-jdbc` |
[Download](https://central.sonatype.com/artifact/com.google.cloud/google-cloud-spanner-jdbc)
|
Review Comment:
Good point, the catalog needs `SpannerOptions` and `ConnectionOptions`,
which are only bundled in the shaded jar. The driver table now names
`google-cloud-spanner-jdbc` with the `single-jar-with-dependencies` classifier
and links to the Maven Central directory, in both en and zh.
##########
flink-connector-jdbc-spanner/src/main/java/org/apache/flink/connector/jdbc/spanner/database/catalog/SpannerTypeMapper.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.connector.jdbc.core.database.catalog.JdbcCatalogTypeMapper;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.types.DataType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+
+/** SpannerTypeMapper util class. */
+@Internal
+public class SpannerTypeMapper implements JdbcCatalogTypeMapper {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(SpannerTypeMapper.class);
+
+ private static final String SPANNER_BOOL = "BOOL";
+ private static final String SPANNER_INT64 = "INT64";
+ private static final String SPANNER_NUMERIC = "NUMERIC";
+ private static final String SPANNER_FLOAT64 = "FLOAT64";
+ private static final String SPANNER_FLOAT32 = "FLOAT32";
+ private static final String SPANNER_STRING = "STRING";
+ private static final String SPANNER_JSON = "JSON";
+ private static final String SPANNER_PROTO = "PROTO";
+ private static final String SPANNER_ENUM = "ENUM";
+ private static final String SPANNER_BYTES = "BYTES";
+ private static final String SPANNER_TIMESTAMP = "TIMESTAMP";
+ private static final String SPANNER_DATE = "DATE";
+ private static final String SPANNER_ARRAY = "ARRAY";
+ private static final String SPANNER_STRUCT = "STRUCT";
+
+ private static final String SPANNER_ARRAY_BOOL =
+ "[Ljava.lang.Boolean;"; // Boolean[].class.getName()
+ private static final String SPANNER_ARRAY_BYTES = "[[B"; //
byte[][].class.getName()
+ // private static final String SPANNER_ARRAY_PROTO = "[[B";
+ private static final String SPANNER_ARRAY_DATE = "[Ljava.sql.Date;"; //
Date[].class.getName()
+ private static final String SPANNER_ARRAY_FLOAT32 =
+ "[Ljava.lang.Float;"; // Float[].class.getName()
+ private static final String SPANNER_ARRAY_FLOAT64 =
+ "[Ljava.lang.Double;"; // Double[].class.getName()
+ private static final String SPANNER_ARRAY_INT64 = "[Ljava.lang.Long;"; //
Long[].class.getName()
+ // private static final String SPANNER_ARRAY_ENUM = "[Ljava.lang.Long;";
+ private static final String SPANNER_ARRAY_NUMERIC =
+ "[Ljava.math.BigDecimal;"; // BigDecimal[].class.getName()
+ private static final String SPANNER_ARRAY_STRING =
+ "[Ljava.lang.String;"; // String[].class.getName()
+ // private static final String SPANNER_ARRAY_JSON = "[Ljava.lang.String;";
+ private static final String SPANNER_ARRAY_TIMESTAMP =
+ "[Ljava.sql.Timestamp;"; // Timestamp[].class.getName()
+
+ @Override
+ public DataType mapping(ObjectPath tablePath, ResultSetMetaData metadata,
int colIndex)
+ throws SQLException {
+ final String spannerType =
normalizeTypeName(metadata.getColumnTypeName(colIndex));
+ final String spannerClassName = metadata.getColumnClassName(colIndex);
+ final int precision = metadata.getPrecision(colIndex);
+ final int scale = metadata.getScale(colIndex);
+ return getMapping(spannerType, spannerClassName, precision, scale);
+ }
+
+ /**
+ * Newer versions of the Spanner JDBC driver return the full type name
(e.g. {@code ARRAY<BOOL>}
+ * or {@code STRUCT<...>}) instead of only the type code. Strip the type
parameters so that the
+ * mapping only depends on the type code.
+ */
+ private static String normalizeTypeName(String spannerType) {
+ final int index = spannerType.indexOf('<');
+ return index == -1 ? spannerType : spannerType.substring(0, index);
+ }
+
+ private DataType getMapping(
+ String spannerType, String spannerClassName, int precision, int
scale) {
+ switch (spannerType) {
+ case SPANNER_BOOL:
+ return DataTypes.BOOLEAN();
+ case SPANNER_BYTES:
+ case SPANNER_PROTO:
+ // The default column display size is returned for precision,
+ // so it is handled as a BYTES type, not a VARBINARY type.
+ //
https://github.com/googleapis/java-spanner-jdbc/blob/v2.26.1/src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSetMetaData.java#L157
+ return DataTypes.BYTES();
+ case SPANNER_DATE:
+ return DataTypes.DATE();
+ case SPANNER_FLOAT32:
+ return DataTypes.FLOAT();
+ case SPANNER_FLOAT64:
+ return DataTypes.DOUBLE();
+ case SPANNER_INT64:
+ case SPANNER_ENUM:
+ return DataTypes.BIGINT();
+ case SPANNER_NUMERIC:
+ // The precision for the numeric type is 14 and the scale is
15, both of which are
+ // fixed values.
+ //
https://github.com/googleapis/java-spanner-jdbc/blob/v2.26.1/src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSetMetaData.java#L149
+ //
https://github.com/googleapis/java-spanner-jdbc/blob/v2.26.1/src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSetMetaData.java#L168
+ // But the document describes it as follows.
+ // The GoogleSQL NUMERIC is an exact numeric data type capable
of representing an
+ // exact numeric value
+ // with a precision of 38 and scale of 9.
+ // https://cloud.google.com/spanner/docs/working-with-numerics
+ return DataTypes.DECIMAL(38, 9);
+ case SPANNER_STRING:
+ // The default column display size is returned for precision,
+ // so it is handled as a STRING type, not a VARCHAR type.
+ //
https://github.com/googleapis/java-spanner-jdbc/blob/v2.26.1/src/main/java/com/google/cloud/spanner/jdbc/JdbcResultSetMetaData.java#L157
+ case SPANNER_JSON:
+ case SPANNER_STRUCT:
+ return DataTypes.STRING();
+ case SPANNER_TIMESTAMP:
+ return DataTypes.TIMESTAMP();
Review Comment:
Agreed. The catalog now maps Spanner `TIMESTAMP` to `TIMESTAMP_LTZ(9)` and
`ARRAY<TIMESTAMP>` to `ARRAY<TIMESTAMP_LTZ(9)>`, like Postgres `timestamptz`.
The read path already converts through `Instant` since the previous fix.
`SpannerCatalogITCase` now uses nanosecond test data and expects `Instant`
values. I also added `testTimestampIsIndependentOfDefaultTimeZone`, which reads
through the catalog with the JVM in `Asia/Tokyo`. With the old mapping, the
catalog tests failed with `2025-01-01T09:00:00.123456789`. The docs type table
now lists Spanner `TIMESTAMP` in the `TIMESTAMP WITH TIMEZONE` row.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/database/catalog/AbstractJdbcCatalog.java:
##########
@@ -141,6 +167,10 @@ public AbstractJdbcCatalog(
this.connectionProperties =
Preconditions.checkNotNull(connectionProperties);
this.defaultUrl = this.urlFunction.apply(defaultDatabase);
+ validateConnectionProperties(connectionProperties);
+ }
+
+ protected void validateConnectionProperties(Properties
connectionProperties) {
Review Comment:
Added Javadoc to both `validateConnectionProperties` and
`calculateUrlFunction`. It describes what each hook is for and notes that it is
called from the constructor before subclass fields are initialized, so
implementations must only depend on the given argument.
--
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]
