[SPARK-55617][SQL] Add VariantGet to V2ExpressionBuilder for DSv2 filter pushdown

[qlong]

What changes were proposed in this pull request?

Add a dedicated VariantGet connector expression class and wire it through V2ExpressionBuilder so that variant_get and try_variant_get predicates can be translated into V2 expressions and pushed down to connectors via SupportsPushDownV2Filters.

Key changes:

• Add org.apache.spark.sql.connector.expressions.VariantGet following the GetArrayItem pattern, carrying child (column reference), path, targetType, failOnError, and optional timeZoneId
• Add a VariantGet case in V2ExpressionBuilder.generateExpression() that translates catalyst VariantGet to the new connector class; only foldable paths and direct column references are supported
• Add visitVariantGet dispatch in V2ExpressionSQLBuilder and a ToStringSQLBuilder override that renders variant_get(col, '$.path', type) / try_variant_get(col, '$.path', type, tz=...) for EXPLAIN output
• When isPredicate=true and the target type is BooleanType, wrap the result in a BOOLEAN_EXPRESSION predicate to satisfy the V2Predicate contract

Jira: https://issues.apache.org/jira/browse/SPARK-55617

Why are the changes needed?

This is prerequisite to support file-level skipping & rowgroup skipping for shredded variant columns in Iceberg (and other DSv2 connectors). When a query filters on variant_get(v, '$.city', string) = 'NYC', connectors can now receive and evaluate that predicate against manifest/statistics metadata instead of fetching every file.

Does this PR introduce any user-facing change?

No. The translation is internal to V2ExpressionBuilder. Connectors that do not implement SupportsPushDownV2Filters are unaffected. Connectors that do will now receive VariantGet expressions where previously none were pushed; they can choose to handle or ignore them.

Does this PR introduce any user-facing change?

No. The translation is internal to V2ExpressionBuilder. Connectors that do not implement SupportsPushDownV2Filters are unaffected. Connectors that do will now receive VariantGet expressions where previously none were pushed; they can choose to handle or ignore them.

How was this patch tested?

Added unit tests.

Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Sonnet.

[qlong]

This is the PR on the iceberg side for variant filter pushdown for file skipping. apache/iceberg#15385

[qlong]

@huaxingao Can you help review? Thanks

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

[qlong]

This is needed to variant filter pushdown.

[huaxingao]

cc @viirya @gengliangwang

[gengliangwang]

2 blocking, 4 non-blocking, 0 nits.
Focused, useful PR with a sound goal; the wire-protocol design choice needs an explicit decision and the boolean-predicate crash needs a guard before merge.

Design / architecture (2)

• V2ExpressionBuilder.scala:348: fabricated UserDefinedScalarFunc for a built-in breaks the catalog-UDF channel contract (collision-prone bare canonical name, undocumented 3-arg/catalogString encoding); recommend a dedicated connector expression like GetArrayItem/V2Cast — see inline
• V2ExpressionBuilder.scala:343: session-timezone dependence of timestamp-target variant_get is dropped from the translation — gate or document? — see inline

Correctness (2)

• V2ExpressionBuilder.scala:338: boolean targetType ignores isPredicate — planning-time AssertionError under And/Or/Not inside the builder (e.g. CHECK-constraint translation) — see inline
• V2ExpressionBuilder.scala:342: v.path.eval().toString NPEs on a foldable null path when the builder runs on un-optimized trees — see inline

Suggestions (2)

• DataSourceV2StrategySuite.scala:822: assertions don't pin the wire contract (child order/values, canonicalName); guard branches and boolean targetType untested — see inline
• V2ExpressionBuilder.scala:341: translate the child via generateExpression(v.child) instead of isInstanceOf/asInstanceOf + manual FieldReference.column — see inline

Verification

Traced the VariantGet -> UserDefinedScalarFunc(name, name, [col, path, type]) encoding across the input space: non-foldable paths and non-Attribute children are gated out (V2ExpressionBuilder.scala:339-340); failOnError is conveyed via the name; catalogString round-trips for every type VariantGet.checkDataType admits. Three cells are neither equivalent nor gated — boolean targetType in predicate position (AssertionError), foldable-null path (NPE), and timezone-dependent target types (session zone not conveyed) — and those are exactly the findings above.

PR description suggestions

• Document: why UserDefinedScalarFunc rather than a dedicated connector expression class (the GetArrayItem/V2Cast pattern), and the resulting name/argument wire contract connectors must implement.
• Fix: PR title is missing the component tag — should be [SPARK-55617][SQL] ....

[gengliangwang]

This fabricates a UserDefinedScalarFunc for a built-in expression, which diverges from both established patterns in this file: built-ins with non-expression payload get a dedicated connector expression class (GetArrayItem in the case right above this one — SPARK-54240 — carries failOnError as a typed field; V2Cast carries a DataType), while the three existing UserDefinedScalarFunc producers all pass a catalog ScalarFunction's name()/canonicalName(). BoundFunction.canonicalName's javadoc requires collision-resistant qualified names (e.g. com.mycompany.bucket(string)) — a bare variant_get is indistinguishable from a genuine catalog UDF of that name, and the 3-arg/catalogString encoding is a wire contract that exists only implicitly between this code and apache/iceberg#15385. I'd recommend a dedicated VariantGet connector expression carrying targetType: DataType and failOnError: Boolean — self-documenting and collision-free; the cost is a (small, @Evolving) connector-API addition, which GetArrayItem already paid in 4.1.0. If you keep the UDSF route, could you say why in the PR description, document the encoding for connector implementers, and qualify the canonical name?

[qlong]

Thanks for review. This is a good call out, dedicated connector class instead of UDSF is much cleaner and more correct. I was using UDSF to avoid compile time dependence between iceberg and spark but i agree it is anti-pattern. Added VariantGet by following GetArrayItem. Also fix timezone and failOnError.

[gengliangwang]

VariantGet is a TimeZoneAwareExpression — a timestamp-target cast uses the session zone (VariantCastArgs.zoneId) — but the translation drops timeZoneId entirely. A connector that fully consumes the pushed predicate and evaluates the cast in a different zone would produce wrong skipping/results. The type literal does let a connector decline timestamp targets, but nothing tells it that it must. Should translation be gated when targetType (recursively) contains a timestamp type, or is the intent to document this caveat as part of the contract?

[qlong]

Fixed with new connector expression. timezoneId is preserved.

[gengliangwang]

This case ignores isPredicate, unlike its peers (generateExpressionWithNameByChildren wraps boolean results in a V2Predicate; boolean columns wrap as = TRUE). For a boolean targetType — variant_get(v, '$.flag', 'boolean') is the most natural variant filter — the case returns a non-Predicate, and when that happens under And/Or/Not inside the builder it trips the live assert(... isInstanceOf[V2Predicate]) calls and crashes planning with AssertionError instead of declining translation (the pre-PR behavior). This is concretely reachable via CHECK-constraint translation (constraints.scala:132), which calls buildPredicate on whole un-optimized conditions, e.g. CHECK (variant_get(v, '$.ok', 'boolean') OR x > 0). Mirroring buildPredicate's own escape hatch fixes it:

val udf = new UserDefinedScalarFunc(canonName, canonName, Array[V2Expression](colRef, pathLit, typeLit))
if (isPredicate && v.dataType.isInstanceOf[BooleanType]) {
  Some(new V2Predicate("BOOLEAN_EXPRESSION", Array[V2Expression](udf)))
} else {
  Some(udf)
}

[qlong]

Good call out. Fixed.

[gengliangwang]

v.path.eval() returns null for a foldable null path — variant_get(v, null) passes analysis (StringTypeWithCollation admits a null string literal) — so .toString throws a bare NPE. The main filter pipeline is protected because NullPropagation folds the null-intolerant VariantGet away before pushdown, but this builder also runs on analyzed-but-un-optimized trees (CHECK-constraint translation, constraints.scala:132), where the NPE is reachable from the planner. Checking the eval result for null and declining translation keeps the failure mode graceful.

[qlong]

fixed.

[gengliangwang]

These tests assert only name() and children().length, but the children encoding is exactly the wire contract the Iceberg consumer matches on — a regression that swapped the path/type argument order, or changed catalogString to sql, would pass all three tests. Worth asserting the children themselves (FieldReference("v"), LiteralValue("$.city", StringType), LiteralValue("string", StringType)) and canonicalName(). Also missing: negative tests pinning the guards (non-foldable path and non-Attribute child should yield None), and a boolean targetType case — which would have surfaced the predicate-wrapping issue flagged above.

[qlong]

agree. Strengthened testing.

[gengliangwang]

Consider translating the child through the existing path instead of the isInstanceOf guard + asInstanceOf + manual FieldReference.column: generateExpression(v.child) already produces the FieldReference for an Attribute (via ColumnOrField), so matching on a Some(ref: FieldReference) result reuses the shared infrastructure, drops both casts, and leaves the door open for struct-nested variant columns later without changing the encoding.

[qlong]

Fixed. Changed to generateExpression(v.child) + FieldReference match

[gengliangwang]

6 addressed, 0 remaining, 2 new. (2 newly introduced, 0 late catches.)
Clean revision — the prior round's design and correctness findings are all resolved correctly; two minor non-blocking items remain.

Suggestions (1)

• V2ExpressionBuilder.scala:341: timezone passthrough (v.timeZoneId.orNull) is untested — only the None case is exercised; add a builder test with a resolved tz — see inline

Nits: 1 minor item (see inline comments).

Verification

Traced the catalyst VariantGet → connector V2VariantGet translation across the input space: path/targetType/failOnError/timeZoneId are conveyed verbatim; non-foldable paths, foldable-null paths, and non-FieldReference children are gated to None (Spark then re-applies the original filter); a boolean target in predicate position wraps in BOOLEAN_EXPRESSION, satisfying the And/Or/Not and CHECK-constraint (constraints.scala:132) V2Predicate asserts. Pushed predicates a connector can't guarantee are rebuilt and re-applied post-scan (PushDownUtils.scala:95-135), so faithful conveyance is the correctness crux — and it holds. Residual: a null timeZoneId with a timestamp target (reachable only on un-resolved trees) conveys no zone, documented on the connector field.

PR description suggestions

• Fix: PR title has a stray space — [SPARK-55617] [SQL] → [SPARK-55617][SQL] (no space between tags).
• Remove: the "Does this PR introduce any user-facing change?" section is duplicated in the body.

[gengliangwang]

This comment is written against the PR's history — once merged there's no "the fix" to contrast with. Reframe to the current invariant:

Suggested change

	// Without the fix, And/Or assert V2Predicate and crash with AssertionError.
	// With the fix, boolExpr is wrapped in BOOLEAN_EXPRESSION and Or translates.
	// A boolean-typed VariantGet in predicate position must translate to a V2Predicate, or the
	// enclosing And/Or's `isInstanceOf[V2Predicate]` assert crashes planning; the BOOLEAN_EXPRESSION
	// wrapper provides that.

[qlong]

fixed comment

[gengliangwang]

v.timeZoneId.orNull is only ever exercised with None here: every test builds the catalyst VariantGet with the default timeZoneId = None, and the two toString tests construct V2VariantGet directly. A regression that dropped this to a hardcoded null would pass the whole suite. Worth a builder test that sets a resolved timezone and asserts it reaches V2VariantGet.timeZoneId() — and, since the Some(colRef: FieldReference) match now admits nested columns, a struct-nested variant column case would pin that path too.

[qlong]

added two more tests:

1. assert non-null timezoneId reaches V2VariantGet.timeZoneId()
2. asserts the resulting FieldReference carries both the parent struct name and the field name for struct-nested variant column

[gengliangwang]

Thanks, merging to master/4.x

[qlong]

Thanks for review and merge. This unblocks a few read optimization work on iceberg side.

[pan3793]

@gengliangwang this needs to be revised

Jira ticket is resolved with Fix Version 5.0.0, commit goes master and branch-4.x, API is marked @since 4.1.0

[gengliangwang]

@pan3793 thanks, I just created #56537 for this.