Fix Variant Parquet scans on Spark 4.1+

[nartal1]

Fixes #15952.

Description

Spark 4.1.1+ enables Variant shredding and pushes Variant extraction into Parquet scans by default. The pushed read schema contains synthetic Variant structs understood by Spark's CPU Parquet reader but not by the GPU Parquet reader. These scans could therefore fail with missing schema-index or empty-footer errors.

This PR detects pushed Variant structs using Spark's VariantMetadata and prevents the Parquet scan from running on GPU, allowing it to fall back cleanly to CPU. Spark 4.0 behavior is unchanged.

The integration tests now cover:

• Shredding and pushdown disabled: try_variant_get executes using GpuVariantGet for Parquet V1 and V2.
• Shredding and pushdown enabled: the Parquet scan falls back to CPU without failure.

Validation:

• Spark 4.2 variant_test.py: 36 passed.
• Spark 4.1.1 focused Variant tests: 3 passed.

Performance:

This correctness fix intentionally falls back to CPU when a raw Variant column may use Spark's shredded Parquet representation. The affected shredded path failed before this change, so it has no successful GPU baseline to compare against. GPU execution for explicitly unshredded Variant input remains unchanged.

GPU support for Variant Parquet scans is tracked by #15180. Pushed Variant extraction support is tracked separately by #14251.

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Confidence Score: 5/5

The PR appears safe to merge; unsupported Variant Parquet representations fall back explicitly while compatible downstream aggregation can resume on GPU.

Findings

1.  Performance impact lacks validation ▶

Summary

• Adds version-specific detection through VariantMetadata and the shredded-read configuration.
• Keeps only the Variant-carrying scan prefix on CPU, allowing compatible downstream operators to return to GPU.
• Extends V1, V2, pushdown, shredded-input, AQE, and downstream GPU aggregate integration coverage.
• Preserves pre-Spark-4.1 behavior through no-op compatibility shims.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Parquet scan schema] --> B{Pushed Variant struct?}
  B -->|Yes| C[Keep scan on CPU]
  B -->|No| D{Raw Variant and shredded reads allowed?}
  D -->|No| E[Normal GPU eligibility checks]
  D -->|Yes| F[Keep Variant-carrying prefix on CPU]
  F --> G{Plan output still contains Variant?}
  G -->|Yes| F
  G -->|No| H[Allow compatible downstream operators on GPU]

Loading

_{Reviews (4) · Last reviewed commit: "Limit Variant CPU fallback to scan prefi..."}

[nartal1]

build

[gerashegalov]

Spark only creates a VariantMetadata struct when PushVariantIntoScan can rewrite the access. It deliberately leaves a raw nested VariantType when the enclosing struct is selected. With a default-shredded Spark 4.1.1 file, SELECT s therefore reaches this block as struct<s:struct<v:variant>>, remains a GPU scan, and fails at runtime because the Parquet group also contains typed_value; I reproduced this against this head. Could we also fall back raw Variant schemas whenever shredded files may be read, unless the footer proves the layout is unshredded?

Suggested coverage:

• Write named_struct('payload', parse_json(...)) AS nested with shredding enabled, select the entire nested struct, and assert both FileSourceScanExec CPU fallback and CPU/GPU result parity. Please parameterize this for V1 and V2.
• Write a shredded top-level Variant, read it with spark.sql.variant.pushVariantIntoScan=false, and assert the same fallback and result parity.

[greptile-apps]

Performance impact lacks validation

When shredded Variant input is allowed, this runtime change can move the Parquet scan and all its ancestor operators from GPU to CPU. The PR marks performance testing as unnecessary because the extraction implementation is unchanged, but that does not establish that the expanded CPU fallback cannot affect runtime performance. This violates the repository directive requiring performance validation or a verifiable exemption for runtime changes.

Rule Used: Report Performance: Not required as a high-severity finding unless the pull request is documentation-only or test-only, or its description gives a verifiable reason the change cannot affect runtime performance. A bug-fix label, small diff, or rarel... (source)

Knowledge Base Used: File format and file I/O support

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

[nartal1]

Updated the PR description.

[nartal1]

build

[gerashegalov]

This keeps walking even after the raw Variant has been consumed. I tried sum(try_variant_get(v, ...)) on a shredded file with scan pushdown disabled, and the entire Project -> partial HashAggregate -> Exchange -> final HashAggregate stayed on CPU. The Project already outputs only int, so a row-to-columnar transition above it should be safe. Could we stop after tagging the first ancestor whose output no longer contains VariantType? It would also be good to add an aggregate-above-extraction test that asserts only the scan/extraction prefix stays on CPU while the downstream aggregate returns to GPU.

[nartal1]

build