From c5ce1cb187d552cf11c015f8b8b64a5803fb466d Mon Sep 17 00:00:00 2001 From: Kent Yao Date: Fri, 15 Mar 2024 12:25:45 +0800 Subject: [PATCH 1/2] [SPARK-47406][SQL] Handle TIMESTAMP and DATETIME in MYSQLDialect --- .../sql/jdbc/MySQLIntegrationSuite.scala | 14 ++++++ .../datasources/jdbc/JdbcUtils.scala | 10 +++-- .../apache/spark/sql/jdbc/JdbcDialects.scala | 7 +++ .../apache/spark/sql/jdbc/MySQLDialect.scala | 45 +++++++++++-------- 4 files changed, 54 insertions(+), 22 deletions(-) diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala index 48b94cf28a633..b1d239337aa01 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLIntegrationSuite.scala @@ -19,6 +19,7 @@ package org.apache.spark.sql.jdbc import java.math.BigDecimal import java.sql.{Connection, Date, Timestamp} +import java.time.LocalDateTime import java.util.Properties import org.apache.spark.sql.Row @@ -134,6 +135,19 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationSuite { } } + test("SPARK-47406: MySQL datetime types with preferTimestampNTZ") { + withDefaultTimeZone(UTC) { + val df = sqlContext.read.option("preferTimestampNTZ", true) + .jdbc(jdbcUrl, "dates", new Properties) + checkAnswer(df, Row( + Date.valueOf("1991-11-09"), + LocalDateTime.of(1970, 1, 1, 13, 31, 24), + LocalDateTime.of(1996, 1, 1, 1, 23, 45), + Timestamp.valueOf("2009-02-13 23:31:30"), + Date.valueOf("2001-01-01"))) + } + } + test("String types") { val df = sqlContext.read.jdbc(jdbcUrl, "strings", new Properties) val rows = df.collect() diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala index b037d862fa1a3..393f09b6075e2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala @@ -167,6 +167,10 @@ object JdbcUtils extends Logging with SQLConfHelper { throw QueryExecutionErrors.cannotGetJdbcTypeError(dt)) } + def getTimestampType(isTimestampNTZ: Boolean): DataType = { + if (isTimestampNTZ) TimestampNTZType else TimestampType + } + /** * Maps a JDBC type to a Catalyst type. This function is called only when * the JdbcDialect class corresponding to your database driver returns null. @@ -211,10 +215,8 @@ object JdbcUtils extends Logging with SQLConfHelper { case java.sql.Types.SMALLINT => IntegerType case java.sql.Types.SQLXML => StringType case java.sql.Types.STRUCT => StringType - case java.sql.Types.TIME if isTimestampNTZ => TimestampNTZType - case java.sql.Types.TIME => TimestampType - case java.sql.Types.TIMESTAMP if isTimestampNTZ => TimestampNTZType - case java.sql.Types.TIMESTAMP => TimestampType + case java.sql.Types.TIME => getTimestampType(isTimestampNTZ) + case java.sql.Types.TIMESTAMP => getTimestampType(isTimestampNTZ) case java.sql.Types.TINYINT => IntegerType case java.sql.Types.VARBINARY => BinaryType case java.sql.Types.VARCHAR if conf.charVarcharAsString => StringType diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala index 7c5e476d97869..7d2812d48caed 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala @@ -781,6 +781,13 @@ abstract class JdbcDialect extends Serializable with Logging { def getFullyQualifiedQuotedTableName(ident: Identifier): String = { (ident.namespace() :+ ident.name()).map(quoteIdentifier).mkString(".") } + + /** + * Return TimestampType/TimestampNTZType based on the metadata. + */ + protected final def getTimestampType(md: Metadata): DataType = { + JdbcUtils.getTimestampType(md.getBoolean("isTimestampNTZ")) + } } /** diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala index 7c63477d45846..42b1a3a2854ed 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala @@ -33,7 +33,7 @@ import org.apache.spark.sql.connector.catalog.index.TableIndex import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, NamedReference, NullOrdering, SortDirection} import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils} -import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, FloatType, LongType, MetadataBuilder, StringType} +import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, FloatType, LongType, MetadataBuilder, StringType, TimestampType} private case object MySQLDialect extends JdbcDialect with SQLConfHelper { @@ -92,23 +92,32 @@ private case object MySQLDialect extends JdbcDialect with SQLConfHelper { override def getCatalystType( sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = { - if (sqlType == Types.VARBINARY && typeName.equals("BIT") && size != 1) { - // This could instead be a BinaryType if we'd rather return bit-vectors of up to 64 bits as - // byte arrays instead of longs. - md.putLong("binarylong", 1) - Option(LongType) - } else if (sqlType == Types.BIT && typeName.equals("TINYINT")) { - Option(BooleanType) - } else if ("TINYTEXT".equalsIgnoreCase(typeName)) { - // TINYTEXT is Types.VARCHAR(63) from mysql jdbc, but keep it AS-IS for historical reason - Some(StringType) - } else if (sqlType == Types.VARCHAR && typeName.equals("JSON")) { - // Some MySQL JDBC drivers converts JSON type into Types.VARCHAR with a precision of -1. - // Explicitly converts it into StringType here. - Some(StringType) - } else if (sqlType == Types.TINYINT && typeName.equals("TINYINT")) { - Some(ByteType) - } else None + sqlType match { + case Types.VARBINARY if "BIT".equalsIgnoreCase(typeName) && size != 1 => + // This could instead be a BinaryType if we'd rather return bit-vectors of up to 64 bits as + // byte arrays instead of longs. + md.putLong("binarylong", 1) + Some(LongType) + case Types.BIT if "TINYINT".equalsIgnoreCase(typeName) => + Some(BooleanType) + case Types.VARCHAR if "TINYTEXT".equalsIgnoreCase(typeName) => + // TINYTEXT is Types.VARCHAR(63) from mysql jdbc, but keep it AS-IS for historical reason + Some(StringType) + case Types.VARCHAR if "JSON".equalsIgnoreCase(typeName) => + // Some MySQL JDBC drivers converts JSON type into Types.VARCHAR with a precision of -1. + // Explicitly converts it into StringType here. + Some(StringType) + case Types.TINYINT if "TINYINT".equalsIgnoreCase(typeName) => + Some(ByteType) + case Types.TIMESTAMP if "DATETIME".equalsIgnoreCase(typeName) => + // scalastyle:off line.size.limit + // In MYSQL, DATETIME is TIMESTAMP WITHOUT TIME ZONE + // https://github.com/mysql/mysql-connector-j/blob/8.3.0/src/main/core-api/java/com/mysql/cj/MysqlType.java#L251 + // scalastyle:on line.size.limit + Some(getTimestampType(md.build())) + case Types.TIMESTAMP => Some(TimestampType) + case _ => None + } } override def quoteIdentifier(colName: String): String = { From 6ace4be60e58d22abbad6b069dca2b12ef215152 Mon Sep 17 00:00:00 2001 From: Kent Yao Date: Fri, 15 Mar 2024 12:34:46 +0800 Subject: [PATCH 2/2] [SPARK-47406][SQL] Handle TIMESTAMP and DATETIME in MYSQLDialect --- docs/sql-migration-guide.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 28fa19c351fc3..27b62a6bd7929 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -41,6 +41,7 @@ license: | - Since Spark 4.0, the SQL config `spark.sql.legacy.allowZeroIndexInFormatString` is deprecated. Consider to change `strfmt` of the `format_string` function to use 1-based indexes. The first argument must be referenced by "1$", the second by "2$", etc. - Since Spark 4.0, the function `to_csv` no longer supports input with the data type `STRUCT`, `ARRAY`, `MAP`, `VARIANT` and `BINARY` (because the `CSV specification` does not have standards for these data types and cannot be read back using `from_csv`), Spark will throw `DATATYPE_MISMATCH.UNSUPPORTED_INPUT_TYPE` exception. - Since Spark 4.0, JDBC read option `preferTimestampNTZ=true` will not convert Postgres TIMESTAMP WITH TIME ZONE and TIME WITH TIME ZONE data types to TimestampNTZType, which is available in Spark 3.5. +- Since Spark 4.0, JDBC read option `preferTimestampNTZ=true` will not convert MySQL TIMESTAMP to TimestampNTZType, which is available in Spark 3.5. MySQL DATETIME is not affected. ## Upgrading from Spark SQL 3.4 to 3.5