[SPARK-58111][SQL] Scan and write schema narrowing for column-level UPDATE in DSv2 - #55518
[SPARK-58111][SQL] Scan and write schema narrowing for column-level UPDATE in DSv2#55518anuragmantri wants to merge 9 commits into
Conversation
fb14c34 to
ae635f4
Compare
| 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)) |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Could you resolve the conflicts, @anuragmantri ?
ae635f4 to
a99bb2d
Compare
Thanks. I rebased and fixed the conflicts. |
| return new NamedReference[0]; | ||
| } | ||
|
|
||
|
|
There was a problem hiding this comment.
nit. Remove redundant empty line.
| * 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 |
| * <p> | ||
| * When empty (the default), Spark falls back to sending only the non-identity assigned columns. | ||
| * | ||
| * @since 4.2.0 |
| val table = buildOperationTable(tbl, UPDATE, CaseInsensitiveStringMap.empty()) | ||
| val updatedCols = assignments.collect { | ||
| case Assignment(key: AttributeReference, value) | ||
| if !isIdentityAssignment(key, value) => |
There was a problem hiding this comment.
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) =>| // | ||
| // 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. |
There was a problem hiding this comment.
For function description, please follow the community style like the other code path.
/**
* ...
*/
| // 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. |
There was a problem hiding this comment.
For function description, please follow the community style like the other code path.
/**
* ...
*/
| WriteDelta(writeRelation, cond, rowDeltaPlan, relation, projections, groupFilterCond) | ||
| } | ||
|
|
||
| // Builds the row delta projection for the column update path. |
There was a problem hiding this comment.
For function description, please follow the community style like the other code path.
/**
* ...
*/
| dataAttrsResolved(inRowAttrs) | ||
| } | ||
|
|
||
| // Validates the narrow-write-schema row projection output. |
There was a problem hiding this comment.
For function description, please follow the community style like the other code path.
/**
* ...
*/
| table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs) | ||
| table.skipSchemaResolution || | ||
| areCompatible(inRowAttrs, outRowAttrs) || | ||
| dataAttrsResolved(inRowAttrs) |
There was a problem hiding this comment.
nit. Please minimize the change of existing code as much as possible like the following.
table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs) ||
dataAttrsResolved(inRowAttrs)
| * 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 |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
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")
| // | ||
| // 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(). |
There was a problem hiding this comment.
IIUC, for the correctness, we need to throw AnalysisException if requiredDataAttributes is invalid.
There was a problem hiding this comment.
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")
| // 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 |
There was a problem hiding this comment.
Do we have a test coverage for this, AlignUpdateAssignments contract?
There was a problem hiding this comment.
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.
| * whether pk is already in the updated columns list and, if not, add it to | ||
| * requiredDataAttributes(). | ||
| * | ||
| * @since 4.2.0 |
| // 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 |
There was a problem hiding this comment.
This looks like duplications: Let's use one variable instead of mixing two variables, updatedAndRemainingRowsPlan and query.
There was a problem hiding this comment.
Done, used a single variable
| // GroupBasedRowLevelOperationScanPlanning needs explicit column declarations to narrow. | ||
| val rowAttrs: Seq[Attribute] = if (isNarrow) connectorDataAttrs else relation.output | ||
|
|
||
| (readRelation, rowAttrs) |
There was a problem hiding this comment.
Please return metadataAttrs too to avoid the following recomputation in the caller-side.
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operationTable.operation)
There was a problem hiding this comment.
I changed this to return metadataAttrs too.
| // | ||
| // 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. |
There was a problem hiding this comment.
Use function description style.
| * | ||
| * @since 4.2.0 | ||
| */ | ||
| default boolean supportsColumnUpdates() { |
There was a problem hiding this comment.
Given the scope of this PR, shall we mention that DELETE and MERGE ignores this method?
| * | ||
| * @since 4.2.0 | ||
| */ | ||
| default NamedReference[] requiredDataAttributes() { |
There was a problem hiding this comment.
Given the scope of this PR, shall we mention that DELETE and MERGE ignores this method?
There was a problem hiding this comment.
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.
|
I finished the first round review, @anuragmantri . |
There was a problem hiding this comment.
Thanks for the review @dongjoon-hyun. I addressed your comments and cleaned up some AI generated comments which were redundant.
| return new NamedReference[0]; | ||
| } | ||
|
|
||
|
|
| * 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 |
| * 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 |
There was a problem hiding this comment.
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")
| * whether pk is already in the updated columns list and, if not, add it to | ||
| * requiredDataAttributes(). | ||
| * | ||
| * @since 4.2.0 |
| // | ||
| // 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. |
| // 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 |
There was a problem hiding this comment.
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.
| // | ||
| // 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(). |
There was a problem hiding this comment.
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")
| dataAttrsResolved(inRowAttrs) | ||
| } | ||
|
|
||
| // Validates the narrow-write-schema row projection output. |
| table.skipSchemaResolution || areCompatible(inRowAttrs, outRowAttrs) | ||
| table.skipSchemaResolution || | ||
| areCompatible(inRowAttrs, outRowAttrs) || | ||
| dataAttrsResolved(inRowAttrs) |
| * | ||
| * @since 4.2.0 | ||
| */ | ||
| default NamedReference[] requiredDataAttributes() { |
There was a problem hiding this comment.
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.
| .getOrElse { | ||
| throw new AnalysisException( | ||
| errorClass = "_LEGACY_ERROR_TEMP_3075", | ||
| messageParameters = Map( | ||
| "tableAttr" -> tableAttr.toString, | ||
| "scanAttrs" -> scanAttrs.mkString(","))) | ||
| } | ||
| } |
There was a problem hiding this comment.
I believe this is safe because condition-referenced columns are guaranteed to be in the scan. Please correct me if I'm wrong.
There was a problem hiding this comment.
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)
}There was a problem hiding this comment.
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
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
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.
|
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. |
e806004 to
4060cbf
Compare
| val updatedAndRemainingRowsPlan = Union(updatedRowsPlan, remainingRowsPlan) | ||
|
|
||
| // build a plan to replace read groups in the table | ||
| val writeRelation = relation.copy(table = operationTable) |
There was a problem hiding this comment.
Why don't we want to narrow down the expected schema in the write relation too in this case?
There was a problem hiding this comment.
Instead of changing logic in ReplaceData that validates against relation.output?
There was a problem hiding this comment.
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.
4060cbf to
1aed4f3
Compare
06c482e to
ac4474b
Compare
anuragmantri
changed the title
|
@aokolnychyi - This is ready for another review. |
362e2ef to
3fa6d10
Compare
|
CC: @dongjoon-hyun and @peter-toth, since you last reviewed, I changed the following
Please take another look. It will be great to have this in by feature freeze for Spark 4.3. |
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
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-outputsForInsertguard, unlikebuildWriteDeltaProjectionsright below it, so a column-update UPDATE builds a full-widthrowProjectionwith-1column ordinals and reportsschema()as the full table -- while the delta path reports an empty schema. Harmless today (never projected for UPDATE) but a latentget(-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(). OnlyRewriteUpdateTablefills 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 therequiredDataAttributes()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.
| val rowProjection = if (updateRowAttrs.nonEmpty) { | ||
| val outputsForInsert = filterOutputs(outputs, | ||
| OPERATIONS_WITH_ROW -- Set(UPDATE_OPERATION, COPY_OPERATION)) | ||
| newLazyProjection(plan, outputsForInsert, rowAttrs) |
There was a problem hiding this comment.
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:
| newLazyProjection(plan, outputsForInsert, rowAttrs) | |
| if (outputsForInsert.isEmpty) { | |
| ProjectingInternalRow(StructType(Nil), Nil) | |
| } else { | |
| newLazyProjection(plan, outputsForInsert, rowAttrs) | |
| } |
There was a problem hiding this comment.
+1 for the above comment.
There was a problem hiding this comment.
Thanks for catching this. I mirrored the guard in buildWriteDeltaProjections
| Command command(); | ||
|
|
||
| /** | ||
| * Returns the columns being updated by this operation. |
There was a problem hiding this comment.
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):
| * 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. |
| * 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 |
There was a problem hiding this comment.
Stray ) after updatedColumns()} and is -> are:
| * If any of the columns from {@link RowLevelOperationInfo#updatedColumns()}) is | |
| * If any of the columns from {@link RowLevelOperationInfo#updatedColumns()} are |
|
|
||
| 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). |
There was a problem hiding this comment.
| // updatedColumns ⊆ requiredDataAttributes at analysis time (root-column granularity). | |
| // updatedColumns is a subset of requiredDataAttributes at analysis time | |
| // (root-column granularity). |
There was a problem hiding this comment.
Is this auto-generated sentence, @anuragmantri ? AFAIK, Apache Spark CI will fail for non-ASCII characters like this.
There was a problem hiding this comment.
Yes, this was auto generated. I rewrote it as per your suggestion.
| table: RowLevelOperationTable, | ||
| metadataAttrs: Seq[AttributeReference], | ||
| rowIdAttrs: Seq[AttributeReference] = Nil): DataSourceV2Relation = { | ||
|
|
There was a problem hiding this comment.
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.
| } else { | ||
| info.schema() | ||
| } | ||
| PartitionBasedNarrowReplaceData(configuredScan, narrowSchema, info.schema()) |
There was a problem hiding this comment.
| 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) |
| val outputsWithRow = filterOutputs(outputs, OPERATIONS_WITH_ROW) | ||
| Some(newLazyProjection(plan, outputsWithRow, rowAttrs)) | ||
| } else { | ||
| Some(ProjectingInternalRow(StructType(Nil), Nil)) |
There was a problem hiding this comment.
Could we keep the last branch as None and adjust the comment?
| 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 aSupportsColumnUpdates
connector still has non-emptyconnectorDataAttrs(an emptyrequiredDataAttributes()
is already rejected withEMPTY_REQUIRED_DATA_ATTRIBUTES), so it always takes the first
branch (updateRowAttrs.nonEmpty). - The only caller that reaches this branch is
RewriteDeleteFromTable, which passes
rowAttrs = Niland noupdateRowAttrs. For DELETE, this changesrowProjectionfrom
NonetoSome(empty projection). The current consumers happen to be equivalent
(V2Writesuses.map(_.schema).getOrElse(StructType(Nil))andDeltaWritingSparkTask
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.
There was a problem hiding this comment.
Thanks for the explanation. I moved it back to None and updated the comment.
| val table = buildOperationTable(tbl, UPDATE, r.options) | ||
| val updatedCols = assignments.collect { | ||
| case Assignment(key: AttributeReference, value) if !isIdentityAssignment(key, value) => | ||
| FieldReference(key.name) |
There was a problem hiding this comment.
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.
| FieldReference(key.name) | |
| FieldReference(Seq(key.name)) |
There was a problem hiding this comment.
Thanks for the pointer. Done.
There was a problem hiding this comment.
I finished another round of reviews. Please address the above review comments, @anuragmantri .
BTW, please avoid pinging someone who is irrelevant this PR.
anuragmantri
changed the title
|
Thanks for the reviews @dongjoon-hyun and @peter-toth. I addressed your comments. |
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
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.
791bc51 to
d06a569
Compare
| val extraRefs = (cond.references.toSeq ++ nonIdentityRhsRefs) | ||
| .collect { case a: AttributeReference => a } | ||
| .filter(relationSet.contains) | ||
| val partitionRefNames = relation.table.partitioning().toImmutableArraySeq |
There was a problem hiding this comment.
After speaking to @aokolnychyi, the current unconditional inclusion of partition source columns in the narrow scan is incorrect. It should be connector-driven, not implicit on Spark's side (e.g. only requested when the connector actually needs it for V2ScanPartitioningAndOrdering or write-side clustering).
The tension is that connectors may need a column visible in the scan (for planning/clustering resolution) but not in the write payload. I'm considering three options:
- Add
scanOnlyDataAttributes()to the mixin for columns the connector wants in the narrow scan but not in the write payload. - Use
requiredDataAttributes()for everything. Connectors that need scan-side resolution declare it here, but by contract it then also flows into the write row, putting the burden on the connector to ignore/filter it when it's not being updated. - Use existing
requiredMetadataAttributes()(we would need to expand its scope to allow certain data attributes). These don't flow into the narrow write data row, but they do reach the writer via the metadata slot, which conflicts with requiredMetadataAttributes()'s original scope of synthetic columns only.
I'm leaning towards option 1 as it seems cleanest but I'm open to any other suggestions.
There was a problem hiding this comment.
Finding 9. I hit the concrete failure this thread is about, so here it is with numbers, plus my read on the three options.
A SupportsDelta + SupportsColumnUpdates connector declaring requiredDataAttributes() = [pk, id] on a table partitioned by dep, whose write requires Distributions.clustered(dep) and an ordering on dep:
UPDATE t SET id = -1 WHERE pk = 1fails at V2Writes with
AnalysisException: Unable to resolve dep given [__row_operation,id,pk,_partition,index,pk]
at DistributionAndOrderingUtils$.prepareQuery(DistributionAndOrderingUtils.scala:47)
at V2Writes$$anonfun$apply$1.applyOrElse(V2Writes.scala:126)
The same connector on the CoW path resolves fine. The asymmetry is that buildNarrowReplaceDataUpdateProjection maps over all of plan.output, so the partition columns stay in the ReplaceData write query, while buildColumnUpdateProjection emits only assignedValues ++ connectorPassThroughValues ++ metadata ++ rowIds, so they are gone from the WriteDelta query. computeNarrowReadAttrs' comment says the partition refs are kept "so downstream rules ... can resolve the table's partitioning expressions against the scan output" -- that holds for the scan, but the delta write query then drops them again, which is exactly the hole you describe.
Worth noting for whichever option you pick: prepareQuery resolves against query.output, and updateRowProjection selects the payload out of that query by name. So a column can sit in the write query without entering the write payload -- no updateSchema() change needed. That means option 1 is implementable by adding the scan-only attrs to the Project list in buildColumnUpdateProjection and leaving updateRowAttrs = connectorDataAttrs alone, which is what the CoW path effectively already does.
On the options:
- Option 1 (
scanOnlyDataAttributes()) -- my preference. It is the only one that names the actual distinction (needed for planning, not for the write row), and it keepsupdateSchema()meaning "the columns being written". - Option 2 (fold into
requiredDataAttributes()) --requiredDataAttributes()is the write-row schema today:updateSchema()is exactly that list, andReplaceData/WriteDelta.dataAttrsResolvedvalidates the projection against it position for position. Folding scan-only columns in would push them into the payload and makeupdateSchema()stop meaning that, which also silently re-introduces the reconstruct-or-lose problem for those columns. - Option 3 (widen
requiredMetadataAttributes()) -- those are validated againstprojectedMetadataAttrsand delivered through the metadata row, so data columns would arrive in the metadata slot. That is a larger contract change than option 1 for a smaller gain.
Whichever way it goes, the two paths should end up agreeing -- right now CoW carries the columns and delta does not, and nothing in the tests catches the difference because every test connector clusters on the _partition metadata column rather than a data column.
Small aside in the same method: pk appears twice in that projection (connectorPassThroughValues and rowIdValues both emit it, hence the duplicate in the output list above). It is harmless -- findColOrdinal takes the first and both share an exprId, so resolution does not go ambiguous -- but the second one is dead and could be filtered out.
peter-toth
left a comment
peter-toth
left a comment
There was a problem hiding this comment.
Re-checked from scratch through d06a5697ab1 -- findings 1-4 stay resolved, nothing regressed. This round I built the branch and ran the suites (151 tests green across the UPDATE paths, 265 across DELETE/MERGE), which turned up things my earlier confirmation-only pass didn't look for; those are marked (late catch).
@anuragmantri -- on your new question at RewriteUpdateTable.scala:458: I reproduced the exact failure your option list is about and I'd go with option 1. Details on your thread (finding 9). @aokolnychyi's thread on narrowing the write relation (RewriteUpdateTable.scala:157) is still the open design gate, and your :458 question is the same tension from the other side -- worth settling both together.
Blocking
- 5. Stale
@sinceand MiMa version (new):branch-4.xwas bumped to 4.4.0 on 2026-08-03 ([SPARK-58534][4.X][BUILD]), one day before this head, so@since 4.3.0on all four new API members is no longer reachable --dev/next_version_candidates.pynow prints onlymaster 5.0.0/branch-4.x 4.4.0. [inline:SupportsColumnUpdates.java:28] - 6. Narrow REINSERT rows carry no original row ID, so a
representUpdateAsDeleteAndInsertconnector silently loses columns (late catch):UPDATE t SET pk = pk + 10, salary = -1on a split connector declaring[pk, salary]writesdepas NULL; the same query with narrowing off keeps it. Verified locally against both. [inline:RewriteUpdateTable.scala:417] - 7. Description doesn't match the code (late catch): it says
schema()"continues to carry the full table shape" (it is empty -- the PR's own tests assert that) and that the additions are "all@Evolving" (SupportsColumnUpdatesis@Experimental); the sentence introducing the rewrite is also cut off mid-line. - 8.
DataWriter.writeUpdate's default delegation is only correct when narrowing is a no-op (late catch): Spark callswriteUpdateonly whenupdateRowProjectionexists, i.e. only forSupportsColumnUpdates, so the default hands a narrow row to awrite(...)that expects the full table shape. Make it throw likeLogicalWriteInfo.metadataSchema()does. [inline:DataWriter.java:117] - 9. The delta write projection drops the carry-along columns the narrow scan deliberately keeps (late catch): a MoR connector that clusters or orders the write by a data column it did not declare fails with
Unable to resolve dep given [...]; the CoW path resolves the same connector. This is your:458question -- repro and option recommendation posted as a reply on that thread.
Non-blocking
- 10.
inRowAttrs.isEmptyrelaxes INSERT-row validation for every connector, not just column-update ones (late catch): the short-circuit is unconditional, so any empty row projection now skips theareCompatiblecheck againsttable.output. Gate it on the column-update path. [inline:v2Commands.scala:444] - 11.
SupportsColumnUpdatesnever states its UPDATE-only scope (late catch): MERGE and DELETE ignorerequiredDataAttributes()entirely and leaveupdateSchema()absent -- @dongjoon-hyun asked for this note back in May and finding 2 only coveredupdatedColumns(). [inline:SupportsColumnUpdates.java:26]
Minor
- 12. Stale API reference in the suite scaladoc (late catch): it links
RowLevelOperation#supportsColumnUpdates, removed in the redesign, and points atLogicalWriteInfo.schema()instead ofupdateSchema(). [inline:DeltaBasedColumnUpdateTableSuite.scala:28] - 13.
checkLastScanDataColumnsis never called (late catch). [inline:RowLevelOperationSuiteBase.scala:323] - 14. Unnecessary churn in an unrelated suite (late catch): the original
StructType(table.schema.map { ... })form still passes on this head (verified), and the added comment implies a behaviour change that isn't there. [inline:DeltaBasedUpdateTableSuite.scala:49]
| * receive a narrow row containing only the columns declared via {@link #requiredDataAttributes()} | ||
| * for updated, copied, and reinserted records, instead of the full table row. | ||
| * | ||
| * @since 4.3.0 |
There was a problem hiding this comment.
Finding 5. 4.3.0 is no longer a version this can ship in. branch-4.3 was cut on 2026-05-01 and branch-4.x was bumped to 4.4.0-SNAPSHOT on 2026-08-03 by [SPARK-58534][4.X][BUILD] Change version in branch-4.x to 4.4.0 -- one day before this head. dev/next_version_candidates.py on this checkout prints:
master 5.0.0
branch-4.x 4.4.0
So this should be 4.4.0 (the version it first ships in via branch-4.x), or 5.0.0 if the team decides the abstract-method addition on RowLevelOperationInfo makes it master-only. Same fix needed in three more places:
RowLevelOperationInfo.java:49LogicalWriteInfo.java:73DataWriter.java:98and:115
Two related nits in project/MimaExcludes.scala:64-66: the entry sits in v43excludes, whose header says "Exclude rules for 4.3.x from 4.2.0" -- 4.3.0 does not have updatedColumns(), so the filter belongs in the list for the version that does (branch-4.x has a v44excludes for exactly this; master currently has no v50excludes list of its own, so worth confirming the intended home with a committer). And the comment cites [SPARK-56599], the SPIP umbrella, while every neighbouring entry cites the implementing ticket -- that should be [SPARK-58111].
| } | ||
| } | ||
| val deleteOutput = deltaDeleteOutput(rowAttrs, rowIdAttrs, metadataAttrs) | ||
| val insertOutput = deltaReinsertOutput(reinsertAssignments, metadataAttrs) |
There was a problem hiding this comment.
Finding 6. On this path the connector cannot reconstruct the columns it did not declare once the row ID is itself reassigned, and the result is silent NULLs rather than an error.
Repro -- a representUpdateAsDeleteAndInsert + SupportsColumnUpdates connector declaring requiredDataAttributes() = [pk, salary], table pk INT NOT NULL, salary INT, dep STRING partitioned by dep, rows (1, 100, 'hr') and (2, 200, 'software'):
UPDATE t SET pk = pk + 10, salary = -1 WHERE dep = 'hr'gives [11, -1, null] -- dep is gone. The same query against the same table with narrowing off (supports-deltas + split-updates) gives [11, -1, 'hr'], so this is specific to the narrow path.
Why it can't be fixed connector-side: buildNarrowDeletesAndInserts does build the REINSERT output over every narrow-scan row attr (so dep is in the Expand), but updateRowProjection then selects only connectorDataAttrs by name, so the row reaching writer.reinsert(metadata, row) is [pk_new, salary_new]. Unlike writer.update(metadata, rowId, row), which gets the pre-update value through buildOriginalRowIdValues' __original_row_id_pk column, reinsert has no row-ID channel at all -- and deltaDeleteOutput nullifies every non-row-ID column on the paired DELETE. So nothing Spark hands the connector identifies the row being replaced, and no lookup key can recover dep.
Two ways out, smallest first:
// in buildWriteDeltaPlan, before choosing buildNarrowDeletesAndInserts:
// reject the combination that is provably unreconstructible
if (operation.representUpdateAsDeleteAndInsert &&
assignments.exists(a => rowIdAttrSet.contains(a.key.asInstanceOf[Attribute]) &&
!isIdentityAssignment(a.key.asInstanceOf[Attribute], a.value))) {
throw ... // "column-level updates cannot reassign a row ID column when updates are
// represented as delete + insert"
}or thread buildOriginalRowIdValues(rowIdAttrs, assignments) into deltaReinsertOutput so the original row ID travels with the REINSERT row. The first is contained enough for this PR; the second needs the reinsert contract to grow. Either way this needs a test -- there is currently none for split updates plus a row-ID assignment (column-update split: data correctness only reassigns id).
| * | ||
| * @since 4.3.0 | ||
| */ | ||
| default void writeUpdate(T record) throws IOException { |
There was a problem hiding this comment.
Finding 8. The default is only right in the one case where the feature does nothing.
writeUpdate is reached from exactly two places -- DataAndMetadataWritingSparkTask and DataWithProjectionWritingSparkTask, both built by ReplaceDataExec.writingTask -- and both gate on useWriteUpdate = updateDataProj != null. updateRowProjection is Some only when updateRowAttrs.nonEmpty, which only happens for a SupportsColumnUpdates operation. So every call to writeUpdate passes the narrow projection; the default then forwards that narrow row to write(...), which by its own contract takes a row in LogicalWriteInfo.schema() shape. It happens to be correct only if the connector declared every table column, i.e. if narrowing is a no-op.
A connector that mixes in SupportsColumnUpdates but forgets to override the writer therefore gets positionally misaligned rows with no error. The javadoc says "Implementations must override this method", but nothing enforces it, and LogicalWriteInfo right next door already uses the loud default for exactly this situation (metadataSchema() throws DATA_SOURCE_METADATA_SCHEMA_NOT_IMPLEMENTED). Since no existing connector can receive this call, making it throw is backward compatible:
default void writeUpdate(T record) throws IOException {
throw new SparkUnsupportedOperationException(
"DATA_SOURCE_WRITE_UPDATE_NOT_IMPLEMENTED", Map.of("class", getClass().getName()));
}Same for writeUpdate(T metadata, T record) at :100.
| // connectors, so it validates against `projectedDataAttrs` (the connector-declared narrow | ||
| // set) instead. When the connector does not mix in `SupportsColumnUpdates`, | ||
| // `updateRowProjection` is absent and `updateResolved` is trivially true. | ||
| val insertResolved = table.skipSchemaResolution || inRowAttrs.isEmpty || |
There was a problem hiding this comment.
Finding 10. The inRowAttrs.isEmpty term is needed for the column-update path, where rowProjection is deliberately ProjectingInternalRow(StructType(Nil), Nil), but it is written unconditionally, so it also switches off the areCompatible(inRowAttrs, table.output) check for connectors that never opt in. That check is the only thing validating the INSERT-shaped row projection against the table shape, and silently skipping it on an empty projection is a wider behaviour change than this PR needs -- the same class of thing @dongjoon-hyun pushed back on for the Some(empty) -> None restore in buildWriteDeltaProjections.
Gating it keeps every non-column-update plan byte-identical to master:
val insertResolved = table.skipSchemaResolution ||
(inUpdateAttrs.nonEmpty && inRowAttrs.isEmpty) ||
areCompatible(inRowAttrs, table.output)Same at :558 in WriteDelta.rowAttrsResolved.
While you are in here: updateResolved does not honour table.skipSchemaResolution, so an ACCEPT_ANY_SCHEMA connector that opts into column updates now gets strict validation where master validated nothing. Probably worth folding into the same skipSchemaResolution short-circuit.
| /** | ||
| * A mix-in interface for {@link RowLevelOperation}. Data sources can implement this interface to | ||
| * receive a narrow row containing only the columns declared via {@link #requiredDataAttributes()} | ||
| * for updated, copied, and reinserted records, instead of the full table row. |
There was a problem hiding this comment.
Finding 11. The interface doc reads as if the mix-in applies to any RowLevelOperation, but only RewriteUpdateTable honours it: RewriteDeleteFromTable and RewriteMergeIntoTable call the three-argument buildReplaceDataProjections / four-argument buildWriteDeltaProjections, so updateRowProjection stays None, updateSchema() stays absent, and requiredDataAttributes() is never read. I confirmed a MERGE against a SupportsColumnUpdates connector runs correctly with full-width rows -- so this is not a correctness problem, but a connector reading the doc would reasonably expect narrow rows there.
Finding 2 fixed the equivalent note on updatedColumns(); this is the same point on the mix-in itself, which is what @dongjoon-hyun originally asked for ("shall we mention that DELETE and MERGE ignores this method?" on the old RowLevelOperation.java).
| * for updated, copied, and reinserted records, instead of the full table row. | |
| * for updated, copied, and reinserted records, instead of the full table row. | |
| * <p> | |
| * Currently honored only for UPDATE. DELETE and MERGE ignore this interface: those operations | |
| * receive full-width rows and {@link LogicalWriteInfo#updateSchema()} is absent for them. |
Worth a cross-reference from DeltaWriter#update and DeltaWriter#reinsert too -- their row parameter is documented as "a row with updated values" with no hint that it may follow updateSchema() rather than schema().
|
|
||
| /** | ||
| * Tests for UPDATE statements targeting connectors that return true from | ||
| * [[org.apache.spark.sql.connector.write.RowLevelOperation#supportsColumnUpdates]]. |
There was a problem hiding this comment.
Finding 12. Two stale references from before the mix-in redesign: RowLevelOperation#supportsColumnUpdates no longer exists, and the narrowed schema is reported through updateSchema(), not schema() (which this suite itself asserts is empty).
| * [[org.apache.spark.sql.connector.write.RowLevelOperation#supportsColumnUpdates]]. | |
| * Tests for UPDATE statements targeting connectors that mix in | |
| * [[org.apache.spark.sql.connector.write.SupportsColumnUpdates]]. | |
| * | |
| * When a connector supports column updates, Spark narrows the update-row projection | |
| * (LogicalWriteInfo.updateSchema()) to contain only the declared columns rather than | |
| * the full table row. |
| * Metadata columns are filtered out so tests can focus on data-column narrowing without | ||
| * having to enumerate `_partition` / `index` on every assertion. | ||
| */ | ||
| protected def checkLastScanDataColumns(expectedNames: String*): Unit = { |
There was a problem hiding this comment.
Finding 13. Nothing calls this -- both new suites use checkLastScanExcludes / checkLastScanIncludes, and the comment block above the scan-narrowing tests explains why exact-set matching was avoided. Worth dropping so a later reader does not take it as the intended assertion helper.
| Row(1, -1, "hr") :: Row(2, 2, "software") :: Row(3, 3, "hr") :: Nil) | ||
|
|
||
| // info.schema() reflects the row projection: `id` is non-nullable because the assignment | ||
| // supplies a non-null literal, matching master's projection-derived write schema semantics. |
There was a problem hiding this comment.
Finding 14. These two hunks are not needed. I restored the original StructType(table.schema.map { case attr if attr.name == "id" => attr.copy(nullable = false); case attr => attr }) form on this head (both call sites) and the whole suite still passes -- 62 tests, 0 failures. PK_FIELD and the hardcoded fields are the same values table.schema already produces here, so the rewrite is churn in a suite this PR does not otherwise touch, and it drags a widened org.apache.spark.sql.types import along with it.
The comment is the part I would definitely drop either way: "matching master's projection-derived write schema semantics" reads as if the non-opted-in path changed behaviour, and it didn't.
4 participants
What changes were proposed in this pull request?
For SPIP: SPARK-56599
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):requiredDataAttributes() accordingly.
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:
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:
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.