[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2 #44119

huaxingao · 2023-12-02T04:05:53Z

What changes were proposed in this pull request?

Add MergeInto support in DataFrameWriterV2

Why are the changes needed?

Spark currently supports merge into sql statement. We want DataFrame to have the same support.

Does this PR introduce any user-facing change?

Yes. This PR introduces new API like the following:

      spark.table("source")
        .mergeInto("target", $"source.id" === $"target.id")
        .whenNotMatched()
        .insertAll()
        .merge()

How was this patch tested?

new tests

Was this patch authored or co-authored using generative AI tooling?

No

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala

huaxingao · 2023-12-12T16:56:40Z

cc @aokolnychyi @cloud-fan @dongjoon-hyun @viirya
This PR is ready for review. Could you please take a look when you have a moment? Thanks a lot!
There is still a mima check failure. It passed on my local, though. I will figure out what is wrong.

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

cloud-fan · 2023-12-20T05:47:56Z

common/utils/src/main/resources/error/error-classes.json

+    "message" : [
+      "df.mergeInto needs to be followed by at least one of whenMatched/whenNotMatched/whenNotMatchedBySource."
+    ],
+    "sqlState" : "23K02"


Suggested change

"sqlState" : "23K02"

"sqlState" : "42K0E"

Fixed. Thanks

cloud-fan · 2023-12-20T05:50:22Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+    qe.assertCommandExecuted()
+  }
+
+  def withNewMatchedUpdateAction(condition: Option[Expression]): MergeIntoWriter[T] = {


this seems too many helper functions...

def withNewMatchedAction(action: MergeAction): MergeIntoWriter[T] = { this.matchedActions = this.matchedActions :+ action this }

I think 3 helper functions should be good enough for 3 different action types.

Fixed. Thanks

cloud-fan · 2023-12-20T05:51:56Z

sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoDataFrameSuite.scala

+import org.apache.spark.sql.Row
+import org.apache.spark.sql.functions._
+
+class MergeIntoDataFrameSuite extends RowLevelOperationSuiteBase {


+1. We only need to make sure the new scala API works. We don't need to test the underlying v2 sources extensively, which should have been covered already by other tests

viirya · 2023-12-20T06:06:16Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+   *
+   * @since 4.0.0


Similar to def write, @group basic?

Added. Thanks

viirya · 2023-12-20T06:20:49Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   * Initialize a `WhenNotMatched` action without any condition.
+   *
+   * This `WhenNotMatched` can be followed by one of the following merge actions:
+   *   - `insertAll`: Insert all the target table with source dataset records.


Insert all the columns of the target table with ....?

Insert all rows from the source that are not already in the target table.
Please refer https://docs.databricks.com/en/sql/language-manual/delta-merge-into.html#when-not-matched-[by-target]

I have fixed this and a few other places

viirya · 2023-12-20T06:22:18Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   *   - `insert(Map)`: Insert all the target table records while changing only
+   *     a subset of fields based on the provided assignment.


Insert the specified columns ...

viirya · 2023-12-20T06:23:30Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   * Initialize a `WhenNotMatchedBySource` action without any condition.
+   *
+   * This `WhenNotMatchedBySource` can be followed by one of the following merge actions:
+   *   - `updateAll`: Update all the target table fields with source dataset fields.


Update all the columns of the target table ...

viirya · 2023-12-20T06:24:17Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   *
+   * This `WhenNotMatchedBySource` can be followed by one of the following merge actions:
+   *   - `updateAll`: Update all the target table fields with source dataset fields.
+   *   - `update(Map)`: Update all the target table records while changing only


Update the specified columns of the target table ...

viirya · 2023-12-20T06:25:19Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   *   - `updateAll`: Update all the target table fields with source dataset fields.
+   *   - `update(Map)`: Update all the target table records while changing only
+   *     a subset of fields based on the provided assignment.
+   *   - `delete`: Delete all the target table records.


Delete the target table row.

viirya · 2023-12-20T06:25:47Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   *   - `updateAll`: Update all the target table fields with source dataset fields.
+   *   - `update(Map)`: Update all the target table records while changing only
+   *     a subset of fields based on the provided assignment.
+   *   - `delete`: Delete all the target table records.


Delete the matching target table row

How about Delete all target rows that have a match in the source table.?
Please refer https://docs.databricks.com/en/sql/language-manual/delta-merge-into.html#when-matched

Yea, I was referring the doc too. I've tried to combine @huaxingao original sentence and the doc. If @huaxingao wants to use these description from the doc, it is good too.

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

.../src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala

viirya · 2023-12-20T06:41:37Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -4129,6 +4129,36 @@ class Dataset[T] private[sql](
    new DataFrameWriterV2[T](table, this)
  }

+  /**
+   * Create a [[MergeIntoWriter]] for MergeInto action.


This is user facing API doc. Not sure if it is proper to put MergeIntoWriter there. For example, we don't put DataFrameWriter in write API doc.

Maybe just describing what the function is used for. E.g., "Merges a set of updates, insertions, and deletions based on a source table into a target table"

https://docs.databricks.com/en/sql/language-manual/delta-merge-into.html

The key is the MergeIntoWriter is public API or developer API.
cc @cloud-fan

Fixed. Thanks

beliefer · 2023-12-20T12:54:36Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   *   - `updateAll`: Update all the target table fields with source dataset fields.
+   *   - `update(Map)`: Update all the target table records while changing only
+   *     a subset of fields based on the provided assignment.
+   *   - `delete`: Delete all the target table records.


How about Delete all target rows that have a match in the source table.?
Please refer https://docs.databricks.com/en/sql/language-manual/delta-merge-into.html#when-matched

beliefer · 2023-12-20T12:56:59Z

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

+   * Initialize a `WhenNotMatched` action without any condition.
+   *
+   * This `WhenNotMatched` can be followed by one of the following merge actions:
+   *   - `insertAll`: Insert all the target table with source dataset records.


Insert all rows from the source that are not already in the target table.
Please refer https://docs.databricks.com/en/sql/language-manual/delta-merge-into.html#when-not-matched-[by-target]

huaxingao · 2023-12-21T01:42:47Z

The test failure doesn't seem to be related to my changes.

beliefer · 2023-12-21T02:18:55Z

Merged to master.
Thank you @huaxingao @cloud-fan @viirya @HyukjinKwon

huaxingao · 2023-12-21T02:21:02Z

Thank you all very much for reviewing the PR!

[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2

227bd1d

github-actions bot added the SQL label Dec 2, 2023

HyukjinKwon reviewed Dec 4, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

Huaxin Gao added 2 commits December 6, 2023 20:10

add scala doc

d2b5098

exclude for connect-client-jvm module mima check

d439cfa

github-actions bot added the CONNECT label Dec 11, 2023

Huaxin Gao added 2 commits December 11, 2023 00:07

Merge remote-tracking branch 'upstream/master' into mergeinto

6a6bdd4

fix conflict

0309663

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 13, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala Show resolved Hide resolved

Huaxin Gao added 2 commits December 14, 2023 11:36

address comments

acc11d4

fix lint scala error

b20430f

cloud-fan reviewed Dec 16, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala Show resolved Hide resolved

cloud-fan reviewed Dec 16, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Dec 16, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Dec 16, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Dec 16, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala Outdated Show resolved Hide resolved

address comments

8df2041

cloud-fan reviewed Dec 18, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Dec 18, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Dec 18, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

address comments

f37fbea

cloud-fan reviewed Dec 18, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriterV2.scala Outdated Show resolved Hide resolved

change the constructor to private[sql]

f456a7a

beliefer reviewed Dec 19, 2023

View reviewed changes

Huaxin Gao added 2 commits December 19, 2023 13:25

address comments

1f9c027

fix a few test failures in SparkThrowableSuite

d913e33

github-actions bot added the DOCS label Dec 20, 2023

cloud-fan reviewed Dec 20, 2023

View reviewed changes

cloud-fan approved these changes Dec 20, 2023

View reviewed changes

viirya reviewed Dec 20, 2023

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 20, 2023

View reviewed changes

.../src/test/scala/org/apache/spark/sql/connect/client/CheckConnectJvmClientCompatibility.scala Outdated Show resolved Hide resolved

viirya reviewed Dec 20, 2023

View reviewed changes

beliefer approved these changes Dec 20, 2023

View reviewed changes

address comments

24108e6

viirya approved these changes Dec 20, 2023

View reviewed changes

fix connect-client-jvm module mima check failure

da384c0

beliefer closed this in 56dc7f8 Dec 21, 2023

huaxingao deleted the mergeinto branch December 21, 2023 02:21

huaxingao mentioned this pull request Dec 21, 2023

Adding MergeInto into the Spark Scala API apache/iceberg#3665

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2 #44119

[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2 #44119

huaxingao commented Dec 2, 2023 •

edited

Loading

huaxingao commented Dec 12, 2023

cloud-fan Dec 20, 2023

huaxingao Dec 20, 2023

cloud-fan Dec 20, 2023

huaxingao Dec 20, 2023

cloud-fan Dec 20, 2023 •

edited

Loading

viirya Dec 20, 2023

huaxingao Dec 20, 2023

viirya Dec 20, 2023

beliefer Dec 20, 2023

huaxingao Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

beliefer Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

viirya Dec 20, 2023

beliefer Dec 20, 2023

huaxingao Dec 20, 2023

beliefer Dec 20, 2023

beliefer Dec 20, 2023

huaxingao commented Dec 21, 2023

beliefer commented Dec 21, 2023 •

edited

Loading

huaxingao commented Dec 21, 2023

		* - `insert(Map)`: Insert all the target table records while changing only
		* a subset of fields based on the provided assignment.

[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2 #44119

[SPARK-46207][SQL] Support MergeInto in DataFrameWriterV2 #44119

Conversation

huaxingao commented Dec 2, 2023 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

huaxingao commented Dec 12, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Dec 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huaxingao commented Dec 21, 2023

beliefer commented Dec 21, 2023 • edited Loading

huaxingao commented Dec 21, 2023

huaxingao commented Dec 2, 2023 •

edited

Loading

cloud-fan Dec 20, 2023 •

edited

Loading

beliefer commented Dec 21, 2023 •

edited

Loading