Skip to content

[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join #57190

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qlong wants to merge 1 commit into
base: master
Choose a base branch
from
+1,404 −34
Open

[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join #57190

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6490,6 +6490,19 @@ object SQLConf {
.booleanConf
.createWithDefault(true)

val PUSH_VARIANT_INTO_SCAN_PULL_OUT_EXTRACTIONS =
buildConf("spark.sql.variant.pushVariantIntoScan.pullOutExtractions")
.internal()
.doc("When true, extend variant field pushdown to extractions used inside aggregate " +
"functions, join conditions, sort keys, and projections above joins (including through " +
"chained joins), so that only the requested fields are read from the scan for these " +
"cases too instead of the whole variant column. Has no effect unless " +
"spark.sql.variant.pushVariantIntoScan is also true.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
.createWithDefault(true)

val PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR =
buildConf("spark.sql.variant.pushVariantIntoScan.deferCastError")
.internal()
Expand Down
6 changes: 5 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.optimizer._
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.execution.datasources.{MarkSingleTaskExecution, PruneFileSourcePartitions, PushVariantIntoScan, SchemaPruning, V1Writes}
import org.apache.spark.sql.execution.datasources.{MarkSingleTaskExecution, PruneFileSourcePartitions, PullOutVariantExtractions, PushVariantIntoScan, SchemaPruning, V1Writes}
import org.apache.spark.sql.execution.datasources.v2.{GroupBasedRowLevelOperationScanPlanning, OptimizeMetadataOnlyDeleteFromTable, V2ScanPartitioningAndOrdering, V2ScanRelationPushDown, V2Writes}
import org.apache.spark.sql.execution.dynamicpruning.{CleanupDynamicPruningFilters, PartitionPruning, RowLevelOperationRuntimeGroupFiltering}
import org.apache.spark.sql.execution.python.{ExtractGroupingPythonUDFFromAggregate, ExtractPythonUDFFromAggregate, ExtractPythonUDFs, ExtractPythonUDTFs}
Expand All @@ -37,6 +37,10 @@ class SparkOptimizer(
override def earlyScanPushDownRules: Seq[Rule[LogicalPlan]] =
// TODO: move SchemaPruning into catalyst
Seq(
// Hoist variant extractions out of operators that variant-into-scan pushdown cannot see
// through (aggregate function arguments, etc.) into a Project above the scan, so the
// extractions below become visible to V2ScanRelationPushDown and PushVariantIntoScan.
PullOutVariantExtractions,
SchemaPruning,
GroupBasedRowLevelOperationScanPlanning,
V1Writes,
Expand Down
Loading