carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [7/8] carbondata git commit: [CARBONDATA-2532][Integration] Carbon to support spark 2.3.1 version(Make API changes in carbon to be compatible with spark 2.3)
Date Wed, 05 Sep 2018 12:40:15 GMT
[CARBONDATA-2532][Integration] Carbon to support spark 2.3.1 version(Make API changes in carbon to be compatible with spark 2.3)

Fixed spark-carbon integration API compatibility issues
f) Corrected the testcases based on spark 2.3.0 behaviour change
g) Excluded following dependency from pom.xml files net.jpountzlz4 as spark 2.3.0 changed
it to org.lz4, so removed from the test class path of spark2,spark-common-test,spark2-examples

This closes #2642


Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/74c3eb10
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/74c3eb10
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/74c3eb10

Branch: refs/heads/master
Commit: 74c3eb10b860845dacd3542ab72595617dc0df57
Parents: bcef656
Author: sandeep-katta <sandeep.katta2007@gmail.com>
Authored: Thu Jul 5 21:31:29 2018 -0700
Committer: ravipesala <ravi.pesala@gmail.com>
Committed: Wed Sep 5 18:09:16 2018 +0530

----------------------------------------------------------------------
 datamap/mv/core/pom.xml                         |   5 +
 .../apache/carbondata/mv/datamap/MVHelper.scala |  44 ++-
 datamap/mv/plan/pom.xml                         |   5 +
 .../mv/plans/modular/ModularPlan.scala          |   3 +-
 .../mv/plans/modular/ModularRelation.scala      |   5 +-
 .../mv/plans/util/BirdcageOptimizer.scala       |  13 +-
 .../util/LogicalPlanSignatureGenerator.scala    |   5 +-
 .../mv/testutil/ModularPlanTest.scala           |  36 +--
 .../mv/plans/ExtractJoinConditionsSuite.scala   |   4 +-
 .../mv/plans/LogicalToModularPlanSuite.scala    |  18 +-
 examples/spark2/pom.xml                         |  20 ++
 integration/spark-common-test/pom.xml           |  32 +++
 .../testsuite/bigdecimal/TestBigDecimal.scala   |  10 +-
 .../filterexpr/AllDataTypesTestCaseFilter.scala |   7 +-
 .../StandardPartitionGlobalSortTestCase.scala   |  22 +-
 .../sql/commands/StoredAsCarbondataSuite.scala  |   3 +-
 .../carbondata/spark/rdd/CarbonScanRDD.scala    |   2 +-
 .../carbondata/spark/rdd/StreamHandoffRDD.scala |  11 +-
 .../streaming/CarbonAppendableStreamSink.scala  |   2 +-
 .../apache/spark/sql/test/util/PlanTest.scala   |   5 -
 .../apache/spark/sql/util/SparkSQLUtil.scala    |  72 ++++-
 .../spark/util/CarbonReflectionUtils.scala      | 121 +++++---
 .../scala/org/apache/spark/util/SparkUtil.scala |  42 ---
 .../org/apache/spark/util/SparkUtilTest.scala   |  58 ++++
 integration/spark-datasource/pom.xml            | 132 +++++++++
 .../vectorreader/ColumnarVectorWrapper.java     |  88 +++---
 .../VectorizedCarbonRecordReader.java           |  20 +-
 .../scala/org/apache/spark/util/SparkUtil.scala |  60 ++++
 .../org/apache/spark/sql/CarbonVectorProxy.java | 272 ++++++++++++++++++
 .../org/apache/spark/sql/CarbonVectorProxy.java | 276 ++++++++++++++++++
 .../apache/spark/sql/ColumnVectorFactory.java   |  45 +++
 ...tCreateTableUsingSparkCarbonFileFormat.scala |  26 +-
 integration/spark2/pom.xml                      |  80 ++++++
 .../apache/spark/sql/hive/CarbonAnalyzer.scala  |  51 ++++
 .../sql/hive/CarbonInMemorySessionState.scala   | 278 +++++++++++++++++++
 .../spark/sql/hive/CarbonOptimizerUtil.scala    |  44 +++
 .../spark/sql/hive/CarbonSessionState.scala     | 269 ++++++++++++++++++
 .../spark/sql/hive/CarbonSessionUtil.scala      |  96 +++++++
 .../apache/spark/sql/hive/CarbonSqlConf.scala   | 148 ++++++++++
 ...CreateCarbonSourceTableAsSelectCommand.scala | 130 +++++++++
 .../spark/sql/hive/SqlAstBuilderHelper.scala    | 110 ++++++++
 .../stream/CarbonStreamRecordReader.java        | 155 ++++-------
 .../org/apache/spark/CarbonInputMetrics.scala   |   1 -
 .../spark/sql/CarbonCatalystOperators.scala     |  57 ++--
 .../org/apache/spark/sql/CarbonCountStar.scala  |  10 +-
 .../org/apache/spark/sql/CarbonSession.scala    |   7 +-
 .../sql/CustomDeterministicExpression.scala     |   2 -
 .../management/CarbonLoadDataCommand.scala      |  19 +-
 .../strategy/CarbonLateDecodeStrategy.scala     |  82 ++++--
 .../sql/execution/strategy/DDLStrategy.scala    |  17 +-
 .../spark/sql/hive/CarbonAnalysisRules.scala    |  17 +-
 .../spark/sql/hive/CarbonFileMetastore.scala    |  26 +-
 .../sql/hive/CarbonPreAggregateRules.scala      | 147 +++++++---
 .../spark/sql/optimizer/CarbonFilters.scala     |  26 +-
 .../sql/optimizer/CarbonLateDecodeRule.scala    |  16 +-
 .../sql/parser/CarbonSpark2SqlParser.scala      |   4 +-
 .../spark/sql/CarbonToSparkAdapater.scala       |  75 +++++
 .../org/apache/spark/sql/CarbonVectorProxy.java | 226 ---------------
 ...CreateCarbonSourceTableAsSelectCommand.scala |   2 -
 .../spark/sql/CarbonToSparkAdapater.scala       |  82 ++++++
 .../org/apache/spark/sql/CarbonVectorProxy.java | 229 ---------------
 .../apache/spark/sql/hive/CarbonAnalyzer.scala  |  51 ----
 .../sql/hive/CarbonInMemorySessionState.scala   | 278 -------------------
 .../apache/spark/sql/hive/CarbonOptimizer.scala |  44 +--
 .../spark/sql/hive/CarbonSessionState.scala     | 277 ------------------
 .../spark/sql/hive/CarbonSessionUtil.scala      |  97 -------
 .../spark/sql/hive/CarbonSqlAstBuilder.scala    |  85 +-----
 .../apache/spark/sql/hive/CarbonSqlConf.scala   | 148 ----------
 ...CreateCarbonSourceTableAsSelectCommand.scala | 122 --------
 .../spark/sql/CarbonToSparkAdapater.scala       |  84 ++++++
 .../org/apache/spark/sql/CarbonVectorProxy.java | 255 -----------------
 .../apache/spark/sql/ColumnVectorFactory.java   |  45 ---
 .../apache/spark/sql/hive/CarbonOptimizer.scala |  37 +++
 .../spark/sql/hive/CarbonSqlAstBuilder.scala    |  52 ++++
 .../stream/CarbonStreamRecordReaderTest.java    | 100 +++++++
 .../BooleanDataTypesFilterTest.scala            |   2 +-
 .../booleantype/BooleanDataTypesLoadTest.scala  |   4 +-
 .../bucketing/TableBucketingTestCase.scala      |  26 +-
 .../carbondata/query/SubQueryTestSuite.scala    |   1 -
 pom.xml                                         |  60 ++++
 .../scala/org/apache/spark/rpc/Master.scala     |   2 +-
 .../scala/org/apache/spark/rpc/RpcUtil.scala    |  56 ++++
 .../scala/org/apache/spark/rpc/Worker.scala     |   4 +-
 .../streaming/CarbonStreamInputFormat.java      |   2 +-
 .../streaming/CarbonStreamInputFormatTest.java  |  99 -------
 85 files changed, 3368 insertions(+), 2433 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/core/pom.xml
----------------------------------------------------------------------
diff --git a/datamap/mv/core/pom.xml b/datamap/mv/core/pom.xml
index 77b5cc7..f5e5715 100644
--- a/datamap/mv/core/pom.xml
+++ b/datamap/mv/core/pom.xml
@@ -50,6 +50,11 @@
       <artifactId>scalatest_${scala.binary.version}</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
index f677826..9da109a 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import org.apache.spark.sql.{CarbonEnv, SparkSession}
+import org.apache.spark.sql.{CarbonEnv, CarbonToSparkAdapater, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeReference, Cast, Expression, NamedExpression, ScalaUDF, SortOrder}
@@ -300,10 +300,13 @@ object MVHelper {
         case Alias(agg: AggregateExpression, name) =>
           agg.aggregateFunction.collect {
             case attr: AttributeReference =>
-              AttributeReference(attr.name, attr.dataType, attr.nullable, attr
-                .metadata)(attr.exprId,
+              CarbonToSparkAdapater.createAttributeReference(attr.name,
+                attr.dataType,
+                attr.nullable,
+                attr.metadata,
+                attr.exprId,
                 aliasName,
-                attr.isGenerated)
+                attr)
           }.head
         case Alias(child, name) =>
           child
@@ -314,18 +317,25 @@ object MVHelper {
     expressions.map {
         case alias@Alias(agg: AggregateExpression, name) =>
           attrMap.get(AttributeKey(agg)).map { exp =>
-            Alias(getAttribute(exp), name)(alias.exprId,
+            CarbonToSparkAdapater.createAliasRef(
+              getAttribute(exp),
+              name,
+              alias.exprId,
               alias.qualifier,
               alias.explicitMetadata,
-              alias.isGenerated)
+              Some(alias))
           }.getOrElse(alias)
 
         case attr: AttributeReference =>
           val uattr = attrMap.get(AttributeKey(attr)).map{a =>
             if (keepAlias) {
-              AttributeReference(a.name, a.dataType, a.nullable, a.metadata)(a.exprId,
+              CarbonToSparkAdapater.createAttributeReference(a.name,
+                a.dataType,
+                a.nullable,
+                a.metadata,
+                a.exprId,
                 attr.qualifier,
-                a.isGenerated)
+                a)
             } else {
               a
             }
@@ -333,10 +343,9 @@ object MVHelper {
           uattr
         case alias@Alias(expression: Expression, name) =>
           attrMap.get(AttributeKey(expression)).map { exp =>
-            Alias(getAttribute(exp), name)(alias.exprId,
-              alias.qualifier,
-              alias.explicitMetadata,
-              alias.isGenerated)
+            CarbonToSparkAdapater
+              .createAliasRef(getAttribute(exp), name, alias.exprId, alias.qualifier,
+                alias.explicitMetadata, Some(alias))
           }.getOrElse(alias)
         case expression: Expression =>
           val uattr = attrMap.get(AttributeKey(expression))
@@ -376,9 +385,14 @@ object MVHelper {
         case attr: AttributeReference =>
           val uattr = attrMap.get(AttributeKey(attr)).map{a =>
             if (keepAlias) {
-              AttributeReference(a.name, a.dataType, a.nullable, a.metadata)(a.exprId,
-                attr.qualifier,
-                a.isGenerated)
+              CarbonToSparkAdapater
+                .createAttributeReference(a.name,
+                  a.dataType,
+                  a.nullable,
+                  a.metadata,
+                  a.exprId,
+                  attr.qualifier,
+                  a)
             } else {
               a
             }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/pom.xml
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/pom.xml b/datamap/mv/plan/pom.xml
index fcf0e51..ff6976d 100644
--- a/datamap/mv/plan/pom.xml
+++ b/datamap/mv/plan/pom.xml
@@ -44,6 +44,11 @@
       <artifactId>scalatest_${scala.binary.version}</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala
index 1420522..6c82598 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPlan.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.plans.{JoinType, QueryPlan}
 import org.apache.spark.sql.catalyst.plans.logical.Statistics
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.SparkSQLUtil
 
 import org.apache.carbondata.mv.plans._
 import org.apache.carbondata.mv.plans.util.{Printers, Signature, SQLBuilder}
@@ -64,7 +65,7 @@ abstract class ModularPlan
     //    spark.conf.set("spark.sql.cbo.enabled", true)
     val sqlStmt = asOneLineSQL
     val plan = spark.sql(sqlStmt).queryExecution.optimizedPlan
-    plan.stats(conf)
+    SparkSQLUtil.invokeStatsMethod(plan, conf)
   }
 
   override def fastEquals(other: TreeNode[_]): Boolean = {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularRelation.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularRelation.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularRelation.scala
index 3349835..7e1eb05 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularRelation.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularRelation.scala
@@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.plans.JoinType
 import org.apache.spark.sql.catalyst.plans.logical
 import org.apache.spark.sql.catalyst.plans.logical.Statistics
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.SparkSQLUtil
 
 import org.apache.carbondata.mv.plans.modular.Flags._
 
@@ -45,7 +46,7 @@ case class ModularRelation(databaseName: String,
     rest: Seq[Seq[Any]]) extends LeafNode {
   override def computeStats(spark: SparkSession, conf: SQLConf): Statistics = {
     val plan = spark.table(s"${ databaseName }.${ tableName }").queryExecution.optimizedPlan
-    val stats = plan.stats(conf)
+    val stats = SparkSQLUtil.invokeStatsMethod(plan, conf)
     val output = outputList.map(_.toAttribute)
     val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
       table => AttributeMap(table.output.zip(output))
@@ -136,7 +137,7 @@ case class HarmonizedRelation(source: ModularPlan) extends LeafNode {
   // (spark, conf)
   override def computeStats(spark: SparkSession, conf: SQLConf): Statistics = {
     val plan = spark.table(s"${ databaseName }.${ tableName }").queryExecution.optimizedPlan
-    val stats = plan.stats(conf)
+    val stats = SparkSQLUtil.invokeStatsMethod(plan, conf)
     val output = source.asInstanceOf[GroupBy].child.children(0).asInstanceOf[ModularRelation]
       .outputList.map(_.toAttribute)
     val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/BirdcageOptimizer.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/BirdcageOptimizer.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/BirdcageOptimizer.scala
index 6363089..e1e891a 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/BirdcageOptimizer.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/BirdcageOptimizer.scala
@@ -17,12 +17,14 @@
 
 package org.apache.carbondata.mv.plans.util
 
+import org.apache.spark.sql.CarbonToSparkAdapater
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.optimizer._
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, _}
 import org.apache.spark.sql.catalyst.rules.{RuleExecutor, _}
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.SparkSQLUtil
 
 object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {
 
@@ -74,8 +76,8 @@ object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {
       "Operator Optimizations", fixedPoint, Seq(
         // Operator push down
         PushProjectionThroughUnion,
-        ReorderJoin(conf),
-        EliminateOuterJoin(conf),
+        SparkSQLUtil.getReorderJoinObj(conf),
+        SparkSQLUtil.getEliminateOuterJoinObj(conf),
         PushPredicateThroughJoin,
         PushDownPredicate,
         //      LimitPushDown(conf),
@@ -89,7 +91,7 @@ object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {
         CombineLimits,
         CombineUnions,
         // Constant folding and strength reduction
-        NullPropagation(conf),
+        SparkSQLUtil.getNullPropagationObj(conf),
         FoldablePropagation,
         //      OptimizeIn(conf),
         ConstantFolding,
@@ -113,7 +115,7 @@ object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {
                                             extendedOperatorOptimizationRules: _*) ::
     Batch(
       "Check Cartesian Products", Once,
-      CheckCartesianProducts(conf)) ::
+      SparkSQLUtil.getCheckCartesianProductsObj(conf)) ::
     //    Batch("Join Reorder", Once,
     //      CostBasedJoinReorder(conf)) ::
     //    Batch("Decimal Optimizations", fixedPoint,
@@ -126,8 +128,7 @@ object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {
     //      ConvertToLocalRelation,
     //      PropagateEmptyRelation) ::
     Batch(
-      "OptimizeCodegen", Once,
-      OptimizeCodegen(conf)) ::
+      "OptimizeCodegen", Once, CarbonToSparkAdapater.getOptimizeCodegenRule(conf): _*) ::
     Batch(
       "RewriteSubquery", Once,
       RewritePredicateSubquery,

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/LogicalPlanSignatureGenerator.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/LogicalPlanSignatureGenerator.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/LogicalPlanSignatureGenerator.scala
index 2aff5c0..eaa8b04 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/LogicalPlanSignatureGenerator.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/LogicalPlanSignatureGenerator.scala
@@ -30,7 +30,8 @@ object CheckSPJG {
       case a: Aggregate =>
         a.child.collect {
           case Join(_, _, _, _) | Project(_, _) | Filter(_, _) |
-               HiveTableRelation(_, _, _) | LogicalRelation(_, _, _) | LocalRelation(_, _) => true
+               HiveTableRelation(_, _, _) => true
+          case  l: LogicalRelation => true
           case _ => false
         }.forall(identity)
       case _ => false
@@ -55,7 +56,7 @@ object LogicalPlanRule extends SignatureRule[LogicalPlan] {
   def apply(plan: LogicalPlan, childSignatures: Seq[Option[Signature]]): Option[Signature] = {
 
     plan match {
-      case LogicalRelation(_, _, _) =>
+      case l: LogicalRelation =>
         // TODO: implement this (link to BaseRelation)
         None
       case HiveTableRelation(tableMeta, _, _) =>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/testutil/ModularPlanTest.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/testutil/ModularPlanTest.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/testutil/ModularPlanTest.scala
index 0c68a1f..9d4735c 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/testutil/ModularPlanTest.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/testutil/ModularPlanTest.scala
@@ -17,13 +17,11 @@
 
 package org.apache.carbondata.mv.testutil
 
-import java.io.File
-
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util._
-import org.apache.spark.sql.test.util.{PlanTest, QueryTest}
+import org.apache.spark.sql.test.util.QueryTest
 
 import org.apache.carbondata.mv.plans.modular
 import org.apache.carbondata.mv.plans.modular.{ModularPlan, OneRowTable, Select}
@@ -56,30 +54,6 @@ abstract class ModularPlanTest extends QueryTest with PredicateHelper {
   }
 
   /**
-   * Normalizes plans:
-   * - Filter the filter conditions that appear in a plan. For instance,
-   * ((expr 1 && expr 2) && expr 3), (expr 1 && expr 2 && expr 3), (expr 3 && (expr 1 && expr 2)
-   *   etc., will all now be equivalent.
-   * - Sample the seed will replaced by 0L.
-   * - Join conditions will be resorted by hashCode.
-   */
-  protected def normalizePlan(plan: LogicalPlan): LogicalPlan = {
-    plan transform {
-      case filter@Filter(condition: Expression, child: LogicalPlan) =>
-        Filter(
-          splitConjunctivePredicates(condition).map(rewriteEqual(_)).sortBy(_.hashCode())
-            .reduce(And), child)
-      case sample: Sample =>
-        sample.copy(seed = 0L)(true)
-      case join@Join(left, right, joinType, condition) if condition.isDefined =>
-        val newCondition =
-          splitConjunctivePredicates(condition.get).map(rewriteEqual(_)).sortBy(_.hashCode())
-            .reduce(And)
-        Join(left, right, joinType, Some(newCondition))
-    }
-  }
-
-  /**
    * Rewrite [[EqualTo]] and [[EqualNullSafe]] operator to keep order. The following cases will be
    * equivalent:
    * 1. (a = b), (b = a);
@@ -177,4 +151,12 @@ abstract class ModularPlanTest extends QueryTest with PredicateHelper {
            """.stripMargin)
     }
   }
+
+  object MatchLocalRelation {
+    def unapply(localRelation: LocalRelation): Option[(Seq[Attribute], Any)] = localRelation match {
+      case l: LocalRelation => Some(l.output, l.data)
+      case _ => None
+    }
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/ExtractJoinConditionsSuite.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/ExtractJoinConditionsSuite.scala b/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/ExtractJoinConditionsSuite.scala
index e1a3d9f..34ee2ed 100644
--- a/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/ExtractJoinConditionsSuite.scala
+++ b/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/ExtractJoinConditionsSuite.scala
@@ -39,7 +39,7 @@ class ExtractJoinConditionsSuite extends ModularPlanTest {
     val extracted = modularPlan.extractJoinConditions(modularPlan.children(0),modularPlan.children(1))
     
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.LocalRelation(tbl2,_),Inner,Some(cond2)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),MatchLocalRelation(tbl2,_),Inner,Some(cond2)) =>
         Seq(cond2)
     }
     
@@ -58,7 +58,7 @@ class ExtractJoinConditionsSuite extends ModularPlanTest {
       left.join(right,condition = Some("l.c".attr === "r.c".attr)).analyze
       
     val correctAnswer = originalQuery1 match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.Filter(cond2,logical.LocalRelation(tbl2,_)),Inner,Some(cond3)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),logical.Filter(cond2,MatchLocalRelation(tbl2,_)),Inner,Some(cond3)) =>
         Seq(cond3)
     }    
     

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/LogicalToModularPlanSuite.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/LogicalToModularPlanSuite.scala b/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/LogicalToModularPlanSuite.scala
index 082c325..144721c 100644
--- a/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/LogicalToModularPlanSuite.scala
+++ b/datamap/mv/plan/src/test/scala/org/apache/carbondata/mv/plans/LogicalToModularPlanSuite.scala
@@ -43,7 +43,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
         .analyze
     val modularized = originalQuery.modularize 
     val correctAnswer = originalQuery match {
-      case logical.Project(proj,logical.LocalRelation(tbl,_)) =>
+      case logical.Project(proj,MatchLocalRelation(tbl,_)) =>
         ModularRelation(null,null,tbl,NoFlags,Seq.empty).select(proj:_*)(tbl:_*)()(Map.empty)()
     }
     comparePlans(modularized, correctAnswer)
@@ -58,7 +58,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
         
     val modularized = originalQuery.modularize
     val correctAnswer = originalQuery match {
-      case logical.Project(proj1,logical.Aggregate(grp,agg,logical.Project(proj2,logical.LocalRelation(tbl,_)))) =>
+      case logical.Project(proj1,logical.Aggregate(grp,agg,logical.Project(proj2,MatchLocalRelation(tbl,_)))) =>
         ModularRelation(null,null,tbl,NoFlags,Seq.empty).select(proj2:_*)(tbl:_*)()(Map.empty)().groupBy(agg:_*)(proj2:_*)(grp:_*).select(proj1:_*)(proj1:_*)()(Map.empty)() 
     }   
     comparePlans(modularized, correctAnswer)
@@ -73,7 +73,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
         
     val modularized = originalQuery.modularize
     val correctAnswer = originalQuery match {
-      case logical.Project(proj,logical.Filter(cond,logical.LocalRelation(tbl,_))) =>
+      case logical.Project(proj,logical.Filter(cond,MatchLocalRelation(tbl,_))) =>
         ModularRelation(null,null,tbl,NoFlags,Seq.empty).select(proj:_*)(tbl:_*)(cond)(Map.empty)()  
     }
     comparePlans(modularized, correctAnswer)
@@ -87,7 +87,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
       
     val modularized = originalQuery.modularize 
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.LocalRelation(tbl2,_),Inner,Some(cond2)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),MatchLocalRelation(tbl2,_),Inner,Some(cond2)) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty)).select(tbl1++tbl2:_*)(tbl1++tbl2:_*)(Seq(cond1,cond2):_*)(Map.empty)(JoinEdge(0,1,Inner)) 
     } 
     comparePlans(modularized, correctAnswer)
@@ -101,7 +101,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
       
     val modularized = originalQuery.modularize 
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.LocalRelation(tbl2,_),LeftOuter,Some(cond2)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),MatchLocalRelation(tbl2,_),LeftOuter,Some(cond2)) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty)).select(tbl1++tbl2:_*)(tbl1++tbl2:_*)(Seq(cond1,cond2):_*)(Map.empty)(JoinEdge(0,1,LeftOuter)) 
     } 
     comparePlans(modularized, correctAnswer)
@@ -115,7 +115,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
       
     val modularized = originalQuery.modularize 
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.LocalRelation(tbl2,_),RightOuter,Some(cond2)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),MatchLocalRelation(tbl2,_),RightOuter,Some(cond2)) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty)).select(tbl1++tbl2:_*)(tbl1++tbl2:_*)(Seq(cond1,cond2):_*)(Map.empty)(JoinEdge(0,1,RightOuter))  
     } 
     comparePlans(modularized, correctAnswer)
@@ -129,7 +129,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
     
     val modularized = analysis.EliminateSubqueryAliases(originalQuery).modularize
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.LocalRelation(tbl2,_),Inner,Some(cond2)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),MatchLocalRelation(tbl2,_),Inner,Some(cond2)) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty)).select(tbl1++tbl2:_*)(tbl1++tbl2:_*)(Seq(cond1,cond2):_*)(Map.empty)(JoinEdge(0,1,Inner))  
     } 
     comparePlans(modularized, correctAnswer)
@@ -147,7 +147,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
         
     val modularized = originalQuery.modularize
     val correctAnswer = originalQuery match {
-      case logical.Join(logical.Filter(cond1,logical.LocalRelation(tbl1,_)),logical.Join(logical.Filter(cond2,logical.LocalRelation(tbl2,_)),logical.LocalRelation(tbl3,_),Inner,Some(cond3)),Inner,Some(cond4)) =>
+      case logical.Join(logical.Filter(cond1,MatchLocalRelation(tbl1,_)),logical.Join(logical.Filter(cond2,MatchLocalRelation(tbl2,_)),MatchLocalRelation(tbl3,_),Inner,Some(cond3)),Inner,Some(cond4)) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty),ModularRelation(null,null,tbl3,NoFlags,Seq.empty)).select(tbl1++tbl2++tbl3:_*)(tbl1++tbl2++tbl3:_*)(Seq(cond1,cond2,cond3,cond4):_*)(Map.empty)(JoinEdge(0,1,Inner),JoinEdge(1,2,Inner))  
     } 
     comparePlans(modularized, correctAnswer)
@@ -165,7 +165,7 @@ class LogicalToModularPlanSuite extends ModularPlanTest {
         
     val modularized = originalQuery.modularize
     val correctAnswer = originalQuery match {
-      case logical.Project(proj0, logical.Filter(cond1, logical.Project(proj1, logical.Aggregate(grp,agg,logical.Join(logical.Filter(cond2,logical.LocalRelation(tbl1,_)),logical.Filter(cond3,logical.LocalRelation(tbl2,_)),Inner,Some(cond4)))))) =>
+      case logical.Project(proj0, logical.Filter(cond1, logical.Project(proj1, logical.Aggregate(grp,agg,logical.Join(logical.Filter(cond2,MatchLocalRelation(tbl1,_)),logical.Filter(cond3,MatchLocalRelation(tbl2,_)),Inner,Some(cond4)))))) =>
         Seq(ModularRelation(null,null,tbl1,NoFlags,Seq.empty),ModularRelation(null,null,tbl2,NoFlags,Seq.empty)).select(tbl1++tbl2:_*)(tbl1++tbl2:_*)(Seq(cond2,cond3,cond4):_*)(Map.empty)(JoinEdge(0,1,Inner)).groupBy(agg:_*)(tbl1++tbl2:_*)(grp:_*).select(proj0:_*)(proj1:_*)(cond1)(Map.empty)()  
     }   
     comparePlans(modularized, correctAnswer)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/examples/spark2/pom.xml
----------------------------------------------------------------------
diff --git a/examples/spark2/pom.xml b/examples/spark2/pom.xml
index 9d1adde..91cb20f 100644
--- a/examples/spark2/pom.xml
+++ b/examples/spark2/pom.xml
@@ -77,6 +77,26 @@
       <artifactId>scalatest_${scala.binary.version}</artifactId>
       <scope>test</scope>
     </dependency>
+    <!-- in spark 2.3 spark catalyst added dependency on spark-core-->
+    <dependency>
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.carbondata</groupId>
+      <artifactId>carbondata-core</artifactId>
+      <version>${project.version}</version>
+      <exclusions>
+        <!-- need to Exclude net.jpountz jar from this project.
+         Spark has changed this jar to org.lz4:lz4-java
+         net.jpountz and org.lz4 has same class Name -->
+        <exclusion>
+          <groupId>net.jpountz.lz4</groupId>
+          <artifactId>lz4</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
   </dependencies>
 
   <build>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common-test/pom.xml
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/pom.xml b/integration/spark-common-test/pom.xml
index ee87d92..50e5830 100644
--- a/integration/spark-common-test/pom.xml
+++ b/integration/spark-common-test/pom.xml
@@ -127,12 +127,27 @@
       <artifactId>carbondata-lucene</artifactId>
       <version>${project.version}</version>
       <scope>test</scope>
+      <exclusions>
+        <!-- need to Exclude net.jpountz jar from this project.
+         Spark has changed this jar to org.lz4:lz4-java
+         net.jpountz and org.lz4 has same class Name -->
+        <exclusion>
+          <groupId>net.jpountz.lz4</groupId>
+          <artifactId>lz4</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.carbondata</groupId>
       <artifactId>carbondata-bloom</artifactId>
       <version>${project.version}</version>
       <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>net.jpountz.lz4</groupId>
+          <artifactId>lz4</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.carbondata</groupId>
@@ -146,6 +161,23 @@
       <scope>test</scope>
     </dependency>
     <dependency>
+      <!-- spark catalyst added runtime dependency on spark-core,so
+      while executing the testcases spark-core should be present else it
+      will fail to execute -->
+      <groupId>org.apache.spark</groupId>
+      <artifactId>spark-core_${scala.binary.version}</artifactId>
+      <version>${spark.version}</version>
+      <scope>test</scope>
+      <exclusions>
+        <!-- need to Exclude Avro jar from this project,spark core is using
+        the version 1.7.4 which is not compatible with Carbon -->
+        <exclusion>
+          <groupId>org.apache.avro</groupId>
+          <artifactId>avro</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/bigdecimal/TestBigDecimal.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/bigdecimal/TestBigDecimal.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/bigdecimal/TestBigDecimal.scala
index 0dba467..1f7aafe 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/bigdecimal/TestBigDecimal.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/bigdecimal/TestBigDecimal.scala
@@ -73,7 +73,6 @@ class TestBigDecimal extends QueryTest with BeforeAndAfterAll {
     checkAnswer(sql("select min(salary) from carbonTable"),
       sql("select min(salary) from hiveTable"))
   }
-  
   test("test min datatype on big decimal column") {
     val output = sql("select min(salary) from carbonTable").collectAsList().get(0).get(0)
     assert(output.isInstanceOf[java.math.BigDecimal])
@@ -83,7 +82,6 @@ class TestBigDecimal extends QueryTest with BeforeAndAfterAll {
     val output = sql("select max(salary) from carbonTable").collectAsList().get(0).get(0)
     assert(output.isInstanceOf[java.math.BigDecimal])
   }
-  
   test("test count function on big decimal column") {
     checkAnswer(sql("select count(salary) from carbonTable"),
       sql("select count(salary) from hiveTable"))
@@ -152,8 +150,8 @@ class TestBigDecimal extends QueryTest with BeforeAndAfterAll {
   }
 
   test("test sum*10 aggregation on big decimal column with high precision") {
-    checkAnswer(sql("select sum(salary)*10 from carbonBigDecimal_2"),
-      sql("select sum(salary)*10 from hiveBigDecimal"))
+    checkAnswer(sql("select cast(sum(salary)*10 as double) from carbonBigDecimal_2"),
+      sql("select cast(sum(salary)*10 as double) from hiveBigDecimal"))
   }
 
   test("test sum/10 aggregation on big decimal column with high precision") {
@@ -167,8 +165,8 @@ class TestBigDecimal extends QueryTest with BeforeAndAfterAll {
   }
 
   test("test sum-distinct*10 aggregation on big decimal column with high precision") {
-    checkAnswer(sql("select sum(distinct(salary))*10 from carbonBigDecimal_2"),
-      sql("select sum(distinct(salary))*10 from hiveBigDecimal"))
+    checkAnswer(sql("select cast(sum(distinct(salary))*10 as decimal)from carbonBigDecimal_2"),
+      sql("select cast(sum(distinct(salary))*10 as decimal) from hiveBigDecimal"))
   }
 
   test("test sum-distinct/10 aggregation on big decimal column with high precision") {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala
index 15ac1f4..73786c8 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala
@@ -17,6 +17,7 @@
 
 package org.apache.carbondata.spark.testsuite.filterexpr
 
+import org.apache.spark.sql.execution.strategy.CarbonDataSourceScan
 import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.BeforeAndAfterAll
 
@@ -56,13 +57,15 @@ class AllDataTypesTestCaseFilter extends QueryTest with BeforeAndAfterAll {
   test("verify like query ends with filter push down") {
     val df = sql("select * from alldatatypestableFilter where empname like '%nandh'").queryExecution
       .sparkPlan
-    assert(df.metadata.get("PushedFilters").get.contains("CarbonEndsWith"))
+    assert(df.asInstanceOf[CarbonDataSourceScan].metadata
+      .get("PushedFilters").get.contains("CarbonEndsWith"))
   }
 
   test("verify like query contains with filter push down") {
     val df = sql("select * from alldatatypestableFilter where empname like '%nand%'").queryExecution
       .sparkPlan
-    assert(df.metadata.get("PushedFilters").get.contains("CarbonContainsWith"))
+    assert(df.asInstanceOf[CarbonDataSourceScan].metadata
+      .get("PushedFilters").get.contains("CarbonContainsWith"))
   }
   
   override def afterAll {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionGlobalSortTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionGlobalSortTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionGlobalSortTestCase.scala
index 150bc36..564c5f3 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionGlobalSortTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionGlobalSortTestCase.scala
@@ -21,6 +21,7 @@ import java.util.concurrent.{Callable, ExecutorService, Executors}
 
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.test.util.QueryTest
+import org.apache.spark.util.SparkUtil
 import org.scalatest.BeforeAndAfterAll
 
 import org.apache.carbondata.common.constants.LoggerAction
@@ -859,11 +860,22 @@ class StandardPartitionGlobalSortTestCase extends QueryTest with BeforeAndAfterA
     sql("create table scalarissue_hive(a int,salary double) using parquet partitioned by (salary) ")
     sql("set hive.exec.dynamic.partition.mode=nonstrict")
     sql("insert into scalarissue_hive values(23,21.2)")
-    intercept[Exception] {
-      sql(s"select * from scalarissue_hive where salary = (select max(salary) from scalarissue_hive)").show()
-    }
-    intercept[Exception] {
-      sql(s"select * from scalarissue where salary = (select max(salary) from scalarissue)").show()
+    if (SparkUtil.isSparkVersionEqualTo("2.1") || SparkUtil.isSparkVersionEqualTo("2.2")) {
+      intercept[Exception] {
+        sql(s"select * from scalarissue_hive where salary = (select max(salary) from " +
+            s"scalarissue_hive)")
+          .show()
+      }
+      intercept[Exception] {
+        sql(s"select * from scalarissue where salary = (select max(salary) from scalarissue)")
+          .show()
+      }
+    } else {
+      checkAnswer(sql(s"select * from scalarissue_hive where salary = (select max(salary) from " +
+                      s"scalarissue_hive)"), Seq(Row(23, 21.2)))
+      checkAnswer(sql(s"select * from scalarissue where salary = (select max(salary) from " +
+                      s"scalarissue)"),
+        Seq(Row(23, 21.2)))
     }
   }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common-test/src/test/scala/org/apache/carbondata/sql/commands/StoredAsCarbondataSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/sql/commands/StoredAsCarbondataSuite.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/sql/commands/StoredAsCarbondataSuite.scala
index 2029e93..6400ed1 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/sql/commands/StoredAsCarbondataSuite.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/sql/commands/StoredAsCarbondataSuite.scala
@@ -87,7 +87,8 @@ class StoredAsCarbondataSuite extends QueryTest with BeforeAndAfterEach {
       sql("CREATE TABLE carbon_table(key INT, value STRING) STORED AS  ")
     } catch {
       case e: Exception =>
-        assert(e.getMessage.contains("no viable alternative at input"))
+        assert(e.getMessage.contains("no viable alternative at input") ||
+        e.getMessage.contains("mismatched input '<EOF>' expecting "))
     }
   }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index b712109..01c77f0 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -421,7 +421,7 @@ class CarbonScanRDD[T: ClassTag](
           // create record reader for row format
           DataTypeUtil.setDataTypeConverter(dataTypeConverterClz.newInstance())
           val inputFormat = new CarbonStreamInputFormat
-          inputFormat.setVectorReader(vectorReader)
+          inputFormat.setIsVectorReader(vectorReader)
           inputFormat.setInputMetricsStats(inputMetricsStats)
           model.setStatisticsRecorder(
             CarbonTimeStatisticsFactory.createExecutorRecorder(model.getQueryId))

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
index 3cbfab2..ab6e320 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
@@ -49,13 +49,8 @@ import org.apache.carbondata.processing.loading.model.CarbonLoadModel
 import org.apache.carbondata.processing.merger.{CompactionResultSortProcessor, CompactionType}
 import org.apache.carbondata.processing.util.CarbonLoaderUtil
 import org.apache.carbondata.spark.{HandoffResult, HandoffResultImpl}
-<<<<<<< 2f537b724f6f03ab40c95f7ecc8ebd38f6500099
 import org.apache.carbondata.spark.util.CommonUtil
-import org.apache.carbondata.streaming.{CarbonStreamInputFormat, CarbonStreamRecordReader}
-=======
-import org.apache.carbondata.spark.util.{CommonUtil, SparkDataTypeConverterImpl}
 import org.apache.carbondata.streaming.CarbonStreamInputFormat
->>>>>>> [CARBONDATA-2532][Integration] Carbon to support spark 2.3 version, ColumnVector Interface
 
 
 /**
@@ -167,11 +162,11 @@ class StreamHandoffRDD[K, V](
     val format = new CarbonTableInputFormat[Array[Object]]()
     val model = format.createQueryModel(inputSplit, attemptContext)
     val inputFormat = new CarbonStreamInputFormat
-    val streamReader = inputFormat.createRecordReader(inputSplit, attemptContext)
-      .asInstanceOf[RecordReader[Void, Any]]
-    inputFormat.setVectorReader(false)
+    inputFormat.setIsVectorReader(false)
     inputFormat.setModel(model)
     inputFormat.setUseRawRow(true)
+    val streamReader = inputFormat.createRecordReader(inputSplit, attemptContext)
+      .asInstanceOf[RecordReader[Void, Any]]
     streamReader.initialize(inputSplit, attemptContext)
     val iteratorList = new util.ArrayList[RawResultIterator](1)
     iteratorList.add(new StreamingRawResultIterator(streamReader))

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
index 196baa6..5762906 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
@@ -121,7 +121,7 @@ class CarbonAppendableStreamSink(
         className = sparkSession.sessionState.conf.streamingFileCommitProtocolClass,
         jobId = batchId.toString,
         outputPath = fileLogPath,
-        isAppend = false)
+        false)
 
       committer match {
         case manifestCommitter: ManifestFileCommitProtocol =>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/PlanTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/PlanTest.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/PlanTest.scala
index 9883607..00156e9 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/PlanTest.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/PlanTest.scala
@@ -26,11 +26,6 @@ import org.apache.spark.sql.catalyst.util._
  */
 class PlanTest extends CarbonFunSuite {
 
-  /** Fails the test if the two expressions do not match */
-  protected def compareExpressions(e1: Expression, e2: Expression): Unit = {
-    comparePlans(Filter(e1, OneRowRelation), Filter(e2, OneRowRelation))
-  }
-
   /** Fails the test if the two plans do not match */
   protected def comparePlans(plan1: LogicalPlan, plan2: LogicalPlan) {
     val normalized1 = normalizeExprIds(plan1)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
index b5fda85..b7d47a0 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/util/SparkSQLUtil.scala
@@ -17,9 +17,13 @@
 
 package org.apache.spark.sql.util
 
+import java.lang.reflect.Method
+
 import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.internal.SessionState
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, Statistics}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.{SessionState, SQLConf}
+import org.apache.spark.util.{CarbonReflectionUtils, SparkUtil}
 
 object SparkSQLUtil {
   def sessionState(sparkSession: SparkSession): SessionState = sparkSession.sessionState
@@ -31,4 +35,68 @@ object SparkSQLUtil {
   def getSparkSession: SparkSession = {
     SparkSession.getDefaultSession.get
   }
+
+  def invokeStatsMethod(logicalPlanObj: LogicalPlan, conf: SQLConf): Statistics = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val method: Method = logicalPlanObj.getClass.getMethod("stats", classOf[SQLConf])
+      method.invoke(logicalPlanObj, conf).asInstanceOf[Statistics]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      val method: Method = logicalPlanObj.getClass.getMethod("stats")
+      method.invoke(logicalPlanObj).asInstanceOf[Statistics]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
+  def getReorderJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin";
+      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";
+      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
+        .asInstanceOf[Rule[LogicalPlan]]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
+  def getEliminateOuterJoinObj(conf: SQLConf): Rule[LogicalPlan] = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin";
+      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";
+      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
+        .asInstanceOf[Rule[LogicalPlan]]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
+  def getNullPropagationObj(conf: SQLConf): Rule[LogicalPlan] = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation";
+      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";
+      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
+        .asInstanceOf[Rule[LogicalPlan]]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
+  def getCheckCartesianProductsObj(conf: SQLConf): Rule[LogicalPlan] = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";
+      CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts$";
+      CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
+        .asInstanceOf[Rule[LogicalPlan]]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/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 34eeded..32cd201 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
@@ -17,24 +17,25 @@
 
 package org.apache.spark.util
 
-import java.lang.reflect.Field
+import java.lang.reflect.Method
 
 import scala.reflect.runtime._
 import scala.reflect.runtime.universe._
 
-import org.apache.spark.SPARK_VERSION
-import org.apache.spark.SparkContext
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.{SPARK_VERSION, SparkContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
 import org.apache.spark.sql.catalyst.parser.AstBuilder
 import org.apache.spark.sql.catalyst.plans.logical.{InsertIntoTable, LogicalPlan, SubqueryAlias}
-import org.apache.spark.sql.execution.datasources.LogicalRelation
-import org.apache.spark.sql.sources.BaseRelation
+import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan}
+import org.apache.spark.sql.execution.datasources.{DataSource, LogicalRelation}
+import org.apache.spark.sql.sources.{BaseRelation, Filter}
 
-import org.apache.carbondata.common.logging.LogServiceFactory
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 
 /**
@@ -61,12 +62,12 @@ object CarbonReflectionUtils {
       tableIdentifier: TableIdentifier,
       tableAlias: Option[String] = None): UnresolvedRelation = {
     val className = "org.apache.spark.sql.catalyst.analysis.UnresolvedRelation"
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       createObject(
         className,
         tableIdentifier,
         tableAlias)._1.asInstanceOf[UnresolvedRelation]
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
       createObject(
         className,
         tableIdentifier)._1.asInstanceOf[UnresolvedRelation]
@@ -79,13 +80,13 @@ object CarbonReflectionUtils {
       relation: LogicalPlan,
       view: Option[TableIdentifier]): SubqueryAlias = {
     val className = "org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias"
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       createObject(
         className,
         alias.getOrElse(""),
         relation,
         Option(view))._1.asInstanceOf[SubqueryAlias]
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
       createObject(
         className,
         alias.getOrElse(""),
@@ -101,7 +102,7 @@ object CarbonReflectionUtils {
       overwrite: Boolean,
       ifPartitionNotExists: Boolean): InsertIntoTable = {
     val className = "org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable"
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       val overwriteOptions = createObject(
         "org.apache.spark.sql.catalyst.plans.logical.OverwriteOptions",
         overwrite.asInstanceOf[Object], Map.empty.asInstanceOf[Object])._1.asInstanceOf[Object]
@@ -112,7 +113,7 @@ object CarbonReflectionUtils {
         query,
         overwriteOptions,
         ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionXandAbove("2.2") ) {
       createObject(
         className,
         table,
@@ -127,20 +128,28 @@ object CarbonReflectionUtils {
 
   def getLogicalRelation(relation: BaseRelation,
       expectedOutputAttributes: Seq[Attribute],
-      catalogTable: Option[CatalogTable]): LogicalRelation = {
+      catalogTable: Option[CatalogTable],
+      isStreaming: Boolean): LogicalRelation = {
     val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       createObject(
         className,
         relation,
         Some(expectedOutputAttributes),
         catalogTable)._1.asInstanceOf[LogicalRelation]
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionEqualTo("2.2")) {
       createObject(
         className,
         relation,
         expectedOutputAttributes,
         catalogTable)._1.asInstanceOf[LogicalRelation]
+    } else if (SparkUtil.isSparkVersionEqualTo("2.3")) {
+      createObject(
+        className,
+        relation,
+        expectedOutputAttributes,
+        catalogTable,
+        isStreaming.asInstanceOf[Object])._1.asInstanceOf[LogicalRelation]
     } else {
       throw new UnsupportedOperationException("Unsupported Spark version")
     }
@@ -183,27 +192,23 @@ object CarbonReflectionUtils {
   def getAstBuilder(conf: Object,
       sqlParser: Object,
       sparkSession: SparkSession): AstBuilder = {
-    if (SPARK_VERSION.startsWith("2.1") || SPARK_VERSION.startsWith("2.2")) {
-      val className = sparkSession.sparkContext.conf.get(
-        CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,
-        "org.apache.spark.sql.hive.CarbonSqlAstBuilder")
-      createObject(className,
-        conf,
-        sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]
-    } else {
-      throw new UnsupportedOperationException("Spark version not supported")
-    }
+    val className = sparkSession.sparkContext.conf.get(
+      CarbonCommonConstants.CARBON_SQLASTBUILDER_CLASSNAME,
+      "org.apache.spark.sql.hive.CarbonSqlAstBuilder")
+    createObject(className,
+      conf,
+      sqlParser, sparkSession)._1.asInstanceOf[AstBuilder]
   }
 
   def getSessionState(sparkContext: SparkContext,
       carbonSession: Object,
       useHiveMetaStore: Boolean): Any = {
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       val className = sparkContext.conf.get(
         CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
         "org.apache.spark.sql.hive.CarbonSessionState")
       createObject(className, carbonSession)._1
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
       if (useHiveMetaStore) {
         val className = sparkContext.conf.get(
           CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
@@ -225,12 +230,12 @@ object CarbonReflectionUtils {
   }
 
   def hasPredicateSubquery(filterExp: Expression) : Boolean = {
-    if (SPARK_VERSION.startsWith("2.1")) {
+    if (SparkUtil.isSparkVersionEqualTo("2.1")) {
       val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.PredicateSubquery")
       val method = tuple.getMethod("hasPredicateSubquery", classOf[Expression])
       val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]
       hasSubquery
-    } else if (SPARK_VERSION.startsWith("2.2")) {
+    } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
       val tuple = Class.forName("org.apache.spark.sql.catalyst.expressions.SubqueryExpression")
       val method = tuple.getMethod("hasInOrExistsSubquery", classOf[Expression])
       val hasSubquery : Boolean = method.invoke(tuple, filterExp).asInstanceOf[Boolean]
@@ -248,6 +253,54 @@ object CarbonReflectionUtils {
     isFormatted
   }
 
+  def getRowDataSourceScanExecObj(relation: LogicalRelation,
+      output: Seq[Attribute],
+      pushedFilters: Seq[Filter],
+      handledFilters: Seq[Filter],
+      rdd: RDD[InternalRow],
+      partition: Partitioning,
+      metadata: Map[String, String]): RowDataSourceScanExec = {
+    val className = "org.apache.spark.sql.execution.RowDataSourceScanExec"
+    if (SparkUtil.isSparkVersionEqualTo("2.1") || SparkUtil.isSparkVersionEqualTo("2.2")) {
+      createObject(className, output, rdd, relation.relation,
+        partition, metadata,
+        relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]
+    } else if (SparkUtil.isSparkVersionXandAbove("2.3")) {
+      createObject(className, output, output.map(output.indexOf),
+        pushedFilters.toSet, handledFilters.toSet, rdd,
+        relation.relation,
+        relation.catalogTable.map(_.identifier))._1.asInstanceOf[RowDataSourceScanExec]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
+  def invokewriteAndReadMethod(dataSourceObj: DataSource,
+      dataFrame: DataFrame,
+      data: LogicalPlan,
+      session: SparkSession,
+      mode: SaveMode,
+      query: LogicalPlan,
+      physicalPlan: SparkPlan): BaseRelation = {
+    if (SparkUtil.isSparkVersionEqualTo("2.2")) {
+      val method: Method = dataSourceObj.getClass
+        .getMethod("writeAndRead", classOf[SaveMode], classOf[DataFrame])
+      method.invoke(dataSourceObj, mode, dataFrame)
+        .asInstanceOf[BaseRelation]
+    } else if (SparkUtil.isSparkVersionEqualTo("2.3")) {
+      val method: Method = dataSourceObj.getClass
+        .getMethod("writeAndRead",
+          classOf[SaveMode],
+          classOf[LogicalPlan],
+          classOf[Seq[Attribute]],
+          classOf[SparkPlan])
+      method.invoke(dataSourceObj, mode, query, query.output, physicalPlan)
+        .asInstanceOf[BaseRelation]
+    } else {
+      throw new UnsupportedOperationException("Spark version not supported")
+    }
+  }
+
   def createObject(className: String, conArgs: Object*): (Any, Class[_]) = {
     val clazz = Utils.classForName(className)
     val ctor = clazz.getConstructors.head
@@ -255,4 +308,10 @@ object CarbonReflectionUtils {
     (ctor.newInstance(conArgs: _*), clazz)
   }
 
+  def createObjectOfPrivateConstructor(className: String, conArgs: Object*): (Any, Class[_]) = {
+    val clazz = Utils.classForName(className)
+    val ctor = clazz.getDeclaredConstructors.head
+    ctor.setAccessible(true)
+    (ctor.newInstance(conArgs: _*), clazz)
+  }
 }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/main/scala/org/apache/spark/util/SparkUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/SparkUtil.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/SparkUtil.scala
deleted file mode 100644
index 4635fc7..0000000
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/SparkUtil.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.util
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.io.{LongWritable, Text}
-import org.apache.hadoop.mapred.JobConf
-import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
-import org.apache.spark.{SparkContext, TaskContext}
-import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.rdd.{NewHadoopPartition, NewHadoopRDD}
-
-import org.apache.carbondata.processing.loading.csvinput.BlockDetails
-
-/*
- * this object use to handle file splits
- */
-object SparkUtil {
-
-  def setTaskContext(context: TaskContext): Unit = {
-    val localThreadContext = TaskContext.get()
-    if (localThreadContext == null) {
-      TaskContext.setTaskContext(context)
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-common/src/test/scala/org/apache/spark/util/SparkUtilTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/test/scala/org/apache/spark/util/SparkUtilTest.scala b/integration/spark-common/src/test/scala/org/apache/spark/util/SparkUtilTest.scala
new file mode 100644
index 0000000..4810db1
--- /dev/null
+++ b/integration/spark-common/src/test/scala/org/apache/spark/util/SparkUtilTest.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util
+
+import org.apache.spark.SPARK_VERSION
+import org.scalatest.FunSuite
+
+class SparkUtilTest extends FunSuite{
+
+  test("Test Spark Version API with X and Above") {
+    if (SPARK_VERSION.startsWith("2.1")) {
+      assert(SparkUtil.isSparkVersionXandAbove("2.1"))
+      assert(!SparkUtil.isSparkVersionXandAbove("2.2"))
+      assert(!SparkUtil.isSparkVersionXandAbove("2.3"))
+    } else if (SPARK_VERSION.startsWith("2.2")) {
+      assert(SparkUtil.isSparkVersionXandAbove("2.1"))
+      assert(SparkUtil.isSparkVersionXandAbove("2.2"))
+      assert(!SparkUtil.isSparkVersionXandAbove("2.3"))
+    } else {
+      assert(SparkUtil.isSparkVersionXandAbove("2.1"))
+      assert(SparkUtil.isSparkVersionXandAbove("2.2"))
+      assert(SparkUtil.isSparkVersionXandAbove("2.3"))
+      assert(!SparkUtil.isSparkVersionXandAbove("2.4"))
+    }
+  }
+
+  test("Test Spark Version API Equal to X") {
+    if (SPARK_VERSION.startsWith("2.1")) {
+      assert(SparkUtil.isSparkVersionEqualTo("2.1"))
+      assert(!SparkUtil.isSparkVersionEqualTo("2.2"))
+      assert(!SparkUtil.isSparkVersionEqualTo("2.3"))
+    } else if (SPARK_VERSION.startsWith("2.2")) {
+      assert(!SparkUtil.isSparkVersionEqualTo("2.1"))
+      assert(SparkUtil.isSparkVersionEqualTo("2.2"))
+      assert(!SparkUtil.isSparkVersionEqualTo("2.3"))
+    } else {
+      assert(!SparkUtil.isSparkVersionEqualTo("2.1"))
+      assert(!SparkUtil.isSparkVersionEqualTo("2.2"))
+      assert(SparkUtil.isSparkVersionEqualTo("2.3"))
+      assert(!SparkUtil.isSparkVersionEqualTo("2.4"))
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-datasource/pom.xml
----------------------------------------------------------------------
diff --git a/integration/spark-datasource/pom.xml b/integration/spark-datasource/pom.xml
index 38cf629..3d017c8 100644
--- a/integration/spark-datasource/pom.xml
+++ b/integration/spark-datasource/pom.xml
@@ -39,6 +39,15 @@
       <groupId>org.apache.carbondata</groupId>
       <artifactId>carbondata-hadoop</artifactId>
       <version>${project.version}</version>
+      <exclusions>
+        <!-- need to Exclude net.jpountz jar from this project.
+         Spark has changed this jar to org.lz4:lz4-java
+         net.jpountz and org.lz4 has same class Name -->
+        <exclusion>
+          <groupId>net.jpountz.lz4</groupId>
+          <artifactId>lz4</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.carbondata</groupId>
@@ -192,5 +201,128 @@
         <maven.test.skip>true</maven.test.skip>
       </properties>
     </profile>
+    <profile>
+      <id>spark-2.1</id>
+      <properties>
+        <spark.version>2.1.0</spark.version>
+        <scala.binary.version>2.11</scala.binary.version>
+        <scala.version>2.11.8</scala.version>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-compiler-plugin</artifactId>
+            <configuration>
+              <excludes>
+                <exclude>src/main/spark2.3plus</exclude>
+              </excludes>
+            </configuration>
+          </plugin>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <version>3.0.0</version>
+            <executions>
+              <execution>
+                <id>add-source</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>src/main/spark2.1andspark2.2</source>
+                  </sources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+    <profile>
+      <id>spark-2.2</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
+      <properties>
+        <spark.version>2.2.1</spark.version>
+        <scala.binary.version>2.11</scala.binary.version>
+        <scala.version>2.11.8</scala.version>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-compiler-plugin</artifactId>
+            <configuration>
+              <excludes>
+                <exclude>src/main/spark2.3plus</exclude>
+              </excludes>
+            </configuration>
+          </plugin>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <version>3.0.0</version>
+            <executions>
+              <execution>
+                <id>add-source</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>src/main/spark2.1andspark2.2</source>
+                  </sources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+    <profile>
+      <id>spark-2.3</id>
+      <properties>
+        <spark.version>2.3.1</spark.version>
+        <scala.binary.version>2.11</scala.binary.version>
+        <scala.version>2.11.8</scala.version>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-compiler-plugin</artifactId>
+            <configuration>
+              <excludes>
+                <exclude>src/main/spark2.1andspark2.2</exclude>
+              </excludes>
+            </configuration>
+          </plugin>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <version>3.0.0</version>
+            <executions>
+              <execution>
+                <id>add-source</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>src/main/spark2.3plus</source>
+                  </sources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
   </profiles>
 </project>

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java
----------------------------------------------------------------------
diff --git a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java
index 7bab117..bc5a387 100644
--- a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java
+++ b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java
@@ -30,7 +30,7 @@ import org.apache.spark.sql.types.Decimal;
 
 class ColumnarVectorWrapper implements CarbonColumnVector {
 
-  private CarbonVectorProxy writableColumnVector;
+  private CarbonVectorProxy sparkColumnVectorProxy;
 
   private boolean[] filteredRows;
 
@@ -38,32 +38,36 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
 
   private int ordinal;
 
+  private boolean isDictionary;
+
   private boolean filteredRowsExist;
 
   private DataType blockDataType;
 
+  private CarbonColumnVector dictionaryVector;
+
   ColumnarVectorWrapper(CarbonVectorProxy writableColumnVector,
-                        boolean[] filteredRows, int ordinal) {
-    this.writableColumnVector = writableColumnVector;
+      boolean[] filteredRows, int ordinal) {
+    this.sparkColumnVectorProxy = writableColumnVector;
     this.filteredRows = filteredRows;
     this.ordinal = ordinal;
   }
 
   @Override public void putBoolean(int rowId, boolean value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putBoolean(counter++, value, ordinal);
+      sparkColumnVectorProxy.putBoolean(counter++, value, ordinal);
     }
   }
 
   @Override public void putFloat(int rowId, float value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putFloat(counter++, value,ordinal);
+      sparkColumnVectorProxy.putFloat(counter++, value,ordinal);
     }
   }
 
   @Override public void putShort(int rowId, short value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putShort(counter++, value, ordinal);
+      sparkColumnVectorProxy.putShort(counter++, value, ordinal);
     }
   }
 
@@ -71,18 +75,22 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          writableColumnVector.putShort(counter++, value, ordinal);
+          sparkColumnVectorProxy.putShort(counter++, value, ordinal);
         }
         rowId++;
       }
     } else {
-      writableColumnVector.putShorts(rowId, count, value, ordinal);
+      sparkColumnVectorProxy.putShorts(rowId, count, value, ordinal);
     }
   }
 
   @Override public void putInt(int rowId, int value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putInt(counter++, value, ordinal);
+      if (isDictionary) {
+        sparkColumnVectorProxy.putDictionaryInt(counter++, value, ordinal);
+      } else {
+        sparkColumnVectorProxy.putInt(counter++, value, ordinal);
+      }
     }
   }
 
@@ -90,18 +98,18 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          writableColumnVector.putInt(counter++, value, ordinal);
+          sparkColumnVectorProxy.putInt(counter++, value, ordinal);
         }
         rowId++;
       }
     } else {
-      writableColumnVector.putInts(rowId, count, value, ordinal);
+      sparkColumnVectorProxy.putInts(rowId, count, value, ordinal);
     }
   }
 
   @Override public void putLong(int rowId, long value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putLong(counter++, value, ordinal);
+      sparkColumnVectorProxy.putLong(counter++, value, ordinal);
     }
   }
 
@@ -109,19 +117,19 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          writableColumnVector.putLong(counter++, value, ordinal);
+          sparkColumnVectorProxy.putLong(counter++, value, ordinal);
         }
         rowId++;
       }
     } else {
-      writableColumnVector.putLongs(rowId, count, value, ordinal);
+      sparkColumnVectorProxy.putLongs(rowId, count, value, ordinal);
     }
   }
 
   @Override public void putDecimal(int rowId, BigDecimal value, int precision) {
     if (!filteredRows[rowId]) {
       Decimal toDecimal = Decimal.apply(value);
-      writableColumnVector.putDecimal(counter++, toDecimal, precision, ordinal);
+      sparkColumnVectorProxy.putDecimal(counter++, toDecimal, precision, ordinal);
     }
   }
 
@@ -129,7 +137,7 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     Decimal decimal = Decimal.apply(value);
     for (int i = 0; i < count; i++) {
       if (!filteredRows[rowId]) {
-        writableColumnVector.putDecimal(counter++, decimal, precision, ordinal);
+        sparkColumnVectorProxy.putDecimal(counter++, decimal, precision, ordinal);
       }
       rowId++;
     }
@@ -137,7 +145,7 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
 
   @Override public void putDouble(int rowId, double value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putDouble(counter++, value, ordinal);
+      sparkColumnVectorProxy.putDouble(counter++, value, ordinal);
     }
   }
 
@@ -145,25 +153,25 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          writableColumnVector.putDouble(counter++, value, ordinal);
+          sparkColumnVectorProxy.putDouble(counter++, value, ordinal);
         }
         rowId++;
       }
     } else {
-      writableColumnVector.putDoubles(rowId, count, value, ordinal);
+      sparkColumnVectorProxy.putDoubles(rowId, count, value, ordinal);
     }
   }
 
   @Override public void putBytes(int rowId, byte[] value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putByteArray(counter++, value, ordinal);
+      sparkColumnVectorProxy.putByteArray(counter++, value, ordinal);
     }
   }
 
   @Override public void putBytes(int rowId, int count, byte[] value) {
     for (int i = 0; i < count; i++) {
       if (!filteredRows[rowId]) {
-        writableColumnVector.putByteArray(counter++, value, ordinal);
+        sparkColumnVectorProxy.putByteArray(counter++, value, ordinal);
       }
       rowId++;
     }
@@ -171,13 +179,13 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
 
   @Override public void putBytes(int rowId, int offset, int length, byte[] value) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putByteArray(counter++, value, offset, length, ordinal);
+      sparkColumnVectorProxy.putByteArray(counter++, value, offset, length, ordinal);
     }
   }
 
   @Override public void putNull(int rowId) {
     if (!filteredRows[rowId]) {
-      writableColumnVector.putNull(counter++, ordinal);
+      sparkColumnVectorProxy.putNull(counter++, ordinal);
     }
   }
 
@@ -185,18 +193,18 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          writableColumnVector.putNull(counter++, ordinal);
+          sparkColumnVectorProxy.putNull(counter++, ordinal);
         }
         rowId++;
       }
     } else {
-      writableColumnVector.putNulls(rowId, count,ordinal);
+      sparkColumnVectorProxy.putNulls(rowId, count,ordinal);
     }
   }
 
   @Override public void putNotNull(int rowId) {
     if (!filteredRows[rowId]) {
-      columnVector.putNotNull(counter++);
+      sparkColumnVectorProxy.putNotNull(counter++,ordinal);
     }
   }
 
@@ -204,17 +212,17 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     if (filteredRowsExist) {
       for (int i = 0; i < count; i++) {
         if (!filteredRows[rowId]) {
-          columnVector.putNotNull(counter++);
+          sparkColumnVectorProxy.putNotNull(counter++, ordinal);
         }
         rowId++;
       }
     } else {
-      columnVector.putNotNulls(rowId, count);
+      sparkColumnVectorProxy.putNotNulls(rowId, count, ordinal);
     }
   }
 
   @Override public boolean isNull(int rowId) {
-    return writableColumnVector.isNullAt(rowId,ordinal);
+    return sparkColumnVectorProxy.isNullAt(rowId,ordinal);
   }
 
   @Override public void putObject(int rowId, Object obj) {
@@ -236,7 +244,7 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
 
   @Override public DataType getType() {
     return CarbonSparkDataSourceUtil
-        .convertSparkToCarbonDataType(writableColumnVector.dataType(ordinal));
+        .convertSparkToCarbonDataType(sparkColumnVectorProxy.dataType(ordinal));
   }
 
   @Override
@@ -249,20 +257,32 @@ class ColumnarVectorWrapper implements CarbonColumnVector {
     this.blockDataType = blockDataType;
   }
 
-  @Override public void setFilteredRowsExist(boolean filteredRowsExist) {
+  @Override
+  public void setFilteredRowsExist(boolean filteredRowsExist) {
     this.filteredRowsExist = filteredRowsExist;
   }
 
   @Override public void setDictionary(CarbonDictionary dictionary) {
     if (dictionary == null) {
-      columnVector.setDictionary(null);
+      sparkColumnVectorProxy.setDictionary(null, ordinal);
     } else {
-      columnVector.setDictionary(new CarbonDictionaryWrapper(Encoding.PLAIN, dictionary));
+      sparkColumnVectorProxy
+          .setDictionary(new CarbonDictionaryWrapper(Encoding.PLAIN, dictionary),ordinal);
     }
   }
 
+  private void  setDictionaryType(boolean type) {
+    this.isDictionary = type;
+  }
+
   @Override public boolean hasDictionary() {
-    return columnVector.hasDictionary();
+    return sparkColumnVectorProxy.hasDictionary(ordinal);
+  }
+
+  public void reserveDictionaryIds() {
+    sparkColumnVectorProxy.reserveDictionaryIds(sparkColumnVectorProxy.numRows(), ordinal);
+    dictionaryVector = new ColumnarVectorWrapper(sparkColumnVectorProxy, filteredRows, ordinal);
+    ((ColumnarVectorWrapper) dictionaryVector).isDictionary = true;
   }
 
   @Override public CarbonColumnVector getDictionaryVector() {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/74c3eb10/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
----------------------------------------------------------------------
diff --git a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
index 7c98608..d6b6b7e 100644
--- a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
+++ b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
@@ -51,16 +51,13 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.spark.memory.MemoryMode;
-<<<<<<< 2f537b724f6f03ab40c95f7ecc8ebd38f6500099:integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
 import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.execution.vectorized.ColumnVectorUtils;
-import org.apache.spark.sql.execution.vectorized.ColumnarBatch;
-=======
 import org.apache.spark.sql.CarbonVectorProxy;
->>>>>>> [CARBONDATA-2532][Integration] Carbon to support spark 2.3 version, ColumnVector Interface:integration/spark2/src/main/java/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
 import org.apache.spark.sql.types.DecimalType;
 import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
 
 /**
  * A specialized RecordReader that reads into InternalRows or ColumnarBatches directly using the
@@ -282,22 +279,23 @@ public class VectorizedCarbonRecordReader extends AbstractRecordReader<Object> {
         schema = schema.add(field);
       }
     }
-    columnarBatch = ColumnarBatch.allocate(schema, memMode);
+    vectorProxy = new CarbonVectorProxy(DEFAULT_MEMORY_MODE,schema,DEFAULT_BATCH_SIZE);
     if (partitionColumns != null) {
       int partitionIdx = fields.length;
       for (int i = 0; i < partitionColumns.fields().length; i++) {
-        ColumnVectorUtils.populate(columnarBatch.column(i + partitionIdx), partitionValues, i);
-        columnarBatch.column(i + partitionIdx).setIsConstant();
+        ColumnVectorUtils.populate(vectorProxy.column(i + partitionIdx), partitionValues, i);
+        vectorProxy.column(i + partitionIdx).setIsConstant();
       }
     }
-    vectorProxy = new CarbonVectorProxy(MemoryMode.OFF_HEAP,DEFAULT_BATCH_SIZE,fields);
     CarbonColumnVector[] vectors = new CarbonColumnVector[fields.length];
     boolean[] filteredRows = new boolean[vectorProxy.numRows()];
     for (int i = 0; i < fields.length; i++) {
-    if (isNoDictStringField[i]) {
-      vectorProxy.reserveDictionaryIds(vectorProxy.numRows(), i);
-      }
       vectors[i] = new ColumnarVectorWrapper(vectorProxy, filteredRows, i);
+      if (isNoDictStringField[i]) {
+        if (vectors[i] instanceof ColumnarVectorWrapper) {
+          ((ColumnarVectorWrapper) vectors[i]).reserveDictionaryIds();
+        }
+      }
     }
     carbonColumnarBatch = new CarbonColumnarBatch(vectors, vectorProxy.numRows(), filteredRows);
   }


Mime
View raw message