[SPARK-34333][SQL] Fix PostgresDialect to handle money types properly

docs/sql-migration-guide.md

@@ -24,6 +24,8 @@ license: |
 
 ## Upgrading from Spark SQL 3.1 to 3.2
 
+  - In Spark 3.2, PostgreSQL JDBC dialect uses StringType for MONEY and MONEY[] is not supported due to the JDBC driver for PostgreSQL can't handle those types properly. In Spark 3.1 or earlier, DoubleType and ArrayType of DoubleType are used respectively.
+
   - In Spark 3.2, `spark.sql.adaptive.enabled` is enabled by default. To restore the behavior before Spark 3.2, you can set `spark.sql.adaptive.enabled` to `false`.
 
   - In Spark 3.2, the following meta-characters are escaped in the `show()` action. In Spark 3.1 or earlier, the following metacharacters are output as it is.

external/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala

@@ -133,6 +133,11 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite {
     conn.prepareStatement("INSERT INTO char_array_types VALUES " +
       """('{"a", "bcd"}', '{"ef", "gh"}', '{"i", "j", "kl"}', '{"mnop"}', '{"q", "r"}')"""
     ).executeUpdate()
+
+    conn.prepareStatement("CREATE TABLE money_types (" +
+      "c0 money)").executeUpdate()
+    conn.prepareStatement("INSERT INTO money_types VALUES " +
+      "('$1,000.00')").executeUpdate()
   }
 
   test("Type mapping for various types") {
@@ -366,4 +371,12 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite {
     assert(row(0).getSeq[String](3) === Seq("mnop"))
     assert(row(0).getSeq[String](4) === Seq("q", "r"))
   }
+
+  test("SPARK-34333: money type tests") {
+    val df = sqlContext.read.jdbc(jdbcUrl, "money_types", new Properties)
+    val row = df.collect()
+    assert(row.length === 1)
+    assert(row(0).length === 1)
+    assert(row(0).getString(0) === "$1,000.00")
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala

@@ -35,8 +35,12 @@ private object PostgresDialect extends JdbcDialect {
       Some(FloatType)
     } else if (sqlType == Types.SMALLINT) {
       Some(ShortType)
-    } else if (sqlType == Types.BIT && typeName.equals("bit") && size != 1) {
+    } 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 (sqlType == Types.ARRAY) {
@@ -56,7 +60,7 @@ private object PostgresDialect extends JdbcDialect {
     case "int4" => Some(IntegerType)
     case "int8" | "oid" => Some(LongType)
     case "float4" => Some(FloatType)
-    case "money" | "float8" => Some(DoubleType)
+    case "float8" => Some(DoubleType)
     case "text" | "varchar" | "char" | "bpchar" | "cidr" | "inet" | "json" | "jsonb" | "uuid" |
          "xml" | "tsvector" | "tsquery" | "macaddr" | "macaddr8" | "txid_snapshot" | "point" |
          "line" | "lseg" | "box" | "path" | "polygon" | "circle" | "pg_lsn" | "varbit" |
@@ -69,6 +73,11 @@ private object PostgresDialect extends JdbcDialect {
     case "numeric" | "decimal" =>
       // SPARK-26538: handle numeric without explicit precision and scale.
       Some(DecimalType. SYSTEM_DEFAULT)
+    case "money" =>
+      // money[] type seems to be broken and difficult to handle.
+      // So this method returns None for now.
+      // See SPARK-34333 and https://github.com/pgjdbc/pgjdbc/issues/1405
+      None
     case _ => None
   }