[SPARK-47390][SQL] PostgresDialect distinguishes TIMESTAMP from TIMESTAMP_TZ

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-47390: 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 " +

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
 

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))