[SPARK-59317][SQL] Unwrap the cast of a DSv2 scalar subquery runtime filter before pushing it down

[Vivek1106-04]

What changes were proposed in this pull request?

Type coercion of the compared sides may wrap the filtered column in a cast, e.g.
cast(part_col as bigint) = <scalar subquery> when an INT partition column is compared with a
BIGINT scalar subquery. DataSourceV2Strategy.translateScalarSubqueryFilterV2 literalizes the
scalar subquery once its result is known and translates the comparison with translateFilterV2,
which pushes CAST(part_col AS BIGINT) = <value> through V2ExpressionBuilder. The cast is
pushed but never unwrapped, so a source that only prunes on column references ignores the filter.
The optimizer cannot unwrap it either, because the other side is not a literal until runtime.

This PR adds UnwrapCastInBinaryComparison.unwrapCastInExpression, which unwraps the cast of
every binary comparison, In and InSet of an expression. The rule's own apply is rewritten on
top of it, so the optimizer and the runtime path share one function.

translateScalarSubqueryFilterV2 calls it on the literalized filter:

• when the cast is unwrapped, the comparison is pushed in the column type, e.g. part_col = 3;
• when the value is out of the column's range or is rounded by the conversion, the unwrapping
  produces a null-returning form that has no V2 predicate of its own, so it is rewritten into what
  it filters by before translation: and(isnull(col), null) keeps no row and is pushed as false
  (AlwaysFalse), or(isnotnull(col), null) keeps the rows where the column is not null and is
  pushed as isnotnull(col). Only the whole filter is rewritten, as neither is interchangeable
  with its counterpart under a Not;
• when the cast cannot be unwrapped, e.g. a lossy one, the filter is pushed as it was.

This is the sibling of SPARK-59301 (#58580), which did the same for the dynamic partition pruning
IN filter. Unlike the IN case, the comparison rule rewrites out-of-range or rounded values into
different comparisons or constant results, hence the handling above.

InMemoryTableWithV2Filter, the test table the DSv2 pruning tests use, matched its = branch on
any predicate and cast the first child to FieldReference, so a pushed cast failed it with a
ClassCastException. It now takes that branch only for a reference, as a source that prunes on
column references does.

Why are the changes needed?

A DSv2 scan whose partition column is compared with a wider-typed scalar subquery received
CAST(part_col AS BIGINT) = <value>, which a source pruning on column references cannot use, so
it read every partition. Reported in
#58580 (comment).

Reproduction, with the in-memory V2 table taking runtime filters as V2 predicates: a 10-partition
table with an INT partition column and a BIGINT dimension column.

SELECT * FROM tbl WHERE part = (SELECT max(val) FROM dim)

Before this PR all 10 partitions are read, after it 1 is.

Does this PR introduce any user-facing change?

Yes. DSv2 sources now receive scalar subquery runtime filters whose column is wrapped in a
lossless numeric cast, so the scan prunes partitions instead of reading all of them. Query results
do not change.

How was this patch tested?

• New unit tests for translateScalarSubqueryFilterV2 in DataSourceV2StrategySuite: bare and
  nested columns, cast unwrapping for = and >, a value rounded by the conversion, a value
  above the column's range for = and for <, and a lossy cast that must not be unwrapped.
• New end-to-end test in DataSourceV2SQLSuiteV2Filter, which asserts the answer and that 1 of 10
  partitions survives pruning.
• Control run with the two main-source changes reverted and the tests kept: the unit test pushes
  CAST(cint AS long) = 1 instead of cint = 1, and the end-to-end test reads all 10 partitions.
• UnwrapCastInBinaryComparisonSuite, DataSourceV2StrategySuite, DataSourceV2SQLSuiteV2Filter,
  DataSourceV2CatalystRuntimeFilterSuite, DynamicPartitionPruningV1Suite and
  DynamicPartitionPruningV2Suite pass.

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

Generated-by: Claude Code (Claude Opus 5)

[uros-b]

Thank you @Vivek1106-04 for working on this!

simplifyUnwrappedNull only matches the whole filter, and it re-couples pushdown to falseIfNotNull / trueIfNotNull.

Sharing unwrapCast with the optimizer is the right SPARK-59301 follow-up, and the InMemoryTableWithV2Filter FieldReference guard is a real CCE fix. After unwrap, an out-of-range equality is And(IsNull(col), null) and an always-true comparison is Or(IsNotNull(col), null). Those have no V2 predicate of their own, which is why this rewrite exists.

Two problems:

1. Only the root is rewritten. cast(part as bigint) = (SELECT max(val) FROM dim) AND id > 0 with an out-of-range subquery becomes And(And(IsNull(part), null), id > 0). The inner And is left alone. With spark.sql.dataSource.alwaysCreateV2Predicate=true (default), that can be pushed as AND(IS_NULL(part), BOOLEAN_EXPRESSION(null)) instead of AlwaysFalse.

2. Matching And(IsNull(_), Literal(null, BooleanType)) is the encoding SPARK-59301 stopped depending on (unwrapCastInSet was added so DSv2 would not pattern-match it). If that encoding changes, this compiles, tests pass, and pruning silently disappears.

Please walk conjuncts (leave Not alone — these are not interchangeable under Not), or have unwrap return a V2-ready form (EqualTo/GreaterThan/…, FalseLiteral, or IsNotNull) so this file does not match optimizer internals.

Tests to add:

• EqualTo(Cast(cint, LongType), Literal(Int.MaxValue + 1L)) AND other → AlwaysFalse (or not pushed, but not BOOLEAN_EXPRESSION(null))
• LessThan(Cast(cint, LongType), Literal(Int.MaxValue + 1L)) AND other → AND(IS_NOT_NULL(cint), other)
• NOT (EqualTo(Cast(cint, LongType), Literal(Int.MaxValue + 1L))) must not become NOT false