You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "dbtsai (via GitHub)" <gi...@apache.org> on 2026/03/25 21:23:38 UTC
[PR] [WIP][SPARK-XXXXX][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
dbtsai opened a new pull request, #55017:
URL: https://github.com/apache/spark/pull/55017
⏺ ## What changes were proposed in this pull request?
`df.cache()` is backed by `InMemoryRelation`, a pre-DSv2 logical plan node. Queries on cached
DataFrames skip the entire `V2ScanRelationPushDown` optimizer batch, losing column pruning, filter
pushdown, sort-order propagation, and standard statistics reporting that all DSv2-backed sources
benefit from.
This PR introduces a thin DSv2 wrapper layer — `InMemoryCacheTable`, `InMemoryScanBuilder`, and
`InMemoryCacheScan` — in a new file `InMemoryCacheTable.scala`. `CacheManager.useCachedData()` now
substitutes matching plan fragments with `DataSourceV2Relation(InMemoryCacheTable)` instead of bare
`InMemoryRelation`, so the `V2ScanRelationPushDown` batch fires on cached DataFrames just as it does
for any other DSv2 source.
The DSv2 interfaces implemented:
| Interface | Benefit |
|---|---|
| `SupportsPushDownRequiredColumns` | Column pruning: `InMemoryTableScanExec` deserializes only the requested columns |
| `SupportsPushDownV2Filters` | Filter pushdown: predicates are recorded for per-batch min/max pruning via `CachedBatchSerializer.buildFilter`; all predicates are returned (category-2) so a post-scan `FilterExec` is always added for
row-level evaluation |
| `SupportsReportOrdering` | Sort-order propagation: cached sort order is visible to `V2ScanPartitioningAndOrdering`, eliminating redundant sorts on ordered cached data |
| `SupportsReportStatistics` | Statistics: accurate row count, size, and column statistics (including those gathered via `ANALYZE`) are reported to the optimizer and AQE via the standard V2 path |
Physical execution is fully preserved. A new case in `DataSourceV2Strategy` intercepts
`DataSourceV2ScanRelation(InMemoryCacheScan)` before the generic `BatchScanExec` path and routes it
back to `InMemoryTableScanExec`. No change to the columnar scan hot path.
A `CachedRelation` extractor object matches any of the three plan forms a cached relation takes
across query stages — `InMemoryRelation`, `DataSourceV2Relation(InMemoryCacheTable)`, and
`DataSourceV2ScanRelation(InMemoryCacheScan)` — providing backward compatibility for all existing
pattern-match sites.
## Why are the changes needed?
Without this change, every query on a cached DataFrame ignores standard optimizer rules that every
DSv2 source benefits from. The most impactful gap is column pruning: even
`SELECT one_col FROM cached_wide_table` causes all columns to be deserialized from the in-memory
columnar store.
## Does this PR introduce _any_ user-facing change?
Yes, in a positive way:
- Queries that access a subset of columns from a cached DataFrame will be significantly faster.
- Queries on sorted cached DataFrames may eliminate redundant sort operations.
- The `ANALYZE` column statistics API for cached queries now correctly propagates stats through the
optimized plan.
The `InMemoryRelation` type is still used internally and visible in `CachedData.cachedRepresentation`.
The change is transparent: `df.cache()`, `spark.catalog.cacheTable()`, and related APIs behave
identically from the user's perspective.
## How was this patch tested?
All existing tests in `CachedTableSuite` (104 tests), `InMemoryColumnarQuerySuite`,
`DatasetCacheSuite`, `UDFSuite`, `DataSourceV2SQLSuite`, and `LogicalPlanTagInSparkPlanSuite` pass
without modification to test logic (only callsite updates to use the new `CachedRelation` extractor).
A new benchmark `InMemoryCacheDSv2Benchmark` was added. Results on 1M rows (AQE off, single
partition):
Column pruning - 1000000 rows, 10 cols, select 2:
sum 2 of 10 cols (column pruning via DSv2) 19ms 1.0X
sum all 10 cols (no pruning - pre-DSv2) 54ms 0.3X (~2.8x speedup)
Planning overhead - 1000 plan-only iterations:
optimizedPlan (DSv2 path, V2ScanRelationPushDown) 254ms total (~0.25ms/plan)
Column pruning yields a **~3x speedup** when accessing 2 of 10 cached columns. Planning overhead
from the additional optimizer rules is **~0.25ms per query**, negligible in practice.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "dbtsai (via GitHub)" <gi...@apache.org>.
dbtsai commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-4569347704
Sorry for the delay — the changes are now pushed (commit da6ae80f23b on `dfCache`). To summarise what's in place:
**Config toggle**: `spark.sql.inMemoryColumnarStorage.enableDatasourceV2` (default `true`). Setting it to `false` falls back to the pre-DSv2 `InMemoryScans` planner path, which is unchanged from before this PR.
**Migration strategy**: Both the DSv2 path and the fallback path ultimately produce `InMemoryTableScanExec` as the physical node, so existing workloads are transparent to the change at the execution level. The main user-visible difference is in `EXPLAIN` / `withCachedData` output, which shows `DataSourceV2ScanRelation` instead of `InMemoryRelation` under the new path. Users or tooling that inspect plan structure can set the config to `false` to restore the old shape.
I'm happy to flip the default to `false` (opt-in) if you and @viirya prefer a more conservative rollout — just let me know.
---
_This comment was generated with [GitHub MCP](http://go/mcps)._
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "dbtsai (via GitHub)" <gi...@apache.org>.
dbtsai commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-4292387338
Thanks @szehon-ho and @viirya for the detailed review — you're both right that my original description overstated the gaps.
After double-checking with `git blame` and reading `InMemoryScans` carefully:
| Capability | Already works today? | Via |
|---|---|---|
| Column pruning | **Yes** | `InMemoryScans` uses `pruneFilterProject` → `InMemoryTableScanExec(prunedCols, ...)` |
| Filter pushdown (batch min/max) | **Yes** | `InMemoryTableScanExec.predicates` → `CachedBatchSerializer.buildFilter` |
| Sort-order propagation | **Yes** | `InMemoryTableScanExec.outputOrdering` from `cachedPlan.outputOrdering` |
| Statistics | **Yes** | `InMemoryRelation.computeStats()` |
| Dynamic Partition Pruning | **No** | New via `SupportsRuntimeV2Filtering` |
| Per-partition LIMIT pushdown | **No** | New via `SupportsPushDownLimit` |
I've pushed a follow-up commit that:
1. **Adds a config toggle** `spark.sql.inMemoryColumnarStorage.enableDatasourceV2` (default `true`). When set to `false`, `CacheManager.useCachedData` returns `InMemoryRelation` directly, falling back to the existing `InMemoryScans` planner path. This addresses the concern about plan shape and stats behavioral changes.
2. **Fixes the ScalaDoc** in `InMemoryCacheTable` and `InMemoryScanBuilder` to accurately state that the genuinely new capabilities are DPP and per-partition LIMIT; the other interfaces mirror what `InMemoryScans`/`pruneFilterProject` already provides.
3. **Updates the `InMemoryScans` comment** to reference the actual config key.
Regarding the two tradeoffs you flagged:
- **Stats behavioral change**: The column-ratio scaling heuristic in `estimateStatistics()` is indeed new behavior. I'll add a note to the PR description and consider whether to make it match the existing behavior more closely.
- **Plan shape change**: With the config toggle users can opt out if tooling depends on `InMemoryRelation` appearing in `EXPLAIN` output.
Does this approach address the architecture concern, or would you prefer I pursue option 1 (add DPP + limit directly to `InMemoryScans`/`InMemoryTableScanExec` without DSv2 wrapping)?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-4482931962
@dbtsai i dont think you pushed the change? it may be ok to me with a config, but need a migration strategy, not sure what @viirya thinks.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-XXXXX][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-4130562668
> The most impactful gap is column pruning: even
SELECT one_col FROM cached_wide_table causes all columns to be deserialized from the in-memory
columnar store.
Currently InMemoryTableScanExec has column pruning already. It only deserializes necessary column.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #55017: [SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules
URL: https://github.com/apache/spark/pull/55017
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #55017:
URL: https://github.com/apache/spark/pull/55017#discussion_r3089775484
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -703,6 +703,12 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
}
}
+ /**
+ * Fallback strategy for cached in-memory tables when the DSv2 cache path is disabled
+ * (spark.sql.inMemoryColumnarStorage.useDataSourceV2 = false).
Review Comment:
This config `spark.sql.inMemoryColumnarStorage.useDataSourceV2` doesn't exist anywhere in the codebase. Also, after this PR `InMemoryScans` becomes unreachable dead code since `CacheManager` always wraps in `DataSourceV2Relation` — should we add a config toggle, or remove this strategy?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala:
##########
@@ -53,7 +53,9 @@ trait InMemoryTableScanLike extends LeafExecNode {
case class InMemoryTableScanExec(
attributes: Seq[Attribute],
predicates: Seq[Expression],
- @transient relation: InMemoryRelation)
+ @transient relation: InMemoryRelation,
+ limit: Option[Int] = None,
+ runtimeFilters: Seq[Expression] = Nil)
Review Comment:
The case class now has 5 fields — `sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala` line 42 still has `InMemoryTableScanExec(_, _, relation)` which won't compile.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -151,6 +152,24 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
DataSourceV2Strategy.withProjectAndFilter(
project, filters, localScanExec, needsUnsafeConversion = false) :: Nil
+ case PhysicalOperation(project, filters,
Review Comment:
`InMemoryCacheScan` implements `SupportsRuntimeV2Filtering` and reports `filterAttributes()`, but this special case doesn't extract scalar subquery filters on those attributes the way the generic `DataSourceV2ScanRelation` case does (lines 168-183 on master). Should these be handled consistently?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryCacheTable.scala:
##########
@@ -0,0 +1,295 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.columnar
+
+import java.util
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.expressions.{
+ Ascending, Attribute, AttributeReference, Descending, NullsFirst, NullsLast,
+ SortOrder => CatalystSortOrder
+}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability}
+import org.apache.spark.sql.connector.expressions.{
+ Expression => V2Expression, FieldReference, NamedReference,
+ NullOrdering => V2NullOrdering, SortDirection => V2SortDirection,
+ SortOrder => V2SortOrder, SortValue
+}
+import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate}
+import org.apache.spark.sql.connector.read.{
+ Scan, ScanBuilder, Statistics => V2Statistics, SupportsPushDownLimit,
+ SupportsPushDownRequiredColumns, SupportsPushDownV2Filters, SupportsReportOrdering,
+ SupportsReportPartitioning, SupportsReportStatistics, SupportsRuntimeV2Filtering
+}
+import org.apache.spark.sql.connector.read.colstats.ColumnStatistics
+import org.apache.spark.sql.connector.read.partitioning.{
+ KeyGroupedPartitioning, Partitioning, UnknownPartitioning
+}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * A DSv2 [[Table]] wrapper around [[InMemoryRelation]], enabling [[V2ScanRelationPushDown]]
+ * optimizer rules to apply column pruning, filter pushdown, and ordering/statistics reporting
+ * to cached DataFrames.
+ */
+private[sql] class InMemoryCacheTable(val relation: InMemoryRelation)
+ extends Table with SupportsRead {
+
+ // Two InMemoryCacheTable instances wrapping the same CachedRDDBuilder are equal.
+ // All InMemoryRelation copies from the same CachedData share the same cacheBuilder by reference.
+ override def equals(other: Any): Boolean = other match {
+ case t: InMemoryCacheTable => relation.cacheBuilder eq t.relation.cacheBuilder
+ case _ => false
+ }
+ override def hashCode(): Int = System.identityHashCode(relation.cacheBuilder)
+
+ override def name(): String = relation.cacheBuilder.cachedName
+
+ override def schema(): StructType = DataTypeUtils.fromAttributes(relation.output)
+
+ override def capabilities(): util.Set[TableCapability] =
+ util.EnumSet.of(TableCapability.BATCH_READ)
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): InMemoryScanBuilder =
+ new InMemoryScanBuilder(relation)
+}
+
+/**
+ * DSv2 [[ScanBuilder]] for [[InMemoryRelation]].
+ *
+ * - Column pruning via [[SupportsPushDownRequiredColumns]]: only requested columns are
+ * passed to [[InMemoryTableScanExec]], reducing deserialization work.
+ * - Filter pushdown via [[SupportsPushDownV2Filters]]: predicates are recorded for
+ * batch-level pruning using per-batch min/max statistics, but all predicates are
+ * returned (category-2: still need post-scan row-level re-evaluation).
+ */
+private[sql] class InMemoryScanBuilder(relation: InMemoryRelation)
+ extends ScanBuilder
+ with SupportsPushDownRequiredColumns
+ with SupportsPushDownV2Filters
+ with SupportsPushDownLimit {
+
+ private var requiredSchema: StructType = DataTypeUtils.fromAttributes(relation.output)
+ private var _pushedPredicates: Array[V2Predicate] = Array.empty
+ private var _pushedLimit: Option[Int] = None
+
+ override def pruneColumns(required: StructType): Unit = {
+ requiredSchema = required
+ }
+
+ /**
+ * Records predicates so Spark adds a post-scan [[FilterExec]] for row-level evaluation.
+ * Batch-level min/max pruning is handled at physical planning: [[DataSourceV2Strategy]]
+ * passes the Catalyst [[FilterExec]] expressions extracted by [[PhysicalOperation]] directly
+ * to [[InMemoryTableScanExec]], which forwards them to [[CachedBatchSerializer.buildFilter]].
+ * The V2 [[Predicate]]s stored here are not used for batch pruning.
+ */
+ override def pushPredicates(predicates: Array[V2Predicate]): Array[V2Predicate] = {
+ _pushedPredicates = predicates
+ predicates
+ }
+
+ override def pushedPredicates(): Array[V2Predicate] = _pushedPredicates
+
+ /**
+ * Pushes a LIMIT down into the scan. Returns true to indicate the limit was accepted.
+ * Because caching may interleave data across partitions, this is always a partial push:
+ * Spark will still apply a LocalLimit on top to enforce the exact count.
+ */
+ override def pushLimit(limit: Int): Boolean = {
+ _pushedLimit = Some(limit)
+ true
+ }
+
+ /** Always partially pushed: Spark applies a LocalLimit on top. */
+ override def isPartiallyPushed(): Boolean = true
+
+ override def build(): InMemoryCacheScan = {
+ val requiredFieldNames = requiredSchema.fieldNames.toSet
+ val prunedAttrs =
+ if (requiredFieldNames == relation.output.map(_.name).toSet) relation.output
+ else relation.output.filter(a => requiredFieldNames.contains(a.name))
+ new InMemoryCacheScan(relation, prunedAttrs, _pushedPredicates, _pushedLimit)
+ }
+}
+
+/**
+ * DSv2 [[Scan]] for [[InMemoryRelation]].
+ *
+ * Physical execution is handled by [[InMemoryTableScanExec]] via [[DataSourceV2Strategy]]
+ * rather than [[Batch]]/[[InputPartition]] to preserve the existing efficient columnar path.
+ *
+ * Reports:
+ * - Ordering ([[SupportsReportOrdering]]): propagates the ordering of the original cached plan
+ * so the optimizer can eliminate redundant sorts on top of the cache.
+ * - Statistics ([[SupportsReportStatistics]]): exposes accurate row count and size from
+ * accumulated scan metrics once the cache is materialized, feeding AQE decisions.
+ * - Partitioning ([[SupportsReportPartitioning]]): reports [[KeyGroupedPartitioning]] when
+ * the cached plan was hash-partitioned on explicit columns, allowing the optimizer to
+ * skip shuffles for downstream joins/aggregates on the same key.
+ * - Runtime filtering ([[SupportsRuntimeV2Filtering]]): enables Dynamic Partition Pruning
+ * on cached scans; [[DynamicPruning]] expressions are passed via [[InMemoryTableScanExec]]
+ * for batch-level min/max pruning.
+ */
+private[sql] class InMemoryCacheScan(
+ val relation: InMemoryRelation,
+ val prunedAttrs: Seq[Attribute],
+ val pushedPredicates: Array[V2Predicate],
+ val pushedLimit: Option[Int] = None)
+ extends Scan
+ with SupportsReportOrdering
+ with SupportsReportStatistics
+ with SupportsReportPartitioning
+ with SupportsRuntimeV2Filtering {
+
+ override def readSchema(): StructType = DataTypeUtils.fromAttributes(prunedAttrs)
+
+ /**
+ * Converts the Catalyst sort ordering of the cached plan to V2 [[SortOrder]]s.
+ * Only attribute-reference based orderings whose column is present in [[prunedAttrs]] are
+ * emitted; sort keys that were pruned away are dropped so that [[V2ScanPartitioningAndOrdering]]
+ * does not attempt to resolve a column that is no longer in the scan output.
+ */
+ override def outputOrdering(): Array[V2SortOrder] = {
+ val prunedNames = prunedAttrs.map(_.name).toSet
+ relation.outputOrdering.flatMap {
+ case CatalystSortOrder(attr: AttributeReference, direction, nullOrdering, _)
+ if prunedNames.contains(attr.name) =>
+ val v2Dir = direction match {
+ case Ascending => V2SortDirection.ASCENDING
+ case Descending => V2SortDirection.DESCENDING
+ }
+ val v2Nulls = nullOrdering match {
+ case NullsFirst => V2NullOrdering.NULLS_FIRST
+ case NullsLast => V2NullOrdering.NULLS_LAST
+ }
+ Some(SortValue(FieldReference.column(attr.name), v2Dir, v2Nulls))
+ case _ => None
+ }.toArray
+ }
+
+ /**
+ * Reports the output partitioning of the cached plan so the optimizer can skip
+ * shuffles for downstream operations on the same partitioning key.
+ */
+ override def outputPartitioning(): Partitioning = {
+ relation.cachedPlan.outputPartitioning match {
+ case HashPartitioning(expressions, numPartitions) =>
+ val keys = expressions.collect { case a: AttributeReference =>
+ FieldReference.column(a.name).asInstanceOf[V2Expression]
+ }
+ if (keys.size == expressions.size) {
+ new KeyGroupedPartitioning(keys.toArray, numPartitions)
+ } else {
+ new UnknownPartitioning(numPartitions)
+ }
+ case other => new UnknownPartitioning(other.numPartitions)
+ }
+ }
+
+ /**
+ * Exposes hash-partitioning key columns for Dynamic Partition Pruning.
+ * Spark will inject runtime IN-list filters on these attributes when it can
+ * derive them from a broadcast side of a join.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ relation.cachedPlan.outputPartitioning match {
+ case HashPartitioning(exprs, _) =>
+ exprs.collect { case a: AttributeReference =>
+ FieldReference.column(a.name).asInstanceOf[NamedReference]
+ }.toArray
+ case _ => Array.empty
+ }
+ }
+
+ /**
+ * No-op: runtime predicates for cached scans are handled entirely through
+ * [[InMemoryTableScanExec.runtimeFilters]], not through this interface method.
+ * The DPP pipeline injects [[DynamicPruning]] expressions into the plan, which
+ * [[DataSourceV2Strategy]] separates and passes as runtimeFilters to the exec node.
+ */
+ override def filter(predicates: Array[V2Predicate]): Unit = {}
Review Comment:
`filter()` is a no-op — runtime filtering is actually routed through `InMemoryTableScanExec.runtimeFilters` instead. This deviates from the V2 contract where `filter()` is expected to prune `InputPartition`s.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-5746507048
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #55017:
URL: https://github.com/apache/spark/pull/55017#issuecomment-4256100949
just to clarify, it looks like a migration and pointing out some risk. (im ok to do it, if we feel its worth to do)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-56479][SQL] df.cache() with DSv2 interfaces to enable V2ScanRelationPushDown optimizer rules [spark]
Posted by "dbtsai (via GitHub)" <gi...@apache.org>.
dbtsai commented on code in PR #55017:
URL: https://github.com/apache/spark/pull/55017#discussion_r3321347259
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -703,6 +703,12 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
}
}
+ /**
+ * Fallback strategy for cached in-memory tables when the DSv2 cache path is disabled
+ * (spark.sql.inMemoryColumnarStorage.useDataSourceV2 = false).
Review Comment:
Fixed in da6ae80f23b — added `spark.sql.inMemoryColumnarStorage.enableDatasourceV2` (default `true`) to `SQLConf`. The `InMemoryScans` ScalaDoc now explicitly states it is only reached when the config is `false`, and documents that both paths produce `InMemoryTableScanExec` with the same column-pruning, filter-pushdown, and sort-order semantics — the DSv2 path additionally enables DPP and per-partition LIMIT pushdown.
---
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala:
##########
@@ -53,7 +53,9 @@ trait InMemoryTableScanLike extends LeafExecNode {
case class InMemoryTableScanExec(
attributes: Seq[Attribute],
predicates: Seq[Expression],
- @transient relation: InMemoryRelation)
+ @transient relation: InMemoryRelation,
+ limit: Option[Int] = None,
+ runtimeFilters: Seq[Expression] = Nil)
Review Comment:
Fixed in da6ae80f23b — updated the pattern to `InMemoryTableScanExec(_, _, relation, _, _)` to match the new 5-field case class.
---
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryCacheTable.scala:
##########
@@ -0,0 +1,295 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.columnar
+
+import java.util
+import java.util.OptionalLong
+
+import org.apache.spark.sql.catalyst.expressions.{
+ Ascending, Attribute, AttributeReference, Descending, NullsFirst, NullsLast,
+ SortOrder => CatalystSortOrder
+}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability}
+import org.apache.spark.sql.connector.expressions.{
+ Expression => V2Expression, FieldReference, NamedReference,
+ NullOrdering => V2NullOrdering, SortDirection => V2SortDirection,
+ SortOrder => V2SortOrder, SortValue
+}
+import org.apache.spark.sql.connector.expressions.filter.{Predicate => V2Predicate}
+import org.apache.spark.sql.connector.read.{
+ Scan, ScanBuilder, Statistics => V2Statistics, SupportsPushDownLimit,
+ SupportsPushDownRequiredColumns, SupportsPushDownV2Filters, SupportsReportOrdering,
+ SupportsReportPartitioning, SupportsReportStatistics, SupportsRuntimeV2Filtering
+}
+import org.apache.spark.sql.connector.read.colstats.ColumnStatistics
+import org.apache.spark.sql.connector.read.partitioning.{
+ KeyGroupedPartitioning, Partitioning, UnknownPartitioning
+}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+/**
+ * A DSv2 [[Table]] wrapper around [[InMemoryRelation]], enabling [[V2ScanRelationPushDown]]
+ * optimizer rules to apply column pruning, filter pushdown, and ordering/statistics reporting
+ * to cached DataFrames.
+ */
+private[sql] class InMemoryCacheTable(val relation: InMemoryRelation)
+ extends Table with SupportsRead {
+
+ // Two InMemoryCacheTable instances wrapping the same CachedRDDBuilder are equal.
+ // All InMemoryRelation copies from the same CachedData share the same cacheBuilder by reference.
+ override def equals(other: Any): Boolean = other match {
+ case t: InMemoryCacheTable => relation.cacheBuilder eq t.relation.cacheBuilder
+ case _ => false
+ }
+ override def hashCode(): Int = System.identityHashCode(relation.cacheBuilder)
+
+ override def name(): String = relation.cacheBuilder.cachedName
+
+ override def schema(): StructType = DataTypeUtils.fromAttributes(relation.output)
+
+ override def capabilities(): util.Set[TableCapability] =
+ util.EnumSet.of(TableCapability.BATCH_READ)
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): InMemoryScanBuilder =
+ new InMemoryScanBuilder(relation)
+}
+
+/**
+ * DSv2 [[ScanBuilder]] for [[InMemoryRelation]].
+ *
+ * - Column pruning via [[SupportsPushDownRequiredColumns]]: only requested columns are
+ * passed to [[InMemoryTableScanExec]], reducing deserialization work.
+ * - Filter pushdown via [[SupportsPushDownV2Filters]]: predicates are recorded for
+ * batch-level pruning using per-batch min/max statistics, but all predicates are
+ * returned (category-2: still need post-scan row-level re-evaluation).
+ */
+private[sql] class InMemoryScanBuilder(relation: InMemoryRelation)
+ extends ScanBuilder
+ with SupportsPushDownRequiredColumns
+ with SupportsPushDownV2Filters
+ with SupportsPushDownLimit {
+
+ private var requiredSchema: StructType = DataTypeUtils.fromAttributes(relation.output)
+ private var _pushedPredicates: Array[V2Predicate] = Array.empty
+ private var _pushedLimit: Option[Int] = None
+
+ override def pruneColumns(required: StructType): Unit = {
+ requiredSchema = required
+ }
+
+ /**
+ * Records predicates so Spark adds a post-scan [[FilterExec]] for row-level evaluation.
+ * Batch-level min/max pruning is handled at physical planning: [[DataSourceV2Strategy]]
+ * passes the Catalyst [[FilterExec]] expressions extracted by [[PhysicalOperation]] directly
+ * to [[InMemoryTableScanExec]], which forwards them to [[CachedBatchSerializer.buildFilter]].
+ * The V2 [[Predicate]]s stored here are not used for batch pruning.
+ */
+ override def pushPredicates(predicates: Array[V2Predicate]): Array[V2Predicate] = {
+ _pushedPredicates = predicates
+ predicates
+ }
+
+ override def pushedPredicates(): Array[V2Predicate] = _pushedPredicates
+
+ /**
+ * Pushes a LIMIT down into the scan. Returns true to indicate the limit was accepted.
+ * Because caching may interleave data across partitions, this is always a partial push:
+ * Spark will still apply a LocalLimit on top to enforce the exact count.
+ */
+ override def pushLimit(limit: Int): Boolean = {
+ _pushedLimit = Some(limit)
+ true
+ }
+
+ /** Always partially pushed: Spark applies a LocalLimit on top. */
+ override def isPartiallyPushed(): Boolean = true
+
+ override def build(): InMemoryCacheScan = {
+ val requiredFieldNames = requiredSchema.fieldNames.toSet
+ val prunedAttrs =
+ if (requiredFieldNames == relation.output.map(_.name).toSet) relation.output
+ else relation.output.filter(a => requiredFieldNames.contains(a.name))
+ new InMemoryCacheScan(relation, prunedAttrs, _pushedPredicates, _pushedLimit)
+ }
+}
+
+/**
+ * DSv2 [[Scan]] for [[InMemoryRelation]].
+ *
+ * Physical execution is handled by [[InMemoryTableScanExec]] via [[DataSourceV2Strategy]]
+ * rather than [[Batch]]/[[InputPartition]] to preserve the existing efficient columnar path.
+ *
+ * Reports:
+ * - Ordering ([[SupportsReportOrdering]]): propagates the ordering of the original cached plan
+ * so the optimizer can eliminate redundant sorts on top of the cache.
+ * - Statistics ([[SupportsReportStatistics]]): exposes accurate row count and size from
+ * accumulated scan metrics once the cache is materialized, feeding AQE decisions.
+ * - Partitioning ([[SupportsReportPartitioning]]): reports [[KeyGroupedPartitioning]] when
+ * the cached plan was hash-partitioned on explicit columns, allowing the optimizer to
+ * skip shuffles for downstream joins/aggregates on the same key.
+ * - Runtime filtering ([[SupportsRuntimeV2Filtering]]): enables Dynamic Partition Pruning
+ * on cached scans; [[DynamicPruning]] expressions are passed via [[InMemoryTableScanExec]]
+ * for batch-level min/max pruning.
+ */
+private[sql] class InMemoryCacheScan(
+ val relation: InMemoryRelation,
+ val prunedAttrs: Seq[Attribute],
+ val pushedPredicates: Array[V2Predicate],
+ val pushedLimit: Option[Int] = None)
+ extends Scan
+ with SupportsReportOrdering
+ with SupportsReportStatistics
+ with SupportsReportPartitioning
+ with SupportsRuntimeV2Filtering {
+
+ override def readSchema(): StructType = DataTypeUtils.fromAttributes(prunedAttrs)
+
+ /**
+ * Converts the Catalyst sort ordering of the cached plan to V2 [[SortOrder]]s.
+ * Only attribute-reference based orderings whose column is present in [[prunedAttrs]] are
+ * emitted; sort keys that were pruned away are dropped so that [[V2ScanPartitioningAndOrdering]]
+ * does not attempt to resolve a column that is no longer in the scan output.
+ */
+ override def outputOrdering(): Array[V2SortOrder] = {
+ val prunedNames = prunedAttrs.map(_.name).toSet
+ relation.outputOrdering.flatMap {
+ case CatalystSortOrder(attr: AttributeReference, direction, nullOrdering, _)
+ if prunedNames.contains(attr.name) =>
+ val v2Dir = direction match {
+ case Ascending => V2SortDirection.ASCENDING
+ case Descending => V2SortDirection.DESCENDING
+ }
+ val v2Nulls = nullOrdering match {
+ case NullsFirst => V2NullOrdering.NULLS_FIRST
+ case NullsLast => V2NullOrdering.NULLS_LAST
+ }
+ Some(SortValue(FieldReference.column(attr.name), v2Dir, v2Nulls))
+ case _ => None
+ }.toArray
+ }
+
+ /**
+ * Reports the output partitioning of the cached plan so the optimizer can skip
+ * shuffles for downstream operations on the same partitioning key.
+ */
+ override def outputPartitioning(): Partitioning = {
+ relation.cachedPlan.outputPartitioning match {
+ case HashPartitioning(expressions, numPartitions) =>
+ val keys = expressions.collect { case a: AttributeReference =>
+ FieldReference.column(a.name).asInstanceOf[V2Expression]
+ }
+ if (keys.size == expressions.size) {
+ new KeyGroupedPartitioning(keys.toArray, numPartitions)
+ } else {
+ new UnknownPartitioning(numPartitions)
+ }
+ case other => new UnknownPartitioning(other.numPartitions)
+ }
+ }
+
+ /**
+ * Exposes hash-partitioning key columns for Dynamic Partition Pruning.
+ * Spark will inject runtime IN-list filters on these attributes when it can
+ * derive them from a broadcast side of a join.
+ */
+ override def filterAttributes(): Array[NamedReference] = {
+ relation.cachedPlan.outputPartitioning match {
+ case HashPartitioning(exprs, _) =>
+ exprs.collect { case a: AttributeReference =>
+ FieldReference.column(a.name).asInstanceOf[NamedReference]
+ }.toArray
+ case _ => Array.empty
+ }
+ }
+
+ /**
+ * No-op: runtime predicates for cached scans are handled entirely through
+ * [[InMemoryTableScanExec.runtimeFilters]], not through this interface method.
+ * The DPP pipeline injects [[DynamicPruning]] expressions into the plan, which
+ * [[DataSourceV2Strategy]] separates and passes as runtimeFilters to the exec node.
+ */
+ override def filter(predicates: Array[V2Predicate]): Unit = {}
Review Comment:
Updated the ScalaDoc in da6ae80f23b to make the deviation explicit: `filter()` is an intentional no-op because `InMemoryCacheScan` never goes through `BatchScanExec` — `DataSourceV2Strategy` special-cases it and creates `InMemoryTableScanExec` directly, passing `DynamicPruning` expressions via `runtimeFilters`. We implement `SupportsRuntimeV2Filtering` solely to expose `filterAttributes()`, which is what lets the optimizer inject DPP filters into the plan.
---
_This comment was generated with [GitHub MCP](http://go/mcps)._
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala:
##########
@@ -151,6 +152,24 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
DataSourceV2Strategy.withProjectAndFilter(
project, filters, localScanExec, needsUnsafeConversion = false) :: Nil
+ case PhysicalOperation(project, filters,
Review Comment:
Added a comment in da6ae80f23b explaining why scalar subquery filters are intentionally not extracted here: `InMemoryCacheScan` never goes through `BatchScanExec`, so `scan.filter()` is never called for partition pruning. Scalar subquery filters stay in `compiledFilters` and are applied by `FilterExec` above the scan — the same post-scan re-evaluation path used by all other compile-time filters.
---
_This comment was generated with [GitHub MCP](http://go/mcps)._
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org