peter-toth commented on code in PR #55518: URL: https://github.com/apache/spark/pull/55518#discussion_r3720993832
########## sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/SupportsColumnUpdates.java: ########## @@ -0,0 +1,47 @@ +/* + * 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.write; + +import org.apache.spark.annotation.Experimental; +import org.apache.spark.sql.connector.expressions.NamedReference; + +/** + * 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. + * + * @since 4.3.0 Review 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:49` - `LogicalWriteInfo.java:73` - `DataWriter.java:98` and `: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]`. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala: ########## @@ -226,4 +389,98 @@ object RewriteUpdateTable extends RewriteRowLevelCommand { val expandOutput = generateExpandOutput(attrs, outputs) Expand(outputs, expandOutput, matchedRowsPlan) } + + /** + * Variant of `buildDeletesAndInserts` for the `SupportsColumnUpdates` narrow-scan path. + * This variant realigns the assignments to one value per surviving rowAttr padding unassigned + * rowAttrs with identity, so the reinsert output arity matches the delete output arity in + * the resulting Expand. + */ + private def buildNarrowDeletesAndInserts( + matchedRowsPlan: LogicalPlan, + assignments: Seq[Assignment], + rowIdAttrs: Seq[Attribute]): Expand = { + + val (metadataAttrs, rowAttrs) = matchedRowsPlan.output.partition { attr => + MetadataAttribute.isValid(attr.metadata) + } + val assignmentMap = AttributeMap(assignments.collect { + case a @ Assignment(key: Attribute, _) => key -> a + }) + val reinsertAssignments = rowAttrs.map { attr => + assignmentMap.get(attr) match { + case Some(a) => a + case None => Assignment(attr, attr) + } + } + val deleteOutput = deltaDeleteOutput(rowAttrs, rowIdAttrs, metadataAttrs) + val insertOutput = deltaReinsertOutput(reinsertAssignments, metadataAttrs) Review 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')`: ```sql 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: ```scala // 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`). ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala: ########## @@ -418,7 +432,29 @@ case class ReplaceData( // validates row projection output is compatible with table attributes private def rowAttrsResolved: Boolean = { val inRowAttrs = DataTypeUtils.toAttributes(projections.rowProjection.schema) - table.skipSchemaResolution || areCompatible(inRowAttrs, table.output) + val inUpdateAttrs = projections.updateRowProjection match { + case Some(projection) => DataTypeUtils.toAttributes(projection.schema) + case None => Nil + } + // `rowProjection` (INSERT-tagged rows) validates against `table.output` -- the full table + // shape. `updateRowProjection` (UPDATE/COPY-tagged rows) is narrow for column-update + // 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 || Review 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: ```scala 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. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala: ########## @@ -267,13 +268,97 @@ abstract class RowLevelOperationSuiteBase protected def checkLastWriteInfo( expectedRowSchema: StructType = new StructType(), expectedRowIdSchema: Option[StructType] = None, - expectedMetadataSchema: Option[StructType] = None): Unit = { + expectedMetadataSchema: Option[StructType] = None, + expectedUpdateSchema: Option[StructType] = None): Unit = { val info = table.lastWriteInfo assert(info.schema == expectedRowSchema, "row schema must match") val actualRowIdSchema = Option(info.rowIdSchema.orElse(null)) assert(actualRowIdSchema == expectedRowIdSchema, "row ID schema must match") val actualMetadataSchema = Option(info.metadataSchema.orElse(null)) assert(actualMetadataSchema == expectedMetadataSchema, "metadata schema must match") + val actualUpdateSchema = Option(info.updateSchema.orElse(null)) + assert(actualUpdateSchema == expectedUpdateSchema, "update schema must match") + } + + protected def getUpdateSummary(): org.apache.spark.sql.connector.write.UpdateSummary = { + catalog.loadTable(ident).asInstanceOf[InMemoryTable] + .commits.last.writeSummary.get + .asInstanceOf[org.apache.spark.sql.connector.write.UpdateSummary] + } + + /** + * Asserts the last UPDATE's write summary metrics. `deltaUpdate` controls the expected + * COPY count: MoR connectors emit only deltas (no COPY rows) so `numCopiedRows` is forced + * to 0; CoW connectors emit COPY rows for unchanged rows in matched groups. + */ + protected def checkUpdateMetrics( + numUpdatedRows: Long, + numCopiedRows: Long, + deltaUpdate: Boolean = false): Unit = { + val summary = getUpdateSummary() + assert(summary.numUpdatedRows() === numUpdatedRows, + s"Expected numUpdatedRows=$numUpdatedRows, got ${summary.numUpdatedRows()}") + val expectedCopied = if (deltaUpdate) 0L else numCopiedRows + assert(summary.numCopiedRows() === expectedCopied, + s"Expected numCopiedRows=$expectedCopied, got ${summary.numCopiedRows()}") + } + + /** + * Asserts that the column names in RowLevelOperationInfo.updatedColumns() received by the + * last operation match exactly the expected set. Order is ignored. + */ + protected def checkLastUpdatedColumns(expectedNames: String*): Unit = { + val actual = table.lastUpdatedColumns.map(_.describe()).toSet + val expected = expectedNames.toSet + assert(actual == expected, + s"updatedColumns mismatch: expected ${expected.mkString("[", ", ", "]")} " + + s"but got ${actual.mkString("[", ", ", "]")}") + } + + /** + * Asserts the set of top-level column names present in the last connector scan schema. + * 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 = { Review 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. ########## sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/SupportsColumnUpdates.java: ########## @@ -0,0 +1,47 @@ +/* + * 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.write; + +import org.apache.spark.annotation.Experimental; +import org.apache.spark.sql.connector.expressions.NamedReference; + +/** + * 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. Review 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`). ```suggestion * 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()`. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedColumnUpdateTableSuite.scala: ########## @@ -0,0 +1,541 @@ +/* + * 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, TableInfo} +import org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity, reference} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType} + +/** + * Tests for UPDATE statements targeting connectors that return true from + * [[org.apache.spark.sql.connector.write.RowLevelOperation#supportsColumnUpdates]]. Review 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). ```suggestion * 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. ``` ########## sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/DataWriter.java: ########## @@ -82,6 +82,42 @@ default void write(T metadata, T record) throws IOException { write(record); } + /** + * Writes one updated, copied, or reinserted record with metadata. + * <p> + * Connectors that mix in {@link SupportsColumnUpdates} receive records here in the schema + * declared by {@link LogicalWriteInfo#updateSchema()}. Implementations must + * override this method when mixing in {@link SupportsColumnUpdates}; the default delegates + * to {@link #write(Object, Object)} so existing connectors are unaffected. + * <p> + * If this method fails (by throwing an exception), {@link #abort()} will be called and this + * data writer is considered to have been failed. + * + * @throws IOException if failure happens during disk/network IO like writing files. + * + * @since 4.3.0 + */ + default void writeUpdate(T metadata, T record) throws IOException { + write(metadata, record); + } + + /** + * Writes one updated, copied, or reinserted record without metadata. + * <p> + * Equivalent to {@link #writeUpdate(Object, Object)} for writers that do not require metadata. + * The default delegates to {@link #write(Object)}. + * <p> + * If this method fails (by throwing an exception), {@link #abort()} will be called and this + * data writer is considered to have been failed. + * + * @throws IOException if failure happens during disk/network IO like writing files. + * + * @since 4.3.0 + */ + default void writeUpdate(T record) throws IOException { Review 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: ```java 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`. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedUpdateTableSuite.scala: ########## @@ -45,11 +45,13 @@ class DeltaBasedUpdateTableSuite extends DeltaBasedUpdateTableSuiteBase { sql(s"SELECT * FROM $tableNameAsString"), 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. Review 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. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
