carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject carbondata git commit: [CARBONDATA-1868][Spark-2.2]Carbon-Spark2.2 Integration Phase 2
Date Thu, 07 Dec 2017 14:41:09 GMT
Repository: carbondata
Updated Branches:
  refs/heads/master 99ec112f2 -> 78f8aae53


[CARBONDATA-1868][Spark-2.2]Carbon-Spark2.2 Integration Phase 2

Enable Spark2.2 Integration for feature like

Alter Table Add and Modify Columns
Update | Delete
SubQuery Handling
SubqueryAlias Handling
Preaggregate handling for Spark2.2
Small Bug Fixes.

This closes #1595


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

Branch: refs/heads/master
Commit: 78f8aae53f59447857626f8331a807d7fbf4949e
Parents: 99ec112
Author: sounakr <sounakr@gmail.com>
Authored: Wed Nov 29 22:34:08 2017 +0530
Committer: ravipesala <ravi.pesala@gmail.com>
Committed: Thu Dec 7 20:10:47 2017 +0530

----------------------------------------------------------------------
 .../TestPreAggregateTableSelection.scala        | 16 +++-
 .../spark/sql/catalyst/CarbonDDLSqlParser.scala |  2 +-
 .../spark/util/CarbonReflectionUtils.scala      | 17 ++++
 .../strategy/CarbonLateDecodeStrategy.scala     | 21 ++++-
 .../spark/sql/hive/CarbonAnalysisRules.scala    | 35 ++++++-
 .../spark/sql/hive/CarbonFileMetastore.scala    |  9 +-
 .../sql/hive/CarbonPreAggregateRules.scala      | 16 ++--
 .../src/main/spark2.2/CarbonSessionState.scala  | 97 +++++++++++++++++++-
 .../BooleanDataTypesFilterTest.scala            | 44 ++++++---
 .../booleantype/BooleanDataTypesLoadTest.scala  | 94 +++++++++++++------
 .../partition/TestAlterPartitionTable.scala     |  1 +
 .../AlterTableValidationTestCase.scala          | 48 +++++-----
 12 files changed, 311 insertions(+), 89 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index 5dfe447..dc117a5 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -16,8 +16,10 @@
  */
 package org.apache.carbondata.integration.spark.testsuite.preaggregate
 
+import org.apache.spark.sql.catalyst.catalog.CatalogRelation
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hive.CarbonRelation
 import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, Row}
 import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.BeforeAndAfterAll
@@ -52,7 +54,7 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll
{
 
   test("test sum and avg on same column should give proper results") {
     val df = sql("select name, sum(id), avg(id) from maintable group by name")
-    checkAnswer(df, Seq(Row("david",1,1.0), Row("jarry",6,3.0), Row("kunal",4,4.0), Row("eason",2,2.0),Row("vishal",4,4.0)))
+    checkAnswer(df, Seq(Row("david",1,1.0), Row("jarry",6,3.0), Row("kunal",4,4.0), Row("eason",2,2.0),
Row("vishal",4,4.0)))
   }
 
 
@@ -61,12 +63,12 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll
{
     preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0")
   }
 
-  ignore("test PreAggregate table selection 2") {
+  test("test PreAggregate table selection 2") {
     val df = sql("select name from mainTable where name in (select name from mainTable) group
by name")
     preAggTableValidator(df.queryExecution.analyzed, "mainTable")
   }
 
-  ignore("test PreAggregate table selection 3") {
+  test("test PreAggregate table selection 3") {
     val df = sql("select name from mainTable where name in (select name from mainTable group
by name) group by name")
     preAggTableValidator(df.queryExecution.analyzed, "mainTable")
   }
@@ -196,6 +198,14 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll
{
     plan.transform {
       // first check if any preaTable1 scala function is applied it is present is in plan
       // then call is from create preaTable1regate table class so no need to transform the
query plan
+      case ca:CarbonRelation =>
+        if (ca.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+          val relation = ca.asInstanceOf[CarbonDatasourceHadoopRelation]
+          if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
+            isValidPlan = true
+          }
+        }
+        ca
       case logicalRelation:LogicalRelation =>
         if(logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]) {
           val relation = logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
index 081e5cf..f405902 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
@@ -1171,7 +1171,7 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser
{
    * @param values
    * @return
    */
-  protected def parseDataType(dataType: String, values: Option[List[(Int, Int)]]): DataTypeInfo
= {
+  def parseDataType(dataType: String, values: Option[List[(Int, Int)]]): DataTypeInfo = {
     dataType match {
       case "bigint" | "long" =>
         if (values.isDefined) {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/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 9a3f28e..d88f190 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
@@ -25,6 +25,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.parser.AstBuilder
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
 
@@ -158,6 +159,22 @@ object CarbonReflectionUtils {
     }
   }
 
+  def hasPredicateSubquery(filterExp: Expression) : Boolean = {
+    if (SPARK_VERSION.startsWith("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")) {
+      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]
+      hasSubquery
+    } 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

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonLateDecodeStrategy.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonLateDecodeStrategy.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonLateDecodeStrategy.scala
index 2c23c57..79bbfb0 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonLateDecodeStrategy.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/CarbonLateDecodeStrategy.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.optimizer.CarbonDecoderRelation
 import org.apache.spark.sql.sources.{BaseRelation, Filter}
-import org.apache.spark.sql.types.{AtomicType, IntegerType, StringType}
+import org.apache.spark.sql.types._
 import org.apache.spark.sql.CarbonExpressions.{MatchCast => Cast}
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
@@ -40,6 +40,7 @@ import org.apache.carbondata.core.metadata.schema.BucketingInfo
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable
 import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager
 import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.format.DataType
 import org.apache.carbondata.spark.CarbonAliasDecoderRelation
 import org.apache.carbondata.spark.rdd.CarbonScanRDD
 import org.apache.carbondata.spark.util.CarbonScalaUtil
@@ -409,23 +410,39 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
     }
   }
 
+  private def isComplexAttribute(attribute: Attribute) = attribute.dataType match {
+    case ArrayType(dataType, _) => true
+    case StructType(_) => true
+    case _ => false
+  }
+
   protected[sql] def selectFilters(
       relation: BaseRelation,
       predicates: Seq[Expression]): (Seq[Expression], Seq[Filter]) = {
 
+    // In case of ComplexType dataTypes no filters should be pushed down. IsNotNull is being
+    // explicitly added by spark and pushed. That also has to be handled and pushed back
to
+    // Spark for handling.
+    val predicatesWithoutComplex = predicates.filter(predicate =>
+      predicate.collect {
+      case a: Attribute if isComplexAttribute(a) => a
+    }.size == 0 )
+
     // For conciseness, all Catalyst filter expressions of type `expressions.Expression`
below are
     // called `predicate`s, while all data source filters of type `sources.Filter` are simply
called
     // `filter`s.
 
     val translated: Seq[(Expression, Filter)] =
       for {
-        predicate <- predicates
+        predicate <- predicatesWithoutComplex
         filter <- translateFilter(predicate)
       } yield predicate -> filter
 
+
     // A map from original Catalyst expressions to corresponding translated data source filters.
     val translatedMap: Map[Expression, Filter] = translated.toMap
 
+
     // Catalyst predicate expressions that cannot be translated to data source filters.
     val unrecognizedPredicates = predicates.filterNot(translatedMap.contains)
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
index b595896..de5fa7e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalysisRules.scala
@@ -49,10 +49,26 @@ case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[Logica
 
       val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
 
+      val tableRelation = if (SPARK_VERSION.startsWith("2.1")) {
+        relation
+      } else if (SPARK_VERSION.startsWith("2.2")) {
+        alias match {
+          case Some(a) =>
+            CarbonReflectionUtils.getSubqueryAlias(
+              sparkSession,
+              alias,
+              relation,
+              Some(table.tableIdentifier))
+          case _ => relation
+        }
+      } else {
+        throw new UnsupportedOperationException("Unsupported Spark version.")
+      }
+
       CarbonReflectionUtils.getSubqueryAlias(
         sparkSession,
         alias,
-        Project(projList, relation),
+        Project(projList, tableRelation),
         Some(table.tableIdentifier))
     }
 
@@ -148,7 +164,22 @@ case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends
Rule[Logica
 
         val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
         // include tuple id in subquery
-        Project(projList, relation)
+        if (SPARK_VERSION.startsWith("2.1")) {
+          Project(projList, relation)
+        } else if (SPARK_VERSION.startsWith("2.2")) {
+          alias match {
+            case Some(a) =>
+              val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(
+                sparkSession,
+                alias,
+                relation,
+                Some(table.tableIdentifier))
+              Project(projList, subqueryAlias)
+            case _ => Project(projList, relation)
+          }
+        } else {
+          throw new UnsupportedOperationException("Unsupported Spark version.")
+        }
     }
     CarbonProjectForDeleteCommand(
       selectPlan,

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
index cdbdb10..82a9302 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
@@ -24,7 +24,7 @@ import scala.collection.mutable.ArrayBuffer
 
 import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
 import org.apache.spark.SPARK_VERSION
-import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonSource, SparkSession}
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
 import org.apache.spark.sql.CarbonExpressions.{CarbonSubqueryAlias => SubqueryAlias}
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
@@ -154,9 +154,10 @@ class CarbonFileMetastore extends CarbonMetaStore {
           case Some(name) if name.equals("org.apache.spark.sql.CarbonSource") => name
           case _ => throw new NoSuchTableException(database, tableIdentifier.table)
         }
-        new CarbonSource().createRelation(sparkSession.sqlContext,
-          catalogTable.storage.properties
-        ).asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation
+        val identifier: AbsoluteTableIdentifier = AbsoluteTableIdentifier.from(
+           catalogTable.location.toString, database, tableIdentifier.table)
+        CarbonEnv.getInstance(sparkSession).carbonMetastore.
+          createCarbonRelation(catalogTable.storage.properties, identifier, sparkSession)
       case _ => throw new NoSuchTableException(database, tableIdentifier.table)
     }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
index 426048f..09e66de 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala
@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeRef
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, LogicalRelation}
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.CarbonException
 import org.apache.spark.sql.CarbonExpressions.MatchCast
@@ -145,8 +145,7 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends
Rule
             carbonTable,
             tableName,
             list)
-          // TODO need to handle filter predicate subquery scenario
-          // isValidPlan = !PredicateSubquery.hasPredicateSubquery(filterExp)
+          isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
           // getting the columns from filter expression
           if(isValidPlan) {
             filterExp.transform {
@@ -210,8 +209,7 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends
Rule
             carbonTable,
             tableName,
             list)
-          // TODO need to handle filter predicate subquery scenario
-//          isValidPlan = !PredicateSubquery.hasPredicateSubquery(filterExp)
+          isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
           if (isValidPlan) {
             list ++
             extractQueryColumnForOrderBy(Some(projectList), sortOrders, carbonTable, tableName)
@@ -258,8 +256,7 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends
Rule
             carbonTable,
             tableName,
             list)
-          // TODO need to handle filter predicate subquery scenario
-//          isValidPlan = !PredicateSubquery.hasPredicateSubquery(filterExp)
+          isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp)
           if(isValidPlan) {
             list ++ extractQueryColumnForOrderBy(sortOrders = sortOrders,
               carbonTable = carbonTable,
@@ -311,8 +308,9 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends
Rule
               val relation = sparkSession.sessionState.catalog.lookupRelation(identifier)
               (selectedDataMapSchema, carbonRelation, relation)
             }.minBy(f => f._2.sizeInBytes)
+          val newRelation = new FindDataSourceTable(sparkSession).apply(relation)
           // transform the query plan based on selected child schema
-          transformPreAggQueryPlan(plan, aggDataMapSchema, relation)
+          transformPreAggQueryPlan(plan, aggDataMapSchema, newRelation)
         } else {
           plan
         }
@@ -1223,6 +1221,4 @@ case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends
Rule[Logi
         "Cannot insert into target table because number of columns mismatch")
     }
   }
-
 }
-

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/spark2.2/CarbonSessionState.scala b/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
index 61149eb..e10feb1 100644
--- a/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
+++ b/integration/spark2/src/main/spark2.2/CarbonSessionState.scala
@@ -16,18 +16,23 @@
  */
 package org.apache.spark.sql.hive
 
+
+import scala.collection.generic.SeqFactory
+
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry}
 import org.apache.spark.sql.catalyst.catalog._
-import org.apache.spark.sql.catalyst.expressions.ScalarSubquery
+import org.apache.spark.sql.catalyst.expressions.{Exists, In, ListQuery, ScalarSubquery}
 import org.apache.spark.sql.catalyst.optimizer.Optimizer
 import org.apache.spark.sql.catalyst.parser.ParserInterface
 import org.apache.spark.sql.catalyst.parser.ParserUtils.string
-import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{CreateHiveTableContext, CreateTableContext}
+import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{AddTableColumnsContext, ChangeColumnContext,
CreateHiveTableContext, CreateTableContext}
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, SubqueryAlias}
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.command.schema.{CarbonAlterTableAddColumnCommand, CarbonAlterTableDataTypeChangeCommand}
 import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, LogicalRelation,
PreWriteCheck, ResolveSQLOnFile, _}
 import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
 import org.apache.spark.sql.execution.{SparkOptimizer, SparkSqlAstBuilder}
@@ -35,10 +40,12 @@ import org.apache.spark.sql.hive.client.HiveClient
 import org.apache.spark.sql.internal.{SQLConf, SessionState}
 import org.apache.spark.sql.optimizer.CarbonLateDecodeRule
 import org.apache.spark.sql.parser.{CarbonHelperSqlAstBuilder, CarbonSpark2SqlParser, CarbonSparkSqlParser}
+import org.apache.spark.sql.types.DecimalType
 
 import org.apache.carbondata.core.datamap.DataMapStoreManager
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
 import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
 
 /**
  * This class will have carbon catalog and refresh the relation from cache if the carbontable
in
@@ -135,6 +142,18 @@ class CarbonSessionCatalog(
   }
 }
 
+
+class CarbonAnalyzer(catalog: SessionCatalog,
+    conf: SQLConf,
+    sparkSession: SparkSession,
+    analyzer: Analyzer) extends Analyzer(catalog, conf) {
+  override def execute(plan: LogicalPlan): LogicalPlan = {
+    val logicalPlan = analyzer.execute(plan)
+    CarbonPreAggregateQueryRules(sparkSession).apply(logicalPlan)
+  }
+}
+
+
 /**
  * Session state implementation to override sql parser and adding strategies
  *
@@ -186,7 +205,7 @@ class CarbonSessionStateBuilder(sparkSession: SparkSession,
 
   override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)
 
-  override protected def analyzer: Analyzer = {
+  override protected def analyzer: Analyzer = new CarbonAnalyzer(catalog, conf, sparkSession,
     new Analyzer(catalog, conf) {
 
       override val extendedResolutionRules: Seq[Rule[LogicalPlan]] =
@@ -194,7 +213,7 @@ class CarbonSessionStateBuilder(sparkSession: SparkSession,
         new FindDataSourceTable(session) +:
         new ResolveSQLOnFile(session) +:
         new CarbonIUDAnalysisRule(sparkSession) +:
-        new CarbonPreAggregateQueryRules(sparkSession) +:
+        CarbonPreAggregateDataLoadingRules +:
         new CarbonPreInsertionCasts(sparkSession) +: customResolutionRules
 
       override val extendedCheckRules: Seq[LogicalPlan => Unit] =
@@ -209,7 +228,7 @@ class CarbonSessionStateBuilder(sparkSession: SparkSession,
         HiveAnalysis +:
         customPostHocResolutionRules
     }
-  }
+  )
 
   override protected def newBuilder: NewBuilder = new CarbonSessionStateBuilder(_, _)
 
@@ -236,6 +255,23 @@ class CarbonOptimizer(
                 lr
             }
             ScalarSubquery(tPlan, s.children, s.exprId)
+          case e: Exists =>
+            val tPlan = e.plan.transform {
+              case lr: LogicalRelation
+                if lr.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
+                lr.relation.asInstanceOf[CarbonDatasourceHadoopRelation].isSubquery += true
+                lr
+            }
+            Exists(tPlan, e.children.map(_.canonicalized), e.exprId)
+
+          case In(value, Seq(l@ListQuery(sub, _, exprId))) =>
+            val tPlan = sub.transform {
+              case lr: LogicalRelation
+                if lr.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
+                lr.relation.asInstanceOf[CarbonDatasourceHadoopRelation].isSubquery += true
+                lr
+            }
+            In(value, Seq(ListQuery(tPlan, l.children , exprId)))
         }
     }
     super.execute(transFormedPlan)
@@ -266,6 +302,57 @@ class CarbonSqlAstBuilder(conf: SQLConf, parser: CarbonSpark2SqlParser)
extends
     }
   }
 
+  override def visitChangeColumn(ctx: ChangeColumnContext): LogicalPlan = {
+
+    val newColumn = visitColType(ctx.colType)
+    if (!ctx.identifier.getText.equalsIgnoreCase(newColumn.name)) {
+      throw new MalformedCarbonCommandException(
+        "Column names provided are different. Both the column names should be same")
+    }
+
+    val (typeString, values) : (String, Option[List[(Int, Int)]]) = newColumn.dataType match
{
+      case d:DecimalType => ("decimal", Some(List((d.precision, d.scale))))
+      case _ => (newColumn.dataType.typeName.toLowerCase, None)
+    }
+
+    val alterTableChangeDataTypeModel =
+      AlterTableDataTypeChangeModel(new CarbonSpark2SqlParser().parseDataType(typeString,
values),
+        new CarbonSpark2SqlParser()
+          .convertDbNameToLowerCase(Option(ctx.tableIdentifier().db).map(_.getText)),
+        ctx.tableIdentifier().table.getText.toLowerCase,
+        ctx.identifier.getText.toLowerCase,
+        newColumn.name.toLowerCase)
+
+    CarbonAlterTableDataTypeChangeCommand(alterTableChangeDataTypeModel)
+  }
+
+
+  override def visitAddTableColumns(ctx: AddTableColumnsContext): LogicalPlan = {
+
+    val cols = Option(ctx.columns).toSeq.flatMap(visitColTypeList)
+    val fields = parser.getFields(cols)
+    val tblProperties = scala.collection.mutable.Map.empty[String, String]
+    val tableModel = new CarbonSpark2SqlParser().prepareTableModel (false,
+      new CarbonSpark2SqlParser().convertDbNameToLowerCase(Option(ctx.tableIdentifier().db)
+        .map(_.getText)),
+      ctx.tableIdentifier.table.getText.toLowerCase,
+      fields,
+      Seq.empty,
+      tblProperties,
+      None,
+      true)
+
+    val alterTableAddColumnsModel = AlterTableAddColumnsModel(
+      Option(ctx.tableIdentifier().db).map(_.getText),
+      ctx.tableIdentifier.table.getText,
+      tblProperties.toMap,
+      tableModel.dimCols,
+      tableModel.msrCols,
+      tableModel.highcardinalitydims.getOrElse(Seq.empty))
+
+    CarbonAlterTableAddColumnCommand(alterTableAddColumnsModel)
+  }
+
   override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = {
     super.visitCreateTable(ctx)
   }

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
index 3312ee4..6149ac9 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesFilterTest.scala
@@ -19,6 +19,7 @@ package org.apache.carbondata.spark.testsuite.booleantype
 import java.io.File
 
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.test.Spark2TestQueryExecutor
 import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
@@ -79,15 +80,37 @@ class BooleanDataTypesFilterTest extends QueryTest with BeforeAndAfterEach
with
     checkAnswer(sql("select count(*) from carbon_table where booleanField = true"),
       Row(4))
 
-//    checkAnswer(sql("select count(*) from carbon_table where booleanField = 'true'"),
-//      Row(0))
-
-    checkAnswer(sql(
-      s"""
-         |select count(*)
-         |from carbon_table where booleanField = \"true\"
-         |""".stripMargin),
-      Row(0))
+    if (!Spark2TestQueryExecutor.spark.version.startsWith("2.2")) {
+      checkAnswer(sql("select count(*) from carbon_table where booleanField = 'true'"),
+        Row(0))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from carbon_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(0))
+
+      checkAnswer(sql("select count(*) from carbon_table where booleanField = 'false'"),
+        Row(0))
+
+    } else {
+      // On Spark-2.2 onwards the filter values are eliminated from quotes and pushed to
carbon
+      // layer. So 'true' will be converted to true and pushed to carbon layer. So in case
of
+      // condition 'true' and true both output same results.
+      checkAnswer(sql("select count(*) from carbon_table where booleanField = 'true'"),
+        Row(4))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from carbon_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(4))
+
+      checkAnswer(sql("select count(*) from carbon_table where booleanField = 'false'"),
+        Row(4))
+    }
 
     checkAnswer(sql("select * from carbon_table where booleanField = false"),
       Seq(Row(false), Row(false), Row(false), Row(false)))
@@ -95,9 +118,6 @@ class BooleanDataTypesFilterTest extends QueryTest with BeforeAndAfterEach
with
     checkAnswer(sql("select count(*) from carbon_table where booleanField = false"),
       Row(4))
 
-    checkAnswer(sql("select count(*) from carbon_table where booleanField = 'false'"),
-      Row(0))
-
     checkAnswer(sql("select count(*) from carbon_table where booleanField = null"),
       Row(0))
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesLoadTest.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesLoadTest.scala
b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesLoadTest.scala
index ef26919..fe11e43 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesLoadTest.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/booleantype/BooleanDataTypesLoadTest.scala
@@ -21,6 +21,7 @@ import java.io.File
 import org.apache.carbondata.core.constants.CarbonCommonConstants
 import org.apache.carbondata.core.util.CarbonProperties
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.test.Spark2TestQueryExecutor
 import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 
@@ -270,30 +271,51 @@ class BooleanDataTypesLoadTest extends QueryTest with BeforeAndAfterEach
with Be
     checkAnswer(sql("select count(*) from boolean_table where booleanField = true"),
       Row(4))
 
-//    checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
-//      Row(0))
-
-    checkAnswer(sql(
-      s"""
-         |select count(*)
-         |from boolean_table where booleanField = \"true\"
-         |""".stripMargin),
-      Row(0))
-
     checkAnswer(sql("select booleanField from boolean_table where booleanField = false"),
       Seq(Row(false), Row(false), Row(false), Row(false), Row(false), Row(false)))
 
     checkAnswer(sql("select count(*) from boolean_table where booleanField = false"),
       Row(6))
 
-    checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
-      Row(0))
-
     checkAnswer(sql("select count(*) from boolean_table where booleanField = null"),
       Row(0))
 
     checkAnswer(sql("select count(*) from boolean_table where booleanField = false or booleanField
= true"),
       Row(10))
+
+    if (!Spark2TestQueryExecutor.spark.version.startsWith("2.2")) {
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
+        Row(0))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from boolean_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(0))
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
+        Row(0))
+
+    } else {
+      // On Spark-2.2 onwards the filter values are eliminated from quotes and pushed to
carbon
+      // layer. So 'true' will be converted to true and pushed to carbon layer. So in case
of
+      // condition 'true' and true both output same results.
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
+        Row(4))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from boolean_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(4))
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
+        Row(6))
+
+    }
   }
 
   test("Loading table: load with DELIMITER, QUOTECHAR, COMMENTCHAR, MULTILINE, ESCAPECHAR,
COMPLEX_DELIMITER_LEVEL_1, SINGLE_PASS") {
@@ -339,30 +361,50 @@ class BooleanDataTypesLoadTest extends QueryTest with BeforeAndAfterEach
with Be
     checkAnswer(sql("select count(*) from boolean_table where booleanField = true"),
       Row(4))
 
-//    checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
-//      Row(0))
-
-    checkAnswer(sql(
-      s"""
-         |select count(*)
-         |from boolean_table where booleanField = \"true\"
-         |""".stripMargin),
-      Row(0))
-
     checkAnswer(sql("select booleanField from boolean_table where booleanField = false"),
       Seq(Row(false), Row(false), Row(false), Row(false), Row(false), Row(false)))
 
     checkAnswer(sql("select count(*) from boolean_table where booleanField = false"),
       Row(6))
 
-    checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
-      Row(0))
-
     checkAnswer(sql("select count(*) from boolean_table where booleanField = null"),
       Row(0))
 
     checkAnswer(sql("select count(*) from boolean_table where booleanField = false or booleanField
= true"),
       Row(10))
+
+    if (!Spark2TestQueryExecutor.spark.version.startsWith("2.2")) {
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
+        Row(0))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from boolean_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(0))
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
+        Row(0))
+
+    } else {
+      // On Spark-2.2 onwards the filter values are eliminated from quotes and pushed to
carbon
+      // layer. So 'true' will be converted to true and pushed to carbon layer. So in case
of
+      // condition 'true' and true both output same results.
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'true'"),
+        Row(4))
+
+      checkAnswer(sql(
+        s"""
+           |select count(*)
+           |from boolean_table where booleanField = \"true\"
+           |""".stripMargin),
+        Row(4))
+
+      checkAnswer(sql("select count(*) from boolean_table where booleanField = 'false'"),
+        Row(6))
+    }
   }
 
   test("Loading table: bad_records_action is FORCE") {

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
index 1115a21..b5325ef 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
@@ -836,6 +836,7 @@ class TestAlterPartitionTable extends QueryTest with BeforeAndAfterAll
{
     val exception_test_add_partition: Exception = intercept[Exception] {
       sql("CREATE DATABASE IF NOT EXISTS carbondb")
       sql("USE default")
+      sql("drop table if exists carbon_table_in_default_db")
       sql(
         """
           | CREATE TABLE carbon_table_in_default_db(id INT, name STRING)

http://git-wip-us.apache.org/repos/asf/carbondata/blob/78f8aae5/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
index 4dc3ee3..92d337f 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
@@ -98,19 +98,19 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     sql("select distinct(tmpstmp) from restructure").show(200,false)
     checkAnswer(sql("select distinct(tmpstmp) from restructure"),
       Row(new java.sql.Timestamp(107, 0, 17, 0, 0, 0, 0)))
-    checkExistence(sql("desc restructure"), true, "tmpstmptimestamp")
+    checkExistence(sql("desc restructure"), true, "tmpstmp", "timestamp")
   }
 
-  ignore ("test add timestamp direct dictionary column") {
+  test ("test add timestamp direct dictionary column") {
     sql(
       "alter table restructure add columns(tmpstmp1 timestamp) TBLPROPERTIES ('DEFAULT.VALUE"
+
       ".tmpstmp1'= '17-01-3007','DICTIONARY_INCLUDE'='tmpstmp1')")
     checkAnswer(sql("select distinct(tmpstmp1) from restructure"),
       Row(null))
-    checkExistence(sql("desc restructure"), true, "tmpstmptimestamp")
+    checkExistence(sql("desc restructure"), true, "tmpstmp", "timestamp")
   }
 
-  ignore("test add timestamp column and load as dictionary") {
+  ignore ("test add timestamp column and load as dictionary") {
     sql("create table table1(name string) stored by 'carbondata'")
     sql("insert into table1 select 'abc'")
     sql("alter table table1 add columns(tmpstmp timestamp) TBLPROPERTIES " +
@@ -121,19 +121,19 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
         Row("name",Timestamp.valueOf("2007-01-17 00:00:00.0"))))
   }
 
-  ignore("test add msr column") {
+  test("test add msr column") {
     sql(
       "alter table restructure add columns(msrField decimal(5,2))TBLPROPERTIES ('DEFAULT.VALUE"
+
       ".msrfield'= '123.45')")
     sql("desc restructure").show(2000,false)
-    checkExistence(sql("desc restructure"), true, "msrfielddecimal(5,2)")
+    checkExistence(sql("desc restructure"), true, "msrfield", "decimal(5,2)")
     val output = sql("select msrField from restructure").collect
     sql("select distinct(msrField) from restructure").show(2000,false)
     checkAnswer(sql("select distinct(msrField) from restructure"),
       Row(new BigDecimal("123.45").setScale(2, RoundingMode.HALF_UP)))
   }
 
-  ignore("test add all datatype supported dictionary column") {
+  test("test add all datatype supported dictionary column") {
     sql(
       "alter table restructure add columns(strfld string, datefld date, tptfld timestamp,
" +
       "shortFld smallInt, " +
@@ -142,14 +142,14 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
       ".dblFld'= '12345')")
     checkAnswer(sql("select distinct(dblFld) from restructure"),
       Row(java.lang.Double.parseDouble("12345")))
-    checkExistence(sql("desc restructure"), true, "strfldstring")
-    checkExistence(sql("desc restructure"), true, "dateflddate")
-    checkExistence(sql("desc restructure"), true, "tptfldtimestamp")
-    checkExistence(sql("desc restructure"), true, "shortfldsmallint")
-    checkExistence(sql("desc restructure"), true, "intfldint")
-    checkExistence(sql("desc restructure"), true, "longfldbigint")
-    checkExistence(sql("desc restructure"), true, "dblflddouble")
-    checkExistence(sql("desc restructure"), true, "dcmldecimal(5,4)")
+    checkExistence(sql("desc restructure"), true, "strfld", "string")
+    checkExistence(sql("desc restructure"), true, "datefld", "date")
+    checkExistence(sql("desc restructure"), true, "tptfld", "timestamp")
+    checkExistence(sql("desc restructure"), true, "shortfld", "smallint")
+    checkExistence(sql("desc restructure"), true, "intfld", "int")
+    checkExistence(sql("desc restructure"), true, "longfld", "bigint")
+    checkExistence(sql("desc restructure"), true, "dblfld", "double")
+    checkExistence(sql("desc restructure"), true, "dcml", "decimal(5,4)")
   }
 
   test("test drop all keycolumns in a table") {
@@ -170,7 +170,7 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     "used")
   {
     sql("alter table restructure add columns(dcmldefault decimal)")
-    checkExistence(sql("desc restructure"), true, "dcmldefaultdecimal(10,0)")
+    checkExistence(sql("desc restructure"), true, "dcmldefault", "decimal(10,0)")
   }
 
   test("test adding existing measure as dimension") {
@@ -292,11 +292,11 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     }
   }
 
-  ignore("test drop dimension, measure column") {
+  test("test drop dimension, measure column") {
     sql("alter table default.restructure drop columns(empno, designation, doj)")
-    checkExistence(sql("desc restructure"), false, "empnoint")
-    checkExistence(sql("desc restructure"), false, "designationstring")
-    checkExistence(sql("desc restructure"), false, "dojtimestamp")
+    checkExistence(sql("desc restructure"), false, "empno")
+    checkExistence(sql("desc restructure"), false, "designation")
+    checkExistence(sql("desc restructure"), false, "doj")
     assert(sql("select * from restructure").schema
              .filter(p => p.name.equalsIgnoreCase("empno") ||
                           p.name.equalsIgnoreCase("designation") || p.name.equalsIgnoreCase("doj"))
@@ -304,7 +304,7 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     sql("alter table restructure add columns(empno int, designation string, doj timestamp)")
   }
 
-  ignore ("test drop & add same column multiple times as dict, nodict, timestamp and
msr") {
+  test ("test drop & add same column multiple times as dict, nodict, timestamp and msr")
{
     // drop and add dict column
     sql("alter table restructure drop columns(designation)")
     sql(
@@ -332,10 +332,10 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     checkAnswer(sql("select distinct(designation) from restructure"), Row(67890))
   }
 
-  ignore("test change datatype of int and decimal column") {
+  test("test change datatype of int and decimal column") {
     sql("alter table restructure add columns(intfield int, decimalfield decimal(10,2))")
     sql("alter table default.restructure change intfield intField bigint")
-    checkExistence(sql("desc restructure"), true, "intfieldbigint")
+    checkExistence(sql("desc restructure"), true, "intfield", "bigint")
     sql("alter table default.restructure change decimalfield deciMalfield Decimal(11,3)")
     sql("alter table default.restructure change decimalfield deciMalfield Decimal(12,3)")
     intercept[RuntimeException] {
@@ -404,7 +404,7 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
     assert(result.count().equals(10L))
   }
 
-  test("test to check if bad record folder name is changed") {
+  ignore("test to check if bad record folder name is changed") {
     sql("alter table restructure_bad rename to restructure_badnew")
     val oldLocation = new File("./target/test/badRecords/default/restructure_bad")
     val newLocation = new File("./target/test/badRecords/default/restructure_badnew")


Mime
View raw message