Spark 4.1: Add scoped replacement writes (INSERT INTO ... REPLACE USING)

[liucao-dd]

Part 1 work of #16785.

What

This adds scoped replacement writes to the Iceberg Spark 4.1 extensions. A scoped replacement treats the source as the complete replacement for the scopes it touches: it reads the distinct scope tuples from the source, deletes target rows whose scope shows up in the source, appends every source row once, and commits the delete and append as a single Iceberg snapshot. Scopes that the source never mentions are left alone. This is the "full group refresh" operation from #16785: atomic, with no two-commit DELETE + INSERT and no stretched MERGE INTO ... WHEN NOT MATCHED BY SOURCE.

There are two ways to invoke it.

DataFrame API (merge-eligible API for spark 4.1)

Dataframe API - DataFrameWriterV2 with the replace-using write option and an unconditional overwrite:

source.writeTo("target_table")
  .option("replace-using", "scope_col_1, scope_col_2")
  .overwrite(lit(true))

overwrite(lit(true)) is required: the option only converts a full overwrite into a scoped replacement, so any other overwrite filter (or append / overwritePartitions) is rejected with a clear error rather than silently ignoring the option. Normal Iceberg write options are preserved and flow through unchanged — for example target file size, compression, distribution mode, and branch selection:

source.writeTo("target_table")
  .option("replace-using", "scope_col_1, scope_col_2")
  .option("branch", "audit")
  .option(SparkWriteOptions.TARGET_FILE_SIZE_BYTES, "134217728")
  .overwrite(lit(true))

This surface works uniformly from Scala, Java, and PySpark because all three route through the same DataFrameWriterV2 → OverwriteByExpression logical plan. It is gated strictly to Iceberg tables; a replace-using option on a non-Iceberg DSv2 table is left untouched.

SQL (demonstration only, need Spark grammar addition ideally)

INSERT INTO target_table
REPLACE USING (scope_col_1, scope_col_2)
<query>

Scope of this PR

This PR covers part of #16785. A few things are left for later on purpose:

• Only the column-list form (REPLACE USING (cols) / replace-using) is implemented. The arbitrary-condition form (REPLACE ON <expr>) from the issue is not here yet.
• There is no native Spark grammar for the SQL form. Spark's insertInto grammar does not accept REPLACE USING, and the Iceberg extension grammar cannot own an arbitrary trailing Spark query. So the SQL path uses an in-place text split as a stopgap: the Iceberg ANTLR grammar parses only the command head (singleScopedReplaceHead: INSERT INTO t REPLACE USING (cols)), and the remaining query tail goes to the wrapped Spark parser. Detection is anchored to the statement head with literals and comments masked, so a REPLACE USING ( later in the query body (say, a join alias) does not trigger the path.

Note

The DataFrame API is self-contained and does not depend on the SQL grammar. Both DF and SQL reuses the same underlying ReplaceScopedData andRewriteScopedReplace. Because the .g4 text-split SQL parsing is a stopgap (the real end state is to propose REPLACE USING / REPLACE ON for Spark's native insertInto grammar), the SQL grammar / parser / AST-builder pieces and the SQL-only tests may be removed before this PR is merged to keep it focused on the DataFrame API.

How it works

• A new logical node, ReplaceScopedData(table, scopeColumns, source).
• RewriteDataFrameScopedReplace detects a replace-using OverwriteByExpression on an Iceberg table, validates the unconditional overwrite, merges the command and relation write options, eagerly resolves the write branch (Spark's post-hoc resolution batch runs only once, so the branch is pinned here rather than relying on a later pass), parses the scope columns, and emits ReplaceScopedData.
• RewriteScopedReplace lowers ReplaceScopedData through Iceberg's existing row-level write path. It requests the operation as MERGE so the table configuration decides between copy-on-write and merge-on-read (and deletion vectors):
  • COW → ReplaceData
  • MOR → WriteDelta
• It reuses the standard row-level group filtering and runtime filtering for performance. Note that there is no planning-time static pruning yet.

Tests

• TestReplaceScopedDataDataFrame covers the DataFrame surface: scope deletion, full source append, untouched scopes, empty source, duplicate scope keys, null scope values, multi-column scopes, write-option and branch passthrough, and validation (non-true overwrite, append/overwritePartitions, non-Iceberg tables, empty scope tokens). TestCopyOnWriteReplaceScopedDataDataFrame and TestMergeOnReadReplaceScopedDataDataFrame exercise both write modes.
• TestReplaceScopedData covers the shared semantics via SQL; TestCopyOnWriteReplaceScopedData and TestMergeOnReadReplaceScopedData exercise both write modes. (These SQL-only suites go away if the SQL grammar path is dropped per the note above.)

Docs

• docs/docs/spark-writes.md documents the scoped replacement DataFrame API and the supported REPLACE USING SQL form, and notes that REPLACE ON and native Spark grammar support are future work.

Follow-ups

1. Decide whether to drop the SQL text-split grammar from this PR (keeping the DataFrame API) and propose native Spark grammar for REPLACE USING / REPLACE ON separately.
2. Implement the REPLACE ON <boolean_expr> form.
3. When the source's distinct scope values are small/known, pre-evaluate them and synthesize a static scope IN (...) predicate to feed Iceberg's static filter.

[aokolnychyi]

This is something that must be done in Spark. Happy to help, but extensions are not the right fit.

[aokolnychyi]

Spark added support for the syntax recently, btw.

[liucao-dd]

realized apache/spark#54722 is done and this was internally built before that. Will look into porting over to spark instead, thx for the quick note @aokolnychyi

[liucao-dd]

opened spark PR instead to implement this properly apache/spark#56705

iceberg adoption will only be possible after spark releases