peter-toth commented on code in PR #55518:
URL: https://github.com/apache/spark/pull/55518#discussion_r3741376788
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -109,6 +120,22 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
relation)
}
+ /**
+ * Resolves the connector-declared required data attributes for
column-update writes against
+ * the given relation. Non-existent columns are rejected with an
`AnalysisException`.
+ */
+ protected def resolveRequiredDataAttrs(
+ relation: DataSourceV2Relation,
+ operation: SupportsColumnUpdates): Seq[AttributeReference] = {
+ val refs = operation.requiredDataAttributes
+ if (refs.isEmpty) {
Review Comment:
**Finding 23.** `resolveRefs[AttributeReference]` casts the result of
`plan.resolve`, which is an `Alias` — not an `AttributeReference` — for any
nested path, so a nested declaration escapes as an unhandled
`ClassCastException`.
Measured on `774412c`: `column-update-req-attrs = "pk,s.c1"` on `pk INT NOT
NULL, s STRUCT<c1: INT, c2: INT>, dep STRING` with `UPDATE t SET s.c1 = 9 WHERE
pk = 1` gives
```
java.lang.ClassCastException: class
org.apache.spark.sql.catalyst.expressions.Alias
cannot be cast to class
org.apache.spark.sql.catalyst.expressions.AttributeReference
```
`SupportsColumnUpdates`' javadoc does say to declare the root column at
root-column granularity, so this is a connector mistake — but it is an easy
one, because `FieldReference(String)` parses dots, so `FieldReference("s.c1")`
produces the two-part path without the author intending it. A one-line guard
next to the existing empty check turns it into a real error:
```scala
val nested = refs.filter(_.fieldNames.length != 1).map(_.describe())
if (nested.nonEmpty) {
throw QueryCompilationErrors.<nestedRequiredDataAttributeError>(
operation.getClass.getName, nested)
}
```
Same treatment for `scanOnlyDataAttributes()` in
`RewriteUpdateTable.scala:472`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala:
##########
@@ -226,4 +421,186 @@ 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)
+ val outputs = Seq(deleteOutput, insertOutput)
+ val operationTypeAttr = AttributeReference(OPERATION_COLUMN, IntegerType,
nullable = false)()
+ val attrs = operationTypeAttr +: matchedRowsPlan.output
+ val expandOutput = generateExpandOutput(attrs, outputs)
+ Expand(outputs, expandOutput, matchedRowsPlan)
+ }
+
+ /**
+ * Resolves the connector's `requiredDataAttributes()` if the operation opts
into column
+ * updates. Returns `Nil` otherwise.
+ */
+ private def resolveConnectorDataAttrs(
+ relation: DataSourceV2Relation,
+ operation: RowLevelOperation): Seq[AttributeReference] = operation match
{
+ case scu: SupportsColumnUpdates => resolveRequiredDataAttrs(relation, scu)
+ case _ => Nil
+ }
+
+ /**
+ * Resolves the connector's `scanOnlyDataAttributes()` if the operation opts
into column
+ * updates. Returns `Nil` otherwise.
+ */
+ private def resolveScanOnlyDataAttrs(
+ relation: DataSourceV2Relation,
+ operation: RowLevelOperation): Seq[AttributeReference] = operation match
{
+ case scu: SupportsColumnUpdates =>
+ V2ExpressionUtils.resolveRefs[AttributeReference](
+ scu.scanOnlyDataAttributes.toImmutableArraySeq, relation)
+ case _ => Nil
+ }
+
+ /**
+ * Computes the narrow set of data columns that must be present in the scan
for a column-update
+ * write: connector-declared attrs (both `requiredDataAttributes()` and
+ * `scanOnlyDataAttributes()`), unioned with any table columns referenced by
non-identity
+ * assignment RHS expressions and the operation condition.
+ */
+ private def computeNarrowReadAttrs(
+ relation: DataSourceV2Relation,
+ connectorDataAttrs: Seq[AttributeReference],
+ scanOnlyDataAttrs: Seq[AttributeReference],
+ assignments: Seq[Assignment],
+ cond: Expression): Seq[AttributeReference] = {
+ val relationSet = relation.outputSet
+ val nonIdentityRhsRefs = assignments.iterator
+ .filterNot(a => a.key.isInstanceOf[Attribute] &&
+ isIdentityAssignment(a.key.asInstanceOf[Attribute], a.value))
+ .flatMap(_.value.references.toSeq)
+ .toSeq
+ val extraRefs = (cond.references.toSeq ++ nonIdentityRhsRefs)
+ .collect { case a: AttributeReference => a }
+ .filter(relationSet.contains)
+ dedupAttrs(connectorDataAttrs ++ scanOnlyDataAttrs ++ extraRefs)
+ }
+
+ /**
+ * Enforces that every column being assigned (non-identity) is present in
the connector-declared
+ * `requiredDataAttributes()`. Comparison is at root-column granularity
+ */
+ private def validateUpdatedColumnsSubset(
+ operation: RowLevelOperation,
+ assignments: Seq[Assignment],
+ connectorDataAttrs: Seq[AttributeReference]): Unit = {
+ val declaredIds = connectorDataAttrs.map(_.exprId).toSet
+ val missing = assignments.collect {
+ case Assignment(key: AttributeReference, value)
+ if !isIdentityAssignment(key, value) &&
!declaredIds.contains(key.exprId) =>
+ key.name
+ }.distinct
+ if (missing.nonEmpty) {
+ throw
QueryCompilationErrors.requiredDataAttributesMissingUpdatedColumnsError(
+ operation.getClass.getName, missing)
+ }
+ }
+
+ /**
+ * Enforces that `requiredDataAttributes()` and `scanOnlyDataAttributes()`
are disjoint.
+ */
+ private def validateNoOverlap(
+ operation: RowLevelOperation,
+ connectorDataAttrs: Seq[AttributeReference],
+ scanOnlyDataAttrs: Seq[AttributeReference]): Unit = {
+ val requiredIds = connectorDataAttrs.map(_.exprId).toSet
+ val overlapping = scanOnlyDataAttrs.collect {
+ case attr if requiredIds.contains(attr.exprId) => attr.name
+ }.distinct
+ if (overlapping.nonEmpty) {
+ throw
QueryCompilationErrors.requiredDataAttributesOverlapScanOnlyAttributesError(
+ operation.getClass.getName, overlapping)
+ }
+ }
+
+ /**
+ * Enforces that every partition-source column is declared in either
`requiredDataAttributes()`
+ * or `scanOnlyDataAttributes()`. Spark no longer adds partition refs to the
narrow scan
+ * implicitly, so a connector that needs them for partitioning resolution or
write-side
+ * clustering must declare them in one of the two methods.
+ */
+ private def validatePartitionAttrsDeclared(
+ operation: RowLevelOperation,
+ relation: DataSourceV2Relation,
+ connectorDataAttrs: Seq[AttributeReference],
+ scanOnlyDataAttrs: Seq[AttributeReference]): Unit = {
+ val partitionRefNames = relation.table.partitioning().toImmutableArraySeq
+ .flatMap(_.references.toImmutableArraySeq)
+ .map(_.fieldNames.head)
+ .toSet
+ val declared = AttributeSet(connectorDataAttrs ++ scanOnlyDataAttrs)
+ val undeclared = relation.output
+ .filter(a => partitionRefNames.exists(name => conf.resolver(name,
a.name)))
+ .filterNot(declared.contains).map(_.name)
+ if (undeclared.nonEmpty) {
+ throw
QueryCompilationErrors.requiredDataAttributesMissingPartitionColumnsError(
+ operation.getClass.getName, undeclared)
+ }
+ }
+
+ /**
+ * For connectors that opt into narrow column updates AND represent UPDATE
as delete + insert,
+ * reject reassignment of any row-ID column as the REINSERT path has no
row-ID channel to
+ * reconstruct columns outside `requiredDataAttributes()`.
+ */
+ private def validateNoRowIdReassignment(
Review Comment:
**Finding 22.** The error message tells the connector two ways out, and the
guard only implements one.
`SPLIT_UPDATE_ROW_ID_REASSIGNMENT` reads: "Either avoid reassigning row ID
columns, **or include every table column in `requiredDataAttributes()`**." But
`validateNoRowIdReassignment` takes `(operation, assignments, rowIdAttrs)` and
throws on any non-identity assignment to a row-ID column — `connectorDataAttrs`
is not a parameter, so the second remedy cannot possibly be honoured. A
connector author who follows it declares every column and still gets rejected.
Two ways to make them agree, and I think the first is right:
- Implement the remedy — skip the check when the declaration already covers
the relation, since then the REINSERT payload from
`buildNarrowDeletesAndInserts` really is the full row with the new row ID, and
the DELETE branch still carries the original row ID via
`newLazyRowIdProjection`:
```scala
val declared = connectorDataAttrs.map(_.exprId).toSet
if (!relation.output.forall(a => declared.contains(a.exprId))) {
// existing reassignment check
}
```
That also matters because a plain `supports-deltas` + `split-updates`
connector *can* do `SET pk = ...` today, so opting into the mix-in currently
costs a capability.
- Or, if the intent is to reject it unconditionally for now, drop the second
clause so the message stops promising it.
I have not measured that the full-declaration case is actually safe end to
end, so treat the first option as a proposal rather than a verified fix — but
the message/code mismatch stands on its own either way.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedColumnUpdateTableSuite.scala:
##########
@@ -0,0 +1,683 @@
+/*
+ * 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 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.
+ */
+class DeltaBasedColumnUpdateTableSuite extends RowLevelOperationSuiteBase {
Review Comment:
**Finding 25.** This suite extends `RowLevelOperationSuiteBase` directly,
and so does `GroupBasedColumnUpdateTableSuite` — but the existing UPDATE
battery lives one level down, in `UpdateTableSuiteBase` (`:31`) and
`DeltaBasedUpdateTableSuiteBase` (`:22`). So none of it runs against the narrow
path.
That matters more here than it would for a smaller change, because
`SupportsColumnUpdates` is not a tweak to the existing rewrite — it is a second
set of builders alongside it: `buildNarrowReplaceDataUpdateProjection`,
`buildColumnUpdateProjection`, `buildNarrowDeletesAndInserts`,
`buildNarrowRelationWithAttrs`, `computeNarrowReadAttrs`, plus
`dataAttrsResolved` / `updateResolved` in `v2Commands.scala`. Everything the
battery covers for the wide path — aliases, `DEFAULT` values, char/varchar
padding, `EXISTS` / `NOT IN` / correlated subquery conditions, transactions,
explain, nested-field assignment — is unexercised against those builders. The
two new suites test the narrowing *mechanism* thoroughly and the UPDATE
*semantics* barely.
The cheap version is to make these two suites extend the corresponding
battery base with the fixture property set, so the existing tests run twice —
once wide, once narrow. Wiring that up locally suggests the large majority pass
as-is, with the failures tracing to fixture limitations rather than the rewrite
(finding 26 is one of them; another is that the fixture's `doCommit`
reconstructs rows by indexing the current schema, which breaks for rows written
before an `ALTER TABLE ADD COLUMN`). Those are worth fixing on the fixture
regardless, since they are what currently makes the battery unreusable.
If that is too much for this PR, it is a reasonable follow-up — but it
should be a named follow-up rather than left implicit, because the narrow
builders are where the next regression will land.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/write/SupportsColumnUpdates.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ * <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.
+ *
+ * @since 4.3.0
+ */
+@Experimental
+public interface SupportsColumnUpdates extends RowLevelOperation {
+ /**
+ * Returns the data column references required to perform this row-level
operation.
+ * <p>
+ * The returned columns become the schema of updated, copied, and reinserted
rows, in declared
+ * order. Implementations must include every column they want to receive
(typically the columns
Review Comment:
**Finding 20.** This says "Implementations must include every column they
want to receive", which reads as permissive — but `RewriteUpdateTable` now
enforces four hard requirements, and none of them are stated on the two methods
that carry them:
1. every column reported by `RowLevelOperationInfo.updatedColumns()`
(`REQUIRED_DATA_ATTRIBUTES_MISSING_UPDATED_COLUMNS`);
2. every row-ID column, when `representUpdateAsDeleteAndInsert()` is `true`
(`SPLIT_UPDATE_ROW_ID_NOT_DECLARED`);
3. no overlap with `scanOnlyDataAttributes()`
(`REQUIRED_DATA_ATTRIBUTES_OVERLAP_SCAN_ONLY_ATTRIBUTES`);
4. every partition-source column, across the two methods
(`REQUIRED_DATA_ATTRIBUTES_MISSING_PARTITION_COLUMNS`).
A connector author currently discovers each one by being rejected.
Requirement 4 in particular is worth a sentence of its own, because it bounds
what the feature can deliver: on a partitioned table the narrow scan always has
to carry the partition columns, so narrowing can never get below them. I
checked that this is a real constraint rather than defensive over-strictness —
with the three `validatePartitionAttrsDeclared` call sites disabled locally,
the same UPDATE fails at write resolution instead, with `AnalysisException:
Unable to resolve dep given [pk,id,_partition,index]`. So the guard is
converting an obscure downstream failure into a clear one, which is the right
call; it just needs to be predictable from the javadoc.
One related doc point while you are here: `DataWriter#writeUpdate`'s javadoc
(`DataWriter.java:113`, `:136`) and `DATA_SOURCE_WRITE_UPDATE_NOT_IMPLEMENTED`
both say a connector mixing in `SupportsColumnUpdates` must override
`writeUpdate`. That holds for group-based operations only — `writeUpdate` is
reached from `ReplaceDataExec`'s writing tasks, so a `SupportsDelta` connector
receives narrow rows through `DeltaWriter.update` / `reinsert` and never sees
`writeUpdate` at all. Worth qualifying both to group-based.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -253,6 +304,382 @@ class InMemoryRowLevelOperationTable private (
}
}
+ // A delta-based operation that supports column-level updates: Spark sends
only the
+ // declared + assigned columns in the row projection instead of the full row
schema. The base
+ // class composes its required-attrs set as `pk` (the row-lookup key) plus
whatever columns
+ // Spark reports as being assigned via
`RowLevelOperationInfo#updatedColumns()`.
+ class DeltaBasedColumnUpdateOperation(
+ command: Command,
+ updatedCols: Seq[NamedReference] = Nil)
+ extends DeltaBasedOperation(command, CaseInsensitiveStringMap.empty())
Review Comment:
**Finding 26.** `DeltaBasedColumnUpdateOperation` hardcodes
`CaseInsensitiveStringMap.empty()` instead of forwarding `info.options`, so
every column-update fixture loses the row-level operation options.
The consequence is invisible in this PR's own suites because they never
assert on options — but it means no column-update test can observe whether the
dynamic-options plumbing from SPARK-57681 still reaches a connector that opts
into narrowing. Pointing the existing dynamic-options tests at this fixture
fails them on a null option value, which is the same blind spot as finding 25
and has the same one-line cause:
```scala
extends DeltaBasedOperation(command, info.options)
```
That needs `info` threaded in (or the options passed at the construction
sites in `newRowLevelOperationBuilder`), which is a small change to the fixture
rather than to the feature.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -109,6 +120,22 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
relation)
}
+ /**
+ * Resolves the connector-declared required data attributes for
column-update writes against
+ * the given relation. Non-existent columns are rejected with an
`AnalysisException`.
+ */
+ protected def resolveRequiredDataAttrs(
+ relation: DataSourceV2Relation,
+ operation: SupportsColumnUpdates): Seq[AttributeReference] = {
+ val refs = operation.requiredDataAttributes
+ if (refs.isEmpty) {
+ throw
QueryCompilationErrors.emptyRequiredDataAttributesError(operation.getClass.getName)
+ }
+ V2ExpressionUtils.resolveRefs[AttributeReference](
Review Comment:
**Finding 21.** A connector's declaration is resolved case-insensitively but
then keeps the *declared* spelling, and that spelling becomes the narrow scan's
output names.
`AttributeSeq.resolve` returns `a.withName(name)` for the requested name,
preserving `exprId` (`expressions/package.scala:308`). So
`resolveRequiredDataAttrs` can hand back an attribute named `PK` for a table
column `pk`. `buildNarrowRelationWithAttrs` (`:98`) then builds the read
relation as `dedupAttrs(dataAttrs ++ rowIdAttrs ++ metadataAttrs)` with the
renamed attrs **first** — and `dedupAttrs` keys on `exprId`, so the renamed
copy wins and the table's own attribute is discarded. Column pruning pushes
those names to the scan, which does not have them.
Measured on `774412c` with the `column-update-req-attrs` fixture, table `pk
INT NOT NULL, salary INT, dep STRING, id INT`, query `UPDATE t SET salary =
salary + 1 WHERE pk = 1` — only the declared letter case differs between arms:
| declared | result |
| --- | --- |
| `pk,salary,dep` | passes; rows `[1,101,hr,10], [2,200,hr,20]`;
`updateSchema = struct<pk,salary,dep>` |
| `PK,salary,dep` | `[INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find
PK#162 in [salary#163,dep#164,_partition#168]` |
| `pk,SALARY,dep` | `[INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find
SALARY#273 in [pk#272,dep#274,_partition#278]` |
An internal error is the wrong response to a connector's declaration either
way, and case-insensitive matching is the behaviour a `NamedReference`-based
API should have — `validatePartitionAttrsDeclared` already uses `conf.resolver`
for exactly this reason.
The pre-existing `buildRelationWithAttrs` (`:88`) is immune only by
accident: it puts `relation.output` first, so `dedupAttrs` drops the renamed
copies.
Fix: after resolving, map each attribute back to the relation's own,
preserving declared order (order matters — `updateSchema()` and `areCompatible`
both depend on it):
```scala
val resolved = V2ExpressionUtils.resolveRefs[AttributeReference](refs,
relation)
val byId = relation.output.map(a => a.exprId -> a).toMap
resolved.map(a => byId.getOrElse(a.exprId,
a).asInstanceOf[AttributeReference])
```
Worth applying at all three resolution sites rather than just this one —
`resolveScanOnlyDataAttrs` (`RewriteUpdateTable.scala:472`) and
`RowLevelWrite.projectedDataAttrs` (`v2Commands.scala:375`) resolve the same
declarations, and `dataAttrsResolved`'s `areCompatible` compares `inAttr.name
== outAttr.name`, so fixing only the read relation would leave the two sides
disagreeing on the name.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteRowLevelCommand.scala:
##########
@@ -109,6 +120,22 @@ trait RewriteRowLevelCommand extends Rule[LogicalPlan] {
relation)
}
+ /**
+ * Resolves the connector-declared required data attributes for
column-update writes against
+ * the given relation. Non-existent columns are rejected with an
`AnalysisException`.
+ */
+ protected def resolveRequiredDataAttrs(
+ relation: DataSourceV2Relation,
+ operation: SupportsColumnUpdates): Seq[AttributeReference] = {
+ val refs = operation.requiredDataAttributes
+ if (refs.isEmpty) {
+ throw
QueryCompilationErrors.emptyRequiredDataAttributesError(operation.getClass.getName)
Review Comment:
**Finding 24.** This checks for empty but not for duplicates, and a
duplicate survives all the way into the public `updateSchema()`.
Measured on `774412c`: `column-update-req-attrs = "pk,pk,salary"` gives
```
updateSchema = Optional[StructType(StructField(pk,IntegerType,false),
StructField(pk,IntegerType,false),
StructField(salary,IntegerType,true))]
```
and the UPDATE succeeds with correct data — the row happens to be written
correctly because the projection resolves ordinals positionally. The hazard is
on the connector side: `updateSchema().fieldNames.zipWithIndex.toMap`, or
anything else keyed by field name, silently collapses two columns into one and
writes the wrong ordinal. A `StructType` with a repeated field name is not
something a connector should have to defend against.
Rejecting it belongs next to the empty check, since both are "the
declaration itself is malformed":
```scala
val dupes = refs.groupBy(_.describe()).filter(_._2.size > 1).keys.toSeq
if (dupes.nonEmpty) {
throw QueryCompilationErrors.<duplicateRequiredDataAttributesError>(
operation.getClass.getName, dupes)
}
```
--
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]
