This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f3c676cc3bc0 [SPARK-60577][SQL] Fix `ClassCastException` of `IN`
filters with scalar subqueries in V1 file source scans
f3c676cc3bc0 is described below
commit f3c676cc3bc0aba6fef620d6e5490a5edf9484e9
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Oct 27 06:27:18 2026 -0700
[SPARK-60577][SQL] Fix `ClassCastException` of `IN` filters with scalar
subqueries in V1 file source scans
### What changes were proposed in this pull request?
`FileSourceScanLike.metadata` builds `PushedFilters` for display. The
scalar subqueries have not run yet at that point, so it replaces each one with
the string literal `ScalarSubquery#<id>` and translates the data filters to V1
filters. The `In` arm of the V1 translator
(`DataSourceStrategy.translateLeafNodeFilter`) accepts an `In` once all its
list items are literals, and converts them with the Scala converter of the
column type. For a `DATE`, `TIME`, `TIMESTAMP`, `TIMESTAMP_NTZ`, `D [...]
This PR leaves out of `PushedFilters` a data filter that has an `In` with a
scalar subquery in its list. The filter is still pushed down at run time, where
`pushedDownFilters` replaces each subquery with a literal of its result.
Comparisons with scalar subqueries, like `GreaterThan(c,ScalarSubquery#1)`, are
displayed as before, since they convert a value by the type of the literal.
### Why are the changes needed?
With Parquet tables `t1(d DATE)` and `t2(d DATE)`, this query fails before
any job runs:
```sql
SELECT * FROM t1 WHERE d IN (DATE'2020-01-05', (SELECT max(d) FROM t2))
```
```
java.lang.ClassCastException: class
org.apache.spark.unsafe.types.UTF8String cannot be cast to class
java.lang.Integer (org.apache.spark.unsafe.types.UTF8String is in unnamed
module of loader 'app'; java.lang.Integer is in module java.base of loader
'bootstrap')
at
org.apache.spark.sql.catalyst.CatalystTypeConverters$DateConverter$.toScala(CatalystTypeConverters.scala:425)
...
at
org.apache.spark.sql.execution.datasources.DataSourceStrategy$.translateLeafNodeFilter(DataSourceStrategy.scala:680)
...
at
org.apache.spark.sql.execution.FileSourceScanLike.metadata(DataSourceScanExec.scala:553)
...
at
org.apache.spark.sql.execution.QueryExecution.explainString(QueryExecution.scala:531)
at
org.apache.spark.sql.execution.SQLExecution$.planDescription(SQLExecution.scala:197)
...
at org.apache.spark.sql.classic.Dataset.collect(Dataset.scala:1623)
```
`SQLExecution` builds the plan description of a query, which shows
`PushedFilters`, when the query starts. `Dataset.explain` throws the same
exception, and SQL `EXPLAIN` prints `Error occurred during query planning`
instead of the plan. `NOT IN`, an `IN` under `OR`, and `d IN ((SELECT min(d)
FROM t2), (SELECT max(d) FROM t2))` with
`spark.sql.execution.reuseSubquery=false` fail the same way. With the default
`true`, the two subqueries of the last query are merged into one, and the `IN
[...]
This fails since
[v4.0.0](https://github.com/apache/spark/releases/tag/v4.0.0) (2025-05-19),
which added the pushdown of data filters with scalar subqueries (SPARK-43402)
and their display as `ScalarSubquery#<id>` (SPARK-44655). Before that, such
filters were not given to the scan.
### Does this PR introduce _any_ user-facing change?
Yes. These queries run, and EXPLAIN shows their plans. `PushedFilters` no
longer lists a filter with an `IN` that has scalar subqueries, whatever the
column type. For example, `In(c, [5,ScalarSubquery#12])` on an `INT` column was
displayed before. The filter is still pushed down to the data source, and query
results do not change.
### How was this patch tested?
Added a test to `SubquerySuite`, "SPARK-60577: FileSourceScanExec supports
push down IN data filter with scalar subquery". On `DATE`, `TIMESTAMP` and
`DECIMAL` columns of Parquet tables, it runs an `IN` with a literal and a
subquery, and an `IN` with two subqueries under
`spark.sql.execution.reuseSubquery=false`. It checks that EXPLAIN (simple and
formatted) and the query run, that `PushedFilters` has no `In(`, and through
the scan's `numOutputRows` that the `IN` still skips the row g [...]
These also pass locally: `SubquerySuite`, `ExplainSuite`, `ExplainSuiteAE`,
the TPC-DS and TPC-H plan stability suites (no approved plan changes), and the
`explain*.sql` cases of `SQLQueryTestSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59818 from dongjoon-hyun/SPARK-60577.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 522140fad2cf3c08950cd6a16e5290d917b24d0b)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/execution/DataSourceScanExec.scala | 10 +++-
.../scala/org/apache/spark/sql/SubquerySuite.scala | 60 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
index a727ccf56506..77f058f6269b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
@@ -544,9 +544,15 @@ 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.
+ // we get the pretty string of the scalar subquery, for display purpose
only. A filter with an
+ // `In` that has a scalar subquery in its list is not displayed, since the
values of an `In`
+ // are converted to the type of its column, which the string does not have.
val pushedFiltersForDisplay = translateToV1Filters(
- dataFilters, s => Literal("ScalarSubquery#" + s.exprId.id))
+ dataFilters.filterNot(_.exists {
+ case in: In =>
in.list.exists(_.exists(_.isInstanceOf[execution.ScalarSubquery]))
+ case _ => false
+ }),
+ s => Literal("ScalarSubquery#" + s.exprId.id))
val metadata =
Map(
"Format" -> relation.fileFormat.toString,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
index c402475eed6b..bd2bf1caa9c2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -2510,6 +2510,66 @@ class SubquerySuite extends SharedSparkSession
}
}
+ test("SPARK-60577: FileSourceScanExec supports push down IN data filter with
scalar subquery") {
+ def checkFileSourceScan(query: String, numScalarSubqueries: Int): Unit = {
+ val df = sql(query)
+ // `PushedFilters` is a part of the output of EXPLAIN, and of the plan
description that is
+ // built when the query starts.
+ Seq(SimpleMode,
FormattedMode).foreach(df.queryExecution.explainString(_))
+ checkAnswer(df, Seq.empty)
+ val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+ case f: FileSourceScanExec => f
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+ assert(fileSourceScanExec.size === 1)
+ val scalarSubquery =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+ case s: ScalarSubquery => s
+ })
+ assert(scalarSubquery.length === numScalarSubqueries)
+ // The `In` is not displayed in `PushedFilters`, but it is pushed down
to skip the row group.
+
assert(!fileSourceScanExec.head.metadata("PushedFilters").contains("In("))
+ assert(fileSourceScanExec.head.metrics("numOutputRows").value === 0)
+ }
+
+ withTable("t1", "t2") {
+ // Parquet does not skip row groups by INT96 timestamps, which are the
default.
+ withSQLConf(
+ SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1",
+ SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> "TIMESTAMP_MICROS") {
+ Seq("t1" -> Seq(1, 2, 3), "t2" -> Seq(4, 5, 6)).foreach { case (table,
ids) =>
+ ids.toDF("id").selectExpr(
+ "DATE_FROM_UNIX_DATE(id) AS d",
+ "TIMESTAMP_SECONDS(id) AS ts",
+ "CAST(id AS DECIMAL(10, 2)) AS
dec").write.format("parquet").saveAsTable(table)
+ }
+
+ Seq(
+ "d" -> "DATE_FROM_UNIX_DATE(5)",
+ "ts" -> "TIMESTAMP_SECONDS(5)",
+ "dec" -> "CAST(5 AS DECIMAL(10, 2))").foreach { case (column,
literal) =>
+ checkFileSourceScan(
+ s"SELECT * FROM t1 WHERE $column IN ($literal, (SELECT
max($column) FROM t2))", 1)
+ // Otherwise, the two scalar subqueries are merged into one.
+ withSQLConf(SQLConf.SUBQUERY_REUSE_ENABLED.key -> "false") {
+ checkFileSourceScan(
+ s"SELECT * FROM t1 WHERE $column IN " +
+ s"((SELECT min($column) FROM t2), (SELECT max($column) FROM
t2))", 2)
+ }
+ }
+
+ // `NOT IN` and an `IN` under `OR` hold the `In` below the top of the
filter.
+ Seq(
+ "SELECT * FROM t1 WHERE d NOT IN (DATE_FROM_UNIX_DATE(5), (SELECT
max(d) FROM t2))",
+ "SELECT * FROM t1 WHERE d IN (DATE_FROM_UNIX_DATE(5), (SELECT max(d)
FROM t2)) " +
+ "OR dec IS NULL").foreach { query =>
+ val df = sql(query)
+ Seq(SimpleMode,
FormattedMode).foreach(df.queryExecution.explainString(_))
+ df.collect()
+ }
+ }
+ }
+ }
+
test("SPARK-44562: Add OptimizeOneRowRelationSubquery in batch of Subquery")
{
withTempView("v1", "v2") {
sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
