peter-toth commented on code in PR #58851:
URL: https://github.com/apache/spark/pull/58851#discussion_r4034497348
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedPartitionFilterSuite.scala:
##########
@@ -403,6 +404,26 @@ class DataSourceV2EnhancedPartitionFilterSuite
}
}
+ test("SPARK-59572: a failed second-pass PartitionPredicate must not keep the
partition") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (part_col string, data string)
USING $v2Source " +
+ "PARTITIONED BY (part_col)")
+ sql(s"INSERT INTO $partFilterTableName VALUES ('hr', 'x'), ('1', 'y')")
+
+ spark.udf.register("to_int", (s: String) => s.toInt)
+
+ // `to_int('hr')` throws, so the filter is never true for that
partition. The filter is
+ // untranslatable, so it is pushed as a PartitionPredicate and accepted,
which removes it
+ // from the post-scan filters. The source is then its only evaluator, so
reporting a failed
+ // evaluation as a match returns the 'hr' row. The error surfaces while
the scan is built.
+ val e = intercept[SparkException] {
Review Comment:
**Finding 1.** This test passes whether or not the `PartitionPredicate` is
the evaluator, so it does not pin the path its comment describes.
Any path that leaves `to_int(part_col) = 1` in the post-scan filters raises
the same error. The `FilterExec` above the scan evaluates it on the `'hr'` row
and gets `FAILED_EXECUTE_UDF` with a `NumberFormatException` cause, so both
assertions hold. Setting `accept-partition-predicates` to `false` on the
fixture is the cheapest way to see it: `pushPredicates` returns the predicate,
`pushPartitionPredicates` restores the original filter, and the test still goes
green. It does fail on master today, so the fix is proven. What is missing is
that a later change which stops pushing this filter leaves the test green and
silent.
The comment already names the discriminator: on the pushed path the error
comes out of planning, on the post-scan path only out of execution. Asserting
that is a one-line change:
```scala
val e = intercept[SparkException] {
sql(s"SELECT * FROM $partFilterTableName WHERE to_int(part_col) = 1")
.queryExecution.optimizedPlan
}
```
The DELETE test at
`sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala:255`
has the same gap and needs a different fix. If the metadata-only rewrite does
not fire, the row-level DELETE evaluates the same UDF, throws the same error
and writes nothing, so `checkAnswer` passes too. `executeAndKeepPlan` cannot
capture the plan there because its `onFailure` is a no-op; recording
`qe.executedPlan` in `onFailure` as well lets the test assert
`DeleteFromTableExec` around the `intercept`.
Both of these are derived from reading, not from running them.
##########
docs/sql-migration-guide.md:
##########
@@ -55,6 +55,11 @@ license: |
- Since Spark 4.3, the Spark Connect session errors
`INVALID_HANDLE.SESSION_CHANGED`/`SESSION_CLOSED`/`SESSION_NOT_FOUND` carry
SQLSTATE `08003` instead of `HY000`; the condition names are unchanged. Code
matching these errors on SQLSTATE should match `08003` or class `08`.
- Since Spark 4.3, [Declarative
Pipelines](declarative-pipelines-programming-guide.html) honors
`spark.sql.caseSensitive` when inferring and evolving pipeline table schemas.
Under case-insensitive resolution (the default), column names that differ only
in case now identify the same column: flows writing to one table contribute a
single column rather than one per spelling, and a column that differs only in
case from one already persisted in the target is written to that column instead
of being added alongside it. Previously such names were always treated as
distinct, producing a table schema that Spark's own resolver could not
disambiguate and that could fail later with errors such as
`COLUMN_ALREADY_EXISTS` or `AMBIGUOUS_REFERENCE`. When two flows' columns fold
together but their types are incompatible, the update now fails at validation
with `UNABLE_TO_INFER_PIPELINE_TABLE_SCHEMA`. Where the spellings differ, the
surviving one comes from the flow with the lowest identifier. An expl
icitly declared table schema keeps its spelling over the inferred one.
Incremental streaming tables keep the persisted spelling of an existing column;
materialized views re-infer the schema on every update. Set
`spark.sql.caseSensitive` to `true` to keep names differing only in case
distinct, as before.
- Since Spark 4.3, all flows writing to the same pipeline table must agree on
the effective `spark.sql.caseSensitive`, which each flow takes from its own SQL
configuration (a `SET` in pipeline source, which never reaches the session) and
otherwise from the session. A disagreement fails the update with
`CONFLICTING_PIPELINE_FLOW_CASE_SENSITIVITY`, because that value decides
whether names differing only in case identify the same column and would
otherwise make the table's schema depend on the order the flows are evaluated
in. Set `spark.sql.caseSensitive` to the same value for every flow writing to
the table, remembering that a flow which does not set it inherits the session's
value.
+- Since Spark 4.3, when a data source cannot evaluate a partition predicate
that Spark pushed down, for example because an ANSI cast of a partition value
fails, the error is raised instead of the partition being reported as a match.
Previously the failure was ignored and the partition was kept, so a scan could
return rows the filter does not accept and a metadata-only `DELETE` could
remove a partition its condition never matched. Queries that appeared to
succeed may now fail with the same error the query raises without pushdown, and
whether a given query fails depends on the order in which the data source
evaluates the predicates it accepted, which Spark does not define. The error
can also surface earlier than before, while the scan is built rather than while
rows are read, so `EXPLAIN` may report it in place of a plan and a partition
holding no rows can raise it. Runtime filters are unaffected: a partition they
cannot evaluate is still kept, because their rows are filtered after th
e scan or by the join the filter came from.
Review Comment:
**Finding 3.** The description's user-facing section lists two changes and
says both are recorded here, but only the first is.
"A partition key whose width does not match `Table.partitioning()` now
raises an internal error" appears in neither this entry nor the 4.2.1 one. For
a connector that passes a subset key it turns a warning plus a kept partition
into a failed query, so a user of such a connector sees a query that used to
work start failing.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -481,7 +485,10 @@ object PushDownUtils extends Logging {
partitionFields: Seq[PartitionPredicateField]):
Seq[PartitionPredicateImpl] = {
val catalystExprs = runtimeFilters.flatMap(unwrapRuntimeFilterExpression)
val flattened = flattenNestedPartitionFilters(catalystExprs,
partitionFields).keys
- createPartitionPredicates(flattened.toSeq, partitionFields)._1
+ // A runtime filter only prunes: its rows are filtered anyway, by the
post-scan `FilterExec`
+ // for a scalar subquery filter and by the join it was derived from for a
DPP filter. So a
Review Comment:
**Finding 2.** A third source of runtime `PartitionPredicate`s is missing
from this list, and neither reason given covers it.
`RowLevelOperationRuntimeGroupFiltering.buildDynamicPruningCond` injects
`DynamicPruningExpression(InSubquery(pruningKeys, ...))`, where `pruningKeys`
resolve from the scan's own `filterAttributes()`
(`sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala:146`).
A partition column there reaches `createRuntimePartitionPredicates` like any
other DPP filter. `DataSourceV2Strategy` partitions it out as a
`DynamicPruning`, so it has no post-scan `FilterExec`, and it was not derived
from a join: it comes from the row-level operation's own condition.
Keeping the partition is still safe there, for a third reason. A
`ReplaceData` or `WriteDelta` rewrite re-applies that condition to the rows it
reads, so an extra group is rewritten unchanged rather than mis-written. Worth
spelling out, because this same two-case list is the whole safety argument in
the `keepOnEvalFailure` scaladoc and in the `InMemoryBaseTable` comment, and
the next reader will check it against this path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
