[
https://issues.apache.org/jira/browse/SPARK-59642?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59642:
-----------------------------------
Labels: pull-request-available (was: )
> Validate the arity of connector-reported partition keys
> -------------------------------------------------------
>
> Key: SPARK-59642
> URL: https://issues.apache.org/jira/browse/SPARK-59642
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 5.0.0
> Reporter: Tim Meehan
> Priority: Major
> Labels: pull-request-available
>
> h3. Problem
> A DSv2 scan that implements {{SupportsReportPartitioning}} reports N
> partition expressions, and
> each of its input partitions reports a key row through
> {{HasPartitionKey.partitionKey()}}. Nothing
> validates that a key row actually holds N fields.
> {{HasPartitionKey}} leaves a key's arity implicit, and nothing downstream
> reads a key defensively --
> Spark indexes it at the arity the scan reported. So an inconsistent key does
> not fail where it is
> produced. It fails later, in interpreted or generated code that names neither
> the data source nor
> the contract it broke, and the symptom depends on the shape of the key:
> || reported key || behaviour on master ||
> | too wide, >= 2 partitions | {{ArrayIndexOutOfBoundsException}} from
> {{Murmur3HashFunction}} ({{hash.scala:817}}), via the {{distinct}} in
> {{KeyedPartitioning.apply}} |
> | too narrow, ties with another key on its leading fields |
> {{ArrayIndexOutOfBoundsException}} from the generated ordering, in the sort
> in {{DataSourceV2ScanExecBase}} |
> | too narrow, no tie | accepted; fails at whichever positional read comes
> first, or groups partitions wrong |
> | too wide, exactly 1 partition | accepted ({{distinct}} short-circuits);
> later {{AssertionError: assertion failed}}, or the extra field is silently
> ignored |
> h3. Reproduction
> One reported partition expression, a second key row with two fields:
> {code:scala}
> KeyedPartitioning(Seq($"a".int), Seq(InternalRow(1), InternalRow(2, 99)))
> {code}
> On master, via the {{distinct}} in {{KeyedPartitioning.apply}}:
> {code}
> java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds for length 1
> at
> org.apache.spark.sql.catalyst.expressions.InterpretedHashFunction.hash(hash.scala:817)
> at
> org.apache.spark.sql.catalyst.util.InternalRowComparableWrapper.hashCode(InternalRowComparableWrapper.scala:74)
> ...
> {code}
> h3. Relation to SPARK-59123
> SPARK-59123 replaced a per-key {{InternalRow.toSeq(dataTypes)}} in
> {{KeyedPartitioning.reduceKeys}}
> with an indexed loop plus a single assert on the head key, and it can look
> like it dropped a per-key
> arity check. It did not: {{GenericInternalRow}} overrides
> {{toSeq(fieldTypes)}} and ignores the
> types, and partition key rows are {{GenericInternalRow}}s, so that assert
> never ran for them. This
> missing validation predates SPARK-59123 and belongs at ingestion rather than
> in the reduction loop.
> h3. Proposed fix
> Validate arity where raw connector key rows first become a
> {{KeyedPartitioning}}: reject any key row
> whose field count differs from the number of reported partition expressions,
> throwing a
> {{SparkException}} that names the contract. Also cover the two callers that
> read raw key rows before
> a partitioning exists -- the sort in {{DataSourceV2ScanExecBase}} and the
> runtime-filter
> re-ingestion path in {{PushDownUtils.replanWithRuntimeFilters}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
