Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sarutak wants to merge 6 commits into from
Closed

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

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
66d4292
Fix for money and money[] types.
sarutak Feb 2, 2021
59dbb59
Assign JIRA ID.
sarutak Feb 2, 2021
2cef36b
Merge branch 'master' of https://github.com/apache/spark into fix-for…
sarutak Feb 9, 2021
c63121a
Add note about the previous behavior.
sarutak Feb 9, 2021
7e8dfb8
Fix conflict.
sarutak Feb 10, 2021
b838218
Change the expression in the migration guide.
sarutak Feb 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/sql-migration-guide.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ license: |

## Upgrading from Spark SQL 3.1 to 3.2

- In Spark 3.2, money type in PostgreSQL table is converted to `StringType` and money[] type is not supported due to the JDBC driver for PostgreSQL can't handle those types properly.

@srowen srowen Feb 3, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, DecimalType fixes the accuracy problem with money, but can't account for units. I suppose we could one day support it as a struct of DecimalType and StringType for currency, but string seems fine now.
Why not a string array for a money array?

@sarutak sarutak Feb 3, 2021 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not a string array for a money array?

For money type, Spark SQL calls PgResultSet.getDouble causing the error.

[info]   org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0) (192.168.1.204 executor driver): org.postgresql.util.PSQLException: Bad value for type double : 1,000.00
[info] 	at org.postgresql.jdbc.PgResultSet.toDouble(PgResultSet.java:3104)
[info] 	at org.postgresql.jdbc.PgResultSet.getDouble(PgResultSet.java:2432)
[info] 	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$makeGetter$5(JdbcUtils.scala:418)

So, we can avoid this issue by mapping money type to StringType to let Spark SQL call getString rather than getDouble.

For money[] type, on the other hand, the PostgreSQL's JDBC driver calls PgResultSet.toDouble internally.

[info] 	at org.postgresql.jdbc.PgResultSet.toDouble(PgResultSet.java:3104)
[info] 	at org.postgresql.jdbc.ArrayDecoding$5.parseValue(ArrayDecoding.java:235)
[info] 	at org.postgresql.jdbc.ArrayDecoding$AbstractObjectStringArrayDecoder.populateFromString(ArrayDecoding.java:122)
[info] 	at org.postgresql.jdbc.ArrayDecoding.readStringArray(ArrayDecoding.java:764)

We can control how Spark SQL gets the value from the array obtained by PgResultSet.getArray, but it's difficult to control how the JDBC driver handles the elements in the array which is to be returned.

@cloud-fan cloud-fan Feb 8, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In the migration guide, we should also mention the previous behavior.

@sarutak sarutak Feb 9, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks. I've updated.


- 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.
Expand Down
13 changes: 13 additions & 0 deletions ...integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,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") {
Expand Down Expand Up @@ -262,4 +267,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")
}
}
13 changes: 11 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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" =>
Some(StringType)
case "bytea" => Some(BinaryType)
Expand All @@ -66,6 +70,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
}

Expand Down