You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/11/29 23:37:32 UTC
carbondata git commit: [CARBONDATA-1592] Refactored CarbonSession
Repository: carbondata
Updated Branches:
refs/heads/master f2dcb6d87 -> 7250ff114
[CARBONDATA-1592] Refactored CarbonSession
Description : Refactored common code to avoid duplication code in extended classes.
This closes #1563
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/7250ff11
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/7250ff11
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/7250ff11
Branch: refs/heads/master
Commit: 7250ff1141d49dfb68b3fc3a14d37369c9b04d5b
Parents: f2dcb6d
Author: Manohar <ma...@gmail.com>
Authored: Fri Nov 24 15:39:09 2017 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Nov 30 07:37:21 2017 +0800
----------------------------------------------------------------------
.../core/constants/CarbonCommonConstants.java | 2 +
.../spark/util/CarbonReflectionUtils.scala | 14 ++++--
.../src/main/spark2.1/CarbonSessionState.scala | 52 +++++++++++++-------
3 files changed, 46 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7250ff11/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index fc20e6d..429ac7f 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -1290,6 +1290,8 @@ public final class CarbonCommonConstants {
public static final String DATA_MANAGEMENT_DRIVER_DEFAULT = "false";
+ public static final String CARBON_SESSIONSTATE_CLASSNAME = "spark.carbon.sessionstate.classname";
+
@CarbonProperty
public static final String CARBON_LEASE_RECOVERY_RETRY_COUNT =
"carbon.lease.recovery.retry.count";
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7250ff11/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
index d212db1..76b9e30 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/CarbonReflectionUtils.scala
@@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.parser.AstBuilder
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.constants.CarbonCommonConstants
/**
* Reflection APIs
@@ -140,12 +141,15 @@ object CarbonReflectionUtils {
def getSessionState(sparkContext: SparkContext, carbonSession: Object): Any = {
if (sparkContext.version.startsWith("2.1")) {
- createObject("org.apache.spark.sql.hive.CarbonSessionState", carbonSession)._1
+ val className = sparkContext.conf.get(
+ CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
+ "org.apache.spark.sql.hive.CarbonSessionState")
+ createObject(className, carbonSession)._1
} else if (sparkContext.version.startsWith("2.2")) {
- val tuple =
- createObject("org.apache.spark.sql.hive.CarbonSessionStateBuilder",
- carbonSession,
- None)
+ val className = sparkContext.conf.get(
+ CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
+ "org.apache.spark.sql.hive.CarbonSessionStateBuilder")
+ val tuple = createObject(className, carbonSession, None)
val method = tuple._2.getMethod("build")
method.invoke(tuple._1)
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/7250ff11/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/spark2.1/CarbonSessionState.scala b/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
index eadae6a..0d6ba66 100644
--- a/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
+++ b/integration/spark2/src/main/spark2.1/CarbonSessionState.scala
@@ -16,6 +16,8 @@
*/
package org.apache.spark.sql.hive
+import java.lang.reflect.Constructor
+
import org.apache.hadoop.conf.Configuration
import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, ExperimentalMethods, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -27,6 +29,7 @@ import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.parser.ParserUtils._
import org.apache.spark.sql.catalyst.parser.SqlBaseParser.CreateTableContext
import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{SparkOptimizer, SparkSqlAstBuilder}
import org.apache.spark.sql.execution.command.datamap.{DataMapDropTablePostListener, DropDataMapPostListener}
import org.apache.spark.sql.execution.command.preaaggregate._
@@ -35,11 +38,11 @@ import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStr
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.optimizer.CarbonLateDecodeRule
import org.apache.spark.sql.parser.{CarbonHelperSqlAstBuilder, CarbonSpark2SqlParser, CarbonSparkSqlParser}
+import org.apache.spark.util.Utils
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.events._
/**
* This class will have carbon catalog and refresh the relation from cache if the carbontable in
@@ -147,22 +150,30 @@ class CarbonSessionState(sparkSession: SparkSession) extends HiveSessionState(sp
override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)
+ def extendedAnalyzerRules: Seq[Rule[LogicalPlan]] = Nil
+ def internalAnalyzerRules: Seq[Rule[LogicalPlan]] = {
+ catalog.ParquetConversions ::
+ catalog.OrcConversions ::
+ CarbonPreInsertionCasts(sparkSession) ::
+ CarbonPreAggregateQueryRules(sparkSession) ::
+ CarbonIUDAnalysisRule(sparkSession) ::
+ AnalyzeCreateTable(sparkSession) ::
+ PreprocessTableInsertion(conf) ::
+ DataSourceAnalysis(conf) ::
+ (if (conf.runSQLonFile) {
+ new ResolveDataSource(sparkSession) :: Nil
+ } else { Nil }
+ )
+ }
+
override lazy val analyzer: Analyzer = {
new Analyzer(catalog, conf) {
override val extendedResolutionRules =
- catalog.ParquetConversions ::
- catalog.OrcConversions ::
- CarbonPreInsertionCasts(sparkSession) ::
- CarbonPreAggregateQueryRules(sparkSession) ::
- CarbonIUDAnalysisRule(sparkSession) ::
- AnalyzeCreateTable(sparkSession) ::
- PreprocessTableInsertion(conf) ::
- DataSourceAnalysis(conf) ::
- (if (conf.runSQLonFile) {
- new ResolveDataSource(sparkSession) :: Nil
- } else { Nil }
- )
-
+ if (extendedAnalyzerRules.nonEmpty) {
+ extendedAnalyzerRules ++ internalAnalyzerRules
+ } else {
+ internalAnalyzerRules
+ }
override val extendedCheckRules = Seq(
PreWriteCheck(conf, catalog))
}
@@ -190,8 +201,15 @@ class CarbonOptimizer(
extends SparkOptimizer(catalog, conf, experimentalMethods) {
override def execute(plan: LogicalPlan): LogicalPlan = {
- // In case scalar subquery add flag in relation to skip the decoder plan in optimizer rule, And
- // optimize whole plan at once.
+ val transFormedPlan: LogicalPlan = CarbonOptimizerUtil.transformForScalarSubQuery(plan)
+ super.execute(transFormedPlan)
+ }
+}
+
+object CarbonOptimizerUtil {
+ def transformForScalarSubQuery(plan: LogicalPlan) : LogicalPlan = {
+ // In case scalar subquery add flag in relation to skip the decoder plan in optimizer rule,
+ // And optimize whole plan at once.
val transFormedPlan = plan.transform {
case filter: Filter =>
filter.transformExpressions {
@@ -213,7 +231,7 @@ class CarbonOptimizer(
PredicateSubquery(tPlan, p.children, p.nullAware, p.exprId)
}
}
- super.execute(transFormedPlan)
+ transFormedPlan
}
}