fix: align nested collection buffer nullability before spilling - #5903
fix: align nested collection buffer nullability before spilling#5903ErikBPF wants to merge 4 commits into
Conversation
Force final aggregation to spill so nested nullability mismatches cannot hide behind shuffle-only spills. Refs apache#5239
Match native collection state so final aggregation can spill structs with required fields. Closes apache#5239
Flat struct coverage misses nullability below array elements.
|
Extended the forced-spill regression to cover arrays of structs as well as structs. Required fields are constructed after the Parquet scan, so the new case exercises recursive nullability beneath an array. Each shape runs both The PR description also records the DataFusion #24767 integration boundary: when upgrading that dependency, re-check outer collection-element nullability against Comet's intermediate buffer schema and rerun this regression. This follow-up does not change the current production declaration. The new array-only case failed with the original strict buffer declaration restored, reporting the expected nested list/struct nullability mismatch. Restoring the fix made the expanded targeted regression pass on both Spark 4.1.3 and Spark 3.5.9, with one test passed per profile and no failures or cancellations. Both profiles passed formatting/style checks; Spark 3.5 used a clean full-reactor build. The native release library was rebuilt from the current PR source. |
Which issue does this PR close?
Closes #5239.
Rationale for this change
Grouped collection aggregation over structs with required fields can fail
when final aggregation spills. Native collection state normalizes nested
fields to nullable, but Comet's intermediate output schema retains the
original required fields. The exchange casts the state back to that strict
schema, and final aggregation subsequently rejects nullable state emitted
for spilling.
What changes are included in this PR?
Use Spark's existing
asNullableconversion for the element type of bothCollectListandCollectSetintermediate buffers. Add one regressioncovering both functions over structs and arrays of structs, with required
fields created after the Parquet scan, Spark result comparison, native
execution, and a positive aggregate spill assertion for every case.
This fixes Comet's intermediate schema declaration independently of
apache/datafusion#24767, which changes upstream Spark accumulator typing.
When upgrading DataFusion to include that change, re-check compatibility
between its non-nullable collection elements and Comet's intermediate
buffer declaration (
containsNull=true), using the forced-spill regression.That declared-type difference is not a demonstrated failure at this PR's
current dependency pin.
How are these changes tested?
The follow-up extends the same regression with
array<struct<required fields>>, keeping native execution, Spark-result comparison, and actualspilling assertions for each shape. An array-only probe with the original
strict buffer declaration restored failed with a nested list/struct schema
nullability mismatch, confirming that the new case exercises the bug.
With the fix restored, the expanded targeted regression passed on both
Spark 4.1.3 and Spark 3.5.9: one test passed per profile, with zero failures
or cancellations. Both shapes execute both collection functions. The
Spark 3.5 run used a clean full-reactor build, and formatting/style checks
passed in both profiles. These runs used the release native library built
from the current PR source. The full aggregate suite was not rerun for
this test-only extension; its earlier results are recorded below.
The original struct-only regression failed on base
f69c4c81b9429e327ea95658530ae4ed4ed19635with a strict-versus-nullablestruct schema mismatch during final aggregation, then passed with this
change using the release native library and Spark 4.1.3.
On the original PR revision, the complete
CometAggregateSuitepassed:99 succeeded, zero failed,
and the two pre-existing metric tests remained ignored. The full Maven
reactor, including Scalastyle and Spotless checks, succeeded.
The original struct-only regression also passed with Spark 3.5.9 / Scala
2.12.18 after a clean
full-reactor build using
-Prelease -Pspark-3.5, including style andformatting checks. Both Spark versions loaded the same release native
library. The full aggregate suite was run on Spark 4.1.3 only.
The first full-suite attempt hit the test unit's 1,024-task limit while
Spark created an AQE thread. Raising that task limit to 4,096 allowed the
suite to finish; the CPU and 32 GiB memory caps were unchanged.