[SPARK-40193][SQL] Merge subquery plans with different filters

Author: peter-toth

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -4303,6 +4303,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val PLAN_MERGE_IGNORE_PUSHED_PUSHED_DATA_FILTERS =
+    buildConf("spark.sql.planMerge.ignorePushedDataFilters")
+      .internal()
+      .doc(s"When set to true plan merging is enabled even if physical scan operations have " +
+        "different data filters pushed down.")
+      .version("4.0.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val ERROR_MESSAGE_FORMAT = buildConf("spark.sql.error.messageFormat")
     .doc("When PRETTY, the error message consists of textual representation of error class, " +
       "message and query context. The MINIMAL and STANDARD formats are pretty JSON formats where " +

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkOptimizer.scala

@@ -58,9 +58,6 @@ class SparkOptimizer(
       new RowLevelOperationRuntimeGroupFiltering(OptimizeSubqueries)) :+
     Batch("InjectRuntimeFilter", FixedPoint(1),
       InjectRuntimeFilter) :+
-    Batch("MergeScalarSubqueries", Once,
-      MergeScalarSubqueries,
-      RewriteDistinctAggregates) :+
     Batch("Pushdown Filters from PartitionPruning", fixedPoint,
       PushDownPredicates) :+
     Batch("Cleanup filters that cannot be pushed down", Once,
@@ -92,6 +89,9 @@ class SparkOptimizer(
       LimitPushDownThroughWindow,
       EliminateLimits) :+
     Batch("User Provided Optimizers", fixedPoint, experimentalMethods.extraOptimizations: _*) :+
+    Batch("Merge Scalar Subqueries", Once,
+      MergeScalarSubqueries,
+      RewriteDistinctAggregates) :+
     Batch("Replace CTE with Repartition", Once, ReplaceCTERefWithRepartition)
 
   override def nonExcludableRules: Seq[String] = super.nonExcludableRules :+

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala

@@ -56,8 +56,14 @@ import org.apache.spark.util.collection.BitSet
  *     is under the threshold with the addition of the next file, add it.  If not, open a new bucket
  *     and add it.  Proceed to the next file.
  */
-object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
+object FileSourceStrategy extends Strategy {
+  def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+    case FileSourceScanPlan(scanPlan, _) => scanPlan :: Nil
+    case _ => Nil
+  }
+}
 
+object FileSourceScanPlan extends PredicateHelper with Logging {
   // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
   private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
     bucketSpec match {
@@ -147,7 +153,7 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
     }
   }
 
-  def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+  def unapply(plan: LogicalPlan): Option[(SparkPlan, FileSourceScanExec)] = plan match {
     case ScanOperation(projects, stayUpFilters, filters,
       l @ LogicalRelation(fsRelation: HadoopFsRelation, _, table, _)) =>
       // Filters on this relation fall into four categories based on where we can use them to avoid
@@ -350,8 +356,8 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
         execution.ProjectExec(projects, withFilter)
       }
 
-      withProjections :: Nil
+      Some(withProjections, scan)
 
-    case _ => Nil
+    case _ => None
   }
 }