spark-test commented on code in PR #59810:
URL: https://github.com/apache/spark/pull/59810#discussion_r4383934071
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala:
##########
@@ -548,9 +547,14 @@ trait FileSourceScanLike extends DataSourceScanExec with
SessionStateHelper {
location.getClass.getSimpleName +
Utils.buildLocationMetadata(location.rootPaths, maxMetadataValueLength)
// `metadata` is accessed during planning and the scalar subquery is not
executed yet. Here
- // we get the pretty string of the scalar subquery, for display purpose
only.
- val pushedFiltersForDisplay = translateToV1Filters(
- dataFilters, s => Literal("ScalarSubquery#" + s.exprId.id))
+ // we get the pretty string of the scalar subquery, for display purpose
only. A field read out
+ // of a subquery that `MergeSubplans` merged with others gets the string
as a whole, since the
+ // string has no field to read.
+ val pushedFiltersForDisplay = translateToV1Filters(dataFilters,
_.transform {
+ case GetStructField(s: execution.ScalarSubquery, _, _) =>
Review Comment:
**Blocking (P1):** This case maps a merged read to the StringType
placeholder `Literal("ScalarSubquery#<id>")`, whatever the field's type. When
every item of an `IN` list is such a read, the `In` arm of
`DataSourceStrategy.translateLeafNodeFilter` now passes its all-literals guard
and converts the items with `createToScalaConverter(col.dataType)`. For a DATE
column that is `DateConverter.toScala`, which runs `asInstanceOf[Int]` on the
`UTF8String` and throws `ClassCastException`. TIMESTAMP, TIMESTAMP_NTZ and
DECIMAL converters cast the same way.
For example, with default configuration and Parquet tables `t1(d DATE)` and
`t2(d DATE)`:
```sql
SELECT * FROM t1 WHERE d IN ((SELECT min(d) FROM t2), (SELECT max(d) FROM
t2))
```
`MergeSubplans` merges the two subqueries, so both items become
`GetStructField(ScalarSubquery(merged), i)`. `metadata` is evaluated when
`SQLExecution` builds the plan description at query start (default
`spark.sql.ui.explainMode=formatted`) and by every `EXPLAIN` mode. Neither
`ExplainUtils` nor `QueryExecution.simpleString` catches a
`ClassCastException`, so the query fails before any job runs. Before this
change these items rendered as `GetStructField(Literal(str), i)`, failed the
guard, and were simply omitted from `PushedFilters`, so the query ran.
Only the V1 file scan's display-only `PushedFilters` is affected. Run-time
pushdown through `replaceWithLiterals` builds literals of the real type and is
correct. Comparison filters are safe because their arms convert by the
literal's own type. The bare-subquery form of this `IN` already fails the same
way before this change, so `spark.sql.execution.reuseSubquery=false` does not
avoid it. The new V1 test uses only `>` and `<` on an INT column, so it cannot
catch this. Traced through the source; not executed.
**Recommended change:** In the V1 file scan's EXPLAIN metadata, stop
substituting the display-only ScalarSubquery#<id> string placeholder into IN
lists. When an IN filter's list contains a scalar-subquery read, leave that IN
untranslated for display rather than handing the V1 translator a string where
it expects the column's type. The simplest form of the rule applies to every
scalar-subquery read in an IN list, whether a field read out of a merged
subquery or a bare subquery. Only the merged form is caused by this change; the
same rule also stops the failure that unmerged subqueries in an IN list already
hit before it, but that is not required here. Add a scalar subquery pushdown
test for V1 file sources that covers an IN list of merged scalar subqueries on
a DATE column.
**Why this works:** The V1 translator's IN case converts every list item
with the column type's converter, which is correct for real filters because
analysis requires IN items to have the column's type, but it casts a string
placeholder to the column's internal representation and fails for DATE,
TIMESTAMP, DECIMAL and similar columns. If the IN keeps its subquery items in
the display path, it fails the translator's all-literals guard and is simply
omitted from the displayed PushedFilters, as merged IN items were before this
change, instead of throwing while the plan description is built. Comparison
filters keep receiving the placeholder, which they convert by its own string
type, so their display stays as the PR intends. The run-time translation keeps
using literals of the real type, so the IN is still pushed to the reader. A
top-down transform still descends into the children of a node that a rule
returns unchanged, so returning the IN as-is from a transform case would not
keep i
ts subquery items; the display rewrite has to stop at the IN itself.
**Scope:** A display-only change to how the V1 file scan renders
PushedFilters for IN lists over scalar subqueries, plus a regression test for
an IN list of merged scalar subqueries on a DATE column; run-time pushdown and
comparison-filter display are unchanged.
**Compatibility:** Run-time pushdown of IN and comparison filters on merged
and unmerged scalar subqueries to V1 file readers, and the PushedFilters
display of comparison filters as ScalarSubquery#<exprId>, must stay as the PR
implements them.
**Risks:** EXPLAIN keeps omitting an IN filter over merged scalar subqueries
from the file scan's PushedFilters even though it is pushed at run time, which
is the display limitation such IN filters already had before this change.
**Constraints:** The run-time pushed-filter translation must be unchanged,
and comparison filters on merged and unmerged scalar subqueries must keep
appearing in PushedFilters as ScalarSubquery#<exprId>, as the existing scalar
subquery pushdown tests assert.
**Exclusions:**
- Changing how the shared V1 filter translator converts IN list values,
which would affect every V1 source and is not needed to stop the display path
from producing mistyped input.
**Success:** With default configuration, a query on a V1 file source that
filters a DATE, TIMESTAMP or DECIMAL data column with IN over scalar subqueries
merged by subquery merging runs and returns correct rows, and its EXPLAIN
renders in every mode without a ClassCastException. At run time the scan still
receives that IN filter with values of the column's type, so data skipping
still applies. Comparison filters on merged and unmerged scalar subqueries
still show in PushedFilters as ScalarSubquery#<exprId>.
**Verification:**
- **Regression:** A Parquet-backed query that filters a DATE data column
with IN over two scalar subqueries on the same table, which subquery merging
combines, returns the expected rows under default configuration, and both
formatted and simple EXPLAIN of it complete without an exception.
- **Behavior:** For that IN query the file scan still prunes data at run
time when the subquery values exclude the table's row group, observable as
fewer rows produced by the scan than the table holds.
- **Compatibility:** Comparison filters on merged and unmerged scalar
subqueries keep their ScalarSubquery#<exprId> entries in the scan's
PushedFilters, and approved plan-stability outputs stay unchanged.
--
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]
