[GitHub] carbondata pull request #1563: [WIP][CARBONDATA-1592] Refactored CarbonSessi...

[qiuchenjian-2]

GitHub user ManoharVanam opened a pull request:

    https://github.com/apache/carbondata/pull/1563

    [WIP][CARBONDATA-1592] Refactored CarbonSession

    Be sure to do all of the following checklist to help us incorporate
    your contribution quickly and easily:
   
     - [ ] Any interfaces changed?
     
     - [ ] Any backward compatibility impacted?
     
     - [ ] Document update required?
   
     - [ ] Testing done
            Please provide details on
            - Whether new unit test cases have been added or why no new tests are required?
            - How it is tested? Please attach test report.
            - Is it a performance related change? Please attach the performance test report.
            - Any additional information to help reviewers in testing this change.
           
     - [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
   


You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/ManoharVanam/incubator-carbondata SessionChanges

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/carbondata/pull/1563.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #1563
   
----
commit 8e2336194cc1c111c7c25db1b62b9bcd002f0052
Author: Manohar <[hidden email]>
Date:   2017-11-24T10:09:09Z

    [CARBONDATA-1592] Refactored CarbonSession

----


---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    retest this please


---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    retest sdv please


---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1856/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1433/



---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153139798
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala ---
    @@ -189,8 +201,16 @@ 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
    +{
    --- End diff --
   
    move it previous line


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153140116
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala ---
    @@ -50,7 +50,11 @@ class CarbonSession(@transient val sc: SparkContext,
       }
     
       @transient
    -  override lazy val sessionState: SessionState = new CarbonSessionState(this)
    +  override lazy val sessionState: SessionState = SessionStateFactory
    --- End diff --
   
    chang to
    ```
    SessionStateFactory.getSessionState(
           this,
           sc.conf.get(
                 "spark.carbon.sessionstate.classname",
                 "org.apache.spark.sql.hive.CarbonSessionState"))
    ```


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153140267
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala ---
    @@ -212,6 +232,21 @@ class CarbonOptimizer(
                 PredicateSubquery(tPlan, p.children, p.nullAware, p.exprId)
             }
         }
    -    super.execute(transFormedPlan)
    +    transFormedPlan
    +  }
    +}
    +
    +object SessionStateFactory {
    +  def getSessionState(sparkSession: SparkSession, className: String): HiveSessionState = {
    --- End diff --
   
    Add comment for this function


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153140542
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala ---
    @@ -212,6 +232,21 @@ class CarbonOptimizer(
                 PredicateSubquery(tPlan, p.children, p.nullAware, p.exprId)
             }
         }
    -    super.execute(transFormedPlan)
    +    transFormedPlan
    +  }
    +}
    +
    +object SessionStateFactory {
    +  def getSessionState(sparkSession: SparkSession, className: String): HiveSessionState = {
    --- End diff --
   
    I think you can add this func in CarbonSession directly instead of creating a new object


---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1497/



---

[qiuchenjian-2]

Github user manishgupta88 commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153193171
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala ---
    @@ -146,22 +149,31 @@ 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]] = {
    --- End diff --
   
    can we rename the method from internalAnalyzerRules to analyzerRules?...As we already have 1 method with name extendedAnalyzerRules, so this name looks more suitable to me


---

[qiuchenjian-2]

Github user manishgupta88 commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    Apart from 1 comment given, LGTM


---

[qiuchenjian-2]

Github user ManoharVanam commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153194116
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala ---
    @@ -146,22 +149,31 @@ 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]] = {
    --- End diff --
   
    ok


---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1914/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    Build Failed  with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1541/



---

[qiuchenjian-2]

Github user ravipesala commented on the issue:

    https://github.com/apache/carbondata/pull/1563
 
    SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1942/



---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153711119
 
    --- Diff: integration/spark2/src/main/spark2.1/CarbonSessionState.scala ---
    @@ -239,3 +258,21 @@ class CarbonSqlAstBuilder(conf: SQLConf, parser: CarbonSpark2SqlParser) extends
         }
       }
     }
    +
    +/**
    + * returns session state object based on passed class name
    + */
    +object SessionStateFactory {
    +  def getSessionState(sparkSession: SparkSession, className: String): HiveSessionState = {
    --- End diff --
   
    seems no one is using this


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153711278
 
    --- Diff: integration/spark2/src/main/spark2.1/CarbonSessionState.scala ---
    @@ -190,8 +202,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 = {
    --- End diff --
   
    why not put it in `CarbonOptimizer` as private function?


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153711427
 
    --- Diff: integration/spark2/src/main/spark2.1/CarbonSessionState.scala ---
    @@ -147,22 +150,31 @@ 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.isEmpty) {
    --- End diff --
   
    change to nonEmpty


---

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1563#discussion_r153711458
 
    --- Diff: integration/spark2/src/main/spark2.1/CarbonSessionState.scala ---
    @@ -147,22 +150,31 @@ 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.isEmpty) {
    +          extendedAnalyzerRules ++ internalAnalyzerRules
    +        }
    +        else {
    --- End diff --
   
    move to previous line


---