[SPARK-58111][SQL] Scan and write schema narrowing for column-level UPDATE in DSv2 by anuragmantri · Pull Request #55518 · apache/spark

anuragmantri · 2026-04-23T18:30:03Z

What changes were proposed in this pull request?

This PR adds an opt-in DSv2 mix-in for connectors to receive narrow rows on column-level UPDATE, so connectors can avoid reading and writing full-table rows when only a subset of columns is being updated.

Public API additions (all @Evolving, since 4.3.0):

SupportsColumnUpdates — a new mix-in on RowLevelOperation. Connectors implement requiredDataAttributes(): NamedReference[] to declare the exact set of columns they need to receive.
RowLevelOperationInfo.updatedColumns(): NamedReference[] — non-identity assignment column names (root-column granularity for nested field updates), populated by Spark before the connector's operation builder runs, so the connector can size
requiredDataAttributes() accordingly.
LogicalWriteInfo.updateSchema(): Optional — narrow row schema for writeUpdate()-bound rows; schema() continues to carry the full table shape for INSERT-tagged rows.
DataWriter.writeUpdate(record) and writeUpdate(metadata, record) — narrow write channel; default implementations delegate to write(...) so existing connectors are unaffected.

When a connector mixes in SupportsColumnUpdates, Spark's UPDATE rewrite (RewriteUpdateTable) narrows both sides of the plan:

Why are the changes needed?

Schema narrowing helps connectors request for only updated columns enabling efficient column-level updates of wide tables.

Does this PR introduce any user-facing change?

Yes, new public DSv2 connector APIs:

RowLevelOperation gains a new mix-in SupportsColumnUpdates (requiredDataAttributes())
RowLevelOperationInfo.updatedColumns()
LogicalWriteInfo.updateSchema()
DataWriter.writeUpdate(...)

All are @evolving and opt-in with backward-compatible defaults. Connectors that do not mix in SupportsColumnUpdates see no behavior change.

How was this patch tested?

New tests in:

DeltaBasedColumnUpdateTableSuite
GroupBasedColumnUpdateTableSuite

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

I used Claude Code to generate code and tests and manually reviewed the generated code.

peter-toth · 2026-04-27T16:26:08Z

+      val required =
+        AttributeSet(dataAttrs) ++ AttributeSet(Seq(cond)) ++ AttributeSet(rowIdAttrs)
+      val narrowOutput = relation.output.filter(required.contains)
+      relation.copy(table = table, output = dedupAttrs(narrowOutput ++ rowIdAttrs ++ metadataAttrs))


Can an attribute in required be missing from relation.output?
rowIdAttrs seems to be added 2 times.
If we already have a dedupAttrs() then probably doesn't make sense build AttributeSets.

Can an attribute in required be missing from relation.output?

No. dataAttrs come from the connector's requiredDataAttributes() which are resolved against relation (via V2ExpressionUtils.resolveRefs), so they're guaranteed to be present. The condition's referenced columns are also table columns from the user's WHERE clause. rowIdAttrs and metadataAttrs can be absent from relation.output (they're resolved separately), but they're not part of the filter. They're appended unconditionally afterward via dedupAttrs(narrowOutput ++ rowIdAttrs ++ metadataAttrs)

rowIdAttrs seems to be added 2 times. If we already have dedupAttrs() then probably doesn't make sense to build AttributeSets.

Agreed. I fixed it.

dongjoon-hyun

Could you resolve the conflicts, @anuragmantri ?

anuragmantri · 2026-05-05T23:38:47Z

Could you resolve the conflicts, @anuragmantri ?

Thanks. I rebased and fixed the conflicts.

dongjoon-hyun · 2026-05-06T00:34:42Z

    return new NamedReference[0];
  }
+
+


nit. Remove redundant empty line.

dongjoon-hyun · 2026-05-06T00:35:34Z

+   * including the columns being updated. If {@link #requiredDataAttributes()} returns an empty
+   * array, Spark sends only the non-identity assigned columns (heuristic path).
+   *
+   * @since 4.2.0


4.2.0 -> 4.3.0

dongjoon-hyun · 2026-05-06T00:35:41Z

+   * <p>
+   * When empty (the default), Spark falls back to sending only the non-identity assigned columns.
+   *
+   * @since 4.2.0


ditto. 4.3.0

dongjoon-hyun · 2026-05-06T00:40:28Z

-          val table = buildOperationTable(tbl, UPDATE, CaseInsensitiveStringMap.empty())
+          val updatedCols = assignments.collect {
+            case Assignment(key: AttributeReference, value)
+                if !isIdentityAssignment(key, value) =>


One liner doesn't violate the line-length rule, does it?

- case Assignment(key: AttributeReference, value) - if !isIdentityAssignment(key, value) => + case Assignment(key: AttributeReference, value) if !isIdentityAssignment(key, value) =>

dongjoon-hyun · 2026-05-06T00:44:55Z

+  //
+  // When dataAttrs is non-empty, the relation output is narrowed to include only columns
+  // required for a column-update write. When dataAttrs is empty, the full relation.output is
+  // preserved.


For function description, please follow the community style like the other code path.

/** * ... */

dongjoon-hyun · 2026-05-06T00:45:06Z

+  // When the connector supports column updates and declares required data attributes,
+  // the read relation is narrowed at analysis time so that
+  // GroupBasedRowLevelOperationScanPlanning uses only the needed columns for the scan.
+  // Otherwise the full relation output is used.


For function description, please follow the community style like the other code path.

/** * ... */

dongjoon-hyun · 2026-05-06T00:45:15Z

    WriteDelta(writeRelation, cond, rowDeltaPlan, relation, projections, groupFilterCond)
  }

+  // Builds the row delta projection for the column update path.


For function description, please follow the community style like the other code path.

/** * ... */

dongjoon-hyun · 2026-05-06T00:45:28Z

+      dataAttrsResolved(inRowAttrs)
+  }
+
+  // Validates the narrow-write-schema row projection output.


For function description, please follow the community style like the other code path.

/** * ... */

dongjoon-hyun · 2026-05-06T00:46:36Z

-    table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs)
+    table.skipSchemaResolution ||
+      areCompatible(inRowAttrs, outRowAttrs) ||
+      dataAttrsResolved(inRowAttrs)


nit. Please minimize the change of existing code as much as possible like the following.

table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs) || dataAttrsResolved(inRowAttrs)

dongjoon-hyun · 2026-05-06T00:51:39Z

+   * is ignored and the full table row is sent (the default behavior).
+   * <p>
+   * When non-empty, the returned columns become the write schema in declared order.
+   * The connector must declare all columns it wants to receive, including the columns being


This is very strong assumption, but it seems that this PR didn't have a protection. May I ask if we have some kind of assertion or a test coverage for this?

Each column the connector returns passes through V2ExpressionUtils.resolveRefs which throws AnalysisException if the column is non existent.

I added a test test("column-update: requiredDataAttributes throws AnalysisException for invalid column")

dongjoon-hyun · 2026-05-06T00:56:47Z

+  //
+  // ColumnPruning observes exactly these references and narrows the physical scan accordingly.
+  // Connectors that need additional columns in the scan (e.g., partition columns for
+  // distribution) should declare them in requiredDataAttributes().


IIUC, for the correctness, we need to throw AnalysisException if requiredDataAttributes is invalid.

Each column the connector returns passes through V2ExpressionUtils.resolveRefs which throws AnalysisException if the column is non existent.

I added a test test("column-update: requiredDataAttributes throws AnalysisException for invalid column")

dongjoon-hyun · 2026-05-06T00:58:39Z

+  // Connectors that need additional columns in the scan (e.g., partition columns for
+  // distribution) should declare them in requiredDataAttributes().
+  //
+  // Note: AlignUpdateAssignments guarantees all assignment keys are top-level


Do we have a test coverage for this, AlignUpdateAssignments contract?

I added a new test test("column-update: nested struct field update narrows to the root struct column") that updates an inner field in a struct, the AlignUpdateAssignment returns only the root key.

dongjoon-hyun · 2026-05-06T00:59:09Z

+   * whether pk is already in the updated columns list and, if not, add it to
+   * requiredDataAttributes().
+   *
+   * @since 4.2.0


dongjoon-hyun · 2026-05-06T01:01:27Z

-    // build a plan to replace read groups in the table
    val writeRelation = relation.copy(table = operationTable)
-    val projections = buildReplaceDataProjections(query, relation.output, metadataAttrs)
+    val query = updatedAndRemainingRowsPlan


This looks like duplications: Let's use one variable instead of mixing two variables, updatedAndRemainingRowsPlan and query.

Done, used a single variable

dongjoon-hyun · 2026-05-06T01:03:37Z

+    //   GroupBasedRowLevelOperationScanPlanning needs explicit column declarations to narrow.
+    val rowAttrs: Seq[Attribute] = if (isNarrow) connectorDataAttrs else relation.output
+
+    (readRelation, rowAttrs)


Please return metadataAttrs too to avoid the following recomputation in the caller-side.

val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)

I changed this to return metadataAttrs too.

dongjoon-hyun · 2026-05-06T01:04:29Z

+  //
+  // Works for both the full-scan and narrow-scan CoW paths.  In the narrow case,
+  // readRelation.output is already restricted by buildCoWReadSetup, so projecting
+  // all plan.output gives the correct narrow write schema.


Use function description style.

dongjoon-hyun · 2026-05-06T01:07:04Z

+   *
+   * @since 4.2.0
+   */
+  default boolean supportsColumnUpdates() {


Given the scope of this PR, shall we mention that DELETE and MERGE ignores this method?

dongjoon-hyun · 2026-05-06T01:07:18Z

+   *
+   * @since 4.2.0
+   */
+  default NamedReference[] requiredDataAttributes() {


Given the scope of this PR, shall we mention that DELETE and MERGE ignores this method?

Even though the scope of this PR is UPDATE only, we'd like this API to work for MERGE as well (DELETE doesn't benefit since it doesn't write data columns). I'm still assessing what it takes and will add a section in the SPIP on how it could be implemented.

Happy to add a "currently only consulted for UPDATE" note in the Javadoc for now and remove it when MERGE support lands.

dongjoon-hyun · 2026-05-06T01:08:26Z

I finished the first round review, @anuragmantri .

anuragmantri

Thanks for the review @dongjoon-hyun. I addressed your comments and cleaned up some AI generated comments which were redundant.

anuragmantri · 2026-05-06T17:53:45Z

    return new NamedReference[0];
  }
+
+


anuragmantri · 2026-05-06T17:53:50Z

+   * including the columns being updated. If {@link #requiredDataAttributes()} returns an empty
+   * array, Spark sends only the non-identity assigned columns (heuristic path).
+   *
+   * @since 4.2.0


anuragmantri · 2026-05-06T22:50:19Z

+   * is ignored and the full table row is sent (the default behavior).
+   * <p>
+   * When non-empty, the returned columns become the write schema in declared order.
+   * The connector must declare all columns it wants to receive, including the columns being


Each column the connector returns passes through V2ExpressionUtils.resolveRefs which throws AnalysisException if the column is non existent.

I added a test test("column-update: requiredDataAttributes throws AnalysisException for invalid column")

anuragmantri · 2026-05-06T22:51:55Z

+   * whether pk is already in the updated columns list and, if not, add it to
+   * requiredDataAttributes().
+   *
+   * @since 4.2.0


anuragmantri · 2026-05-06T22:52:05Z

+  //
+  // When dataAttrs is non-empty, the relation output is narrowed to include only columns
+  // required for a column-update write. When dataAttrs is empty, the full relation.output is
+  // preserved.


anuragmantri · 2026-05-06T23:48:12Z

+  // Connectors that need additional columns in the scan (e.g., partition columns for
+  // distribution) should declare them in requiredDataAttributes().
+  //
+  // Note: AlignUpdateAssignments guarantees all assignment keys are top-level


I added a new test test("column-update: nested struct field update narrows to the root struct column") that updates an inner field in a struct, the AlignUpdateAssignment returns only the root key.

anuragmantri · 2026-05-06T23:50:05Z

+  //
+  // ColumnPruning observes exactly these references and narrows the physical scan accordingly.
+  // Connectors that need additional columns in the scan (e.g., partition columns for
+  // distribution) should declare them in requiredDataAttributes().


Each column the connector returns passes through V2ExpressionUtils.resolveRefs which throws AnalysisException if the column is non existent.

I added a test test("column-update: requiredDataAttributes throws AnalysisException for invalid column")

anuragmantri · 2026-05-07T00:29:51Z

+      dataAttrsResolved(inRowAttrs)
+  }
+
+  // Validates the narrow-write-schema row projection output.


anuragmantri · 2026-05-07T00:30:56Z

-    table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs)
+    table.skipSchemaResolution ||
+      areCompatible(inRowAttrs, outRowAttrs) ||
+      dataAttrsResolved(inRowAttrs)


anuragmantri · 2026-05-07T01:00:12Z

+   *
+   * @since 4.2.0
+   */
+  default NamedReference[] requiredDataAttributes() {


Even though the scope of this PR is UPDATE only, we'd like this API to work for MERGE as well (DELETE doesn't benefit since it doesn't write data columns). I'm still assessing what it takes and will add a section in the SPIP on how it could be implemented.

Happy to add a "currently only consulted for UPDATE" note in the Javadoc for now and remove it when MERGE support lands.

anuragmantri · 2026-05-07T01:15:18Z

-        .getOrElse {
-          throw new AnalysisException(
-            errorClass = "_LEGACY_ERROR_TEMP_3075",
-            messageParameters = Map(
-              "tableAttr" -> tableAttr.toString,
-              "scanAttrs" -> scanAttrs.mkString(",")))
-        }
    }


I believe this is safe because condition-referenced columns are guaranteed to be in the scan. Please correct me if I'm wrong.

No. Unfortunately, this PR should not remove this because the existing sanity check is used for other code path in the existing test cases. Please recover it.

I guess you may achieve your goal via the following. Please review and revise the following example for your purpose.

private def buildTableToScanAttrMap( tableAttrs: Seq[Attribute], scanAttrs: Seq[Attribute], requiredAttrs: AttributeSet): AttributeMap[Attribute] = { // Table attrs may be legitimately absent from a column-update narrowed scan, so map only // those that have a matching scan attribute. Attrs referenced by the condition must always // be present (computeNarrowReadAttrs keeps them in the scan); failing to map one would // leave a dangling reference in the group filter, so keep the strict check for them. val attrMapping = tableAttrs.flatMap { tableAttr => val matched = scanAttrs.find(scanAttr => conf.resolver(scanAttr.name, tableAttr.name)) if (matched.isEmpty && requiredAttrs.contains(tableAttr)) { throw new AnalysisException( errorClass = "_LEGACY_ERROR_TEMP_3075", messageParameters = Map( "tableAttr" -> tableAttr.toString, "scanAttrs" -> scanAttrs.mkString(","))) } matched.map(scanAttr => tableAttr -> scanAttr) } AttributeMap(attrMapping) }

Makes sense. Similar to other changes for column updates paths, I created a conditional method buildNarrowTableToScanAttrMap() which is called only during column updates and throws when any of the condition references are missing. My rationale is that the runtime filtering applies to only the filters so it is sufficient remap the filters only. Let me know if this understanding is incorrect.

dongjoon-hyun

Thank you for updating, @anuragmantri .

BTW, I cannot find the vote for the mentioned SPIP. Does pass the vote officially, @anuragmantri ? For SPIP, we need an official vote result to move forward including merging something, don't we? (cc @huaxingao as the Shepherd of SPARK-56599 JIRA issue)

What changes were proposed in this pull request?

For SPIP: SPARK-56599

cc @aokolnychyi too because RowLevelOperation.java has been never changed since being added 4 years ago via the following.

#36304

anuragmantri · 2026-05-08T13:29:15Z

Thanks for the review @dongjoon-hyun. For the SPIP, we are waiting for a few more maintainers to also review the design as well as the PR before going for a vote.

Addressed.

aokolnychyi · 2026-06-16T20:12:09Z

+   * @since 4.3.0
+   */
+  default NamedReference[] updatedColumns() {
+    return new NamedReference[0];


Hm... Defaulting to an empty array is fragile. Instances of this type are never constructed by connectors. Can Spark always fill the right value so that we don't have to default here?

Do we handle nested column update correctly? The doc must be clear and implementation consistent.

Added docs for nested column updates.

aokolnychyi · 2026-06-16T20:36:01Z

+   * required for a column-update write. When dataAttrs is empty, the full relation.output is
+   * preserved.
+   */
  protected def buildRelationWithAttrs(


Not a big fan of this change, it makes the method fragile. I would consider adding another one instead.
Something like buildNarrowRelationWithAttrs() or something similar.

I added a buildNarrowRelationWithAttrs() version and kept the original unchanged.

aokolnychyi · 2026-06-16T20:38:56Z

-    relation.copy(table = table, output = attrs)
+      rowIdAttrs: Seq[AttributeReference] = Nil,
+      dataAttrs: Seq[AttributeReference] = Nil,
+      cond: Expression = TrueLiteral): DataSourceV2Relation = {


Don't think cond should be passed here, I'd compute prior to calling.

I added a narrow version of this method which has a different implementation.

aokolnychyi · 2026-06-16T20:43:38Z

+    }
+  }
+
+  private def isIdentityAssignment(key: Attribute, value: Expression): Boolean = {


Does this handle nested columns?

Yes, I added a test for this.

aokolnychyi · 2026-06-16T20:48:51Z

+   * plan.output may be narrowed by buildReplaceDataReadRelation, so only columns present in the
+   * plan are projected.
+   */
  private def buildReplaceDataUpdateProjection(


This is fragile too.

Added a new method for narrow projection

aokolnychyi · 2026-06-16T20:59:47Z

+    val updatedAndRemainingRowsPlan = Union(updatedRowsPlan, remainingRowsPlan)

-    // build a plan to replace read groups in the table
    val writeRelation = relation.copy(table = operationTable)


Why don't we want to narrow down the expected schema in the write relation too in this case?

Instead of changing logic in ReplaceData that validates against relation.output?

I thought about narrowing the write relation here. I preferred not to because it works for UPDATE but when we implement MERGE later, this would not work and we would have to have the full write relation because INSERT rows will need it.

Also, if we narrow it here, the writeSchema() and updateSchema() would look the same making updateSchema() redundant for UPDATEs. I would like to have writeSchema() always represent full width and updateSchema() be the narrow width to be consistent.

anuragmantri · 2026-07-14T01:05:09Z

@aokolnychyi - This is ready for another review.

…DSv2

anuragmantri · 2026-07-21T22:57:55Z

CC: @dongjoon-hyun and @peter-toth, since you last reviewed, I changed the following

SupportsColumnUpdate is now a mixin for RowLevelOperations
LogicalWriteInfo.updateSchema(): Optional — narrow row schema for writeUpdate()
DataWriter.writeUpdate(record) and writeUpdate(metadata, record) for writing column update records with narrow schema.

Please take another look. It will be great to have this in by feature freeze for Spark 4.3.

peter-toth

Took another look after the redesign, @anuragmantri -- SupportsColumnUpdates is now an opt-in mix-in, with LogicalWriteInfo.updateSchema() carrying the narrow row shape and DataWriter.writeUpdate(...) as the narrow write channel. Reconstructed it fresh: RewriteUpdateTable narrows the scan to [connector-declared + cond + RHS + partition] refs and builds a separate narrow update projection, while the write relation stays full-width and updateRowProjection validates against the connector-declared set. The UPDATE paths (delta MoR, delta split, CoW) and their test coverage look solid.

Nothing blocking from my end -- the items below are all non-blocking. The one thing I'd like to see closed before merge is @aokolnychyi's open thread on narrowing the write relation: you've explained the full-width choice, but he hasn't re-reviewed since the redesign, so it'd be good to land his sign-off on that direction (finding 1 sits downstream of it).

Non-blocking

info.schema() differs between the CoW and delta paths for the same UPDATE. buildReplaceDataProjections' narrow branch has no empty-outputsForInsert guard, unlike buildWriteDeltaProjections right below it, so a column-update UPDATE builds a full-width rowProjection with -1 column ordinals and reports schema() as the full table -- while the delta path reports an empty schema. Harmless today (never projected for UPDATE) but a latent get(-1) once MERGE/INSERT rows reuse this path. This is downstream of the full-width write-relation choice @aokolnychyi asked about. [inline: RewriteRowLevelCommand.scala:257]
Document the UPDATE-only scope of updatedColumns(). Only RewriteUpdateTable fills it, so DELETE and MERGE report an empty array even though MERGE does update columns -- worth stating on the public method, as @dongjoon-hyun asked earlier. [inline: RowLevelOperationInfo.java:43]

Minor

Javadoc typo. Stray ) and "is" -> "are" in the requiredDataAttributes() doc. [inline: SupportsColumnUpdates.java:40]
Title undersells the change. The PR narrows both the scan and the write, but the title says only "Write schema narrowing" -- suggest "scan and write schema narrowing", matching your own MimaExcludes comment.

peter-toth · 2026-07-22T17:47:53Z

+    val rowProjection = if (updateRowAttrs.nonEmpty) {
+      val outputsForInsert = filterOutputs(outputs,
+        OPERATIONS_WITH_ROW -- Set(UPDATE_OPERATION, COPY_OPERATION))
+      newLazyProjection(plan, outputsForInsert, rowAttrs)


This narrow branch doesn't guard an empty outputsForInsert, unlike buildWriteDeltaProjections right below (lines 294-296). For a column-update UPDATE there are no INSERT/REINSERT rows, so outputsForInsert is empty and rowAttrs is the full relation.output; newLazyProjection then resolves ordinals against the narrow query and returns -1 for every column absent from the narrow scan, and info.schema() ends up as the full table shape -- whereas the delta path reports an empty schema for the same UPDATE.

It's harmless today because this projection is never applied for UPDATE, but the -1 ordinals become a real row.get(-1) once MERGE/INSERT rows flow through this path. Mirroring the WriteDelta guard fixes both the inconsistency and the latent case:

Suggested change

newLazyProjection(plan, outputsForInsert, rowAttrs)

if (outputsForInsert.isEmpty) {

ProjectingInternalRow(StructType(Nil), Nil)

} else {

newLazyProjection(plan, outputsForInsert, rowAttrs)

}

+1 for the above comment.

Thanks for catching this. I mirrored the guard in buildWriteDeltaProjections

peter-toth · 2026-07-22T17:47:53Z

  Command command();
+
+  /**
+   * Returns the columns being updated by this operation.


updatedColumns() is only populated by RewriteUpdateTable; DELETE and MERGE fall through to the default Nil, so a connector running a MERGE sees an empty array even though columns are being updated. Worth stating the UPDATE-only scope on the public method (the note @dongjoon-hyun asked for earlier):

Suggested change

* Returns the columns being updated by this operation.

* Returns the columns being updated by this operation. Currently populated only for UPDATE;

* DELETE and MERGE report an empty array.

peter-toth · 2026-07-22T17:47:54Z

+   * reported by {@link RowLevelOperationInfo#updatedColumns()} plus any columns needed for row
+   * lookup or routing, e.g. a primary key).
+   * <p>
+   * If any of the columns from {@link RowLevelOperationInfo#updatedColumns()}) is


Stray ) after updatedColumns()} and is -> are:

Suggested change

* If any of the columns from {@link RowLevelOperationInfo#updatedColumns()}) is

* If any of the columns from {@link RowLevelOperationInfo#updatedColumns()} are

dongjoon-hyun · 2026-07-22T21:25:30Z

+
+  test("column-update: analysis fails when assignment key is outside requiredDataAttributes") {
+    // Connector declares only [pk] but the user assigns to `id`. We enforce
+    // updatedColumns ⊆ requiredDataAttributes at analysis time (root-column granularity).


Suggested change

// updatedColumns ⊆ requiredDataAttributes at analysis time (root-column granularity).

// updatedColumns is a subset of requiredDataAttributes at analysis time

// (root-column granularity).

Is this auto-generated sentence, @anuragmantri ? AFAIK, Apache Spark CI will fail for non-ASCII characters like this.

Yes, this was auto generated. I rewrote it as per your suggestion.

dongjoon-hyun · 2026-07-22T21:29:21Z

      table: RowLevelOperationTable,
      metadataAttrs: Seq[AttributeReference],
      rowIdAttrs: Seq[AttributeReference] = Nil): DataSourceV2Relation = {
-


Please recover this empty line removal. This removal seems that a leftover when you recover the original method. @aokolnychyi 's request was to keep the original method unchanged completely.

dongjoon-hyun · 2026-07-22T21:34:46Z

+            } else {
+              info.schema()
+            }
+            PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, info.schema())


Suggested change

PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, info.schema())

// info.schema() is empty for column-update UPDATE writes (no INSERT rows);

// use the table schema to align any INSERT-tagged rows.

PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, schema)

dongjoon-hyun · 2026-07-22T21:39:19Z

      val outputsWithRow = filterOutputs(outputs, OPERATIONS_WITH_ROW)
      Some(newLazyProjection(plan, outputsWithRow, rowAttrs))
+    } else {
+      Some(ProjectingInternalRow(StructType(Nil), Nil))


Could we keep the last branch as None and adjust the comment?

Suggested change

Some(ProjectingInternalRow(StructType(Nil), Nil))

None

This else branch is unreachable for the column-update path, so the change here is an unnecessary behavior change to DELETE:

The comment says the Some(...) is needed for identity-only column updates, but that
case never reaches this branch: an identity-only UPDATE on a SupportsColumnUpdates
connector still has non-empty connectorDataAttrs (an empty requiredDataAttributes()
is already rejected with EMPTY_REQUIRED_DATA_ATTRIBUTES), so it always takes the first
branch (updateRowAttrs.nonEmpty).

The only caller that reaches this branch is RewriteDeleteFromTable, which passes
rowAttrs = Nil and no updateRowAttrs. For DELETE, this changes rowProjection from
None to Some(empty projection). The current consumers happen to be equivalent
(V2Writes uses .map(_.schema).getOrElse(StructType(Nil)) and DeltaWritingSparkTask
uses .orNull), so nothing breaks today, but it silently alters the DELETE plan shape
for no benefit and widens the blast radius of this PR.

Reverting this branch to None keeps the DELETE path byte-for-byte identical to master.

Thanks for the explanation. I moved it back to None and updated the comment.

dongjoon-hyun · 2026-07-22T21:53:30Z

-          val table = buildOperationTable(tbl, UPDATE, r.options)
+          val updatedCols = assignments.collect {
+            case Assignment(key: AttributeReference, value) if !isIdentityAssignment(key, value) =>
+              FieldReference(key.name)


FieldReference(key.name) goes through the parser — FieldReference.apply(String) is
LogicalExpressions.parseReference(column) — rather than wrapping the name as-is. So a
top-level column whose name contains a dot (e.g. CREATE TABLE t (`a.b` INT, ...)) gets
reported to the connector as a nested reference a.b instead of a single column named
a.b, and names containing backticks can even fail to parse.

Since key is an already-resolved AttributeReference at this point, there is no reason
to re-parse its name. This matters because the value flows into
RowLevelOperationInfo.updatedColumns(), which connectors use to size
requiredDataAttributes(), and it also feeds validateUpdatedColumnsSubset — a
mis-parsed reference would surface as a spurious validation failure or a wrong narrow
schema.

Suggested change

FieldReference(key.name)

FieldReference(Seq(key.name))

Thanks for the pointer. Done.

dongjoon-hyun

I finished another round of reviews. Please address the above review comments, @anuragmantri .

BTW, please avoid pinging someone who is irrelevant this PR.

anuragmantri · 2026-07-23T06:12:31Z

Thanks for the reviews @dongjoon-hyun and @peter-toth. I addressed your comments.

peter-toth

Thanks for addressing the review, @anuragmantri -- re-checked through 791bc516be4; my findings 1-4 are all resolved (empty outputsForInsert guard now mirrored in buildReplaceDataProjections, with the test connector aligning INSERT-tagged rows to the table schema; updatedColumns() UPDATE-only scope documented; the SupportsColumnUpdates javadoc typo; and the title now covers scan + write). I also read the code that changed since my last round -- the new buildNarrowTableToScanAttrMap split and the Some(empty) -> None restore in the delta row projection both look correct (the None matches master's DELETE behavior, and the narrow group-filter path is exercised by the "runtime group filtering data correctness" test). Nothing new from me.

anuragmantri force-pushed the dsv2-required-data-attrs branch from fb14c34 to ae635f4 Compare April 23, 2026 20:51

peter-toth reviewed Apr 27, 2026

View reviewed changes

dongjoon-hyun reviewed May 5, 2026

View reviewed changes

anuragmantri force-pushed the dsv2-required-data-attrs branch from ae635f4 to a99bb2d Compare May 5, 2026 23:32

dongjoon-hyun reviewed May 6, 2026

View reviewed changes

dongjoon-hyun previously requested changes May 6, 2026

View reviewed changes

dongjoon-hyun reviewed May 6, 2026

View reviewed changes

anuragmantri commented May 7, 2026

View reviewed changes

dongjoon-hyun reviewed May 8, 2026

View reviewed changes

anuragmantri force-pushed the dsv2-required-data-attrs branch from e806004 to 4060cbf Compare May 29, 2026 06:44

aokolnychyi reviewed Jun 16, 2026

View reviewed changes

anuragmantri force-pushed the dsv2-required-data-attrs branch from 4060cbf to 1aed4f3 Compare June 16, 2026 23:16

anuragmantri force-pushed the dsv2-required-data-attrs branch 2 times, most recently from 06c482e to ac4474b Compare July 14, 2026 00:25

anuragmantri changed the title ~~[SPARK-56599][SQL] Add scan narrowing for column-level UPDATEs in DSv2~~ [SPARK-58111][SQL] Write schema narrowing for column-level UPDATE in DSv2 Jul 14, 2026

anuragmantri added 6 commits July 21, 2026 15:43

[SPARK-58111][SQL] Write schema narrowing for column-level UPDATE in …

f530540

…DSv2

Add new replace data narrow projection

dcbcb8e

Add tests for nested column updates

be05544

Update comments

9aee23e

Add mima excludes

5b9caf5

Scala fmt

3fa6d10

anuragmantri force-pushed the dsv2-required-data-attrs branch from 362e2ef to 3fa6d10 Compare July 21, 2026 22:54

peter-toth reviewed Jul 22, 2026

View reviewed changes

dongjoon-hyun reviewed Jul 22, 2026

View reviewed changes

dongjoon-hyun requested changes Jul 22, 2026

View reviewed changes

Address review comments

791bc51

anuragmantri changed the title ~~[SPARK-58111][SQL] Write schema narrowing for column-level UPDATE in DSv2~~ [SPARK-58111][SQL] Scan and write schema narrowing for column-level UPDATE in DSv2 Jul 23, 2026

peter-toth reviewed Jul 23, 2026

View reviewed changes

	* Returns the columns being updated by this operation.
	* Returns the columns being updated by this operation. Currently populated only for UPDATE;
	* DELETE and MERGE report an empty array.

	* If any of the columns from {@link RowLevelOperationInfo#updatedColumns()}) is
	* If any of the columns from {@link RowLevelOperationInfo#updatedColumns()} are

	// updatedColumns ⊆ requiredDataAttributes at analysis time (root-column granularity).
	// updatedColumns is a subset of requiredDataAttributes at analysis time
	// (root-column granularity).

-            PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, info.schema())
+            // info.schema() is empty for column-update UPDATE writes (no INSERT rows);
+            // use the table schema to align any INSERT-tagged rows.
+            PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, schema)

Uh oh!

Conversation

anuragmantri commented Apr 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

anuragmantri commented May 5, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun May 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anuragmantri commented Apr 23, 2026 •

edited

Loading

dongjoon-hyun May 6, 2026 •

edited

Loading

anuragmantri left a comment •

edited

Loading