Skip to content

[SPARK-19335][SPARK-38200][SQL] Add upserts for writing to JDBC using MERGE INTO with temp table #41611

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

[SPARK-19335][SPARK-38200][SQL] Add upserts for writing to JDBC using MERGE INTO with temp table #41611

Changes from 1 commit
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
b54c3b2
Add framework to implement upsert via temp table
EnricoMi Jun 5, 2023
e93239e
Implementation for MsSqlServer
EnricoMi Jun 14, 2023
e7449ec
Implementation for Postgres
EnricoMi Jun 15, 2023
817ca0b
Implementation for Derby
EnricoMi Jun 15, 2023
12c46d1
Implemenation for H2
EnricoMi Jun 15, 2023
c22e6ea
Modify tests so they work for Oracle
EnricoMi Jun 15, 2023
7943498
Implementation for Oracle
EnricoMi Jun 14, 2023
a3fb51d
Add dialect to upsert not supported error
EnricoMi Jun 23, 2023
66ee0fe
Fix after merge with master HEAD
EnricoMi Jul 18, 2023
0e0bcac
Ensure H2Dialect is registered for upsert test
EnricoMi Jul 26, 2023
18b1ee0
Fix after rebase master
EnricoMi Oct 25, 2023
44eb0be
Add sqlState to new error messages
EnricoMi Jan 11, 2024
9883cf9
Temp tables in Derby use SESSION schema
EnricoMi Jan 23, 2024
c38aa3a
Fix minor typo
EnricoMi Jan 23, 2024
8e6106e
Remove derby upsert support
EnricoMi Jan 23, 2024
eb9571b
Fix JDBC+Oservation test
EnricoMi Jan 23, 2024
583d349
Fix lint and test compile error after merge
EnricoMi Sep 2, 2024
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
Prev Previous commit
Next Next commit
Fix JDBC+Oservation test
  • Loading branch information
@EnricoMi
EnricoMi committed Apr 28, 2025
commit eb9571b37f80190331236ebe5f3d22dadb0e692b
11 changes: 5 additions & 6 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1065,9 +1065,8 @@ object JdbcUtils extends Logging with SQLConfHelper {
case Some(n) if n < df.rdd.getNumPartitions => df.coalesce(n)
case _ => df
}
repartitionedDF.rdd.foreachPartition { iterator =>
savePartition(None, iterator, rddSchema, insertStmt, batchSize, dialect,
isolationLevel, options)
repartitionedDF.foreachPartition { iterator: Iterator[Row] => savePartition(
None, iterator, rddSchema, insertStmt, batchSize, dialect, isolationLevel, options)
}
}

Expand Down Expand Up @@ -1105,9 +1104,9 @@ object JdbcUtils extends Logging with SQLConfHelper {
case Some(n) if n < df.rdd.getNumPartitions => df.coalesce(n)
case _ => df
}
repartitionedDF.rdd.foreachPartition { iterator => upsertPartition(
table, iterator, rddSchema, tableSchema, isCaseSensitive, batchSize,
dialectWithMerge, isolationLevel, options)
repartitionedDF.foreachPartition { iterator: Iterator[Row] => upsertPartition(
table, iterator, rddSchema, tableSchema, isCaseSensitive, batchSize, dialectWithMerge,
isolationLevel, options)
}
}

Expand Down