[SPARK-59410][SQL] Derive PartitionPredicate from identity fields of a mixed partitioning

[pan3793]

Depends on #58756 (SPARK-59457), included here as the first commit.

What changes were proposed in this pull request?

PushDownUtils.getPartitionPredicateSchema returned a schema only when every transform in
Table.partitioning() is an identity transform. It now accepts any partitioning with at least
one identity transform: identity fields can be referenced by a PartitionPredicate, other
fields keep their ordinal but are never referenced, so a filter on the source column of a
non-identity transform stays a data filter.

The connector contract is unchanged: PartitionPredicate.eval still receives the full
partition key, and references() still reports ordinals into Table.partitioning().

This applies to all users of the schema: the static second pass, runtime filter pushdown
(DPP and scalar subqueries), the metadata-only DELETE rewrite, and the scan of a group-based
UPDATE, MERGE or DELETE.

The in-memory V2 filter test table now accepts only predicates of the shape it can evaluate,
a column against a literal, and returns anything else, e.g. a predicate over a cast, as a
real connector would.

Why are the changes needed?

A table partitioned by, for example, dt STRING (identity) and bucket(16, user_id) received
no PartitionPredicate at all. A filter such as dt = DATE'2026-09-01' is analyzed as
cast(dt AS DATE) = DATE'2026-09-01'; a connector that does not evaluate casts returns it
from the first pass (and without ANSI mode it is not translatable at all), so only a
PartitionPredicate can prune with it. On such a table it could not prune partitions in
either the static or the runtime path, and could not drive a metadata-only DELETE, while the
same filter on an identity-only table can. Mixed partitionings are the common case for
connectors that support partition transforms.

The all-identity check was raised in the SPARK-55596 review (#54459) and kept only because
the partition-key contract was still open then. The contract that shipped (full key,
partitioning order) is what makes lifting the check safe.

Does this PR introduce any user-facing change?

No. A connector that opts into iterative pushdown and has a mixed partitioning now receives
PartitionPredicates over its identity fields; the eval contract is unchanged.

How was this patch tested?

New tests on a partitioning of one identity column plus a bucket transform: a predicate on
the identity field is pushed and prunes in the static pass, via DPP, in a metadata-only
DELETE, and in the scan of a group-based UPDATE and MERGE, including the dt STRING vs
DATE literal case whose type-coercion cast the source cannot evaluate; an identity field
placed after the bucket transform binds to its own ordinal; a filter on the bucket source
column is not turned into a predicate; a bucket-only partitioning yields no predicate; the
predicate survives Java and Kryo round-trips.

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

Generated-by: Claude Fable 5.1

[pan3793]

cc @szehon-ho

[uros-b]

Looks clean to me, thank you @pan3793! But yeah let's definitely add @szehon-ho who has more context in PartitionPredicate

[dongjoon-hyun]

+1, LGTM.

A few suggestions:

1. getPartitionPredicateSchema has a fourth caller, GroupBasedRowLevelOperationScanPlanning (group-based UPDATE/MERGE/DELETE scans), which the description does not mention. Could you mention it and add a mixed-partitioning MERGE or UPDATE test?
2. In "filter on the source column of a bucket transform stays post-scan", exists(_.isInstanceOf[FilterExec]) is a weak check. Could you assert that the FilterExec condition references id?
3. nit: "mixed partitioning: no identity transform" uses a bucket-only table, so it is not mixed.
4. nit: the warning in PartitionPredicateImpl.apply lists non-identity fields (e.g. bucket(4, id)) as partition fields a filter could reference.

[szehon-ho]

lgtm, some suggestions

[szehon-ho]

nit: we can say 'for now Spark doesnt support'. it was in the plan but never implemented yet

[pan3793]

Done in d0d6b52.

[szehon-ho]

suggestion:

def supportsPredicates(predicates: Array[Predicate]): Boolean = {
  predicates.flatMap(splitAnd).forall { p =>
    (p.name(), p.children().toSeq) match {
      case ("=" | "<=>", Seq(_: NamedReference, _: LiteralValue[_])) => true
      case ("IS_NULL" | "IS_NOT_NULL", Seq(_: NamedReference)) => true
      case ("ALWAYS_TRUE", _) => true
      case _ => false
    }
  }
}

[pan3793]

Applied as suggested in d0d6b52.

[pan3793]

Thanks, all four addressed in d0d6b52.

1. The description now names the fourth caller, and there are UPDATE and MERGE tests on a mixed partitioning through the row-level scan. Writing them surfaced that RowLevelOperationTable never delegated partitioning(), so getPartitionPredicateSchema has seen an empty partitioning for every row-level scan since SPARK-55596. That is a bug in its own right and applies to 4.2/4.3, so it is split out as [SPARK-59457][SQL] RowLevelOperationTable should delegate partitioning() to the wrapped table #58756 (SPARK-59457); this PR is rebased on it and carries it as the first commit.
2. The post-scan FilterExec condition must now reference id and not part_col.
3. Renamed to bucket-only partitioning: ....
4. The warning lists only the identity partition fields.