From 47d817df08e3600b9513bd9e08bd0a76fce791de Mon Sep 17 00:00:00 2001 From: Kent Yao Date: Thu, 14 Mar 2024 16:42:45 +0800 Subject: [PATCH 1/2] [SPARK-47390][SQL] PostgresDialect distinguishes TIMESTAMP from TIMESTAMP_TZ --- .../sql/jdbc/PostgresIntegrationSuite.scala | 17 ++++--- docs/sql-migration-guide.md | 1 + .../spark/sql/jdbc/PostgresDialect.scala | 45 ++++++++++--------- 3 files changed, 33 insertions(+), 30 deletions(-) diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala index 04e31679f386..54ba912a6d5f 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala @@ -284,15 +284,14 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite { assert(schema(1).dataType == ShortType) } - test("SPARK-20557: column type TIMESTAMP with TIME ZONE and TIME with TIME ZONE " + - "should be recognized") { - // When using JDBC to read the columns of TIMESTAMP with TIME ZONE and TIME with TIME ZONE - // the actual types are java.sql.Types.TIMESTAMP and java.sql.Types.TIME - val dfRead = sqlContext.read.jdbc(jdbcUrl, "ts_with_timezone", new Properties) - val rows = dfRead.collect() - val types = rows(0).toSeq.map(x => x.getClass.toString) - assert(types(1).equals("class java.sql.Timestamp")) - assert(types(2).equals("class java.sql.Timestamp")) + test("SPARK-20557: Convert TIMESTAMP/TIME WITH TIME ZONE regardless of preferTimestampNTZ") { + Seq(true, false).foreach { prefer => + val rows = sqlContext.read + .option("preferTimestampNTZ", prefer) + .jdbc(jdbcUrl, "ts_with_timezone", new Properties) + .collect() + rows.head.toSeq.tail.foreach(c => assert(c.isInstanceOf[java.sql.Timestamp])) + } } test("SPARK-22291: Conversion error when transforming array types of " + diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 325e58bf753d..28fa19c351fc 100644 --- a/docs/sql-migration-guide.md +++ b/docs/sql-migration-guide.md @@ -40,6 +40,7 @@ license: | - Since Spark 4.0, `SELECT (*)` is equivalent to `SELECT struct(*)` instead of `SELECT *`. To restore the previous behavior, set `spark.sql.legacy.ignoreParenthesesAroundStar` to `true`. - 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. ## Upgrading from Spark SQL 3.4 to 3.5 diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala index d19b5ba3e0eb..7d8ed70b2bd1 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala @@ -49,26 +49,28 @@ private object PostgresDialect extends JdbcDialect with SQLConfHelper { override def getCatalystType( sqlType: Int, typeName: String, size: Int, md: MetadataBuilder): Option[DataType] = { - if (sqlType == Types.REAL) { - Some(FloatType) - } else if (sqlType == Types.SMALLINT) { - Some(ShortType) - } else if (sqlType == Types.BIT && typeName == "bit" && size != 1) { - Some(BinaryType) - } else if (sqlType == Types.DOUBLE && typeName == "money") { - // money type seems to be broken but one workaround is to handle it as string. - // See SPARK-34333 and https://github.com/pgjdbc/pgjdbc/issues/100 - Some(StringType) - } else if (sqlType == Types.OTHER) { - Some(StringType) - } else if ("text".equalsIgnoreCase(typeName)) { - Some(StringType) // sqlType is Types.VARCHAR - } else if (sqlType == Types.ARRAY) { - val scale = md.build().getLong("scale").toInt - val isTimestampNTZ = md.build().getBoolean("isTimestampNTZ") - // postgres array type names start with underscore - toCatalystType(typeName.drop(1), size, scale, isTimestampNTZ).map(ArrayType(_)) - } else None + sqlType match { + case Types.REAL => Some(FloatType) + case Types.SMALLINT => Some(ShortType) + case Types.BIT if typeName == "bit" && size != 1 => Some(BinaryType) + case Types.DOUBLE if typeName == "money" => + // money type seems to be broken but one workaround is to handle it as string. + // See SPARK-34333 and https://github.com/pgjdbc/pgjdbc/issues/100 + Some(StringType) + case Types.TIMESTAMP + if "timestamptz".equalsIgnoreCase(typeName) || "timetz".equalsIgnoreCase(typeName) => + // timestamptz represents timestamp with time zone, currently it maps to Types.TIMESTAMP. + // We need to change to Types.TIMESTAMP_WITH_TIMEZONE if the upstream changes. + Some(TimestampType) + case Types.OTHER => Some(StringType) + case _ if "text".equalsIgnoreCase(typeName) => Some(StringType) // sqlType is Types.VARCHAR + case Types.ARRAY => + val scale = md.build().getLong("scale").toInt + val isTimestampNTZ = md.build().getBoolean("isTimestampNTZ") + // postgres array type names start with underscore + toCatalystType(typeName.drop(1), size, scale, isTimestampNTZ).map(ArrayType(_)) + case _ => None + } } private def toCatalystType( @@ -91,7 +93,8 @@ private object PostgresDialect extends JdbcDialect with SQLConfHelper { "interval" | "pg_snapshot" => Some(StringType) case "bytea" => Some(BinaryType) - case "timestamp" | "timestamptz" | "time" | "timetz" => + case "timestamptz" | "timetz" => Some(TimestampType) + case "timestamp" | "time" => Some(if (isTimestampNTZ) TimestampNTZType else TimestampType) case "date" => Some(DateType) case "numeric" | "decimal" if precision > 0 => Some(DecimalType.bounded(precision, scale)) From 579bc06db747cbc7affb2e4171b75527abd43ae6 Mon Sep 17 00:00:00 2001 From: Kent Yao Date: Thu, 14 Mar 2024 16:47:15 +0800 Subject: [PATCH 2/2] [SPARK-47390][SQL] PostgresDialect distinguishes TIMESTAMP from TIMESTAMP_TZ --- .../org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala index 54ba912a6d5f..4ea0bdc24460 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala @@ -284,7 +284,7 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite { assert(schema(1).dataType == ShortType) } - test("SPARK-20557: Convert TIMESTAMP/TIME WITH TIME ZONE regardless of preferTimestampNTZ") { + test("SPARK-47390: Convert TIMESTAMP/TIME WITH TIME ZONE regardless of preferTimestampNTZ") { Seq(true, false).foreach { prefer => val rows = sqlContext.read .option("preferTimestampNTZ", prefer)