pan3793 commented on code in PR #58851:
URL: https://github.com/apache/spark/pull/58851#discussion_r4034741382
##########
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:
Confirmed by running it, and fixed in 38a683b. With
`accept-partition-predicates` set to `false` the scan query throws the same
`FAILED_EXECUTE_UDF` with the same cause, and the DELETE fallback throws and
leaves both rows, so both tests passed on the path they were written to exclude.
The scan test now asserts against `queryExecution.optimizedPlan`, which
raises on the pushed path and raises nothing on the fallback. For the DELETE I
split `executeAndKeepPlan` into a `keepingPlan` helper that records the plan in
`onFailure` too, so the test asserts `DeleteFromTableExec` around the intercept.
##########
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:
Right, and the third reason is the one that makes it safe. Fixed in 38a683b:
both places that enumerate now name the row-level group filter and say a
`ReplaceData` or `WriteDelta` rewrite re-applies its own condition. The
`InMemoryBaseTable` comment needed no change, since it says "whose rows are
filtered anyway" without listing the sources.
Thanks also for verifying the runtime half independently. That matches what
I traced before opening the PR, so it now has two confirmations.
##########
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:
Added to both entries in 38a683b. I also shortened them while there, since
they had grown well past their neighbours.
--
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]
