Tim Meehan created SPARK-59642:
----------------------------------
Summary: 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
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]
