[SPARK-59572][SQL] PartitionPredicate evaluation failure must not be reported as a partition match

[pan3793]

What changes were proposed in this pull request?

PartitionPredicateImpl.eval reported a partition as matching whenever it could not evaluate
the predicate, either because evaluating the expression threw or because the partition key width
did not match the schema. It now propagates that failure.

The old behavior is kept for runtime filters only, which merely prune: their rows are filtered
anyway, by the post-scan FilterExec for a scalar subquery filter, and by the join it was
derived from for a dynamic partition pruning filter, which has no post-scan FilterExec.
createRuntimePartitionPredicates is the only caller that asks for it.

This also documents the eval contract for callers, and corrects a comment in the in-memory
test table that stated the old justification as a general rule.

Why are the changes needed?

Failing open is sound only while something else still evaluates the filter. Two paths have not
met that condition since PartitionPredicate was introduced in 4.2.0:

• the static second pass, where a PartitionPredicate the connector accepts is removed from the
  post-scan filters;
• the metadata-only DELETE rewrite, which has no post-scan filter at all.

On both, the predicate is Spark's only evaluator, so reporting a match on failure is a wrong
answer. On a table partitioned by dep STRING holding 'hr' and '1', with a filter on dep
whose evaluation fails for 'hr', an ANSI cast error for example:

• SELECT ... WHERE <filter> returns the 'hr' row, for which the filter is never true, and
  reports no error;
• DELETE FROM ... WHERE <filter> deletes the 'hr' partition, for which the condition is never
  true.

The second is silent data loss.

The invariant was identified during the SPARK-58523 review, where fullyPushedFilterAttributes
was described as the first thing in Spark to remove the post-scan FilterExec. The two paths
above already removed it.

Keeping the pushed filter instead of propagating is not available here: by the time eval fails,
the connector has accepted the predicate and Spark has already dropped the filter, and Spark
never learns that an evaluation failed.

Does this PR introduce any user-facing change?

Yes, recorded in docs/sql-migration-guide.md for 4.3.0 and 4.2.1.

A query whose partition predicate the source cannot evaluate now fails with the error it would
raise without pushdown, instead of silently returning rows the filter rejects or deleting
partitions its condition never matched. Which queries fail depends on the order the source
evaluates the predicates it accepted; the error can surface while the scan is built, so EXPLAIN
may report it in place of a plan, and a partition with no rows can raise it.

A partition key whose width does not match Table.partitioning() now raises an internal error.
Runtime filtering is unchanged.

How was this patch tested?

New tests, all of which fail without the fix:

• a scan whose partition predicate fails to evaluate must surface the error rather than return
  the partition it could not evaluate;
• a metadata-only DELETE in the same situation must surface the error and leave both partitions
  in place;
• a unit test over PartitionPredicateImpl pinning that the same predicate propagates by default
  and keeps the partition when constructed for a runtime filter;
• a PushDownUtilsSuite test pinning the wiring, so that only the runtime factory produces a
  predicate that keeps a key it cannot evaluate.

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

Generated-by: Claude Opus 5

[pan3793]

cc @szehon-ho @peter-toth @cloud-fan

[dongjoon-hyun]

Thank you for the fix, @pan3793. The direction looks right to me. I'd like to raise two user-facing points that may be worth documenting in the PR description (or covering with a test).

1. Queries that currently succeed may fail, depending on conjunct evaluation order.
   For example, WHERE dep <> 'hr' AND to_int(dep) = 1. Without pushdown, FilterExec evaluates the translated conjunct first (prioritizeFilters) and short-circuits, so to_int is never called on 'hr' and the query succeeds. With pushdown, dep <> 'hr' (first pass) and to_int(dep) = 1 (second-pass PartitionPredicate) are evaluated independently inside the connector, in whatever order the connector chooses. Before this PR, fail-open happened to give the correct answer here. After it, the query may fail, depending on the connector. Since Spark doesn't guarantee evaluation order, this is not a correctness issue, but the set of queries that "start failing" is wider than the description currently suggests.

2. The error can now surface earlier, and on empty partitions.
   As the new test notes, the error is raised while the scan is built. A connector that evaluates partition predicates during scan building or planning will therefore make EXPLAIN SELECT ... fail with a data error. Also, the non-pushdown path fails only when a row in the partition is actually evaluated, whereas partition-metadata evaluation can fail even for a partition with no rows. Both seem acceptable, but it would be good to mention them in the user-facing change section.

[tdcmeehan]

This change overall LGTM, but does this require an update to docs/sql-migration-guide.md? Per the contributing guide.

[szehon-ho]

thank you for the fix! some style nits

It's my fault, i started with pushdown of static/ runtime filters where its safe to return true, and added metadata delete later and forgot this issue.

Yes agree to add warning doc on 4.2 , and it'd be great to have it in 4.2.1 if we have that

[szehon-ho]

can we rename it for clarity? maybe 'keepPartitionOnEvaluationFailure'

[pan3793]

Renamed to keepOnEvalFailure in 9b8d884. Kept it shorter than the suggestion since the class is already PartitionPredicateImpl, so "Partition" reads as redundant.

[szehon-ho]

can we make the message once? like for example

val message =
  log"Cannot evaluate partition predicate ${MDC(LogKeys.EXPR, catalystExpr.sql)}: " +
    log"partition value field count " +
    log"(${MDC(LogKeys.COUNT, partitionValues.numFields)}) does not match schema " +
    log"(${MDC(LogKeys.NUM_PARTITIONS, partitionFields.length)})."

if (!keepPartitionOnEvaluationFailure) {
  throw SparkException.internalError(message.message)
}

logWarning(message + log" Including partition in scan result.")
return true

[pan3793]

Applied in 9b8d884, as suggested. The exception text is unchanged, so the assertion that pins it still matches.

[szehon-ho]

nit: not sure what do you mean 'is the only evaluator'. do you mean :

// otherwise, let the exception propagate.```

if we rename the variable as  i suggest, i think its more clear and we may not need comment?

[pan3793]

Dropped it in 9b8d884. You were right that the rename carries it.

[pan3793]

@dongjoon-hyun both points are real, and both are now in the description and the migration guide.

1. Confirmed, and connector-dependent as you say. I measured your example: dep <> 'hr' AND to_int(dep) = 1 still succeeds against the in-memory source, because its predicate loop short-circuits before reaching the UDF, while data <> 'x' AND to_int(dep) = 1 throws, since the data-column conjunct stays post-scan and cannot shield the partition predicate. Spark defines no order here, so the guide says the outcome depends on the source.

2. Confirmed with one correction worth recording: EXPLAIN does not fail. It prints Error occurred during query planning: followed by the error in place of the plan. The empty-partition case follows from evaluating partition metadata rather than rows, and is noted too.

@tdcmeehan added in 9b8d884. Since this targets 4.3.0 and 4.2.1, there are two entries: one under "Upgrading from Spark SQL 4.2 to 4.3" and one under a new "Upgrading from Spark SQL 4.2.0 to 4.2.1" section.