[CARMEL-6734] Fix reorder filter condition issue (#1293)

* Update PredicateReorder.scala * fix * fix
apache · Apr 14, 2023 · 905523e · 905523e
1 parent b3036d6
commit 905523e
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PredicateReorder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PredicateReorder.scala
@@ -136,10 +136,10 @@ object PredicateReorder extends Rule[LogicalPlan] with PredicateHelper {
   }
 
   def apply(plan: LogicalPlan): LogicalPlan = {
-    if (conf.predicateReorder) {
+    if (conf.predicateReorder && !plan.isStreaming) {
       plan transform {
         case f @ Filter(cond, _) =>
-          val filterEstimation = if (conf.cboEnabled || conf.planStatsEnabled) {
+          val filterEstimation = if (plan.stats.attributeStats.nonEmpty) {
             Some(FilterEstimation(f))
           } else {
             None

diff --git a/...lyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala b/...lyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelperSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.analysis
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet}
 import org.apache.spark.sql.catalyst.optimizer.SimpleTestOptimizer
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, Filter, LeafNode, LocalRelation}
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, Filter, LeafNode, Statistics}
 import org.apache.spark.sql.types.{IntegerType, MetadataBuilder, TimestampType}
 
 class StreamingJoinHelperSuite extends AnalysisTest {
@@ -39,6 +39,7 @@ class StreamingJoinHelperSuite extends AnalysisTest {
     case class DummyLeafNode() extends LeafNode {
       override def output: Seq[Attribute] =
         attributesToFindConstraintFor ++ attributesWithWatermark
+      override def computeStats(): Statistics = Statistics(sizeInBytes = BigInt(0))
     }
 
     def watermarkFrom(