Skip to content

[SPARK-43267][JDBC] Handle postgres unknown user-defined column as string in array #40953

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
wants to merge 3 commits into from
Closed

[SPARK-43267][JDBC] Handle postgres unknown user-defined column as string in array #40953

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ea266a1
[SPARK-43267][JDBC] Handle postgres unknown user-defined column array…
Hisoka-X Apr 26, 2023
faa82aa
Merge branch 'master' into SPARK-43267_pg_array
Hisoka-X Jun 1, 2023
07030c4
add test
Hisoka-X Jun 1, 2023
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
16 changes: 16 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 @@ -147,6 +147,13 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite {
|('2013-04-05 12:01:02'),
|('2013-04-05 18:01:02.123'),
|('2013-04-05 18:01:02.123456')""".stripMargin).executeUpdate()

conn.prepareStatement("CREATE DOMAIN not_null_text AS TEXT DEFAULT ''").executeUpdate()
conn.prepareStatement("create table custom_type(type_array not_null_text[]," +
"type not_null_text)").executeUpdate()
conn.prepareStatement("INSERT INTO custom_type (type_array, type) VALUES" +
"('{1,fds,fdsa}','fdasfasdf')").executeUpdate()

}

test("Type mapping for various types") {
Expand Down Expand Up @@ -416,4 +423,13 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite {
val df_actual = sqlContext.read.jdbc(jdbcUrl, "timestamp_ntz_roundtrip", prop)
assert(df_actual.collect()(0) == df_expected.collect()(0))
}

test("SPARK-43267: user-defined column in array test") {
val df = sqlContext.read.jdbc(jdbcUrl, "custom_type", new Properties)
val row = df.collect()
assert(row.length === 1)
assert(row(0).length === 2)
assert(row(0).getSeq[String](0) == Seq("1", "fds", "fdsa"))
assert(row(0).getString(1) == "fdasfasdf")
}
}
6 changes: 4 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 @@ -94,13 +94,15 @@ private object PostgresDialect extends JdbcDialect with SQLConfHelper {
case "numeric" | "decimal" if precision > 0 => Some(DecimalType.bounded(precision, scale))
case "numeric" | "decimal" =>
// SPARK-26538: handle numeric without explicit precision and scale.
Some(DecimalType. SYSTEM_DEFAULT)
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
case _ =>
// SPARK-43267: handle unknown types in array as string, because there are user-defined types
Some(StringType)
}

override def convertJavaTimestampToTimestampNTZ(t: Timestamp): LocalDateTime = {
Expand Down