[SPARK-57636][SQL] Support REPLACE USING with row-level DSv2 writes by liucao-dd · Pull Request #56705 · apache/spark

liucao-dd · 2026-06-23T20:06:51Z

What changes were proposed in this pull request?

SPARK-56001 added parser and logical syntax support.

This PR is a continuation of that work, and adds row-level DSv2 execution support for INSERT INTO ... REPLACE USING.

It introduces a dedicated RowLevelOperation.Command.REPLACE command and rewrites scoped replace operations onto Spark's existing row-level write machinery:

copy-on-write tables use ReplaceData
merge-on-read tables use WriteDelta
replace write metrics are reported through ReplaceSummary
existing row-level connectors are protected by a new SupportsRowLevelReplace opt-in marker

The PR also updates the in-memory DSv2 test fixtures to cover positive and negative opt-in behavior, including copy, ALTER, and transactional wrapper paths.

Key design decisions

Use a new RowLevelOperation.Command.REPLACE instead of reusing MERGE. REPLACE USING deletes target rows by source-defined scope tuples and inserts all source rows, including duplicate source scopes, which is not the same cardinality model as MERGE.
Keep the execution path in Spark's row-level DSv2 framework. Group-based/copy-on-write connectors reuse ReplaceData; delta-based/merge-on-read connectors reuse WriteDelta.
Resolve the source with normal INSERT INTO alignment rules. By-position and BY NAME behavior continue to use Spark's standard output resolution before the rewrite builds scope joins and insert rows by ordinal.
Treat the replacement scope as dynamic source state. Static planning predicates stay broad, while runtime group filtering is populated when enabled so capable scans can prune affected groups dynamically.
Require deterministic source queries. The source determines both the delete scope and inserted rows, so non-deterministic sources are rejected during analysis.
Gate connector adoption explicitly with SupportsRowLevelReplace. Non-row-level DSv2 tables and row-level tables that do not opt in fail with an unsupported table operation error before Spark calls the connector's row-level operation builder with REPLACE.
Report REPLACE-specific metrics separately through ReplaceSummary so inserted, deleted, and copied rows are not derived from ambiguous generic row-level task counters.
Targets spark 4.2 in current code, but if missed release window, need to modify @since annotations to use 4.3 instead

Out of scope

REPLACE ON execution support.
DataFrame API support for scoped replace.

Why are the changes needed?

INSERT INTO ... REPLACE USING needs a row-level execution path so DSv2 tables can replace source-defined scopes without relying on table-wide overwrite behavior.

The explicit SupportsRowLevelReplace opt-in prevents existing row-level connectors that only implemented DELETE, UPDATE, or MERGE from unexpectedly receiving the new REPLACE command.

Does this PR introduce any user-facing change?

Yes. DSv2 connectors that implement SupportsRowLevelReplace can support INSERT INTO ... REPLACE USING (cols) through row-level operations. Row-level tables that do not opt in fail analysis with an unsupported table operation error instead of invoking the connector's row-level operation builder with an unexpected command.

How was this patch tested?

Tested with:

export JAVA_HOME=/Library/Java/JavaVirtualMachines/amazon-corretto-17.jdk/Contents/Home
export SPARK_LOCAL_IP=127.0.0.1
build/sbt 'catalyst/Test/compile' 'sql/testOnly org.apache.spark.sql.connector.GroupBasedReplaceUsingTableSuite org.apache.spark.sql.connector.DeltaBasedReplaceUsingTableSuite org.apache.spark.sql.connector.ReplaceUsingUnsupportedTableSuite'

Also ran representative row-level regression suites during development:

build/sbt 'sql/testOnly *DeltaBasedReplaceUsingTableSuite *ReplaceUsingUnsupportedTableSuite *GroupBasedDeleteFromTableSuite *DeltaBasedMergeIntoTableSuite'

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

co-authored with the help of a combination of GPT-5.5 and Claude Opus 4.8

szehon-ho · 2026-06-23T20:29:14Z

@longvu-db fyi

uros-b

Thank you @liucao-dd, I left a few comments, but adding @aokolnychyi for further review in DSv2 row-level operations.

longvu-db · 2026-06-29T16:46:58Z

Hi @liucao-dd, thanks for the PR, this is a great addition to Spark!

I'm reviewing the code now and will drop my feedback soon.

uros-b · 2026-06-30T08:53:54Z

      case DELETE =>
        MetadataAttribute.isPreservedOnDelete(attr)
+      case REPLACE =>
+        // REPLACE emits delete rows for matched target rows and insert rows for source rows.
+        // Insert rows carry no metadata, so metadata nullability only needs delete preservation.
+        MetadataAttribute.isPreservedOnDelete(attr)


Nit: this still has two identical adjacent case DELETE => / case REPLACE => arms collapsible to case DELETE | REPLACE =>.

uros-b

Thank you for addressing the comments @liucao-dd. I left just one more note. Let's please wait for @longvu-db's review too.

longvu-db

Thanks for working on this impactful feature @liucao-dd!

I suggest splitting the PR into as many self-contained smaller PRs, i.e.:

INSERT REPLACE USING ReplaceData
INSERT REPLACE USING WriteDelta
INSERT REPLACE USING write metrics
INSERT REPLACE USING with other SQL Clause (BY NAME, WITH OPTIONS, WITH SCHEMA EVOLUTION,..)

So that this PR focuses purely on getting an E2E working version of INSERT REPLACE USING.

Since

INSERT, INSERT REPLACE USING, and DSv2, each requires a lot of testing, since they write data, and they interact with a lot of adjacent features, which combine together will be even more.
This PR introduces quite some user-facing changes, which requires careful reviews.
INSERT REPLACE USING is not a trivial feature.

longvu-db · 2026-07-02T08:29:53Z

   */
  enum Command {
-    DELETE, UPDATE, MERGE
+    DELETE, UPDATE, MERGE, REPLACE


Suggested change

DELETE, UPDATE, MERGE, REPLACE

DELETE, UPDATE, MERGE, INSERT_INTO_REPLACE

To better distinguish against the SQL REPLACE TABLE command

How about ROW_LEVEL_REPLACE? The enum is describing the kind of row-level operation and not necessarily the SQL grammar. Though the existing 3 are all non-ambiguously matching the SQL grammar. So happy to adopt your suggestion as well if we do not anticipate this being used by anything other than INSERT INTO REPLACE USING/ON

longvu-db · 2026-07-02T08:39:45Z

    ],
    "sqlState" : "0A000"
  },
+  "INSERT_REPLACE_USING_DUPLICATE_SCOPE_COLUMN" : {


I think let's not block duplicate columns, users would expect INSERT REPLACE USING to have the same matching semantics as JOIN USING, and JOIN USING allows duplicate columns.

Also the term "scope column" is likely to be unfamiliar to users.

longvu-db · 2026-07-06T08:40:23Z

+
+    // All source rows are inserted. Carryover supplies target metadata; insert rows null it out.
+    val insertData = rowAttrs.indices.map { i =>
+      Alias(insertRef.output(i), rowAttrs(i).name)()


I'm not sure if we need all these Alias-ing in the INSERT steps, I think it should already be handled by TableOutputResolver.resolveOutputColumns, or something in the Analyzer, INSERT SQLs in DSv1 already has a way to allow INSERT source query with different names into the table via aliasing (or projecting)

longvu-db · 2026-07-06T08:56:23Z

+    props
+  }
+
+  test("replace using deletes matching scopes and inserts all source rows") {


We should add the following tests:

Self-INSERTing: INSERT INTO t REPLACE USING ... SELECT ... FROM t

Non-Deterministic formats such as Parquet, CSV,...

INSERT REPLACE USING analysis Exceptions: too many source query columns than table columns, empty REPLACE USING clause,...

INSERT adjacent features: WITH SCHEMA EVOLUTION, constraint check, BY NAME (matching on a column in a different position in the table and the source query

Matching on multiple columns, where there exists one that carries the NULL value, if a tuple has a NULL, then it shouldn't match.

INSERT REPLACE USING on a partitioned table, behavior should be similar to DPO, except not matching on NULL partitions

Palindrome on REPLACE USING clauses, i.e. (col1, col2) and (col2, col1)

Different source query shape (SELECT * FROM table), SELECT * FROM TABLE with filter, VALUES (x, y, z) (in this case the source query's columns are col1, col2, ...), source table exists but empty

Empty target table

REPLACE USING column exists in table but not in source query, exists in source query but not in table, exist in neither,...

INSERT REPLACE USING with optionsClause, should still respect optionsClause
https://github.com/apache/spark/blob/master/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4#L594C82-L594C96

REPLACE USING with struct columns in the USING clause, top-level name, or nested level column name

Column in the REPLACE USING clause has different data types between the source and target.

Not enough source query columns compared to table columns, this should still succeed, and the missing columns get NULL

CTCC1 · 2026-07-08T06:59:09Z

thank you for the detailed review! @longvu-db - it will take me a while to check and address these, will let you know.

As for the suggestion on splitting PRs, I will see I can address the suggestions and get a E2E working version first, and split as needed.

Add a dedicated RowLevelOperation.Command.REPLACE and resolve INSERT INTO ... REPLACE USING into a row-level command that reuses Spark's ReplaceData and WriteDelta execution paths. REPLACE stays SQL-only in this change: REPLACE ON and DataFrame APIs remain out of scope, and write summaries are left unset until row-level tasks expose accurate REPLACE metrics. The rewrite treats the source scope as dynamic state: target rows matching source scope tuples are deleted, all source rows are inserted, source queries must be deterministic, and runtime group filtering is populated when enabled. Existing DSv2 connectors are unaffected unless they participate in row-level rewrite planning for REPLACE; non-row-level tables retain the unsupported-operation error path.

Expose REPLACE-specific write summaries so DSv2 connectors receive accurate inserted, deleted, and copied row counts for scoped replacement.

…e opt-in for existing DSV2 connector safe adoption fix style

Merge the identical DELETE/REPLACE and MERGE/REPLACE match arms in WriteDelta.isMetadataNullabilityPreserved and RowLevelOperationRuntimeGroupFiltering.buildMatchingRowsPlan. Pure simplification with no behavior change.

…fallback Tighten the ResolveInsertInto unsupported-criteria guard to match InsertReplaceOn explicitly instead of any defined replace criteria, so the fallback documents exactly what remains unsupported for DSv2 tables. Behavior-preserving: InsertReplaceUsing is already handled by the preceding case.

Refer to INSERT INTO ... REPLACE USING rather than a bare REPLACE in the SLAM metric comment, to avoid confusion with the REPLACE TABLE DDL command. Comment-only; no behavior change.

Extract a narrow buildReplacementQuery helper for the copy-on-write union of carryover and inserted source rows, and reuse scopeEquality inside scopeExists by widening its parameters to Expression and passing the full outer-referenced target output. No behavior change; equality semantics are unchanged.

…supported Rename INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE to INSERT_REPLACE_USING_NON_DETERMINISTIC_SOURCE_QUERY_NOT_SUPPORTED and change its sqlState from 42K0E to 0A000, reflecting that a non-deterministic source query is a currently-unsupported feature rather than an invalid expression. Reword the message accordingly and update the error method and test.

…UsingCols Align the raw column-name identifiers with the parser's replaceUsingCols naming: the ReplaceUsingTable parameter, the ResolveInsertInto local, and the RewriteReplaceUsing pattern binding and resolveScopeOrdinals parameter. The derived matching-tuple helpers (scopeOrdinals/scopeEquality/scopeExists) keep the scope terminology.

liucao-dd mentioned this pull request Jun 23, 2026

Spark 4.1: Add scoped replacement writes (INSERT INTO ... REPLACE USING) apache/iceberg#16795

Open

liucao-dd force-pushed the spark-support-insert-into-replace-using-execution-dsv2 branch from 4492b4e to ec6526f Compare June 23, 2026 21:53