Conversation
tdcmeehan
force-pushed
the
spark-59123-partition-key-shape-clean
branch
from
September 17, 2026 20:31
41d370c to
8e4a630
Compare
… keys Reject a DSv2 partition key row whose field count differs from the number of reported partition expressions, at the KeyedPartitioning construction boundary and on the runtime-filter re-ingestion path. Co-authored-by: Isaac <no-reply@databricks.com>
tdcmeehan
force-pushed
the
spark-59123-partition-key-shape-clean
branch
from
September 18, 2026 16:53
8e4a630 to
20182e1
Compare
The rationale comment above `checkPartitionKeyArity` named only the opaque `ArrayIndexOutOfBoundsException`, which is what a too-narrow key produces. A too-wide key never crashed: the grouped-key ordering and the comparable-wrapper grouping are built over `expressions.length`, so its trailing fields were silently dropped and keys grouped too loosely. Name both directions so the symmetric exact-arity check reads as contract enforcement, not crash cosmetics. Co-authored-by: Isaac <no-reply@databricks.com>
`DataSourceV2ScanExecBase` is the one caller whose `checkPartitionKeyArity` call is load-bearing rather than redundant with `KeyedPartitioning.apply`: it runs before `keys.sorted(groupedKeyRowOrdering(...))` reads every key at the declared positions, so a too-narrow key becomes a `SparkException` instead of an `ArrayIndexOutOfBoundsException` from inside the ordering. The existing tests drive `KeyedPartitioning.apply` and `PushDownUtils.replanWithRuntimeFilters`, so neither fails if that call is removed or moved after the sort. The new test reports two keys that tie on the leading field, which is what forces the ordering to read the short key's missing field. Removing the pre-sort call makes it fail with `ArrayIndexOutOfBoundsException`. Co-authored-by: Isaac <no-reply@databricks.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
A DSv2 scan that implements
SupportsReportPartitioningreports N partition expressions, and eachof its input partitions reports a key row through
HasPartitionKey.partitionKey(). Nothing checksthat a key row actually holds N fields. This adds that check where raw connector key rows first
become a
KeyedPartitioning.KeyedPartitioning.checkPartitionKeyArity(new) rejects any key row whosenumFieldsdiffersfrom the number of reported partition expressions, throwing a
SparkExceptionthat names thecontract -- matching the sibling connector-contract checks already in
replanWithRuntimeFilters.KeyedPartitioning.apply(Seq[Expression], Seq[InternalRow])calls it, so a partitioning builtfrom raw rows carries the guarantee for everything downstream.
DataSourceV2ScanExecBase.reportedKeyedPartitioning, ahead of the sort it runs on the keys --the generated ordering indexes a key at the declared positions, so it has to be checked before
the sort, not by
applyafterwards;PushDownUtils.replanWithRuntimeFilters, on the partitions a source returns fromfilter().Those are re-reported rows that never pass through
apply, so a malformed replacementpartition would otherwise bypass the check entirely.
Why are the changes needed?
HasPartitionKeyleaves 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 it fails in four different ways depending on the shape:
ArrayIndexOutOfBoundsExceptionfromMurmur3HashFunction(hash.scala:817), via thedistinctinKeyedPartitioning.applyArrayIndexOutOfBoundsExceptionfrom the generated ordering, in the sort inDataSourceV2ScanExecBasedistinctshort-circuits); laterAssertionError: assertion failed, or the extra field is silently ignoredReproduction -- one reported partition expression, a second key row with two fields:
On master, via the
distinctinKeyedPartitioning.apply:With this patch:
Relation to SPARK-59123 (#58421). That commit replaced a per-key
InternalRow.toSeq(dataTypes)in
KeyedPartitioning.reduceKeyswith an indexed loop plus a single assert on the head key, and itcan look like it dropped a per-key arity check, since
InternalRow.toSeq(fieldTypes)assertsnumFields == fieldTypes.length. It did not.GenericInternalRowoverridestoSeq(fieldTypes)(
rows.scala:178) and ignores the types entirely, and partition key rows areGenericInternalRows,so that assert never ran for them. The two
reduceKeysbodies agree on every reachable input, andthe new head assert is strictly stricter in the one single-key case. The missing validation is older
than #58421 and belongs at ingestion rather than in the reduction loop, which is why this centralizes
it at
KeyedPartitioningconstruction instead of adding per-key checks back to the loop.Does this PR introduce any user-facing change?
Yes, for a DSv2 connector that reports partition keys inconsistent with the partitioning it
reported. In most shapes the query already fails and only the diagnostic improves: an opaque
ArrayIndexOutOfBoundsExceptionor a bareAssertionErrorbecomes an error that names thecontract, the arity it got, and the expressions it was measured against.
Two shapes are newly rejected rather than newly diagnosed, so a connector relying on them would see
a query that completes today start to fail:
distinctshort-circuits on asingle element, so nothing hashes the row and the extra field is simply never read;
its own width.
Both are contract violations, and the second can mis-group partitions rather than raise anything,
so failing fast at the boundary is the point of the change. No behaviour change for a connector
whose keys match the partitioning it reported.
How was this patch tested?
New tests, both at boundaries a connector actually reaches:
ShuffleSpecSuite, "a reported partition key of a different arity is rejected at construction" --the construction boundary, in the suite that already holds the other
KeyedPartitioningconstruction-rejection tests. Covers a too-wide key (an AIOOBE from the hash on master), a
too-narrow key (accepted on master, the case nothing caught), and that well-formed keys still
build the partitioning.
DataSourceV2CatalystRuntimeFilterSuite-- extended the existing "data source that breaks thepartitioning it reported -> rejected" case with a partition whose key row carries a field the scan
reported no expression for. This is the runtime-filter re-ingestion path, whose rows never pass
through
KeyedPartitioning.apply; on master it fails with an AIOOBE out ofgroupByinreplanWithRuntimeFilters.There is no end-to-end SQL reproducer because the in-tree test connectors cannot express the bug:
InMemoryBaseTableand its relatives derive each partition key from the table's own partitionschema, so a key's arity is structurally tied to the reported partitioning. Reaching the case
end-to-end would mean adding a deliberately malformed connector for no other purpose. The
runtime-filter suite above is the connector-shaped test -- real
InputPartition/HasPartitionKeyinstances through the production replan path -- and it already exists to host exactly this class of
contract violation.
Suites run locally with sbt on JDK 17, all passing:
catalyst:ShuffleSpecSuite,DistributionSuitesql:DataSourceV2CatalystRuntimeFilterSuite,KeyGroupedPartitioningSuite,EnsureRequirementsSuite,ValidateRequirementsSuite,GroupPartitionsExecSuite,ProjectedOrderingAndPartitioningSuite,SparkThrowableSuitedev/scalastylereports 0 errors for every changed module.dev/lint-scala's scalafmt check isscoped to
sql/apiandsql/connect, so it does not cover any file here.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac
This pull request and its description were written by Isaac.