fix: support struct-typed scalar subquery results - #5889
Conversation
github-actions
Bot
added
bug
|
Local validation at
The Spark 4.1.3 logs show the pre-merge red/green comparison: the same two regressions failed against the original base due to unsupported struct results/projection fallback, then passed with the fix. The baseline failures are expected. Logs are attached below. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed 91820220 against de1eb4f8. No verified P1/P2 findings.
Correctness
Previously, the struct returned by Spark's merged scalar subqueries caused the consuming projection to fall back. The recursive support check now admits non-empty structs with distinct field names and supported leaves. Array/map/interval fields, duplicate names, non-default collations and unsupported decimal scales retain fallback.
The one-row, one-column IPC envelope preserves the distinction between a null struct and a present struct containing null fields. Native decoding validates shape, field order, names, nullability and leaf types before restoring planned metadata. Timestamp fields retain UTC microseconds, while timestamp-without-time-zone fields remain unzoned. Spark still enforces scalar cardinality before native evaluation. Each task iterator builds a fresh native plan, so the owned cache stays within that execution; cache contents do not affect expression equality or hashing.
The SQL fixture covers supported leaves, nested nulls, empty results, separate subqueries and recursive fallback. The optimizer tests verify actual struct subqueries and a native consuming projection, including the Spark 4.2 union/aggregate case. I inspected the author's attached logs: they report three Spark 4.2 tests and five Rust tests passing at the stated head, plus the pre-merge Spark 4.1 red/green comparison. These are author-provided results, not independent CI. CI remains action_required with zero jobs; the sole successful check is labeling. I ran no local JVM/native tests. Canonical source checks covered the maintained Spark 3.5/4.0 branches; the required 3.4/4.1 branches were unavailable.
Performance
The first evaluation pays for row normalization, Arrow allocation, IPC serialization and decoding. Subsequent evaluations reuse the owned scalar, avoiding repeated IPC work per batch for that expression. Repeated references can still own separate caches, so this does not establish one transfer per logical subquery. Primitive transfer paths remain unchanged apart from the empty-cache lookup. No benchmark was run and no measured speedup is established. Could you add a focused comparison against the previous Spark fallback, varying struct width and input-batch count, to quantify first-transfer cost and how well the cache amortizes it?
Design
Reusing the existing Arrow writer keeps value transfer aligned with the project's supported primitive representations. The Scala support gate and native wire validation serve separate purposes: the former decides fallback before execution, while the latter rejects a malformed or mismatched result. Metadata restoration checks the value shape rather than introducing casts. The change addresses the struct-transfer seam and explicitly retains the existing malformed-UTF-8 limitation and separate scan/shuffle limitations.
Abstraction & complexity
The implementation adds a focused serializer, decoder and execution-local cache without a new plan protocol or global cache. The explicit equality/hash implementations preserve the original three identity fields while excluding mutable state. Recursive normalization and metadata alignment are limited to the supported struct shape; the added tests exercise those responsibilities directly. I found no additional abstraction or simplification that warrants a pre-merge change.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Follow-up at unchanged 91820220 / base de1eb4f8:
[P2] Regenerate q9’s approved plan for the newly native struct-subquery projection. All five exec jobs fail only check simplified (tpcds-v1.4/q9): 3.4, 3.5, 4.0, 4.1, 4.2. The unchanged golden expects Project [COMET: Unsupported data type: StructType(...)]; each actual plan has CometColumnarToRow → CometProject and reports 38 accelerated operators instead of 37. Please regenerate and commit the affected approved plan using the documented golden-file workflow, then rerun the supported-profile plan-stability checks.
These are actual plan-comparison failures at the exact tested H/B merge. The native producer and JVM consumers share artifact 10307852860 and its verified digest. The merged-struct result/operator test and scalar_subquery_struct.sql pass on all five profiles; the five Rust struct tests also pass. This supports an expected-plan omission, without demonstrating a query-result regression. I replayed the captured plan comparison locally; I did not run Spark/Rust tests or benchmarks locally. Other CI jobs remain pending.
|
Thanks for the review; I've regenerated q9's shared golden file to reflect the native |
| private def supportedStructField(dt: DataType): Boolean = dt match { | ||
| case s: StructType => | ||
| s.nonEmpty && s.fieldNames.distinct.length == s.length && | ||
| s.fields.forall(f => supportedStructField(f.dataType)) | ||
| case BooleanType | ByteType | ShortType | IntegerType | LongType | FloatType | DoubleType | | ||
| StringType | BinaryType | DateType | TimestampType | TimestampNTZType | NullType => | ||
| true | ||
| case d: DecimalType => d.scale >= 0 && d.scale <= d.precision | ||
| case _ => false | ||
| } | ||
|
|
||
| override def getSupportLevel(expr: ScalarSubquery): SupportLevel = { | ||
| val supported = expr.dataType match { | ||
| case s: StructType => supportedStructField(s) | ||
| case dt => supportedDataType(dt) | ||
| } | ||
| if (supported) { | ||
| Compatible() |
There was a problem hiding this comment.
Related to #5025, which centralizes these recursive type checks. Could we coordinate this with that helper instead of adding another independent predicate? It already covers duplicate names, string types, and interval restrictions. This path would still need its struct-only and decimal-scale restrictions preserved.
| // Concurrent first evaluations may both initialize the same immutable result. | ||
| // Failed evaluations are never cached. | ||
| let _ = self.struct_value.set(value.clone()); |
There was a problem hiding this comment.
Can the same native Subquery instance be evaluated concurrently here? Separate Spark tasks have separate instances. Understanding the sharing would help decide whether lazy initialization is needed.
| if let Some(value) = self.struct_value.get() { | ||
| return Ok(ColumnarValue::Scalar(value.clone())); | ||
| } |
There was a problem hiding this comment.
By native physical planning, Spark has materialized and registered the subquery result. Could we resolve it into an immutable scalar there? What requires deferring initialization until evaluate?
| if reader.next().transpose()?.is_some() { | ||
| return internal_err!("Scalar subquery IPC result contains more than one batch"); | ||
| } | ||
| let value = align_struct_metadata(batch.column(0), data_type)?; |
There was a problem hiding this comment.
The Rust test covers restoring metadata. Could we also exercise a struct with Parquet field IDs through the JVM serializer, showing the schema mismatch that occurs without this alignment?
|
|
||
| /// The JVM bridge emits one row with one struct column. Validate the wire shape and type before | ||
| /// creating the scalar; Arrow IPC validation also keeps malformed strings out of native arrays. | ||
| fn decode_struct_result( |
There was a problem hiding this comment.
The JVM writer always emits one row, one column, and one batch. Can these shape checks fail through a supported production path? If not, could we simplify them?
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
I think GetStructField over the new scalar struct only works when the consuming batch has exactly one row.
Subquery::evaluate returns ColumnarValue::Scalar(ScalarValue::Struct(..)) holding a one-row StructArray. The scalar arm of GetStructField::evaluate handles that by returning ColumnarValue::Array(child_with_parent_nulls(&struct_array, self.ordinal)?), and child_with_parent_nulls hands back the child of that one-row struct, so the result is a length-1 Array rather than a scalar. DataFusion then runs it through ColumnarValue::into_array_of_size(batch.num_rows()), which rejects an Array whose length is not the batch row count with Array length 1 does not match expected length N.
So SELECT c1, (SELECT max(x) FROM t2), (SELECT min(x) FROM t2) FROM t1 should fail as soon as t1 produces a batch with more than one row, and that is the main shape MergeScalarSubqueries creates. Would it make sense for the scalar arm of GetStructField::evaluate to return ColumnarValue::Scalar(ScalarValue::try_from_array(&child, 0)?) so the projection broadcasts it instead?
I think the reason the tests do not catch this is that all of them land on one-row batches. CometTestBase runs local[5], so the 4-row INSERT ... VALUES in scalar_subquery_struct.sql becomes four one-row Parquet files and withParquetTable((0 until 5)...) becomes five one-row files. The Spark 4.2 test's projection sits directly over one-row aggregates, and q9 filters reason down to r_reason_sk = 1. Could you add a fixture where the table is written as a single file so the consuming projection sees a batch with many rows, and check that it fails without the GetStructField change?
4 participants
Which issue does this PR close?
Closes #5834.
Rationale for this change
Spark's merged subqueries produce struct-typed scalar results that currently force the consuming projection to fall back to Spark. Spark 4.2's MergeSubplans broadens this to queries without explicit subqueries, potentially taking Union and downstream aggregates off Comet as well.
What changes are included in this PR?
Duplicate-name structs and unsupported field types retain fallback. This PR does not claim to resolve #4949 or the separate native shuffle/Parquet issues in #5605 and #5783. Existing malformed-UTF-8 limitations still apply.
How are these changes tested?
Local validation on macOS arm64:
918202206: native build succeeded; Spark 4.2.0 focused tests passed (3 tests), and Rust scalar-subquery tests passed (5 tests). Scalastyle and Spotless also passed.The tests compare Spark results and assert native operator coverage, including the consuming projection and the Spark 4.2 Union/aggregate reproducer. The latter uses the existing test setup with Spark-to-columnar input conversion enabled.
Clippy and Apache RAT passed locally at
918202206. Scalafix also passed both the syntactic check and the semantic CHECK using Spark 4.0 / JDK 21.These local checks do not cover the full CI matrix. No benchmarks were run, and no measured performance improvement is claimed.