Fix Query Performance issue for Spark 3.1

apache · Oct 20, 2021 · f65ccdb · f65ccdb
1 parent b8d9a97
commit f65ccdb
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 22 deletions.
diff --git a/...n/spark/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonSourceStrategy.scala b/...n/spark/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonSourceStrategy.scala
@@ -152,7 +152,7 @@ private[sql] object CarbonSourceStrategy extends SparkStrategy {
     var partitions : (Seq[CatalogTablePartition], Seq[PartitionSpec], Seq[Expression]) =
       (null, null, Seq.empty)
     var filterPredicates = allPredicates
-    if(names.nonEmpty) {
+    if(names.nonEmpty && partitionsFilter.nonEmpty) {
       partitions = CarbonFilters.getCatalogTablePartitions(
         partitionsFilter.filterNot(e => e.find(_.isInstanceOf[PlanExpression[_]]).isDefined),
         SparkSession.getActiveSession.get,

diff --git a/integration/spark/src/main/spark3.1/org/apache/spark/sql/CarbonToSparkAdapter.scala b/integration/spark/src/main/spark3.1/org/apache/spark/sql/CarbonToSparkAdapter.scala
@@ -25,14 +25,14 @@ import org.apache.spark.SparkContext
 import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
 import org.apache.spark.sql.carbondata.execution.datasources.CarbonFileIndexReplaceRule
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, ExternalCatalogWithListener}
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, ExternalCatalogWithListener, SessionCatalog}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, AttributeSet, DynamicPruningSubquery, Expression, ExprId, NamedExpression, Predicate, ScalaUDF, SubqueryExpression}
 import org.apache.spark.sql.catalyst.expressions.BindReferences.bindReference
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
-import org.apache.spark.sql.catalyst.optimizer.Optimizer
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkOptimizer
 import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
 import org.apache.spark.sql.hive.HiveExternalCatalog
 import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonUDFTransformRule, MVRewriteRule}
@@ -237,34 +237,21 @@ object CarbonToSparkAdapter extends SparkVersionAdapter {
   }
 }
 
-class CarbonOptimizer(session: SparkSession, optimizer: Optimizer) extends
-  Optimizer(session.sessionState.catalogManager) {
+class CarbonOptimizer(session: SparkSession, catalog: SessionCatalog) extends
+  SparkOptimizer(session.sessionState.catalogManager, catalog,
+    session.sessionState.experimentalMethods) {
 
   private lazy val mvRules = Seq(Batch("Materialized View Optimizers", Once,
     Seq(new MVRewriteRule(session)): _*))
 
-  private lazy val iudRule = Batch("IUD Optimizers", fixedPoint,
+  private lazy val iudRule = Batch("IUD Optimizers", Once,
     Seq(new CarbonIUDRule(), new CarbonUDFTransformRule(), new CarbonFileIndexReplaceRule()): _*)
 
   private lazy val secondaryIndexRule = Batch("SI Optimizers", Once,
     Seq(new CarbonSITransformationRule(session)): _*)
 
   override def defaultBatches: Seq[Batch] = {
-    mvRules ++ convertedBatch() :+ iudRule :+ secondaryIndexRule
+    mvRules ++ super.defaultBatches :+ iudRule :+ secondaryIndexRule
   }
 
-  def convertedBatch(): Seq[Batch] = {
-    optimizer.batches.map { batch =>
-      Batch(
-        batch.name,
-        batch.strategy match {
-          case optimizer.Once =>
-            Once
-          case _: optimizer.FixedPoint =>
-            fixedPoint
-        },
-        batch.rules: _*
-      )
-    }
-  }
 }
diff --git a/integration/spark/src/main/spark3.1/org/apache/spark/sql/SparkVersionAdapter.scala b/integration/spark/src/main/spark3.1/org/apache/spark/sql/SparkVersionAdapter.scala
@@ -443,7 +443,7 @@ trait SparkVersionAdapter {
 
   def getCarbonOptimizer(session: SparkSession,
     sessionState: SessionState): CarbonOptimizer = {
-    new CarbonOptimizer(session, sessionState.optimizer)
+    new CarbonOptimizer(session, sessionState.catalog)
   }
 
   def isCharType(dataType: DataType): Boolean = {