You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "EamesTrinh (via GitHub)" <gi...@apache.org> on 2026/07/24 11:42:06 UTC
[PR] [SPARK-56942][SQL] Support nested field references as row-level opera… [spark]
EamesTrinh opened a new pull request, #57500:
URL: https://github.com/apache/spark/pull/57500
### 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_index` fails analysis: it resolves to
`Alias(GetStructField(...))`, whose `toAttribute` is not an `AttributeReference`, and the
rewrite (which resolves each `rowId()` reference to an `AttributeReference`) throws a
`ClassCastException`. Connectors could therefore only use top-level columns as row IDs.
This PR adds nested row ID support for DELETE, UPDATE, and MERGE:
- Resolves row ID references as named expressions and materializes nested fields as top-level
attributes on the read plan.
- Carries extracted row IDs through projections that generate new attributes.
- Binds extracted row IDs by output position so they do not collide with data columns that
share the same leaf name, and resolves a reference whose head names a metadata column by its
logical name so a same-named user data column cannot shadow it. The rewrite and
`WriteDelta`'s resolution check share this resolution.
A simplified example (expression IDs are illustrative):
**1. Resolve the row IDs**
Suppose `rowId()` returns `id` and `_metadata.row_index`. The rewrite produces:
```scala
ResolvedRowIdRefs(
rowIdAttrs = Seq(id#1, row_index#20),
extractionAliases = Seq(_metadata#10.row_index AS row_index#20),
scanAttrs = Seq(id#1, _metadata#10))
```
The scan reads `id` and `_metadata`, while a `Project` materializes the nested row ID as
`row_index#20`.
**2. Rebind generated attributes**
`Expand` and `MergeRows` create fresh output attributes. If the extracted field becomes
`row_index#31`, the rewrite updates:
```text
Seq(id#1, row_index#20) -> Seq(id#1, row_index#31)
```
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 a `ClassCastException` on the nested reference. Supporting nested references lets such
connectors implement merge-on-read DELETE / UPDATE / MERGE through the standard DSv2
`SupportsDelta` path 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 is `nested.pk` and the nested values are disjoint from a same-named top-level
`pk` so a wrong bind is observable.
- `V2ExpressionUtilsSuite` -- resolving a metadata-rooted reference past a colliding user
data column of the same physical name.
Existing `DeltaBased*` and `GroupBased*` row-level suites continue to pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3648865677
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNestedRowIdTableSuite.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, StringType, StructField, StructType}
+
+abstract class DeltaBasedNestedRowIdTableSuiteBase(splitUpdates: Boolean)
+ extends RowLevelOperationSuiteBase {
+
+ import testImplicits._
+
+ override protected def extraTableProps: java.util.Map[String, String] = {
+ val props = new java.util.HashMap[String, String]()
+ props.put("supports-deltas", "true")
+ props.put("nested-row-id", "true")
+ if (splitUpdates) props.put("split-updates", "true")
+ props
+ }
+
+ // mirror the metadata marker on file-source row IDs
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "zhipengmao-db (via GitHub)" <gi...@apache.org>.
zhipengmao-db commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3656987519
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -60,6 +60,27 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
refs.map(ref => resolveRef[T](ref, plan))
}
+ // resolve metadata-rooted row IDs by logical name to avoid collisions with data columns
+ def resolveRowIdRef(ref: NamedReference, plan: LogicalPlan): NamedExpression = {
+ val parts = ref.fieldNames.toImmutableArraySeq
+ plan.getMetadataAttributeByNameOpt(parts.head) match {
Review Comment:
This resolves the reference against metadata columns first. The original logic in `V2ExpressionUtils.resolveRefs` seems to resolve against the data columns first. Does it change this semantics / behavior ?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala:
##########
@@ -100,16 +100,22 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand {
// resolve all needed attrs (e.g. row ID and any required metadata attrs)
val operation = operationTable.operation.asInstanceOf[SupportsDelta]
- val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val rowIdRefs = resolveRowIdRefs(relation, operation)
+ val rowIdAttrs = rowIdRefs.rowIdAttrs
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
- // construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable, metadataAttrs, rowIdAttrs)
+ // construct a read plan with all required row ID and metadata columns
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs, rowIdRefs.scanAttrs)
+ val readPlan = withExtractedRowIds(readRelation, rowIdRefs.extractionAliases)
// construct a plan that only contains records to delete
- val deletedRowsPlan = Filter(cond, readRelation)
+ val deletedRowsPlan = Filter(cond, readPlan)
val operationType = Alias(Literal(DELETE_OPERATION), OPERATION_COLUMN)()
- val requiredWriteAttrs = nullifyMetadataOnDelete(dedupAttrs(rowIdAttrs ++ metadataAttrs))
+ // preserve row IDs even when they are also required metadata attributes
+ val rowIdExprIds = rowIdAttrs.map(_.exprId).toSet
Review Comment:
I think it is ambiguous what the behavior should be when there are overlaps between `requiredMetadataAttributes` and `rowId`. Not nullifying the rowId columns here will break the contract of `PRESERVE_ON_DELETE=false`. We could either create separate project slots for `rowId` or redefine the contract.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -100,29 +100,108 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
+ protected case class ResolvedRowIdRefs(
+ // row ID attributes used by the write projections
+ rowIdAttrs: Seq[AttributeReference],
+ // aliases that materialize nested row IDs as top-level attributes
+ extractionAliases: Seq[Alias],
+ // scan attributes needed to read or extract the row IDs
+ scanAttrs: Seq[AttributeReference]) {
+
+ lazy val extractionAttrs: Seq[AttributeReference] = extractionAliases.map { alias =>
+ rowIdAttrs.find(_.exprId == alias.exprId).getOrElse {
+ throw SparkException.internalError(s"Cannot find extracted row ID attribute: $alias")
+ }
+ }
+
+ // rebind nested row IDs after Expand or MergeRows generates fresh output attributes
+ // top-level row IDs retain their original attributes for update handling
+ def rebindExtractions(reboundAttrs: Seq[Attribute]): Seq[Attribute] = {
+ if (extractionAttrs.size != reboundAttrs.size) {
+ throw SparkException.internalError(
+ s"Expected ${extractionAttrs.size} extracted row ID attributes, " +
+ s"but found ${reboundAttrs.size}")
+ }
+ val reboundByExprId = extractionAttrs.zip(reboundAttrs).map { case (attr, reboundAttr) =>
+ attr.exprId -> reboundAttr
+ }.toMap
+ rowIdAttrs.map(attr => reboundByExprId.getOrElse(attr.exprId, attr))
+ }
+ }
+
+ protected case class RowDeltaPlan(plan: LogicalPlan, rowIdAttrs: Seq[Attribute])
+
+ // generate fresh output attributes and rebind nested row IDs to the trailing extracted fields
+ protected def buildRowDeltaPlan(
+ baseAttrs: Seq[Attribute],
+ outputs: Seq[Seq[Expression]],
+ rowIdRefs: ResolvedRowIdRefs)(buildPlan: Seq[Attribute] => LogicalPlan): RowDeltaPlan = {
+ val attrs = baseAttrs ++ rowIdRefs.extractionAttrs
+ outputs.find(_.size != attrs.size).foreach { output =>
+ throw SparkException.internalError(
+ s"Expected ${attrs.size} row delta values, but found ${output.size}")
+ }
+ val output = generateExpandOutput(attrs, outputs)
+ // nested row ID attributes are the output suffix
+ val reboundRowIdAttrs = rowIdRefs.rebindExtractions(
+ output.takeRight(rowIdRefs.extractionAttrs.size))
+ RowDeltaPlan(buildPlan(output), reboundRowIdAttrs)
+ }
+
+ private def resolveRefs(
+ relation: DataSourceV2Relation,
+ refs: Seq[NamedReference]): ResolvedRowIdRefs = {
+ val rowIdAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ val extractionAliases = mutable.ArrayBuffer.empty[Alias]
+ val scanAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ refs.foreach { ref =>
+ V2ExpressionUtils.resolveRowIdRef(ref, relation) match {
+ case attr: AttributeReference =>
+ rowIdAttrs += attr
+ scanAttrs += attr
+ case alias: Alias =>
+ extractionAliases += alias
+ scanAttrs ++= alias.references.collect { case ref: AttributeReference => ref }
Review Comment:
`case ref: AttributeReference` shadows the outside `refs.foreach { ref =>`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNestedRowIdTableSuite.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+abstract class DeltaBasedNestedRowIdTableSuiteBase(splitUpdates: Boolean)
+ extends RowLevelOperationSuiteBase {
+
+ import testImplicits._
+
+ override protected def extraTableProps: java.util.Map[String, String] = {
+ val props = new java.util.HashMap[String, String]()
+ props.put("supports-deltas", "true")
+ props.put("nested-row-id", "true")
+ if (splitUpdates) props.put("split-updates", "true")
+ props
+ }
+
+ // the nested row ID pk shares its leaf name with the top-level pk data column, so a name-based
+ // bind would confuse the two; the suite checks that the nested field is used
+ private val tableSchema = StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("nested", StructType(Seq(
+ StructField("pk", IntegerType, nullable = false))),
+ nullable = false),
+ StructField("id", IntegerType),
+ StructField("dep", StringType)))
+
+ // use different top-level and nested PK values to expose name-based binding
+ private val initialRows =
+ """{ "pk": 10, "nested": { "pk": 1 }, "id": 1, "dep": "hr" }
+ |{ "pk": 20, "nested": { "pk": 2 }, "id": 2, "dep": "software" }
+ |{ "pk": 30, "nested": { "pk": 3 }, "id": 3, "dep": "hr" }
+ |""".stripMargin
+
+ private def createNestedRowIdTable(): Unit = {
+ createTable(CatalogV2Util.structTypeToV2Columns(tableSchema))
+ append(tableSchema.toDDL, initialRows)
+ }
+
+ private def checkTable(expected: Seq[Row]): Unit = {
+ checkAnswer(
+ sql(s"SELECT pk, nested.pk, id, dep FROM $tableNameAsString ORDER BY pk"),
+ expected)
+ }
+
+ test("delete with nested row id") {
+ createNestedRowIdTable()
+ // use a multi-value IN to force the row-level delta path instead of metadata-only deleteWhere
+ sql(s"DELETE FROM $tableNameAsString WHERE id IN (1, 100)")
+ checkTable(Seq(
Review Comment:
Could you also verify `checkLastWriteInfo`, `checkLastWriteLog`, `checkDeleteMetrics` as is done in `DeltaBasedDeleteFromTableSuite`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -100,29 +100,108 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
+ protected case class ResolvedRowIdRefs(
+ // row ID attributes used by the write projections
+ rowIdAttrs: Seq[AttributeReference],
+ // aliases that materialize nested row IDs as top-level attributes
+ extractionAliases: Seq[Alias],
+ // scan attributes needed to read or extract the row IDs
+ scanAttrs: Seq[AttributeReference]) {
+
+ lazy val extractionAttrs: Seq[AttributeReference] = extractionAliases.map { alias =>
+ rowIdAttrs.find(_.exprId == alias.exprId).getOrElse {
+ throw SparkException.internalError(s"Cannot find extracted row ID attribute: $alias")
+ }
+ }
+
+ // rebind nested row IDs after Expand or MergeRows generates fresh output attributes
+ // top-level row IDs retain their original attributes for update handling
+ def rebindExtractions(reboundAttrs: Seq[Attribute]): Seq[Attribute] = {
+ if (extractionAttrs.size != reboundAttrs.size) {
+ throw SparkException.internalError(
+ s"Expected ${extractionAttrs.size} extracted row ID attributes, " +
+ s"but found ${reboundAttrs.size}")
+ }
+ val reboundByExprId = extractionAttrs.zip(reboundAttrs).map { case (attr, reboundAttr) =>
+ attr.exprId -> reboundAttr
+ }.toMap
+ rowIdAttrs.map(attr => reboundByExprId.getOrElse(attr.exprId, attr))
+ }
+ }
+
+ protected case class RowDeltaPlan(plan: LogicalPlan, rowIdAttrs: Seq[Attribute])
+
+ // generate fresh output attributes and rebind nested row IDs to the trailing extracted fields
+ protected def buildRowDeltaPlan(
+ baseAttrs: Seq[Attribute],
+ outputs: Seq[Seq[Expression]],
+ rowIdRefs: ResolvedRowIdRefs)(buildPlan: Seq[Attribute] => LogicalPlan): RowDeltaPlan = {
+ val attrs = baseAttrs ++ rowIdRefs.extractionAttrs
+ outputs.find(_.size != attrs.size).foreach { output =>
+ throw SparkException.internalError(
+ s"Expected ${attrs.size} row delta values, but found ${output.size}")
+ }
+ val output = generateExpandOutput(attrs, outputs)
+ // nested row ID attributes are the output suffix
+ val reboundRowIdAttrs = rowIdRefs.rebindExtractions(
+ output.takeRight(rowIdRefs.extractionAttrs.size))
+ RowDeltaPlan(buildPlan(output), reboundRowIdAttrs)
+ }
+
+ private def resolveRefs(
Review Comment:
The method name is very generic while it returns `ResolvedRowIdRefs`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -100,29 +100,108 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
+ protected case class ResolvedRowIdRefs(
+ // row ID attributes used by the write projections
+ rowIdAttrs: Seq[AttributeReference],
+ // aliases that materialize nested row IDs as top-level attributes
+ extractionAliases: Seq[Alias],
+ // scan attributes needed to read or extract the row IDs
+ scanAttrs: Seq[AttributeReference]) {
+
+ lazy val extractionAttrs: Seq[AttributeReference] = extractionAliases.map { alias =>
+ rowIdAttrs.find(_.exprId == alias.exprId).getOrElse {
+ throw SparkException.internalError(s"Cannot find extracted row ID attribute: $alias")
+ }
+ }
+
+ // rebind nested row IDs after Expand or MergeRows generates fresh output attributes
+ // top-level row IDs retain their original attributes for update handling
+ def rebindExtractions(reboundAttrs: Seq[Attribute]): Seq[Attribute] = {
+ if (extractionAttrs.size != reboundAttrs.size) {
+ throw SparkException.internalError(
+ s"Expected ${extractionAttrs.size} extracted row ID attributes, " +
+ s"but found ${reboundAttrs.size}")
+ }
+ val reboundByExprId = extractionAttrs.zip(reboundAttrs).map { case (attr, reboundAttr) =>
+ attr.exprId -> reboundAttr
+ }.toMap
+ rowIdAttrs.map(attr => reboundByExprId.getOrElse(attr.exprId, attr))
+ }
+ }
+
+ protected case class RowDeltaPlan(plan: LogicalPlan, rowIdAttrs: Seq[Attribute])
+
+ // generate fresh output attributes and rebind nested row IDs to the trailing extracted fields
+ protected def buildRowDeltaPlan(
+ baseAttrs: Seq[Attribute],
+ outputs: Seq[Seq[Expression]],
+ rowIdRefs: ResolvedRowIdRefs)(buildPlan: Seq[Attribute] => LogicalPlan): RowDeltaPlan = {
+ val attrs = baseAttrs ++ rowIdRefs.extractionAttrs
+ outputs.find(_.size != attrs.size).foreach { output =>
+ throw SparkException.internalError(
+ s"Expected ${attrs.size} row delta values, but found ${output.size}")
+ }
+ val output = generateExpandOutput(attrs, outputs)
+ // nested row ID attributes are the output suffix
+ val reboundRowIdAttrs = rowIdRefs.rebindExtractions(
+ output.takeRight(rowIdRefs.extractionAttrs.size))
+ RowDeltaPlan(buildPlan(output), reboundRowIdAttrs)
+ }
+
+ private def resolveRefs(
+ relation: DataSourceV2Relation,
+ refs: Seq[NamedReference]): ResolvedRowIdRefs = {
+ val rowIdAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ val extractionAliases = mutable.ArrayBuffer.empty[Alias]
+ val scanAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ refs.foreach { ref =>
+ V2ExpressionUtils.resolveRowIdRef(ref, relation) match {
+ case attr: AttributeReference =>
+ rowIdAttrs += attr
+ scanAttrs += attr
+ case alias: Alias =>
+ extractionAliases += alias
+ scanAttrs ++= alias.references.collect { case ref: AttributeReference => ref }
+ alias.toAttribute match {
+ case attr: AttributeReference => rowIdAttrs += attr
+ case other =>
+ throw SparkException.internalError(s"Row ID reference did not resolve: $other")
+ }
+ case other =>
+ throw SparkException.internalError("Unexpected resolved row-level reference: " + other)
+ }
+ }
+ ResolvedRowIdRefs(
+ rowIdAttrs.toSeq, extractionAliases.toSeq, dedupAttrs(scanAttrs.toSeq))
+ }
+
protected def resolveRequiredMetadataAttrs(
relation: DataSourceV2Relation,
operation: RowLevelOperation): Seq[AttributeReference] = {
-
Review Comment:
nit: revert irrelevant changes.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNestedRowIdTableSuite.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+abstract class DeltaBasedNestedRowIdTableSuiteBase(splitUpdates: Boolean)
Review Comment:
Could you also add e2e tests for data column & metadata column conflicts for each command?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala:
##########
@@ -100,16 +100,22 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand {
// resolve all needed attrs (e.g. row ID and any required metadata attrs)
val operation = operationTable.operation.asInstanceOf[SupportsDelta]
- val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val rowIdRefs = resolveRowIdRefs(relation, operation)
+ val rowIdAttrs = rowIdRefs.rowIdAttrs
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
- // construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable, metadataAttrs, rowIdAttrs)
+ // construct a read plan with all required row ID and metadata columns
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs, rowIdRefs.scanAttrs)
+ val readPlan = withExtractedRowIds(readRelation, rowIdRefs.extractionAliases)
// construct a plan that only contains records to delete
- val deletedRowsPlan = Filter(cond, readRelation)
+ val deletedRowsPlan = Filter(cond, readPlan)
val operationType = Alias(Literal(DELETE_OPERATION), OPERATION_COLUMN)()
- val requiredWriteAttrs = nullifyMetadataOnDelete(dedupAttrs(rowIdAttrs ++ metadataAttrs))
+ // preserve row IDs even when they are also required metadata attributes
+ val rowIdExprIds = rowIdAttrs.map(_.exprId).toSet
+ val metadataOnlyAttrs = metadataAttrs.filterNot(attr => rowIdExprIds.contains(attr.exprId))
+ val requiredWriteAttrs = rowIdAttrs ++ nullifyMetadataOnDelete(metadataOnlyAttrs)
Review Comment:
If `rowId()` returns `a.b` and required metadata contains top-level `b` (`PRESERVE_ON_DELETE=false`), `a.b` is then materialized as a top-level `b`, and nullification gives the metadata attribute a fresh expression ID. Could `newLazyProjection` fail to find the expression id and then fall back to name look up and bind metadata to the row id `b` slot?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3678216346
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala:
##########
@@ -100,16 +100,22 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand {
// resolve all needed attrs (e.g. row ID and any required metadata attrs)
val operation = operationTable.operation.asInstanceOf[SupportsDelta]
- val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val rowIdRefs = resolveRowIdRefs(relation, operation)
+ val rowIdAttrs = rowIdRefs.rowIdAttrs
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
- // construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable, metadataAttrs, rowIdAttrs)
+ // construct a read plan with all required row ID and metadata columns
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs, rowIdRefs.scanAttrs)
+ val readPlan = withExtractedRowIds(readRelation, rowIdRefs.extractionAliases)
// construct a plan that only contains records to delete
- val deletedRowsPlan = Filter(cond, readRelation)
+ val deletedRowsPlan = Filter(cond, readPlan)
val operationType = Alias(Literal(DELETE_OPERATION), OPERATION_COLUMN)()
- val requiredWriteAttrs = nullifyMetadataOnDelete(dedupAttrs(rowIdAttrs ++ metadataAttrs))
+ // preserve row IDs even when they are also required metadata attributes
+ val rowIdExprIds = rowIdAttrs.map(_.exprId).toSet
Review Comment:
I think you make a valid point, but row id must be preserved and non-null since the connector uses it to locate the row, so it can't also be nulled. I.e. row id has to win over `PRESERVE_ON_DELETE`. The old code nulled it (the deduped attr still carried the marker), corrupting the delete. Slots would work too, but I'd rather document this on rowId().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3665507421
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -100,29 +100,108 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
+ protected case class ResolvedRowIdRefs(
+ // row ID attributes used by the write projections
+ rowIdAttrs: Seq[AttributeReference],
+ // aliases that materialize nested row IDs as top-level attributes
+ extractionAliases: Seq[Alias],
+ // scan attributes needed to read or extract the row IDs
+ scanAttrs: Seq[AttributeReference]) {
+
+ lazy val extractionAttrs: Seq[AttributeReference] = extractionAliases.map { alias =>
+ rowIdAttrs.find(_.exprId == alias.exprId).getOrElse {
+ throw SparkException.internalError(s"Cannot find extracted row ID attribute: $alias")
+ }
+ }
+
+ // rebind nested row IDs after Expand or MergeRows generates fresh output attributes
+ // top-level row IDs retain their original attributes for update handling
+ def rebindExtractions(reboundAttrs: Seq[Attribute]): Seq[Attribute] = {
+ if (extractionAttrs.size != reboundAttrs.size) {
+ throw SparkException.internalError(
+ s"Expected ${extractionAttrs.size} extracted row ID attributes, " +
+ s"but found ${reboundAttrs.size}")
+ }
+ val reboundByExprId = extractionAttrs.zip(reboundAttrs).map { case (attr, reboundAttr) =>
+ attr.exprId -> reboundAttr
+ }.toMap
+ rowIdAttrs.map(attr => reboundByExprId.getOrElse(attr.exprId, attr))
+ }
+ }
+
+ protected case class RowDeltaPlan(plan: LogicalPlan, rowIdAttrs: Seq[Attribute])
+
+ // generate fresh output attributes and rebind nested row IDs to the trailing extracted fields
+ protected def buildRowDeltaPlan(
+ baseAttrs: Seq[Attribute],
+ outputs: Seq[Seq[Expression]],
+ rowIdRefs: ResolvedRowIdRefs)(buildPlan: Seq[Attribute] => LogicalPlan): RowDeltaPlan = {
+ val attrs = baseAttrs ++ rowIdRefs.extractionAttrs
+ outputs.find(_.size != attrs.size).foreach { output =>
+ throw SparkException.internalError(
+ s"Expected ${attrs.size} row delta values, but found ${output.size}")
+ }
+ val output = generateExpandOutput(attrs, outputs)
+ // nested row ID attributes are the output suffix
+ val reboundRowIdAttrs = rowIdRefs.rebindExtractions(
+ output.takeRight(rowIdRefs.extractionAttrs.size))
+ RowDeltaPlan(buildPlan(output), reboundRowIdAttrs)
+ }
+
+ private def resolveRefs(
+ relation: DataSourceV2Relation,
+ refs: Seq[NamedReference]): ResolvedRowIdRefs = {
+ val rowIdAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ val extractionAliases = mutable.ArrayBuffer.empty[Alias]
+ val scanAttrs = mutable.ArrayBuffer.empty[AttributeReference]
+ refs.foreach { ref =>
+ V2ExpressionUtils.resolveRowIdRef(ref, relation) match {
+ case attr: AttributeReference =>
+ rowIdAttrs += attr
+ scanAttrs += attr
+ case alias: Alias =>
+ extractionAliases += alias
+ scanAttrs ++= alias.references.collect { case ref: AttributeReference => ref }
Review Comment:
Renamed the inner binding to scanRef
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3665687014
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNestedRowIdTableSuite.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
+
+abstract class DeltaBasedNestedRowIdTableSuiteBase(splitUpdates: Boolean)
+ extends RowLevelOperationSuiteBase {
+
+ import testImplicits._
+
+ override protected def extraTableProps: java.util.Map[String, String] = {
+ val props = new java.util.HashMap[String, String]()
+ props.put("supports-deltas", "true")
+ props.put("nested-row-id", "true")
+ if (splitUpdates) props.put("split-updates", "true")
+ props
+ }
+
+ // the nested row ID pk shares its leaf name with the top-level pk data column, so a name-based
+ // bind would confuse the two; the suite checks that the nested field is used
+ private val tableSchema = StructType(Seq(
+ StructField("pk", IntegerType, nullable = false),
+ StructField("nested", StructType(Seq(
+ StructField("pk", IntegerType, nullable = false))),
+ nullable = false),
+ StructField("id", IntegerType),
+ StructField("dep", StringType)))
+
+ // use different top-level and nested PK values to expose name-based binding
+ private val initialRows =
+ """{ "pk": 10, "nested": { "pk": 1 }, "id": 1, "dep": "hr" }
+ |{ "pk": 20, "nested": { "pk": 2 }, "id": 2, "dep": "software" }
+ |{ "pk": 30, "nested": { "pk": 3 }, "id": 3, "dep": "hr" }
+ |""".stripMargin
+
+ private def createNestedRowIdTable(): Unit = {
+ createTable(CatalogV2Util.structTypeToV2Columns(tableSchema))
+ append(tableSchema.toDDL, initialRows)
+ }
+
+ private def checkTable(expected: Seq[Row]): Unit = {
+ checkAnswer(
+ sql(s"SELECT pk, nested.pk, id, dep FROM $tableNameAsString ORDER BY pk"),
+ expected)
+ }
+
+ test("delete with nested row id") {
+ createNestedRowIdTable()
+ // use a multi-value IN to force the row-level delta path instead of metadata-only deleteWhere
+ sql(s"DELETE FROM $tableNameAsString WHERE id IN (1, 100)")
+ checkTable(Seq(
Review Comment:
Done, except for `checkDeleteMetrics`/`checkUpdateMetrics` which live in `DeleteFromTableSuiteBase`/`UpdateTableSuiteBase`. I can mix these traits in if you really think it's necessary (or otherwise refactor)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3678210926
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala:
##########
@@ -60,6 +60,27 @@ object V2ExpressionUtils extends SQLConfHelper with Logging {
refs.map(ref => resolveRef[T](ref, plan))
}
+ // resolve metadata-rooted row IDs by logical name to avoid collisions with data columns
+ def resolveRowIdRef(ref: NamedReference, plan: LogicalPlan): NamedExpression = {
+ val parts = ref.fieldNames.toImmutableArraySeq
+ plan.getMetadataAttributeByNameOpt(parts.head) match {
Review Comment:
Only metadata-rooted refs resolve metadata-first. `getMetadataAttributeByNameOpt` returns `None` for data columns and top-level row ids, so we fall through to `resolveRef` and existing behavior is unchanged. The metadata-first branch is needed so a nested id like `_metadata.row_index` binds to `_metadata` even when a data column shares that name. `WriteDelta`'s check uses the same helper, so both agree.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3678206200
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala:
##########
@@ -100,16 +100,22 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand {
// resolve all needed attrs (e.g. row ID and any required metadata attrs)
val operation = operationTable.operation.asInstanceOf[SupportsDelta]
- val rowIdAttrs = resolveRowIdAttrs(relation, operation)
+ val rowIdRefs = resolveRowIdRefs(relation, operation)
+ val rowIdAttrs = rowIdRefs.rowIdAttrs
val metadataAttrs = resolveRequiredMetadataAttrs(relation, operation)
- // construct a read relation and include all required metadata columns
- val readRelation = buildRelationWithAttrs(relation, operationTable, metadataAttrs, rowIdAttrs)
+ // construct a read plan with all required row ID and metadata columns
+ val readRelation = buildRelationWithAttrs(
+ relation, operationTable, metadataAttrs, rowIdRefs.scanAttrs)
+ val readPlan = withExtractedRowIds(readRelation, rowIdRefs.extractionAliases)
// construct a plan that only contains records to delete
- val deletedRowsPlan = Filter(cond, readRelation)
+ val deletedRowsPlan = Filter(cond, readPlan)
val operationType = Alias(Literal(DELETE_OPERATION), OPERATION_COLUMN)()
- val requiredWriteAttrs = nullifyMetadataOnDelete(dedupAttrs(rowIdAttrs ++ metadataAttrs))
+ // preserve row IDs even when they are also required metadata attributes
+ val rowIdExprIds = rowIdAttrs.map(_.exprId).toSet
+ val metadataOnlyAttrs = metadataAttrs.filterNot(attr => rowIdExprIds.contains(attr.exprId))
+ val requiredWriteAttrs = rowIdAttrs ++ nullifyMetadataOnDelete(metadataOnlyAttrs)
Review Comment:
You're right, so I added a failing regression (`DeltaBasedNestedRowIdMetadataCollisionSuite`): row id `nested.index` colliding with the `index` metadata column wrote `Row("hr", 1)` instead of `Row("hr", null)`. Cause was `nullifyMetadata` creating a new exprId, causing `findColOrdinalByExprId` to miss and fall back to the name, matching the row-id column. Fix preserves the metadata exprId (using an alias) so the projection binds by id.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3665505483
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -100,29 +100,108 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
}
}
+ protected case class ResolvedRowIdRefs(
+ // row ID attributes used by the write projections
+ rowIdAttrs: Seq[AttributeReference],
+ // aliases that materialize nested row IDs as top-level attributes
+ extractionAliases: Seq[Alias],
+ // scan attributes needed to read or extract the row IDs
+ scanAttrs: Seq[AttributeReference]) {
+
+ lazy val extractionAttrs: Seq[AttributeReference] = extractionAliases.map { alias =>
+ rowIdAttrs.find(_.exprId == alias.exprId).getOrElse {
+ throw SparkException.internalError(s"Cannot find extracted row ID attribute: $alias")
+ }
+ }
+
+ // rebind nested row IDs after Expand or MergeRows generates fresh output attributes
+ // top-level row IDs retain their original attributes for update handling
+ def rebindExtractions(reboundAttrs: Seq[Attribute]): Seq[Attribute] = {
+ if (extractionAttrs.size != reboundAttrs.size) {
+ throw SparkException.internalError(
+ s"Expected ${extractionAttrs.size} extracted row ID attributes, " +
+ s"but found ${reboundAttrs.size}")
+ }
+ val reboundByExprId = extractionAttrs.zip(reboundAttrs).map { case (attr, reboundAttr) =>
+ attr.exprId -> reboundAttr
+ }.toMap
+ rowIdAttrs.map(attr => reboundByExprId.getOrElse(attr.exprId, attr))
+ }
+ }
+
+ protected case class RowDeltaPlan(plan: LogicalPlan, rowIdAttrs: Seq[Attribute])
+
+ // generate fresh output attributes and rebind nested row IDs to the trailing extracted fields
+ protected def buildRowDeltaPlan(
+ baseAttrs: Seq[Attribute],
+ outputs: Seq[Seq[Expression]],
+ rowIdRefs: ResolvedRowIdRefs)(buildPlan: Seq[Attribute] => LogicalPlan): RowDeltaPlan = {
+ val attrs = baseAttrs ++ rowIdRefs.extractionAttrs
+ outputs.find(_.size != attrs.size).foreach { output =>
+ throw SparkException.internalError(
+ s"Expected ${attrs.size} row delta values, but found ${output.size}")
+ }
+ val output = generateExpandOutput(attrs, outputs)
+ // nested row ID attributes are the output suffix
+ val reboundRowIdAttrs = rowIdRefs.rebindExtractions(
+ output.takeRight(rowIdRefs.extractionAttrs.size))
+ RowDeltaPlan(buildPlan(output), reboundRowIdAttrs)
+ }
+
+ private def resolveRefs(
Review Comment:
Renamed to `resolveRowIdRefsToAttrs`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3652280989
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedNestedRowIdTableSuite.scala:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
+import org.apache.spark.sql.connector.catalog.CatalogV2Util
+import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, StringType, StructField, StructType}
+
+abstract class DeltaBasedNestedRowIdTableSuiteBase(splitUpdates: Boolean)
+ extends RowLevelOperationSuiteBase {
+
+ import testImplicits._
+
+ override protected def extraTableProps: java.util.Map[String, String] = {
+ val props = new java.util.HashMap[String, String]()
+ props.put("supports-deltas", "true")
+ props.put("nested-row-id", "true")
+ if (splitUpdates) props.put("split-updates", "true")
+ props
+ }
+
+ // mirror the metadata marker on file-source row IDs
Review Comment:
Good catch, removed it. Corrected the comment to describe what the suite actually covers (the `nested.pk` vs top-level `pk` leaf-name collision); the metadata-name-collision path is covered by `V2ExpressionUtilsSuite`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3645789732
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -279,7 +358,11 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
plan: LogicalPlan,
outputs: Seq[Seq[Expression]],
attrs: Seq[Attribute]): ProjectingInternalRow = {
- val colOrdinals = attrs.map(attr => findColOrdinal(plan, attr.name))
+ val colOrdinals = attrs.map { attr =>
+ val byExprId = findColOrdinalByExprId(plan, attr.exprId)
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56942][SQL] Support nested field references as row-level operation row IDs [spark]
Posted by "EamesTrinh (via GitHub)" <gi...@apache.org>.
EamesTrinh commented on code in PR #57500:
URL: https://github.com/apache/spark/pull/57500#discussion_r3645789732
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -279,7 +358,11 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
plan: LogicalPlan,
outputs: Seq[Seq[Expression]],
attrs: Seq[Attribute]): ProjectingInternalRow = {
- val colOrdinals = attrs.map(attr => findColOrdinal(plan, attr.name))
+ val colOrdinals = attrs.map { attr =>
+ val byExprId = findColOrdinalByExprId(plan, attr.exprId)
Review Comment:
`newLazyProjection` doesn't have to strictly also start using `findColOrdinalByExprId`, because this PR is only targeting 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org