[SPARK-56942][SQL] Support nested field references as row-level operation row IDs#57500
[SPARK-56942][SQL] Support nested field references as row-level operation row IDs#57500EamesTrinh wants to merge 1 commit into
Conversation
…tion row IDs Row-level DELETE/UPDATE/MERGE resolve each SupportsDelta.rowId() reference to an attribute. A nested reference such as _metadata.row_index resolves to an Alias(GetStructField(...)), not an AttributeReference, so it previously failed with a ClassCastException and connectors could only use top-level columns as row IDs. Resolve row-id references as NamedExpression and materialize any nested field as a top-level extraction column on the read plan. Expand and MergeRows regenerate expression IDs, so carry the extraction as a trailing output column and rebind the row-id projection to it by output position. Resolve a reference whose head names a metadata column by its logical name so a user data column of the same physical name cannot shadow it, and share that resolution between the rewrite and WriteDelta's resolution check.
| attrs: Seq[Attribute]): ProjectingInternalRow = { | ||
| val colOrdinals = attrs.map(attr => findColOrdinal(plan, attr.name)) | ||
| val colOrdinals = attrs.map { attr => | ||
| val byExprId = findColOrdinalByExprId(plan, attr.exprId) |
There was a problem hiding this comment.
newLazyProjection doesn't have to strictly also start using findColOrdinalByExprId, since this PR only targets row id columns (this function handles everything except row id columns (i.e. it will handle normal metadata columns too)). However, it seems fairly innocuous to apply the same logic in both places. The fallback to by-name searching is still there.
| props | ||
| } | ||
|
|
||
| // mirror the metadata marker on file-source row IDs |
There was a problem hiding this comment.
nestedPkMetadata sets METADATA_COL_ATTR_KEY on a leaf StructField inside the nested struct. getMetadataAttribute ByNameOpt / MetadataAttributeWithLogicalName inspect only top-level AttributeReference metadata (verified against master), so this marker is never read on any production path; it is dead code, and the comment "mirror the metadata marker on file-source row IDs" is misleading (real file-source markers live on the top-level _metadata attribute, not on inner struct fields). Remove it, or replace it with an accurate comment / a test that actually reaches the metadata path.
What changes were proposed in this pull request?
DataSource V2 row-level operations use connector-provided row ID references to identify affected rows. The rewrite currently assumes row IDs resolve to top-level attributes, so a nested reference such as
_metadata.row_indexfails analysis: it resolves toAlias(GetStructField(...)), whosetoAttributeis not anAttributeReference, and the rewrite (which resolves eachrowId()reference to anAttributeReference) throws aClassCastException. Connectors could therefore only use top-level columns as row IDs.This PR adds nested row ID support for DELETE, UPDATE, and MERGE:
WriteDelta's resolution check share this resolution.A simplified example (expression IDs are illustrative):
1. Resolve the row IDs
Suppose
rowId()returnsidand_metadata.row_index. The rewrite produces:The scan reads
idand_metadata, while aProjectmaterializes the nested row ID asrow_index#20.2. Rebind generated attributes
ExpandandMergeRowscreate fresh output attributes. If the extracted field becomesrow_index#31, the rewrite updates:Only the extracted row ID is rebound. This keeps it distinct from any data column also named
row_index; top-level row IDs retain their existing handling.Required-metadata references remain top-level only; nested support is limited to
rowId(). Top-level row ID behavior is unchanged.Why are the changes needed?
Connectors that identify rows by file-source metadata (e.g.
_metadata.file_path/_metadata.row_index) cannot currently express that as a DSv2 row ID: the rewrite fails with aClassCastExceptionon the nested reference. Supporting nested references lets such connectors implement merge-on-read DELETE, UPDATE, and MERGE through the standard DSv2SupportsDeltapath instead of reserving synthetic top-level columns.Does this PR introduce any user-facing change?
No. This enables a DSv2 connector capability; existing top-level-row-id behavior is unchanged.
How was this patch tested?
New tests, run for both the merge-on-read and delete-and-insert UPDATE encodings:
DeltaBasedNestedRowIdTableSuite/DeltaBasedNestedRowIdUpdateAsDeleteAndInsertTableSuite: DELETE, UPDATE, UPDATE that replaces the row id's containing struct, and MERGE (matched update, matched delete, not-matched-by-source update, not-matched insert), where the row id isnested.pkand the nested values are disjoint from a same-named top-levelpkso a wrong bind is observable.V2ExpressionUtilsSuite: resolving a metadata-rooted reference past a colliding user data column of the same physical name.Existing
DeltaBased*andGroupBased*row-level suites continue to pass.