[CARBONDATA-3514] Support Spark 2.4.4 integration
This PR adds integration with Spark 2.4.4
This closes #3378
diff --git a/README.md b/README.md
index a34784d..2f661ed 100644
--- a/README.md
+++ b/README.md
@@ -28,8 +28,8 @@
## Status
-Spark2.2:
-[](https://builds.apache.org/view/A-D/view/CarbonData/job/carbondata-master-spark-2.2/lastBuild/testReport)
+Spark2.3:
+[](https://builds.apache.org/view/A-D/view/CarbonData/job/carbondata-master-spark-2.2/lastBuild/testReport)
[](https://coveralls.io/github/apache/carbondata?branch=master)
<a href="https://scan.coverity.com/projects/carbondata">
<img alt="Coverity Scan Build Status"
diff --git a/build/README.md b/build/README.md
index f361a6e..960ccce 100644
--- a/build/README.md
+++ b/build/README.md
@@ -25,11 +25,9 @@
* [Apache Thrift 0.9.3](http://archive.apache.org/dist/thrift/0.9.3/)
## Build command
-Build with different supported versions of Spark, by default using Spark 2.2.1 to build
+Build with different supported versions of Spark, by default using Spark 2.4.4
```
-mvn -DskipTests -Pspark-2.1 -Dspark.version=2.1.0 clean package
-mvn -DskipTests -Pspark-2.2 -Dspark.version=2.2.1 clean package
-mvn -DskipTests -Pspark-2.3 -Dspark.version=2.3.2 clean package
+mvn -DskipTests -Pspark-2.4 clean package
```
Note:
@@ -39,5 +37,5 @@
## For contributors : To build the format code after any changes, please follow the below command.
Note:Need install Apache Thrift 0.9.3
```
-mvn clean -DskipTests -Pbuild-with-format -Pspark-2.2 package
+mvn clean -DskipTests -Pbuild-with-format -Pspark-2.4 package
```
\ No newline at end of file
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
index d0e5a42..c18298d 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
@@ -27,11 +27,11 @@
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Set;
import org.apache.carbondata.common.exceptions.sql.NoSuchDataMapException;
+import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
@@ -42,12 +42,15 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.log4j.Logger;
/**
* Stores datamap schema in disk as json format
*/
public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStorageProvider {
+ private Logger LOG = LogServiceFactory.getLogService(this.getClass().getCanonicalName());
+
private String storePath;
private String mdtFilePath;
@@ -171,17 +174,15 @@
if (!FileFactory.isFileExist(schemaPath)) {
throw new IOException("DataMap with name " + dataMapName + " does not exists in storage");
}
- Iterator<DataMapSchema> iterator = dataMapSchemas.iterator();
- while (iterator.hasNext()) {
- DataMapSchema schema = iterator.next();
- if (schema.getDataMapName().equalsIgnoreCase(dataMapName)) {
- iterator.remove();
- }
- }
+
+ LOG.info(String.format("Trying to delete DataMap %s schema", dataMapName));
+
+ dataMapSchemas.removeIf(schema -> schema.getDataMapName().equalsIgnoreCase(dataMapName));
touchMDTFile();
if (!FileFactory.deleteFile(schemaPath)) {
throw new IOException("DataMap with name " + dataMapName + " cannot be deleted");
}
+ LOG.info(String.format("DataMap %s schema is deleted", dataMapName));
}
private void checkAndReloadDataMapSchemas(boolean touchFile) throws IOException {
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 b32367b..08841b6 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
@@ -373,7 +373,8 @@
def updateColumnName(attr: Attribute, counter: Int): String = {
val name = getUpdatedName(attr.name, counter)
- attr.qualifier.map(qualifier => qualifier + "_" + name).getOrElse(name)
+ val value = attr.qualifier.map(qualifier => qualifier + "_" + name)
+ if (value.nonEmpty) value.head else name
}
def getTables(logicalPlan: LogicalPlan): Seq[CatalogTable] = {
@@ -473,7 +474,7 @@
}
def createAttrReference(ref: NamedExpression, name: String): Alias = {
- Alias(ref, name)(exprId = ref.exprId, qualifier = None)
+ CarbonToSparkAdapter.createAliasRef(ref, name, exprId = ref.exprId)
}
case class AttributeKey(exp: Expression) {
@@ -537,13 +538,13 @@
case attr: AttributeReference =>
val uattr = attrMap.get(AttributeKey(attr)).map{a =>
if (keepAlias) {
- CarbonToSparkAdapter.createAttributeReference(a.name,
- a.dataType,
- a.nullable,
- a.metadata,
- a.exprId,
- attr.qualifier,
- a)
+ CarbonToSparkAdapter.createAttributeReference(
+ name = a.name,
+ dataType = a.dataType,
+ nullable = a.nullable,
+ metadata = a.metadata,
+ exprId = a.exprId,
+ qualifier = attr.qualifier)
} else {
a
}
@@ -575,9 +576,9 @@
outputSel.zip(subsumerOutputList).map{ case (l, r) =>
l match {
case attr: AttributeReference =>
- Alias(attr, r.name)(r.exprId, None)
+ CarbonToSparkAdapter.createAliasRef(attr, r.name, r.exprId)
case a@Alias(attr: AttributeReference, name) =>
- Alias(attr, r.name)(r.exprId, None)
+ CarbonToSparkAdapter.createAliasRef(attr, r.name, r.exprId)
case other => other
}
}
@@ -594,13 +595,13 @@
val uattr = attrMap.get(AttributeKey(attr)).map{a =>
if (keepAlias) {
CarbonToSparkAdapter
- .createAttributeReference(a.name,
+ .createAttributeReference(
+ a.name,
a.dataType,
a.nullable,
a.metadata,
a.exprId,
- attr.qualifier,
- a)
+ attr.qualifier)
} else {
a
}
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVUtil.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVUtil.scala
index 3ddb0fc..cff5c41 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVUtil.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVUtil.scala
@@ -125,17 +125,18 @@
arrayBuffer += relation
}
var qualifier: Option[String] = None
- if (attr.qualifier.isDefined) {
- qualifier = if (attr.qualifier.get.startsWith("gen_sub")) {
+ if (attr.qualifier.nonEmpty) {
+ qualifier = if (attr.qualifier.headOption.get.startsWith("gen_sub")) {
Some(carbonTable.getTableName)
} else {
- attr.qualifier
+ attr.qualifier.headOption
}
}
fieldToDataMapFieldMap +=
- getFieldToDataMapFields(attr.name,
+ getFieldToDataMapFields(
+ attr.name,
attr.dataType,
- qualifier,
+ qualifier.headOption,
"",
arrayBuffer,
carbonTable.getTableName)
@@ -248,7 +249,8 @@
/**
* Below method will be used to get the fields object for mv table
*/
- private def getFieldToDataMapFields(name: String,
+ private def getFieldToDataMapFields(
+ name: String,
dataType: DataType,
qualifier: Option[String],
aggregateType: String,
@@ -313,7 +315,7 @@
val updatedOutList = outputList.map { col =>
val duplicateColumn = duplicateNameCols
.find(a => a.semanticEquals(col))
- val qualifiedName = col.qualifier.getOrElse(s"${ col.exprId.id }") + "_" + col.name
+ val qualifiedName = col.qualifier.headOption.getOrElse(s"${ col.exprId.id }") + "_" + col.name
if (duplicateColumn.isDefined) {
val attributesOfDuplicateCol = duplicateColumn.get.collect {
case a: AttributeReference => a
@@ -329,7 +331,7 @@
attributeOfCol.exists(a => a.semanticEquals(expr)))
if (!isStrictDuplicate) {
Alias(col, qualifiedName)(exprId = col.exprId)
- } else if (col.qualifier.isDefined) {
+ } else if (col.qualifier.nonEmpty) {
Alias(col, qualifiedName)(exprId = col.exprId)
// this check is added in scenario where the column is direct Attribute reference and
// since duplicate columns select is allowed, we should just put alias for those columns
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
index 7e8eb96..6fbc87f 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/rewrite/DefaultMatchMaker.scala
@@ -18,11 +18,12 @@
package org.apache.carbondata.mv.rewrite
+import org.apache.spark.sql.CarbonToSparkAdapter
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, AttributeReference, AttributeSet, Expression, PredicateHelper, _}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, LeftOuter}
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, Metadata}
import org.apache.carbondata.mv.datamap.MVHelper
import org.apache.carbondata.mv.plans.modular.{JoinEdge, Matchable, ModularPlan, _}
@@ -95,9 +96,12 @@
// Replace all compensation1 attributes with refrences of subsumer attributeset
val compensationFinal = compensation1.transformExpressions {
case ref: Attribute if subqueryAttributeSet.contains(ref) =>
- AttributeReference(ref.name, ref.dataType)(exprId = ref.exprId, qualifier = subsumerName)
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, metadata = Metadata.empty,
+ exprId = ref.exprId, qualifier = subsumerName)
case alias: Alias if subqueryAttributeSet.contains(alias.toAttribute) =>
- Alias(alias.child, alias.name)(exprId = alias.exprId, qualifier = subsumerName)
+ CarbonToSparkAdapter.createAliasRef(
+ alias.child, alias.name, alias.exprId, subsumerName)
}
compensationFinal
} else {
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/Harmonizer.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/Harmonizer.scala
index cb2043e..2b4247e 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/Harmonizer.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/Harmonizer.scala
@@ -17,10 +17,11 @@
package org.apache.carbondata.mv.plans.modular
+import org.apache.spark.sql.{CarbonToSparkAdapter, SQLConf}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.rules._
-import org.apache.spark.sql.SQLConf
+import org.apache.spark.sql.types.Metadata
import org.apache.carbondata.mv.plans
import org.apache.carbondata.mv.plans._
@@ -198,18 +199,18 @@
.isInstanceOf[Attribute]))
val aggOutputList = aggTransMap.values.flatMap(t => t._2)
.map { ref =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(hFactName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(hFactName))
}
val hFactOutputSet = hFact.outputSet
// Update the outputlist qualifier
val hOutputList = (attrOutputList ++ aggOutputList).map {attr =>
attr.transform {
case ref: Attribute if hFactOutputSet.contains(ref) =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(hFactName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(hFactName))
}
}.asInstanceOf[Seq[NamedExpression]]
@@ -217,9 +218,9 @@
val hPredList = s.predicateList.map{ pred =>
pred.transform {
case ref: Attribute if hFactOutputSet.contains(ref) =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(hFactName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(hFactName))
}
}
val hSel = s.copy(
@@ -241,9 +242,9 @@
val wip = g.copy(outputList = gOutputList, inputList = hInputList, child = hSel)
wip.transformExpressions {
case ref: Attribute if hFactOutputSet.contains(ref) =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(hFactName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(hFactName))
}
}
}
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPatterns.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPatterns.scala
index 30857c8..b694e78 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPatterns.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/modular/ModularPatterns.scala
@@ -52,9 +52,11 @@
plan transform {
case g@GroupBy(_, _, _, _, s@Select(_, _, _, aliasmap, _, children, _, _, _, _), _, _, _) =>
val aq = AttributeSet(g.outputList).filter(_.qualifier.nonEmpty)
- val makeupmap = children.zipWithIndex.flatMap {
+ val makeupmap: Map[Int, String] = children.zipWithIndex.flatMap {
case (child, i) =>
- aq.find(child.outputSet.contains(_)).map(_.qualifier).flatten.map((i, _))
+ aq.find(child.outputSet.contains(_))
+ .flatMap(_.qualifier.headOption)
+ .map((i, _))
}.toMap
g.copy(child = s.copy(aliasMap = makeupmap ++ aliasmap))
}
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 0bbacc4..7068b7e 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
@@ -110,10 +110,7 @@
RewriteCorrelatedScalarSubquery,
EliminateSerialization,
SparkSQLUtil.getRemoveRedundantAliasesObj(),
- RemoveRedundantProject,
- SimplifyCreateStructOps,
- SimplifyCreateArrayOps,
- SimplifyCreateMapOps) ++
+ RemoveRedundantProject) ++
extendedOperatorOptimizationRules: _*) ::
Batch(
"Check Cartesian Products", Once,
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Logical2ModularExtractions.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Logical2ModularExtractions.scala
index 3b6c725..2033342 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Logical2ModularExtractions.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Logical2ModularExtractions.scala
@@ -167,7 +167,9 @@
val aq = attributeSet.filter(_.qualifier.nonEmpty)
children.zipWithIndex.flatMap {
case (child, i) =>
- aq.find(child.outputSet.contains(_)).map(_.qualifier).flatten.map((i, _))
+ aq.find(child.outputSet.contains(_))
+ .flatMap(_.qualifier.headOption)
+ .map((i, _))
}.toMap
}
@@ -353,28 +355,13 @@
Seq.empty)
case l: LogicalRelation =>
val tableIdentifier = l.catalogTable.map(_.identifier)
- val database = tableIdentifier.map(_.database).flatten.getOrElse(null)
- val table = tableIdentifier.map(_.table).getOrElse(null)
+ val database = tableIdentifier.flatMap(_.database).orNull
+ val table = tableIdentifier.map(_.table).orNull
Some(database, table, l.output, Nil, NoFlags, Seq.empty)
case l: LocalRelation => // used for unit test
Some(null, null, l.output, Nil, NoFlags, Seq.empty)
case _ =>
- // this check is added as we get MetastoreRelation in spark2.1,
- // this is removed in later spark version
- // TODO: this check can be removed once 2.1 support is removed from carbon
- if (SparkUtil.isSparkVersionEqualTo("2.1") &&
- plan.getClass.getName.equals("org.apache.spark.sql.hive.MetastoreRelation")) {
- val catalogTable = CarbonReflectionUtils.getFieldOfCatalogTable("catalogTable", plan)
- .asInstanceOf[CatalogTable]
- Some(catalogTable.database,
- catalogTable.identifier.table,
- plan.output,
- Nil,
- NoFlags,
- Seq.empty)
- } else {
None
- }
}
}
}
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Printers.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Printers.scala
index d3ce38d..366284b 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Printers.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/Printers.scala
@@ -204,7 +204,7 @@
s.child match {
case a: Alias =>
val qualifierPrefix = a.qualifier
- .map(_ + ".").getOrElse("")
+ .map(_ + ".").headOption.getOrElse("")
s"$qualifierPrefix${
quoteIdentifier(a
.name)
@@ -221,7 +221,7 @@
s.child match {
case a: Alias =>
val qualifierPrefix = a.qualifier.map(_ + ".")
- .getOrElse("")
+ .headOption.getOrElse("")
s"$qualifierPrefix${ quoteIdentifier(a.name) }"
case other => other.sql
diff --git a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/SQLBuilder.scala b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/SQLBuilder.scala
index b17eea2..c3a3a68 100644
--- a/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/SQLBuilder.scala
+++ b/datamap/mv/plan/src/main/scala/org/apache/carbondata/mv/plans/util/SQLBuilder.scala
@@ -21,8 +21,11 @@
import scala.collection.immutable
+import org.apache.spark.sql.CarbonToSparkAdapter
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, AttributeReference, AttributeSet, Cast, Expression, NamedExpression}
import org.apache.spark.sql.catalyst.rules.{Rule, RuleExecutor}
+import org.apache.spark.sql.types.Metadata
+import org.apache.spark.util.SparkUtil
import org.apache.carbondata.mv.expressions.modular._
import org.apache.carbondata.mv.plans._
@@ -116,18 +119,19 @@
if (i > -1) {
// this is a walk around for mystery of spark qualifier
if (aliasMap.nonEmpty && aliasMap(i).nonEmpty) {
- AttributeReference(
- ref.name,
- ref.dataType)(exprId = ref.exprId, qualifier = Option(aliasMap(i)))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, metadata = Metadata.empty,
+ exprId = ref.exprId, qualifier = Some(aliasMap(i)))
} else {
ref
}
} else {
attrMap.get(ref) match {
case Some(alias) =>
- AttributeReference(
+ CarbonToSparkAdapter.createAttributeReference(
alias.child.asInstanceOf[AttributeReference].name,
- ref.dataType)(exprId = ref.exprId,
+ ref.dataType, nullable = true, metadata = Metadata.empty,
+ exprId = ref.exprId,
alias.child.asInstanceOf[AttributeReference].qualifier)
case None => ref
}
@@ -178,13 +182,12 @@
list = list :+ ((index, subqueryName))
newS = newS.transformExpressions {
case ref: Attribute if (subqueryAttributeSet.contains(ref)) =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(subqueryName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(subqueryName))
case alias: Alias if (subqueryAttributeSet.contains(alias.toAttribute)) =>
- Alias(alias.child, alias.name)(
- exprId = alias.exprId,
- qualifier = Some(subqueryName))
+ CarbonToSparkAdapter.createAliasRef(
+ alias.child, alias.name, alias.exprId, Some(subqueryName))
}
case _ =>
@@ -212,13 +215,12 @@
}
newG.transformExpressions {
case ref: AttributeReference if (subqueryAttributeSet.contains(ref)) =>
- AttributeReference(ref.name, ref.dataType)(
- exprId = ref.exprId,
- qualifier = Some(subqueryName))
+ CarbonToSparkAdapter.createAttributeReference(
+ ref.name, ref.dataType, nullable = true, Metadata.empty,
+ ref.exprId, Some(subqueryName))
case alias: Alias if (subqueryAttributeSet.contains(alias.toAttribute)) =>
- Alias(alias.child, alias.name)(
- exprId = alias.exprId,
- qualifier = Some(subqueryName))
+ CarbonToSparkAdapter.createAliasRef(
+ alias.child, alias.name, alias.exprId, Some(subqueryName))
}.copy(alias = Some(subqueryName))
}
}
diff --git a/docs/alluxio-guide.md b/docs/alluxio-guide.md
index b1bfeeb..bad1fc0 100644
--- a/docs/alluxio-guide.md
+++ b/docs/alluxio-guide.md
@@ -50,7 +50,7 @@
### Running spark-shell
- Running the command in spark path
```$command
-./bin/spark-shell --jars ${CARBONDATA_PATH}/assembly/target/scala-2.11/apache-carbondata-1.6.0-SNAPSHOT-bin-spark2.2.1-hadoop2.7.2.jar,${ALLUXIO_PATH}/client/alluxio-1.8.1-client.jar
+./bin/spark-shell --jars ${CARBONDATA_PATH}/assembly/target/scala-2.11/apache-carbondata-1.6.0-SNAPSHOT-bin-spark2.3.4-hadoop2.7.2.jar,${ALLUXIO_PATH}/client/alluxio-1.8.1-client.jar
```
- Testing use alluxio by CarbonSession
```$scala
@@ -98,7 +98,7 @@
--master local \
--jars ${ALLUXIO_PATH}/client/alluxio-1.8.1-client.jar,${CARBONDATA_PATH}/examples/spark2/target/carbondata-examples-1.6.0-SNAPSHOT.jar \
--class org.apache.carbondata.examples.AlluxioExample \
-${CARBONDATA_PATH}/assembly/target/scala-2.11/apache-carbondata-1.6.0-SNAPSHOT-bin-spark2.2.1-hadoop2.7.2.jar \
+${CARBONDATA_PATH}/assembly/target/scala-2.11/apache-carbondata-1.6.0-SNAPSHOT-bin-spark2.3.4-hadoop2.7.2.jar \
false
```
**NOTE**: Please set runShell as false, which can avoid dependency on alluxio shell module.
diff --git a/docs/faq.md b/docs/faq.md
index 16cdfa5..88ca186 100644
--- a/docs/faq.md
+++ b/docs/faq.md
@@ -316,7 +316,7 @@
2. Use the following command :
```
- mvn -Pspark-2.1 -Dspark.version {yourSparkVersion} clean package
+ mvn -Pspark-2.4 -Dspark.version {yourSparkVersion} clean package
```
Note : Refrain from using "mvn clean package" without specifying the profile.
diff --git a/docs/presto-guide.md b/docs/presto-guide.md
index 483585f..0c49f35 100644
--- a/docs/presto-guide.md
+++ b/docs/presto-guide.md
@@ -237,9 +237,9 @@
$ mvn -DskipTests -P{spark-version} -Dspark.version={spark-version-number} -Dhadoop.version={hadoop-version-number} clean package
```
Replace the spark and hadoop version with the version used in your cluster.
- For example, if you are using Spark 2.2.1 and Hadoop 2.7.2, you would like to compile using:
+ For example, if you are using Spark 2.4.4, you would like to compile using:
```
- mvn -DskipTests -Pspark-2.2 -Dspark.version=2.2.1 -Dhadoop.version=2.7.2 clean package
+ mvn -DskipTests -Pspark-2.4 -Dspark.version=2.4.4 -Dhadoop.version=2.7.2 clean package
```
Secondly: Create a folder named 'carbondata' under $PRESTO_HOME$/plugin and
diff --git a/docs/streaming-guide.md b/docs/streaming-guide.md
index b66fce4..d007e03 100644
--- a/docs/streaming-guide.md
+++ b/docs/streaming-guide.md
@@ -37,11 +37,11 @@
- [CLOSE STREAM](#close-stream)
## Quick example
-Download and unzip spark-2.2.0-bin-hadoop2.7.tgz, and export $SPARK_HOME
+Download and unzip spark-2.4.4-bin-hadoop2.7.tgz, and export $SPARK_HOME
-Package carbon jar, and copy assembly/target/scala-2.11/carbondata_2.11-1.3.0-SNAPSHOT-shade-hadoop2.7.2.jar to $SPARK_HOME/jars
+Package carbon jar, and copy assembly/target/scala-2.11/carbondata_2.11-1.6.0-SNAPSHOT-shade-hadoop2.7.2.jar to $SPARK_HOME/jars
```shell
-mvn clean package -DskipTests -Pspark-2.2
+mvn clean package -DskipTests -Pspark-2.4
```
Start a socket data server in a terminal
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
index b6921f2..d5c1188 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
@@ -37,9 +37,9 @@
s"$rootPath/examples/spark2/src/main/resources/log4j.properties")
CarbonProperties.getInstance()
- .addProperty(CarbonCommonConstants.ENABLE_QUERY_STATISTICS, "true")
+ .addProperty(CarbonCommonConstants.ENABLE_QUERY_STATISTICS, "false")
val spark = ExampleUtils.createCarbonSession("CarbonSessionExample")
- spark.sparkContext.setLogLevel("INFO")
+ spark.sparkContext.setLogLevel("error")
exampleBody(spark)
spark.close()
}
diff --git a/integration/flink/pom.xml b/integration/flink/pom.xml
index 73bf941..de3b5bc 100644
--- a/integration/flink/pom.xml
+++ b/integration/flink/pom.xml
@@ -184,26 +184,6 @@
<profiles>
<profile>
- <id>spark-2.2</id>
- <activation>
- <activeByDefault>false</activeByDefault>
- </activation>
- <dependencies>
- <dependency>
- <groupId>org.apache.carbondata</groupId>
- <artifactId>carbondata-spark2</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-exec</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- </dependencies>
- </profile>
- <profile>
<id>spark-2.3</id>
<activation>
<activeByDefault>true</activeByDefault>
diff --git a/integration/spark-common/pom.xml b/integration/spark-common/pom.xml
index 199ff84..0790763 100644
--- a/integration/spark-common/pom.xml
+++ b/integration/spark-common/pom.xml
@@ -191,4 +191,103 @@
</plugins>
</build>
+ <profiles>
+ <profile>
+ <id>build-all</id>
+ <properties>
+ <spark.version>2.3.4</spark.version>
+ <scala.binary.version>2.11</scala.binary.version>
+ <scala.version>2.11.8</scala.version>
+ </properties>
+ </profile>
+ <profile>
+ <id>sdvtest</id>
+ <properties>
+ <maven.test.skip>true</maven.test.skip>
+ </properties>
+ </profile>
+ <profile>
+ <id>spark-2.3</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <properties>
+ <spark.version>2.3.4</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.4</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.3</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>spark-2.4</id>
+ <properties>
+ <spark.version>2.4.4</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.3</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.4</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala
index f629260..2548110 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/CsvRDDHelper.scala
@@ -37,11 +37,12 @@
import org.apache.spark.sql.util.SparkSQLUtil
import org.apache.spark.sql.util.SparkSQLUtil.sessionState
-import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
+import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.util.ThreadLocalSessionInfo
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
+import org.apache.carbondata.spark.adapter.CarbonToSparkAdapter
import org.apache.carbondata.spark.util.CommonUtil
object CsvRDDHelper {
@@ -93,9 +94,9 @@
def closePartition(): Unit = {
if (currentFiles.nonEmpty) {
val newPartition =
- FilePartition(
+ CarbonToSparkAdapter.createFilePartition(
partitions.size,
- currentFiles.toArray.toSeq)
+ currentFiles)
partitions += newPartition
}
currentFiles.clear()
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 13e7c45..a46568a 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,23 +17,22 @@
package org.apache.spark.sql.util
-import java.lang.reflect.Method
-
import org.apache.hadoop.conf.Configuration
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.analysis.EmptyRule
+import org.apache.spark.sql.catalyst.analysis.EliminateView
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeSeq, NamedExpression}
-import org.apache.spark.sql.catalyst.plans.logical
+import org.apache.spark.sql.catalyst.optimizer.{CheckCartesianProducts, EliminateOuterJoin, NullPropagation, PullupCorrelatedPredicates, RemoveRedundantAliases, ReorderJoin}
+import org.apache.spark.sql.catalyst.plans.{logical, QueryPlan}
import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LogicalPlan, Statistics}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.LogicalRDD
import org.apache.spark.sql.internal.{SessionState, SQLConf}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.util.{CarbonReflectionUtils, SerializableConfiguration, SparkUtil, Utils}
+import org.apache.spark.util.SerializableConfiguration
object SparkSQLUtil {
def sessionState(sparkSession: SparkSession): SessionState = sparkSession.sessionState
@@ -51,166 +50,60 @@
}
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")
- }
+ logicalPlanObj.stats
}
def invokeQueryPlannormalizeExprId(r: NamedExpression, input: AttributeSeq)
: NamedExpression = {
- if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- val clazz = Utils.classForName("org.apache.spark.sql.catalyst.plans.QueryPlan")
- clazz.getDeclaredMethod("normalizeExprId", classOf[Any], classOf[AttributeSeq]).
- invoke(null, r, input).asInstanceOf[NamedExpression]
- } else {
- r
- }
+ QueryPlan.normalizeExprId(r, input)
}
def getStatisticsObj(outputList: Seq[NamedExpression],
plan: LogicalPlan, stats: Statistics,
aliasMap: Option[AttributeMap[Attribute]] = None)
: Statistics = {
- val className = "org.apache.spark.sql.catalyst.plans.logical.Statistics"
- if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- val output = outputList.map(_.toAttribute)
- val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
- table => AttributeMap(table.output.zip(output))
- }
- val rewrites = mapSeq.head
- val attributes : AttributeMap[ColumnStat] = CarbonReflectionUtils.
- getField("attributeStats", stats).asInstanceOf[AttributeMap[ColumnStat]]
- var attributeStats = AttributeMap(attributes.iterator
- .map { pair => (rewrites(pair._1), pair._2) }.toSeq)
- if (aliasMap.isDefined) {
- attributeStats = AttributeMap(
- attributeStats.map(pair => (aliasMap.get(pair._1), pair._2)).toSeq)
- }
- val hints = CarbonReflectionUtils.getField("hints", stats).asInstanceOf[Object]
- CarbonReflectionUtils.createObject(className, stats.sizeInBytes,
- stats.rowCount, attributeStats, hints).asInstanceOf[Statistics]
- } else {
- val output = outputList.map(_.name)
- val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
- table => table.output.map(_.name).zip(output).toMap
- }
- val rewrites = mapSeq.head
- val colStats = CarbonReflectionUtils.getField("colStats", stats)
- .asInstanceOf[Map[String, ColumnStat]]
- var attributeStats = colStats.iterator
- .map { pair => (rewrites(pair._1), pair._2) }.toMap
- if (aliasMap.isDefined) {
- val aliasMapName = aliasMap.get.map(x => (x._1.name, x._2.name))
- attributeStats =
- attributeStats.map(pair => (aliasMapName.getOrElse(pair._1, pair._1)
- , pair._2))
- }
- CarbonReflectionUtils.createObject(className, stats.sizeInBytes,
- stats.rowCount, attributeStats).asInstanceOf[Statistics]
+ val output = outputList.map(_.toAttribute)
+ val mapSeq = plan.collect { case n: logical.LeafNode => n }.map {
+ table => AttributeMap(table.output.zip(output))
}
+ val rewrites = mapSeq.head
+ val attributes : AttributeMap[ColumnStat] = stats.attributeStats
+ var attributeStats = AttributeMap(attributes.iterator
+ .map { pair => (rewrites(pair._1), pair._2) }.toSeq)
+ if (aliasMap.isDefined) {
+ attributeStats = AttributeMap(
+ attributeStats.map(pair => (aliasMap.get(pair._1), pair._2)).toSeq)
+ }
+ val hints = stats.hints
+ Statistics(stats.sizeInBytes, stats.rowCount, attributeStats, hints)
}
def getEliminateViewObj(): Rule[LogicalPlan] = {
- if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- val className = "org.apache.spark.sql.catalyst.analysis.EliminateView"
- CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
- } else {
- EmptyRule
- }
+ EliminateView
}
def getPullupCorrelatedPredicatesObj(): Rule[LogicalPlan] = {
- if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.PullupCorrelatedPredicates"
- CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
- } else {
- EmptyRule
- }
+ PullupCorrelatedPredicates
}
def getRemoveRedundantAliasesObj(): Rule[LogicalPlan] = {
- if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases"
- CarbonReflectionUtils.createSingleObject(className).asInstanceOf[Rule[LogicalPlan]]
- } else {
- EmptyRule
- }
+ RemoveRedundantAliases
}
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 if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.ReorderJoin$";
- CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
- .asInstanceOf[Rule[LogicalPlan]]
- }
- else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ ReorderJoin
}
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 if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin$";
- CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
- .asInstanceOf[Rule[LogicalPlan]]
- }
- else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ EliminateOuterJoin
}
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 if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.NullPropagation$";
- CarbonReflectionUtils.createObjectOfPrivateConstructor(className)._1
- .asInstanceOf[Rule[LogicalPlan]]
- } else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ NullPropagation
}
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 if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- val className = "org.apache.spark.sql.catalyst.optimizer.CheckCartesianProducts";
- CarbonReflectionUtils.createObject(className, conf)._1.asInstanceOf[Rule[LogicalPlan]]
- }
- else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ CheckCartesianProducts
}
/**
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 46692df..93e66ea 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,22 @@
package org.apache.spark.util
-import java.lang.reflect.Method
-
import scala.reflect.runtime._
import scala.reflect.runtime.universe._
-import org.apache.spark.{SPARK_VERSION, SparkContext}
+import org.apache.spark.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.Analyzer
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.expressions.{Attribute, AttributeReference, 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.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan}
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.{AlterTableAddColumnsCommand}
import org.apache.spark.sql.execution.datasources.{DataSource, LogicalRelation}
import org.apache.spark.sql.internal.HiveSerDe
import org.apache.spark.sql.sources.{BaseRelation, Filter}
@@ -60,45 +58,19 @@
def getField[T: TypeTag : reflect.ClassTag](name: String, obj: T): Any = {
val im = rm.reflect(obj)
im.symbol.typeSignature.members.find(_.name.toString.equals(name))
- .map(l => im.reflectField(l.asTerm).get).getOrElse(null)
+ .map(l => im.reflectField(l.asTerm).get).orNull
}
def getUnresolvedRelation(
tableIdentifier: TableIdentifier,
tableAlias: Option[String] = None): UnresolvedRelation = {
- val className = "org.apache.spark.sql.catalyst.analysis.UnresolvedRelation"
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- createObject(
- className,
- tableIdentifier,
- tableAlias)._1.asInstanceOf[UnresolvedRelation]
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- createObject(
- className,
- tableIdentifier)._1.asInstanceOf[UnresolvedRelation]
- } else {
- throw new UnsupportedOperationException(s"Unsupported Spark version $SPARK_VERSION")
- }
+ UnresolvedRelation(tableIdentifier)
}
def getSubqueryAlias(sparkSession: SparkSession, alias: Option[String],
relation: LogicalPlan,
view: Option[TableIdentifier]): SubqueryAlias = {
- val className = "org.apache.spark.sql.catalyst.plans.logical.SubqueryAlias"
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- createObject(
- className,
- alias.getOrElse(""),
- relation,
- Option(view))._1.asInstanceOf[SubqueryAlias]
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- createObject(
- className,
- alias.getOrElse(""),
- relation)._1.asInstanceOf[SubqueryAlias]
- } else {
- throw new UnsupportedOperationException("Unsupported Spark version")
- }
+ SubqueryAlias(alias.getOrElse(""), relation)
}
def getInsertIntoCommand(table: LogicalPlan,
@@ -106,58 +78,23 @@
query: LogicalPlan,
overwrite: Boolean,
ifPartitionNotExists: Boolean): InsertIntoTable = {
- val className = "org.apache.spark.sql.catalyst.plans.logical.InsertIntoTable"
- 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]
- createObject(
- className,
- table,
- partition,
- query,
- overwriteOptions,
- ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]
- } else if (SparkUtil.isSparkVersionXandAbove("2.2") ) {
- createObject(
- className,
- table,
- partition,
- query,
- overwrite.asInstanceOf[Object],
- ifPartitionNotExists.asInstanceOf[Object])._1.asInstanceOf[InsertIntoTable]
- } else {
- throw new UnsupportedOperationException("Unsupported Spark version")
- }
+ InsertIntoTable(
+ table,
+ partition,
+ query,
+ overwrite,
+ ifPartitionNotExists)
}
def getLogicalRelation(relation: BaseRelation,
expectedOutputAttributes: Seq[Attribute],
catalogTable: Option[CatalogTable],
isStreaming: Boolean): LogicalRelation = {
- val className = "org.apache.spark.sql.execution.datasources.LogicalRelation"
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- createObject(
- className,
- relation,
- Some(expectedOutputAttributes),
- catalogTable)._1.asInstanceOf[LogicalRelation]
- } 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")
- }
+ new LogicalRelation(
+ relation,
+ expectedOutputAttributes.asInstanceOf[Seq[AttributeReference]],
+ catalogTable,
+ isStreaming)
}
@@ -208,46 +145,28 @@
def getSessionState(sparkContext: SparkContext,
carbonSession: Object,
useHiveMetaStore: Boolean): Any = {
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
+ if (useHiveMetaStore) {
val className = sparkContext.conf.get(
CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
- "org.apache.spark.sql.hive.CarbonSessionState")
- createObject(className, carbonSession)._1
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- if (useHiveMetaStore) {
- val className = sparkContext.conf.get(
- CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
- "org.apache.spark.sql.hive.CarbonSessionStateBuilder")
- val tuple = createObject(className, carbonSession, None)
- val method = tuple._2.getMethod("build")
- method.invoke(tuple._1)
- } else {
- val className = sparkContext.conf.get(
- CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
- "org.apache.spark.sql.hive.CarbonInMemorySessionStateBuilder")
- val tuple = createObject(className, carbonSession, None)
- val method = tuple._2.getMethod("build")
- method.invoke(tuple._1)
- }
+ "org.apache.spark.sql.hive.CarbonSessionStateBuilder")
+ val tuple = createObject(className, carbonSession, None)
+ val method = tuple._2.getMethod("build")
+ method.invoke(tuple._1)
} else {
- throw new UnsupportedOperationException("Spark version not supported")
+ val className = sparkContext.conf.get(
+ CarbonCommonConstants.CARBON_SESSIONSTATE_CLASSNAME,
+ "org.apache.spark.sql.hive.CarbonInMemorySessionStateBuilder")
+ val tuple = createObject(className, carbonSession, None)
+ val method = tuple._2.getMethod("build")
+ method.invoke(tuple._1)
}
}
def hasPredicateSubquery(filterExp: Expression) : Boolean = {
- 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 (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]
- hasSubquery
- } else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ 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
}
def getDescribeTableFormattedField[T: TypeTag : reflect.ClassTag](obj: T): Boolean = {
@@ -265,19 +184,10 @@
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")
- }
+ RowDataSourceScanExec(output, output.map(output.indexOf),
+ pushedFilters.toSet, handledFilters.toSet, rdd,
+ relation.relation,
+ relation.catalogTable.map(_.identifier))
}
def invokewriteAndReadMethod(dataSourceObj: DataSource,
@@ -287,25 +197,7 @@
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[String]],
- classOf[SparkPlan])
- // since spark 2.3.2 version (SPARK-PR#22346),
- // change 'query.output' to 'query.output.map(_.name)'
- method.invoke(dataSourceObj, mode, query, query.output.map(_.name), physicalPlan)
- .asInstanceOf[BaseRelation]
- } else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ dataSourceObj.writeAndRead(mode, query, query.output.map(_.name), physicalPlan)
}
/**
@@ -316,9 +208,7 @@
*/
def invokeAlterTableAddColumn(table: TableIdentifier,
colsToAdd: Seq[StructField]): Object = {
- val caseClassName = "org.apache.spark.sql.execution.command.AlterTableAddColumnsCommand"
- CarbonReflectionUtils.createObject(caseClassName, table, colsToAdd)
- ._1.asInstanceOf[RunnableCommand]
+ AlterTableAddColumnsCommand(table, colsToAdd)
}
def createSingleObject(className: String): Any = {
@@ -385,16 +275,6 @@
def invokeAnalyzerExecute(analyzer: Analyzer,
plan: LogicalPlan): LogicalPlan = {
- if (SparkUtil.isSparkVersionEqualTo("2.1") || SparkUtil.isSparkVersionEqualTo("2.2")) {
- val method: Method = analyzer.getClass
- .getMethod("execute", classOf[LogicalPlan])
- method.invoke(analyzer, plan).asInstanceOf[LogicalPlan]
- } else if (SparkUtil.isSparkVersionEqualTo("2.3")) {
- val method: Method = analyzer.getClass
- .getMethod("executeAndCheck", classOf[LogicalPlan])
- method.invoke(analyzer, plan).asInstanceOf[LogicalPlan]
- } else {
- throw new UnsupportedOperationException("Spark version not supported")
- }
+ analyzer.executeAndCheck(plan)
}
}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/plans/logical/Subquery.scala b/integration/spark-common/src/main/spark2.3/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala
similarity index 65%
rename from integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/plans/logical/Subquery.scala
rename to integration/spark-common/src/main/spark2.3/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala
index 4abf189..a2ca9e6 100644
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/plans/logical/Subquery.scala
+++ b/integration/spark-common/src/main/spark2.3/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala
@@ -15,14 +15,14 @@
* limitations under the License.
*/
-package org.apache.spark.sql.catalyst.plans.logical
+package org.apache.carbondata.spark.adapter
-import org.apache.spark.sql.catalyst.expressions.Attribute
+import scala.collection.mutable.ArrayBuffer
-/**
- * This node is inserted at the top of a subquery when it is optimized. This makes sure we can
- * recognize a subquery as such, and it allows us to write subquery aware transformations.
- */
-case class Subquery(child: LogicalPlan) extends UnaryNode {
- override def output: Seq[Attribute] = child.output
-}
\ No newline at end of file
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+
+object CarbonToSparkAdapter {
+ def createFilePartition(index: Int, files: ArrayBuffer[PartitionedFile]) = {
+ FilePartition(index, files.toArray.toSeq)
+ }
+}
diff --git a/integration/spark-common/src/main/spark2.4/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala b/integration/spark-common/src/main/spark2.4/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala
new file mode 100644
index 0000000..be82907
--- /dev/null
+++ b/integration/spark-common/src/main/spark2.4/org/apache/carbondata/spark/adapter/CarbonToSparkAdapter.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.carbondata.spark.adapter
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, ExprId, Expression, NamedExpression}
+import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
+import org.apache.spark.sql.types.{DataType, Metadata}
+
+object CarbonToSparkAdapter {
+ def createFilePartition(index: Int, files: ArrayBuffer[PartitionedFile]) = {
+ FilePartition(index, files.toArray)
+ }
+
+ def createAttributeReference(
+ name: String,
+ dataType: DataType,
+ nullable: Boolean,
+ metadata: Metadata,
+ exprId: ExprId,
+ qualifier: Option[String],
+ attrRef : NamedExpression = null): AttributeReference = {
+ val qf = if (qualifier.nonEmpty) Seq(qualifier.get) else Seq.empty
+ AttributeReference(
+ name,
+ dataType,
+ nullable,
+ metadata)(exprId, qf)
+ }
+
+ def createAliasRef(
+ child: Expression,
+ name: String,
+ exprId: ExprId = NamedExpression.newExprId,
+ qualifier: Option[String] = None,
+ explicitMetadata: Option[Metadata] = None,
+ namedExpr : Option[NamedExpression] = None ) : Alias = {
+ Alias(child, name)(
+ exprId,
+ if (qualifier.nonEmpty) Seq(qualifier.get) else Seq(),
+ explicitMetadata)
+ }
+}
diff --git a/integration/spark-datasource/pom.xml b/integration/spark-datasource/pom.xml
index cda1954..1f1cac3 100644
--- a/integration/spark-datasource/pom.xml
+++ b/integration/spark-datasource/pom.xml
@@ -192,86 +192,6 @@
</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>
- <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>
<activation>
<activeByDefault>true</activeByDefault>
@@ -286,30 +206,6 @@
<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>
diff --git a/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonDictionaryWrapper.java b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/CarbonDictionaryWrapper.java
similarity index 100%
rename from integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonDictionaryWrapper.java
rename to integration/spark-datasource/src/main/scala/org/apache/spark/sql/CarbonDictionaryWrapper.java
diff --git a/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/CarbonVectorProxy.java
similarity index 100%
rename from integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java
rename to integration/spark-datasource/src/main/scala/org/apache/spark/sql/CarbonVectorProxy.java
diff --git a/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/ColumnVectorFactory.java b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/ColumnVectorFactory.java
similarity index 100%
rename from integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/ColumnVectorFactory.java
rename to integration/spark-datasource/src/main/scala/org/apache/spark/sql/ColumnVectorFactory.java
diff --git a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonDictionaryWrapper.java b/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonDictionaryWrapper.java
deleted file mode 100644
index 605df66..0000000
--- a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonDictionaryWrapper.java
+++ /dev/null
@@ -1,47 +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.sql;
-
-import org.apache.carbondata.core.scan.result.vector.CarbonDictionary;
-
-import org.apache.parquet.column.Dictionary;
-import org.apache.parquet.column.Encoding;
-import org.apache.parquet.io.api.Binary;
-
-public class CarbonDictionaryWrapper extends Dictionary {
-
- private Binary[] binaries;
-
- CarbonDictionaryWrapper(Encoding encoding, CarbonDictionary dictionary) {
- super(encoding);
- binaries = new Binary[dictionary.getDictionarySize()];
- for (int i = 0; i < binaries.length; i++) {
- binaries[i] = Binary.fromReusedByteArray(dictionary.getDictionaryValue(i));
- }
- }
-
- @Override
- public int getMaxId() {
- return binaries.length - 1;
- }
-
- @Override
- public Binary decodeToBinary(int id) {
- return binaries[id];
- }
-}
diff --git a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java b/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java
deleted file mode 100644
index 7d23d7c..0000000
--- a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java
+++ /dev/null
@@ -1,586 +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.sql;
-
-import java.lang.reflect.Field;
-import java.math.BigInteger;
-
-import org.apache.carbondata.core.scan.result.vector.CarbonDictionary;
-import org.apache.carbondata.core.scan.scanner.LazyPageLoader;
-
-import org.apache.parquet.column.Encoding;
-import org.apache.spark.memory.MemoryMode;
-import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.execution.vectorized.ColumnVector;
-import org.apache.spark.sql.execution.vectorized.ColumnarBatch;
-import org.apache.spark.sql.types.CalendarIntervalType;
-import org.apache.spark.sql.types.DataType;
-import org.apache.spark.sql.types.Decimal;
-import org.apache.spark.sql.types.DecimalType;
-import org.apache.spark.sql.types.StructType;
-import org.apache.spark.unsafe.types.CalendarInterval;
-import org.apache.spark.unsafe.types.UTF8String;
-
-/**
- * Adapter class which handles the columnar vector reading of the carbondata
- * based on the spark ColumnVector and ColumnarBatch API. This proxy class
- * handles the complexity of spark 2.3 version related api changes since
- * spark ColumnVector and ColumnarBatch interfaces are still evolving.
- */
-public class CarbonVectorProxy {
-
- private ColumnarBatch columnarBatch;
- private ColumnVectorProxy[] columnVectorProxies;
-
-
- private void updateColumnVectors() {
- try {
- Field field = columnarBatch.getClass().getDeclaredField("columns");
- field.setAccessible(true);
- field.set(columnarBatch, columnVectorProxies);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Adapter class which handles the columnar vector reading of the carbondata
- * based on the spark ColumnVector and ColumnarBatch API. This proxy class
- * handles the complexity of spark 2.3 version related api changes since
- * spark ColumnVector and ColumnarBatch interfaces are still evolving.
- *
- * @param memMode which represent the type onheap or offheap vector.
- * @param outputSchema, metadata related to current schema of table.
- * @param rowNum rows number for vector reading
- * @param useLazyLoad Whether to use lazy load while getting the data.
- */
- public CarbonVectorProxy(MemoryMode memMode, StructType outputSchema, int rowNum,
- boolean useLazyLoad) {
- columnarBatch = ColumnarBatch.allocate(outputSchema, memMode, rowNum);
- columnVectorProxies = new ColumnVectorProxy[columnarBatch.numCols()];
- for (int i = 0; i < columnVectorProxies.length; i++) {
- if (useLazyLoad) {
- columnVectorProxies[i] =
- new ColumnVectorProxyWithLazyLoad(columnarBatch.column(i), rowNum, memMode);
- } else {
- columnVectorProxies[i] = new ColumnVectorProxy(columnarBatch.column(i), rowNum, memMode);
- }
- }
- updateColumnVectors();
- }
-
- public ColumnVectorProxy getColumnVector(int ordinal) {
- return columnVectorProxies[ordinal];
- }
-
- /**
- * Returns the number of rows for read, including filtered rows.
- */
- public int numRows() {
- return columnarBatch.capacity();
- }
-
- /**
- * This API will return a columnvector from a batch of column vector rows
- * based on the ordinal
- *
- * @param ordinal
- * @return
- */
- public ColumnVector column(int ordinal) {
- return columnarBatch.column(ordinal);
- }
-
- /**
- * Resets this column for writing. The currently stored values are no longer accessible.
- */
- public void reset() {
- for (int i = 0; i < columnarBatch.numCols(); i++) {
- ((ColumnVectorProxy) columnarBatch.column(i)).reset();
- }
- }
-
- public void resetDictionaryIds(int ordinal) {
- (((ColumnVectorProxy) columnarBatch.column(ordinal)).getVector()).getDictionaryIds().reset();
- }
-
- /**
- * Returns the row in this batch at `rowId`. Returned row is reused across calls.
- */
- public InternalRow getRow(int rowId) {
- return columnarBatch.getRow(rowId);
- }
-
- /**
- * Returns the row in this batch at `rowId`. Returned row is reused across calls.
- */
- public Object getColumnarBatch() {
- return columnarBatch;
- }
-
- /**
- * Called to close all the columns in this batch. It is not valid to access the data after
- * calling this. This must be called at the end to clean up memory allocations.
- */
- public void close() {
- columnarBatch.close();
- }
-
- /**
- * Sets the number of rows in this batch.
- */
- public void setNumRows(int numRows) {
- columnarBatch.setNumRows(numRows);
- }
-
- public DataType dataType(int ordinal) {
- return columnarBatch.column(ordinal).dataType();
- }
-
- public static class ColumnVectorProxy extends ColumnVector {
-
- private ColumnVector vector;
-
- public ColumnVectorProxy(ColumnVector columnVector, int capacity, MemoryMode mode) {
- super(capacity, columnVector.dataType(), mode);
- try {
- Field childColumns =
- columnVector.getClass().getSuperclass().getDeclaredField("childColumns");
- childColumns.setAccessible(true);
- Object o = childColumns.get(columnVector);
- childColumns.set(this, o);
- Field resultArray =
- columnVector.getClass().getSuperclass().getDeclaredField("resultArray");
- resultArray.setAccessible(true);
- Object o1 = resultArray.get(columnVector);
- resultArray.set(this, o1);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- vector = columnVector;
- }
-
- public void putRowToColumnBatch(int rowId, Object value) {
- org.apache.spark.sql.types.DataType t = dataType();
- if (null == value) {
- putNull(rowId);
- } else {
- if (t == org.apache.spark.sql.types.DataTypes.BooleanType) {
- putBoolean(rowId, (boolean) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.ByteType) {
- putByte(rowId, (byte) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.ShortType) {
- putShort(rowId, (short) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.IntegerType) {
- putInt(rowId, (int) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.LongType) {
- putLong(rowId, (long) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.FloatType) {
- putFloat(rowId, (float) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.DoubleType) {
- putDouble(rowId, (double) value);
- } else if (t == org.apache.spark.sql.types.DataTypes.StringType) {
- UTF8String v = (UTF8String) value;
- putByteArray(rowId, v.getBytes());
- } else if (t instanceof org.apache.spark.sql.types.DecimalType) {
- DecimalType dt = (DecimalType) t;
- Decimal d = Decimal.fromDecimal(value);
- if (dt.precision() <= Decimal.MAX_INT_DIGITS()) {
- putInt(rowId, (int) d.toUnscaledLong());
- } else if (dt.precision() <= Decimal.MAX_LONG_DIGITS()) {
- putLong(rowId, d.toUnscaledLong());
- } else {
- final BigInteger integer = d.toJavaBigDecimal().unscaledValue();
- byte[] bytes = integer.toByteArray();
- putByteArray(rowId, bytes, 0, bytes.length);
- }
- } else if (t instanceof CalendarIntervalType) {
- CalendarInterval c = (CalendarInterval) value;
- vector.getChildColumn(0).putInt(rowId, c.months);
- vector.getChildColumn(1).putLong(rowId, c.microseconds);
- } else if (t instanceof org.apache.spark.sql.types.DateType) {
- putInt(rowId, (int) value);
- } else if (t instanceof org.apache.spark.sql.types.TimestampType) {
- putLong(rowId, (long) value);
- }
- }
- }
-
- public void putBoolean(int rowId, boolean value) {
- vector.putBoolean(rowId, value);
- }
-
- public void putByte(int rowId, byte value) {
- vector.putByte(rowId, value);
- }
-
- public void putBytes(int rowId, int count, byte[] src, int srcIndex) {
- vector.putBytes(rowId, count, src, srcIndex);
- }
-
- public void putShort(int rowId, short value) {
- vector.putShort(rowId, value);
- }
-
- public void putInt(int rowId, int value) {
- vector.putInt(rowId, value);
- }
-
- public void putFloat(int rowId, float value) {
- vector.putFloat(rowId, value);
- }
-
- public void putFloats(int rowId, int count, float[] src, int srcIndex) {
- vector.putFloats(rowId, count, src, srcIndex);
- }
-
- public void putLong(int rowId, long value) {
- vector.putLong(rowId, value);
- }
-
- public void putDouble(int rowId, double value) {
- vector.putDouble(rowId, value);
- }
-
- public void putInts(int rowId, int count, int value) {
- vector.putInts(rowId, count, value);
- }
-
- public void putInts(int rowId, int count, int[] src, int srcIndex) {
- vector.putInts(rowId, count, src, srcIndex);
- }
-
- public void putShorts(int rowId, int count, short value) {
- vector.putShorts(rowId, count, value);
- }
-
- public void putShorts(int rowId, int count, short[] src, int srcIndex) {
- vector.putShorts(rowId, count, src, srcIndex);
- }
-
- public void putLongs(int rowId, int count, long value) {
- vector.putLongs(rowId, count, value);
- }
-
- public void putLongs(int rowId, int count, long[] src, int srcIndex) {
- vector.putLongs(rowId, count, src, srcIndex);
- }
-
- public void putDoubles(int rowId, int count, double value) {
- vector.putDoubles(rowId, count, value);
- }
-
- public void putDoubles(int rowId, int count, double[] src, int srcIndex) {
- vector.putDoubles(rowId, count, src, srcIndex);
- }
-
- public DataType dataType(int ordinal) {
- return vector.dataType();
- }
-
- public void putNotNull(int rowId) {
- vector.putNotNull(rowId);
- }
-
- public void putNotNulls(int rowId, int count) {
- vector.putNotNulls(rowId, count);
- }
-
- public void putDictionaryInt(int rowId, int value) {
- vector.getDictionaryIds().putInt(rowId, value);
- }
-
- public void setDictionary(CarbonDictionary dictionary) {
- if (null != dictionary) {
- CarbonDictionaryWrapper dictionaryWrapper =
- new CarbonDictionaryWrapper(Encoding.PLAIN, dictionary);
- vector.setDictionary(dictionaryWrapper);
- this.dictionary = dictionaryWrapper;
- } else {
- this.dictionary = null;
- vector.setDictionary(null);
- }
- }
-
- public void putNull(int rowId) {
- vector.putNull(rowId);
- }
-
- public void putNulls(int rowId, int count) {
- vector.putNulls(rowId, count);
- }
-
- public boolean hasDictionary() {
- return vector.hasDictionary();
- }
-
- public ColumnVector reserveDictionaryIds(int capacity) {
- this.dictionaryIds = vector.reserveDictionaryIds(capacity);
- return dictionaryIds;
- }
-
- @Override
- public boolean isNullAt(int i) {
- return vector.isNullAt(i);
- }
-
- @Override
- public boolean getBoolean(int i) {
- return vector.getBoolean(i);
- }
-
- @Override
- public byte getByte(int i) {
- return vector.getByte(i);
- }
-
- @Override
- public short getShort(int i) {
- return vector.getShort(i);
- }
-
- @Override
- public int getInt(int i) {
- return vector.getInt(i);
- }
-
- @Override
- public long getLong(int i) {
- return vector.getLong(i);
- }
-
- @Override
- public float getFloat(int i) {
- return vector.getFloat(i);
- }
-
- @Override
- public double getDouble(int i) {
- return vector.getDouble(i);
- }
-
- @Override
- protected void reserveInternal(int capacity) {
- }
-
- @Override
- public void reserve(int requiredCapacity) {
- vector.reserve(requiredCapacity);
- }
-
- @Override
- public long nullsNativeAddress() {
- return vector.nullsNativeAddress();
- }
-
- @Override
- public long valuesNativeAddress() {
- return vector.valuesNativeAddress();
- }
-
- @Override
- public void putBooleans(int rowId, int count, boolean value) {
- vector.putBooleans(rowId, count, value);
- }
-
- @Override
- public void putBytes(int rowId, int count, byte value) {
- vector.putBytes(rowId, count, value);
- }
-
- @Override
- public void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
- vector.putIntsLittleEndian(rowId, count, src, srcIndex);
- }
-
- @Override
- public int getDictId(int rowId) {
- return vector.getDictId(rowId);
- }
-
- @Override
- public void putLongsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
- vector.putLongsLittleEndian(rowId, count, src, srcIndex);
- }
-
- @Override
- public void putFloats(int rowId, int count, float value) {
- vector.putFloats(rowId, count, value);
- }
-
- @Override
- public void putFloats(int rowId, int count, byte[] src, int srcIndex) {
- vector.putFloats(rowId, count, src, srcIndex);
- }
-
- @Override
- public void putDoubles(int rowId, int count, byte[] src, int srcIndex) {
- vector.putDoubles(rowId, count, src, srcIndex);
- }
-
- @Override
- public void putArray(int rowId, int offset, int length) {
- vector.putArray(rowId, offset, length);
- }
-
- @Override
- public int getArrayLength(int rowId) {
- return vector.getArrayLength(rowId);
- }
-
- @Override
- public int getArrayOffset(int rowId) {
- return vector.getArrayOffset(rowId);
- }
-
- @Override
- public void loadBytes(Array array) {
- vector.loadBytes(array);
- }
-
- @Override
- public int putByteArray(int rowId, byte[] value, int offset, int count) {
- return vector.putByteArray(rowId, value, offset, count);
- }
-
- /**
- * It keeps all binary data of all rows to it.
- * Should use along with @{putArray(int rowId, int offset, int length)} to keep lengths
- * and offset.
- */
- public void putAllByteArray(byte[] data, int offset, int length) {
- vector.arrayData().appendBytes(length, data, offset);
- }
-
- @Override
- public void close() {
- vector.close();
- }
-
- public void reset() {
- if (isConstant) {
- return;
- }
- vector.reset();
- }
-
- public void setLazyPage(LazyPageLoader lazyPage) {
- lazyPage.loadPage();
- }
-
- public ColumnVector getVector() {
- return vector;
- }
- }
-
- public static class ColumnVectorProxyWithLazyLoad extends ColumnVectorProxy {
-
- private ColumnVector vector;
-
- private LazyPageLoader pageLoad;
-
- private boolean isLoaded;
-
- public ColumnVectorProxyWithLazyLoad(ColumnVector columnVector, int capacity, MemoryMode mode) {
- super(columnVector, capacity, mode);
- vector = columnVector;
- }
-
- @Override
- public boolean isNullAt(int i) {
- checkPageLoaded();
- return vector.isNullAt(i);
- }
-
- @Override
- public boolean getBoolean(int i) {
- checkPageLoaded();
- return vector.getBoolean(i);
- }
-
- @Override
- public byte getByte(int i) {
- checkPageLoaded();
- return vector.getByte(i);
- }
-
- @Override
- public short getShort(int i) {
- checkPageLoaded();
- return vector.getShort(i);
- }
-
- @Override
- public int getInt(int i) {
- checkPageLoaded();
- return vector.getInt(i);
- }
-
- @Override
- public long getLong(int i) {
- checkPageLoaded();
- return vector.getLong(i);
- }
-
- @Override
- public float getFloat(int i) {
- checkPageLoaded();
- return vector.getFloat(i);
- }
-
- @Override
- public double getDouble(int i) {
- checkPageLoaded();
- return vector.getDouble(i);
- }
-
- @Override
- public int getArrayLength(int rowId) {
- checkPageLoaded();
- return vector.getArrayLength(rowId);
- }
-
- @Override
- public int getArrayOffset(int rowId) {
- checkPageLoaded();
- return vector.getArrayOffset(rowId);
- }
-
- private void checkPageLoaded() {
- if (!isLoaded) {
- if (pageLoad != null) {
- pageLoad.loadPage();
- }
- isLoaded = true;
- }
- }
-
- public void reset() {
- if (isConstant) {
- return;
- }
- isLoaded = false;
- vector.reset();
- }
-
- public void setLazyPage(LazyPageLoader lazyPage) {
- this.pageLoad = lazyPage;
- }
-
- }
-}
diff --git a/integration/spark2/pom.xml b/integration/spark2/pom.xml
index 7b65e0b..dfebf6f 100644
--- a/integration/spark2/pom.xml
+++ b/integration/spark2/pom.xml
@@ -253,92 +253,6 @@
</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.2</exclude>
- <exclude>src/main/spark2.3</exclude>
- <exclude>src/main/commonTo2.2And2.3</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.1</source>
- <source>src/main/commonTo2.1And2.2</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
- <id>spark-2.2</id>
- <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.1</exclude>
- <exclude>src/main/spark2.3</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.2</source>
- <source>src/main/commonTo2.2And2.3</source>
- <source>src/main/commonTo2.1And2.2</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
<id>spark-2.3</id>
<activation>
<activeByDefault>true</activeByDefault>
@@ -355,9 +269,7 @@
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<excludes>
- <exclude>src/main/spark2.1</exclude>
- <exclude>src/main/spark2.2</exclude>
- <exclude>src/main/commonTo2.1And2.2</exclude>
+ <exclude>src/main/spark2.4</exclude>
</excludes>
</configuration>
</plugin>
@@ -375,7 +287,46 @@
<configuration>
<sources>
<source>src/main/spark2.3</source>
- <source>src/main/commonTo2.2And2.3</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>spark-2.4</id>
+ <properties>
+ <spark.version>2.4.4</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.3</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.4</source>
</sources>
</configuration>
</execution>
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonBoundReference.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonBoundReference.scala
index aa650e0..78d6a46 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonBoundReference.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonBoundReference.scala
@@ -17,13 +17,8 @@
package org.apache.spark.sql
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, ExprId, LeafExpression, NamedExpression}
-import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.DataType
-
-import org.apache.carbondata.core.scan.expression.ColumnExpression
case class CastExpr(expr: Expression) extends Filter {
override def references: Array[String] = null
@@ -33,26 +28,6 @@
override def references: Array[String] = null
}
-case class CarbonBoundReference(colExp: ColumnExpression, dataType: DataType, nullable: Boolean)
- extends LeafExpression with NamedExpression with CodegenFallback {
-
- type EvaluatedType = Any
-
- override def toString: String = s"input[" + colExp.getColIndex + "]"
-
- override def eval(input: InternalRow): Any = input.get(colExp.getColIndex, dataType)
-
- override def name: String = colExp.getColumnName
-
- override def toAttribute: Attribute = throw new UnsupportedOperationException
-
- override def exprId: ExprId = throw new UnsupportedOperationException
-
- override def qualifier: Option[String] = null
-
- override def newInstance(): NamedExpression = throw new UnsupportedOperationException
-}
-
case class CarbonEndsWith(expr: Expression) extends Filter {
override def references: Array[String] = null
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index e020a99..23c078a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -258,41 +258,67 @@
s"""
|org.apache.spark.sql.DictTuple $value = $decodeDecimal($dictRef, ${ev.value});
""".stripMargin
- ExprCode(code, s"$value.getIsNull()",
- s"((org.apache.spark.sql.types.Decimal)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((org.apache.spark.sql.types.Decimal)$value.getValue())",
+ expr.dataType)
} else {
getDictionaryColumnIds(index)._3.getDataType match {
case CarbonDataTypes.INT => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeInt($dictRef, ${ ev.value });
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((Integer)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((Integer)$value.getValue())",
+ expr.dataType)
case CarbonDataTypes.SHORT => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeShort($dictRef, ${ ev.value });
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((Short)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((Short)$value.getValue())",
+ expr.dataType)
case CarbonDataTypes.DOUBLE => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeDouble($dictRef, ${ ev.value });
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((Double)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((Double)$value.getValue())",
+ expr.dataType)
case CarbonDataTypes.LONG => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeLong($dictRef, ${ ev.value });
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((Long)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((Long)$value.getValue())",
+ expr.dataType)
case CarbonDataTypes.BOOLEAN => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeBool($dictRef, ${ ev.value });
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((Boolean)$value.getValue())")
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((Boolean)$value.getValue())",
+ expr.dataType)
case _ => code +=
s"""
|org.apache.spark.sql.DictTuple $value = $decodeStr($dictRef, ${ev.value});
""".stripMargin
- ExprCode(code, s"$value.getIsNull()", s"((UTF8String)$value.getValue())")
-
+ CarbonToSparkAdapter.createExprCode(
+ code,
+ s"$value.getIsNull()",
+ s"((UTF8String)$value.getValue())",
+ expr.dataType)
}
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
index 376d121..703df20 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSource.scala
@@ -40,11 +40,9 @@
import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.SchemaEvolutionEntry
import org.apache.carbondata.core.metadata.schema.table.TableInfo
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
-import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
import org.apache.carbondata.spark.CarbonOption
import org.apache.carbondata.spark.util.CarbonScalaUtil
import org.apache.carbondata.streaming.{CarbonStreamException, CarbonStreamingQueryListener, StreamSinkFactory}
@@ -183,12 +181,9 @@
LOGGER.warn("Carbon Table [" +dbName +"] [" +tableName +"] is not found, " +
"Now existing Schema will be overwritten with default properties")
val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
- val identifier = AbsoluteTableIdentifier.from(
- CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession),
- dbName,
- tableName)
+ val tablePath = CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession)
val updatedParams = CarbonSource.updateAndCreateTable(
- identifier, dataSchema, sparkSession, metaStore, parameters, None)
+ dbName, tableName, tablePath, dataSchema, sparkSession, metaStore, parameters, None)
(CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession), updatedParams)
case ex: Exception =>
throw new Exception("do not have dbname and tablename for carbon table", ex)
@@ -278,9 +273,10 @@
lazy val listenerAdded = new mutable.HashMap[Int, Boolean]()
def createTableInfoFromParams(
+ databaseName: String,
+ tableName: String,
parameters: Map[String, String],
dataSchema: StructType,
- identifier: AbsoluteTableIdentifier,
query: Option[LogicalPlan],
sparkSession: SparkSession): TableModel = {
val sqlParser = new CarbonSpark2SqlParser
@@ -301,8 +297,8 @@
sqlParser.getFields(dataSchema)
}
val bucketFields = sqlParser.getBucketFields(map, fields, options)
- sqlParser.prepareTableModel(ifNotExistPresent = false, Option(identifier.getDatabaseName),
- identifier.getTableName, fields, Nil, map, bucketFields)
+ sqlParser.prepareTableModel(ifNotExistPresent = false, Option(databaseName),
+ tableName, fields, Nil, map, bucketFields)
}
/**
@@ -314,7 +310,8 @@
def updateCatalogTableWithCarbonSchema(
tableDesc: CatalogTable,
sparkSession: SparkSession,
- query: Option[LogicalPlan] = None): CatalogTable = {
+ query: Option[LogicalPlan] = None,
+ persistSchema: Boolean = true): CatalogTable = {
val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
val storageFormat = tableDesc.storage
val properties = storageFormat.properties
@@ -322,14 +319,16 @@
val tablePath = CarbonEnv.getTablePath(
tableDesc.identifier.database, tableDesc.identifier.table)(sparkSession)
val dbName = CarbonEnv.getDatabaseName(tableDesc.identifier.database)(sparkSession)
- val identifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableDesc.identifier.table)
val map = updateAndCreateTable(
- identifier,
+ dbName,
+ tableDesc.identifier.table,
+ tablePath,
tableDesc.schema,
sparkSession,
metaStore,
properties,
- query)
+ query,
+ persistSchema)
// updating params
val updatedFormat = CarbonToSparkAdapter
.getUpdatedStorageFormat(storageFormat, map, tablePath)
@@ -351,36 +350,56 @@
}
}
+ def createTableInfo(
+ databaseName: String,
+ tableName: String,
+ tablePath: String,
+ dataSchema: StructType,
+ properties: Map[String, String],
+ query: Option[LogicalPlan],
+ sparkSession: SparkSession
+ ): TableInfo = {
+ val model = createTableInfoFromParams(
+ databaseName, tableName, properties, dataSchema, query, sparkSession)
+ val tableInfo = TableNewProcessor(model)
+ val isTransactionalTable = properties.getOrElse("isTransactional", "true").contains("true")
+ tableInfo.setTablePath(tablePath)
+ tableInfo.setTransactionalTable(isTransactionalTable)
+ tableInfo.setDatabaseName(databaseName)
+ val schemaEvolutionEntry = new SchemaEvolutionEntry
+ schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)
+ tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)
+ tableInfo
+ }
+
def updateAndCreateTable(
- identifier: AbsoluteTableIdentifier,
+ databaseName: String,
+ tableName: String,
+ tablePath: String,
dataSchema: StructType,
sparkSession: SparkSession,
metaStore: CarbonMetaStore,
properties: Map[String, String],
- query: Option[LogicalPlan]): Map[String, String] = {
- val model = createTableInfoFromParams(properties, dataSchema, identifier, query, sparkSession)
- val tableInfo: TableInfo = TableNewProcessor(model)
- val isTransactionalTable = properties.getOrElse("isTransactional", "true").contains("true")
- tableInfo.setTablePath(identifier.getTablePath)
- tableInfo.setTransactionalTable(isTransactionalTable)
- tableInfo.setDatabaseName(identifier.getDatabaseName)
- val schemaEvolutionEntry = new SchemaEvolutionEntry
- schemaEvolutionEntry.setTimeStamp(tableInfo.getLastUpdatedTime)
- tableInfo.getFactTable.getSchemaEvolution.getSchemaEvolutionEntryList.add(schemaEvolutionEntry)
- val map = if (!metaStore.isReadFromHiveMetaStore && isTransactionalTable) {
- metaStore.saveToDisk(tableInfo, identifier.getTablePath)
+ query: Option[LogicalPlan],
+ persistSchema: Boolean = true): Map[String, String] = {
+ val tableInfo = createTableInfo(
+ databaseName, tableName, tablePath, dataSchema, properties, query, sparkSession)
+ val map = if (persistSchema &&
+ !metaStore.isReadFromHiveMetaStore &&
+ tableInfo.isTransactionalTable) {
+ metaStore.saveToDisk(tableInfo, tablePath)
new java.util.HashMap[String, String]()
} else {
CarbonUtil.convertToMultiStringMap(tableInfo)
}
properties.foreach(e => map.put(e._1, e._2))
- map.put("tablepath", identifier.getTablePath)
- map.put("dbname", identifier.getDatabaseName)
+ map.put("tablepath", tablePath)
+ map.put("dbname", databaseName)
if (map.containsKey("tableName")) {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
LOGGER.warn("tableName is not required in options, ignoring it")
}
- map.put("tableName", identifier.getTableName)
+ map.put("tableName", tableName)
map.asScala.toMap
}
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CustomDeterministicExpression.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CustomDeterministicExpression.scala
index 8a37989..233f28d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CustomDeterministicExpression.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CustomDeterministicExpression.scala
@@ -38,5 +38,5 @@
override def genCode(ctx: CodegenContext): ExprCode = nonDt.genCode(ctx)
- override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = ev.copy("")
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = ev.copy()
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
index ded87b9..9f203c8 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
@@ -164,6 +164,7 @@
}
private def dropDataMapFromSystemFolder(sparkSession: SparkSession): Unit = {
+ LOGGER.info("Trying to drop DataMap from system folder")
try {
if (dataMapSchema == null) {
dataMapSchema = DataMapStoreManager.getInstance().getDataMapSchema(dataMapName)
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 130580d..b90faa7 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -690,13 +690,13 @@
attr.dataType
}
}
- CarbonToSparkAdapter.createAttributeReference(attr.name,
+ CarbonToSparkAdapter.createAttributeReference(
+ attr.name,
updatedDataType,
attr.nullable,
attr.metadata,
attr.exprId,
- attr.qualifier,
- attr)
+ attr.qualifier)
}
// Only select the required columns
var output = if (partition.nonEmpty) {
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
index c12ff6c..d3f8079 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
@@ -97,12 +97,12 @@
carbonTable,
schemaConverter.fromWrapperToExternalSchemaEvolutionEntry(schemaEvolutionEntry),
thriftTable)(sparkSession)
- // In case of spark2.2 and above and , when we call
+ // when we call
// alterExternalCatalogForTableWithUpdatedSchema to update the new schema to external catalog
// in case of add column, spark gets the catalog table and then it itself adds the partition
// columns if the table is partition table for all the new data schema sent by carbon,
// so there will be duplicate partition columns, so send the columns without partition columns
- val cols = if (SparkUtil.isSparkVersionXandAbove("2.2") && carbonTable.isHivePartitionTable) {
+ val cols = if (carbonTable.isHivePartitionTable) {
val partitionColumns = carbonTable.getPartitionInfo.getColumnSchemaList.asScala
val carbonColumnsWithoutPartition = carbonColumns.filterNot(col => partitionColumns.contains
(col))
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableColRenameDataTypeChangeCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableColRenameDataTypeChangeCommand.scala
index 7e66d34..cf05a9d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableColRenameDataTypeChangeCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableColRenameDataTypeChangeCommand.scala
@@ -163,7 +163,7 @@
if (isDataTypeChange) {
// if column datatype change operation is on partition column, then fail the datatype change
// operation
- if (SparkUtil.isSparkVersionXandAbove("2.2") && null != carbonTable.getPartitionInfo) {
+ if (null != carbonTable.getPartitionInfo) {
val partitionColumns = carbonTable.getPartitionInfo.getColumnSchemaList
partitionColumns.asScala.foreach {
col =>
@@ -286,14 +286,13 @@
// update the schema changed column at the specific index in carbonColumns based on schema order
carbonColumns
.update(schemaOrdinal, schemaConverter.fromExternalToWrapperColumnSchema(addColumnSchema))
- // In case of spark2.2 and above and , when we call
+ // When we call
// alterExternalCatalogForTableWithUpdatedSchema to update the new schema to external catalog
// in case of rename column or change datatype, spark gets the catalog table and then it itself
// adds the partition columns if the table is partition table for all the new data schema sent
// by carbon, so there will be duplicate partition columns, so send the columns without
// partition columns
- val columns = if (SparkUtil.isSparkVersionXandAbove("2.2") &&
- carbonTable.isHivePartitionTable) {
+ val columns = if (carbonTable.isHivePartitionTable) {
val partitionColumns = carbonTable.getPartitionInfo.getColumnSchemaList.asScala
val carbonColumnsWithoutPartition = carbonColumns.filterNot(col => partitionColumns.contains(
col))
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDropColumnCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDropColumnCommand.scala
index bdc0228..4cb8a2e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDropColumnCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDropColumnCommand.scala
@@ -151,13 +151,12 @@
val cols = carbonTable.getCreateOrderColumn().asScala
.collect { case carbonColumn if !carbonColumn.isInvisible => carbonColumn.getColumnSchema }
.filterNot(column => delCols.contains(column))
- // In case of spark2.2 and above and , when we call
+ // When we call
// alterExternalCatalogForTableWithUpdatedSchema to update the new schema to external catalog
// in case of drop column, spark gets the catalog table and then it itself adds the partition
// columns if the table is partition table for all the new data schema sent by carbon,
// so there will be duplicate partition columns, so send the columns without partition columns
- val columns = if (SparkUtil.isSparkVersionXandAbove("2.2") &&
- carbonTable.isHivePartitionTable) {
+ val columns = if (carbonTable.isHivePartitionTable) {
val partitionColumns = partitionInfo.getColumnSchemaList.asScala
val carbonColumnsWithoutPartition = cols.filterNot(col => partitionColumns.contains(col))
Some(carbonColumnsWithoutPartition)
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateDataSourceTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateDataSourceTableCommand.scala
new file mode 100644
index 0000000..f36e1ac
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateDataSourceTableCommand.scala
@@ -0,0 +1,61 @@
+/*
+ * 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.sql.execution.command.table
+
+import org.apache.spark.sql.{CarbonEnv, CarbonSource, Row, SparkSession}
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.execution.command.{CreateDataSourceTableCommand, MetadataCommand}
+
+/**
+ * Command to create table in case of 'USING CARBONDATA' DDL
+ *
+ * @param catalogTable catalog table created by spark
+ * @param ignoreIfExists ignore if table exists
+ * @param sparkSession spark session
+ */
+case class CarbonCreateDataSourceTableCommand(
+ catalogTable: CatalogTable,
+ ignoreIfExists: Boolean,
+ sparkSession: SparkSession)
+ extends MetadataCommand {
+
+ override def processMetadata(session: SparkSession): Seq[Row] = {
+ // Run the spark command to create table in metastore before saving carbon schema
+ // in table path.
+ // This is required for spark 2.4, because spark 2.4 will fail to create table
+ // if table path is created before hand
+ val updatedCatalogTable =
+ CarbonSource.updateCatalogTableWithCarbonSchema(catalogTable, session, None, false)
+ val sparkCommand = CreateDataSourceTableCommand(updatedCatalogTable, ignoreIfExists)
+ sparkCommand.run(session)
+
+ // save the table info (carbondata's schema) in table path
+ val tableName = catalogTable.identifier.table
+ val dbName = CarbonEnv.getDatabaseName(catalogTable.identifier.database)(session)
+ val tablePath = CarbonEnv.getTablePath(Some(dbName), tableName)(session)
+ val tableInfo = CarbonSource.createTableInfo(
+ dbName, tableName, tablePath, catalogTable.schema, catalogTable.properties, None, session)
+ val metastore = CarbonEnv.getInstance(session).carbonMetaStore
+ metastore.saveToDisk(tableInfo, tablePath)
+
+ Seq.empty
+ }
+
+ override protected def opName: String = "CREATE TABLE USING CARBONDATA"
+
+}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
index 4adb1aa..2a5d78b 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableCommand.scala
@@ -207,8 +207,9 @@
} catch {
case _: Exception => // No operation
}
+ throw e
val msg = s"Create table'$tableName' in database '$dbName' failed"
- throwMetadataException(dbName, tableName, msg.concat(", ").concat(e.getMessage))
+ throwMetadataException(dbName, tableName, s"$msg, ${e.getMessage}")
}
}
val createTablePostExecutionEvent: CreateTablePostExecutionEvent =
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
index 54a5757..8700b29 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
@@ -134,6 +134,7 @@
}
val indexDatamapSchemas =
DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
+ LOGGER.info(s"Dropping DataMaps in table $tableName, size: " + indexDatamapSchemas.size())
if (!indexDatamapSchemas.isEmpty) {
childDropDataMapCommands = indexDatamapSchemas.asScala.map { schema =>
val command = CarbonDropDataMapCommand(schema.getDataMapName,
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 2e1f91f..8acc749 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
@@ -142,10 +142,13 @@
val relation = CarbonDecoderRelation(logicalRelation.attributeMap,
logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation])
val attrs = projectExprsNeedToDecode.map { attr =>
- val newAttr = AttributeReference(attr.name,
+ val newAttr = CarbonToSparkAdapter.createAttributeReference(
+ attr.name,
attr.dataType,
attr.nullable,
- attr.metadata)(attr.exprId, Option(table.carbonRelation.tableName))
+ attr.metadata,
+ attr.exprId,
+ Option(table.carbonRelation.tableName))
relation.addAttribute(newAttr)
newAttr
}
@@ -194,8 +197,7 @@
attr.nullable,
attr.metadata,
attr.exprId,
- attr.qualifier,
- attr)
+ attr.qualifier)
}
}
partitions =
@@ -403,7 +405,7 @@
newProjectList :+= reference
a.transform {
case s: ScalaUDF =>
- ScalaUDF(s.function, s.dataType, Seq(reference), s.inputTypes)
+ CarbonToSparkAdapter.createScalaUDF(s, reference)
}
case other => other
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
index a851bc3..49a3d3b 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
@@ -27,7 +27,7 @@
import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonInsertIntoCommand, CarbonLoadDataCommand, RefreshCarbonTableCommand}
import org.apache.spark.sql.execution.command.partition.{CarbonAlterTableAddHivePartitionCommand, CarbonAlterTableDropHivePartitionCommand}
import org.apache.spark.sql.execution.command.schema._
-import org.apache.spark.sql.execution.command.table.{CarbonCreateTableLikeCommand, CarbonDescribeFormattedCommand, CarbonDropTableCommand}
+import org.apache.spark.sql.execution.command.table.{CarbonCreateDataSourceTableCommand, CarbonCreateTableLikeCommand, CarbonDescribeFormattedCommand, CarbonDropTableCommand}
import org.apache.spark.sql.hive.execution.command.{CarbonDropDatabaseCommand, CarbonResetCommand, CarbonSetCommand}
import org.apache.spark.sql.CarbonExpressions.{CarbonDescribeTable => DescribeTableCommand}
import org.apache.spark.sql.execution.datasources.{RefreshResource, RefreshTable}
@@ -171,7 +171,7 @@
ExecutedCommandExec(addColumn) :: Nil
}
// TODO: remove this else if check once the 2.1 version is unsupported by carbon
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
+ } else {
val structField = (alterTableAddColumnsModel.dimCols ++ alterTableAddColumnsModel.msrCols)
.map {
a =>
@@ -185,8 +185,6 @@
.invokeAlterTableAddColumn(identifier, structField).asInstanceOf[RunnableCommand]) ::
Nil
// TODO: remove this else check once the 2.1 version is unsupported by carbon
- } else {
- throw new MalformedCarbonCommandException("Unsupported alter operation on hive table")
}
case dropColumn@CarbonAlterTableDropColumnCommand(alterTableDropColumnModel) =>
val isCarbonTable = CarbonEnv.getInstance(sparkSession).carbonMetaStore
@@ -275,9 +273,12 @@
if table.provider.get != DDLUtils.HIVE_PROVIDER
&& (table.provider.get.equals("org.apache.spark.sql.CarbonSource")
|| table.provider.get.equalsIgnoreCase("carbondata")) =>
- val updatedCatalog = CarbonSource
- .updateCatalogTableWithCarbonSchema(table, sparkSession)
- val cmd = CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists)
+ val cmd = if (SparkUtil.isSparkVersionEqualTo("2.4")) {
+ CarbonCreateDataSourceTableCommand(table, ignoreIfExists, sparkSession)
+ } else {
+ val updatedCatalog = CarbonSource.updateCatalogTableWithCarbonSchema(table, sparkSession)
+ CreateDataSourceTableCommand(updatedCatalog, ignoreIfExists)
+ }
ExecutedCommandExec(cmd) :: Nil
case AlterTableSetPropertiesCommand(tableName, properties, isView)
if CarbonEnv.getInstance(sparkSession).carbonMetaStore
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/MixedFormatHandler.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/MixedFormatHandler.scala
index fd7defa..5e82eb3 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/MixedFormatHandler.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/MixedFormatHandler.scala
@@ -26,8 +26,8 @@
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{MixedFormatHandlerUtil, SparkSession}
import org.apache.spark.sql.carbondata.execution.datasources.SparkCarbonFileFormat
-import org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, Expression, ExpressionSet, NamedExpression}
import org.apache.spark.sql.execution.{FilterExec, ProjectExec}
import org.apache.spark.sql.execution.datasources.{FileFormat, HadoopFsRelation, InMemoryFileIndex, LogicalRelation}
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 5323293..eda131f 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
@@ -78,9 +78,7 @@
"Update operation is not supported for mv datamap table")
}
}
- val tableRelation = if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- relation
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
+ val tableRelation =
alias match {
case Some(_) =>
CarbonReflectionUtils.getSubqueryAlias(
@@ -90,9 +88,6 @@
Some(table.tableIdentifier))
case _ => relation
}
- } else {
- throw new UnsupportedOperationException("Unsupported Spark version.")
- }
CarbonReflectionUtils.getSubqueryAlias(
sparkSession,
@@ -221,21 +216,15 @@
}
}
// include tuple id in subquery
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- Project(projList, relation)
- } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
- alias match {
- case Some(_) =>
- val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(
- sparkSession,
- alias,
- relation,
- Some(table.tableIdentifier))
- Project(projList, subqueryAlias)
- case _ => Project(projList, relation)
- }
- } else {
- throw new UnsupportedOperationException("Unsupported Spark version.")
+ alias match {
+ case Some(_) =>
+ val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(
+ sparkSession,
+ alias,
+ relation,
+ Some(table.tableIdentifier))
+ Project(projList, subqueryAlias)
+ case _ => Project(projList, relation)
}
}
CarbonProjectForDeleteCommand(
@@ -314,13 +303,7 @@
}
}
val newChild: LogicalPlan = if (newChildOutput == child.output) {
- if (SparkUtil.isSparkVersionEqualTo("2.1")) {
- CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]
- } else if (SparkUtil.isSparkVersionEqualTo("2.2")) {
- CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]
- } else {
- throw new UnsupportedOperationException(s"Spark version $SPARK_VERSION is not supported")
- }
+ throw new UnsupportedOperationException(s"Spark version $SPARK_VERSION is not supported")
} else {
Project(newChildOutput, child)
}
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonAnalyzer.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalyzer.scala
similarity index 99%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonAnalyzer.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalyzer.scala
index 8f4d45e..04beea7 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonAnalyzer.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonAnalyzer.scala
@@ -47,4 +47,4 @@
logicalPlan
}
}
-}
\ No newline at end of file
+}
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 b2ba7f4..849d7ba 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
@@ -200,8 +200,7 @@
carbonDatasourceHadoopRelation: CarbonDatasourceHadoopRelation, _, _) =>
carbonDatasourceHadoopRelation.carbonRelation
case SubqueryAlias(_, c)
- if (SparkUtil.isSparkVersionXandAbove("2.2")) &&
- (c.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") ||
+ if (c.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") ||
c.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.HiveTableRelation") ||
c.getClass.getName.equals(
"org.apache.spark.sql.catalyst.catalog.UnresolvedCatalogRelation")) =>
@@ -524,8 +523,7 @@
carbonDataSourceHadoopRelation: CarbonDatasourceHadoopRelation, _, _) =>
carbonDataSourceHadoopRelation
case SubqueryAlias(_, c)
- if (SparkUtil.isSparkVersionXandAbove("2.2")) &&
- (c.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") ||
+ if (c.getClass.getName.equals("org.apache.spark.sql.catalyst.catalog.CatalogRelation") ||
c.getClass.getName
.equals("org.apache.spark.sql.catalyst.catalog.HiveTableRelation") ||
c.getClass.getName.equals(
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala
similarity index 68%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala
index 72d3ae2..e19966f 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonOptimizerUtil.scala
@@ -1,3 +1,20 @@
+/*
+ * 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.sql.hive
import org.apache.spark.sql.CarbonDatasourceHadoopRelation
@@ -29,7 +46,7 @@
}
Exists(tPlan, e.children.map(_.canonicalized), e.exprId)
- case In(value, Seq(l:ListQuery)) =>
+ case In(value, Seq(l: ListQuery)) =>
val tPlan = l.plan.transform {
case lr: LogicalRelation
if lr.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
@@ -41,4 +58,4 @@
}
transFormedPlan
}
-}
\ No newline at end of file
+}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
index 9b3ff87..68c293b 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
@@ -18,11 +18,11 @@
import java.util.LinkedHashSet
-import scala.Array.canBuildFrom
import scala.collection.JavaConverters._
+import org.apache.spark.sql.CarbonToSparkAdapter
import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, NamedExpression}
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.{CarbonMetastoreTypes, SparkTypeConverter}
@@ -119,7 +119,8 @@
val dataType = SparkTypeConverter.addDecimalScaleAndPrecision(column, dType)
CarbonMetastoreTypes.toDataType(dataType)
}
- AttributeReference(column.getColName, output, nullable = true )(
+ CarbonToSparkAdapter.createAttributeReference(
+ column.getColName, output, nullable = true, Metadata.empty, NamedExpression.newExprId,
qualifier = Option(tableName + "." + column.getColName))
} else {
val output = CarbonMetastoreTypes.toDataType {
@@ -129,7 +130,8 @@
case others => others
}
}
- AttributeReference(column.getColName, output, nullable = true)(
+ CarbonToSparkAdapter.createAttributeReference(
+ column.getColName, output, nullable = true, Metadata.empty, NamedExpression.newExprId,
qualifier = Option(tableName + "." + column.getColName))
}
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionCatalog.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionCatalog.scala
index 20d43df..dcba730 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionCatalog.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionCatalog.scala
@@ -26,11 +26,9 @@
import org.apache.carbondata.core.metadata.schema.table.column.{ColumnSchema => ColumnSchema}
/**
- * This interface defines those common api used by carbon for spark-2.1 and spark-2.2 integration,
+ * This interface defines those common api used by carbon for spark integration,
* but are not defined in SessionCatalog or HiveSessionCatalog to give contract to the
* Concrete implementation classes.
- * For example CarbonSessionCatalog defined in 2.1 and 2.2.
- *
*/
@InterfaceAudience.Internal
@InterfaceStability.Stable
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionState.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
similarity index 92%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionState.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
index f78c785..44b3bfd 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionState.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
@@ -27,12 +27,11 @@
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.optimizer.Optimizer
import org.apache.spark.sql.catalyst.parser.ParserInterface
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, PreWriteCheck, ResolveSQLOnFile, _}
import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
import org.apache.spark.sql.hive.client.HiveClient
-import org.apache.spark.sql.internal.{SQLConf, SessionState}
+import org.apache.spark.sql.internal.{SessionState, SQLConf}
import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonLateDecodeRule, CarbonUDFTransformRule}
import org.apache.spark.sql.parser.CarbonSparkSqlParser
@@ -61,8 +60,8 @@
parser: ParserInterface,
functionResourceLoader: FunctionResourceLoader)
extends HiveSessionCatalog (
- externalCatalog,
- globalTempViewManager,
+ SparkAdapter.getExternalCatalogCatalog(externalCatalog),
+ SparkAdapter.getGlobalTempViewManager(globalTempViewManager),
new HiveMetastoreCatalog(sparkSession),
functionRegistry,
conf,
@@ -111,8 +110,9 @@
* @return
*/
override def getClient(): org.apache.spark.sql.hive.client.HiveClient = {
- sparkSession.asInstanceOf[CarbonSession].sharedState.externalCatalog
- .asInstanceOf[HiveExternalCatalog].client
+ SparkAdapter.getHiveExternalCatalog(
+ sparkSession.asInstanceOf[CarbonSession].sharedState.externalCatalog
+ ).client
}
override def alterAddColumns(tableIdentifier: TableIdentifier,
@@ -174,9 +174,10 @@
* @param identifier
* @return
*/
- override def getPartitionsAlternate(partitionFilters: Seq[Expression],
+ override def getPartitionsAlternate(
+ partitionFilters: Seq[Expression],
sparkSession: SparkSession,
- identifier: TableIdentifier) = {
+ identifier: TableIdentifier): Seq[CatalogTablePartition] = {
CarbonSessionUtil.prunePartitionsByFilter(partitionFilters, sparkSession, identifier)
}
@@ -233,14 +234,14 @@
}
private def externalCatalog: HiveExternalCatalog =
- session.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog]
+ SparkAdapter.getHiveExternalCatalog(session.sharedState.externalCatalog)
/**
* Create a Hive aware resource loader.
*/
override protected lazy val resourceLoader: HiveSessionResourceLoader = {
val client: HiveClient = externalCatalog.client.newSession()
- new HiveSessionResourceLoader(session, client)
+ new HiveSessionResourceLoader(session, SparkAdapter.getHiveClient(client))
}
override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)
@@ -274,4 +275,4 @@
}
override protected def newBuilder: NewBuilder = new CarbonSessionStateBuilder(_, _)
-}
\ No newline at end of file
+}
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionUtil.scala
similarity index 85%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionUtil.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionUtil.scala
index e3f1d3f..e9bcb43 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSessionUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionUtil.scala
@@ -1,39 +1,36 @@
/*
-* 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.
-*/
+ * 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.sql.hive
-import java.util.concurrent.Callable
+import scala.collection.mutable.ArrayBuffer
-import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTablePartition, ExternalCatalogUtils}
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
-import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
-import org.apache.spark.sql.types.MetadataBuilder
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.types.{MetadataBuilder, StructField, StructType}
+import org.apache.spark.sql.util.SparkTypeConverter
import org.apache.spark.util.CarbonReflectionUtils
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
-import org.apache.spark.sql.types.{StructField, StructType}
-import org.apache.spark.sql.util.SparkTypeConverter
-import scala.collection.mutable.ArrayBuffer
-
-import org.apache.spark.sql.execution.datasources.LogicalRelation
/**
* This class refresh the relation from cache if the carbontable in
@@ -177,8 +174,7 @@
def updateCachedPlan(plan: LogicalPlan): LogicalPlan = {
plan match {
- case sa@SubqueryAlias(_,
- MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, catalogTable)) =>
+ case sa@SubqueryAlias(_, MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _)) =>
sa.copy(child = sa.child.asInstanceOf[LogicalRelation].copy())
case MatchLogicalRelation(_: CarbonDatasourceHadoopRelation, _, _) =>
plan.asInstanceOf[LogicalRelation].copy()
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
similarity index 93%
rename from integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
index 73b6790..36ec2c8 100644
--- a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
@@ -38,8 +38,8 @@
fileStorage.equalsIgnoreCase("'carbonfile'") ||
fileStorage.equalsIgnoreCase("'org.apache.carbondata.format'")) {
val createTableTuple = (ctx.createTableHeader, ctx.skewSpec(0),
- ctx.bucketSpec(0), ctx.partitionColumns, ctx.columns, ctx.tablePropertyList(0),ctx.locationSpec(0),
- Option(ctx.STRING(0)).map(string), ctx.AS, ctx.query, fileStorage)
+ ctx.bucketSpec(0), ctx.partitionColumns, ctx.columns, ctx.tablePropertyList(0),
+ ctx.locationSpec(0), Option(ctx.STRING(0)).map(string), ctx.AS, ctx.query, fileStorage)
helper.createCarbonTable(createTableTuple)
} else {
super.visitCreateHiveTable(ctx)
@@ -47,6 +47,6 @@
}
override def visitAddTableColumns(ctx: AddTableColumnsContext): LogicalPlan = {
- visitAddTableColumns(parser,ctx)
+ visitAddTableColumns(parser, ctx)
}
}
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSqlConf.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSqlConf.scala
similarity index 100%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonSqlConf.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSqlConf.scala
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
similarity index 98%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
index ee9fb0f..0335b36 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
@@ -1,4 +1,3 @@
-
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -20,12 +19,12 @@
import java.net.URI
+import org.apache.spark.sql.{AnalysisException, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, CatalogUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.command.{AlterTableRecoverPartitionsCommand, AtomicRunnableCommand, RunnableCommand}
+import org.apache.spark.sql.execution.command.{AlterTableRecoverPartitionsCommand, AtomicRunnableCommand}
import org.apache.spark.sql.execution.datasources.{DataSource, HadoopFsRelation}
import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.{AnalysisException, Dataset, Row, SaveMode, SparkSession}
import org.apache.spark.util.CarbonReflectionUtils
/**
@@ -49,7 +48,7 @@
Seq.empty
}
- override def processData(sparkSession: SparkSession): Seq[Row] ={
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
assert(table.tableType != CatalogTableType.VIEW)
assert(table.provider.isDefined)
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/SqlAstBuilderHelper.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/SqlAstBuilderHelper.scala
similarity index 100%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/SqlAstBuilderHelper.scala
rename to integration/spark2/src/main/scala/org/apache/spark/sql/hive/SqlAstBuilderHelper.scala
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
index 2765c5f..e288e6d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
@@ -315,15 +315,13 @@
* @return
*/
def isStringTrimCompatibleWithCarbon(stringTrim: StringTrim): Boolean = {
- var isCompatible = true
- if (SparkUtil.isSparkVersionXandAbove("2.3")) {
- val trimStr = CarbonReflectionUtils.getField("trimStr", stringTrim)
- .asInstanceOf[Option[Expression]]
- if (trimStr.isDefined) {
- isCompatible = false
- }
+ val trimStr = CarbonReflectionUtils.getField("trimStr", stringTrim)
+ .asInstanceOf[Option[Expression]]
+ if (trimStr.isDefined) {
+ false
+ } else {
+ true
}
- isCompatible
}
def transformExpression(expr: Expression): CarbonExpression = {
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index 10b661a..4866301 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -305,7 +305,7 @@
}
}
val (selectStmt, relation) =
- if (!selectPattern.findFirstIn(sel.toLowerCase).isDefined ||
+ if (selectPattern.findFirstIn(sel.toLowerCase).isEmpty ||
!StringUtils.isEmpty(subQueryResults)) {
// if subQueryResults are not empty means, it is not join with main table.
// so use subQueryResults in update with value flow.
@@ -348,8 +348,6 @@
rel
}
-
-
private def updateRelation(
r: UnresolvedRelation,
tableIdent: Seq[String],
@@ -362,8 +360,6 @@
case Seq(dbName, tableName) => Some(tableName)
case Seq(tableName) => Some(tableName)
}
- // Use Reflection to choose between Spark2.1 and Spark2.2
- // Move UnresolvedRelation(tableIdentifier, tableAlias) to reflection.
CarbonReflectionUtils.getUnresolvedRelation(tableIdentifier, tableAlias)
}
}
@@ -386,8 +382,6 @@
val tableIdentifier: TableIdentifier = toTableIdentifier(tableIdent)
- // Use Reflection to choose between Spark2.1 and Spark2.2
- // Move (UnresolvedRelation(tableIdent, alias), tableIdent, alias) to reflection.
val unresolvedRelation = CarbonReflectionUtils.getUnresolvedRelation(tableIdentifier, alias)
(unresolvedRelation, tableIdent, alias, tableIdentifier)
@@ -678,6 +672,7 @@
val name = field.column.toLowerCase
field.copy(column = name, name = Some(name))
}
+
protected lazy val alterTableDropColumn: Parser[LogicalPlan] =
ALTER ~> TABLE ~> (ident <~ ".").? ~ ident ~ DROP ~ COLUMNS ~
("(" ~> rep1sep(ident, ",") <~ ")") <~ opt(";") ^^ {
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/CarbonToSparkAdapater.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/CarbonToSparkAdapater.scala
deleted file mode 100644
index 79a6240..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/CarbonToSparkAdapater.scala
+++ /dev/null
@@ -1,90 +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.sql
-
-import java.net.URI
-
-import org.apache.spark.SparkContext
-import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
-import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSet, ExprId, Expression, ExpressionSet, NamedExpression, SubqueryExpression}
-import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext
-import org.apache.spark.sql.catalyst.optimizer.OptimizeCodegen
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.command.ExplainCommand
-import org.apache.spark.sql.types.{DataType, Metadata}
-
-object CarbonToSparkAdapter {
-
- def addSparkListener(sparkContext: SparkContext) = {
- sparkContext.addSparkListener(new SparkListener {
- override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
- SparkSession.setDefaultSession(null)
- SparkSession.sqlListener.set(null)
- }
- })
- }
-
- def createAttributeReference(name: String, dataType: DataType, nullable: Boolean,
- metadata: Metadata,exprId: ExprId, qualifier: Option[String],
- attrRef : NamedExpression): AttributeReference = {
- AttributeReference(
- name,
- dataType,
- nullable,
- metadata)(exprId, qualifier,attrRef.isGenerated)
- }
-
- def createAliasRef(child: Expression,
- name: String,
- exprId: ExprId = NamedExpression.newExprId,
- qualifier: Option[String] = None,
- explicitMetadata: Option[Metadata] = None,
- namedExpr: Option[NamedExpression] = None): Alias = {
- val isGenerated:Boolean = if (namedExpr.isDefined) {
- namedExpr.get.isGenerated
- } else {
- false
- }
- Alias(child, name)(exprId, qualifier, explicitMetadata,isGenerated)
- }
-
- def getExplainCommandObj() : ExplainCommand = {
- ExplainCommand(OneRowRelation)
- }
-
- def getPartitionKeyFilter(
- partitionSet: AttributeSet,
- filterPredicates: Seq[Expression]): ExpressionSet = {
- ExpressionSet(
- ExpressionSet(filterPredicates)
- .filter(_.references.subsetOf(partitionSet)))
- }
-
- def getUpdatedStorageFormat(storageFormat: CatalogStorageFormat,
- map: Map[String, String],
- tablePath: String): CatalogStorageFormat = {
- storageFormat.copy(properties = map, locationUri = Some(tablePath))
- }
-
- def getOptimizeCodegenRule(conf :SQLConf): Seq[Rule[LogicalPlan]] = {
- Seq(OptimizeCodegen(conf))
- }
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/MixedFormatHandlerUtil.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/MixedFormatHandlerUtil.scala
deleted file mode 100644
index 69ed477..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/MixedFormatHandlerUtil.scala
+++ /dev/null
@@ -1,44 +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.sql
-
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
-import org.apache.spark.sql.execution.FileSourceScanExec
-import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, HadoopFsRelation}
-import org.apache.spark.sql.types.StructType
-
-object MixedFormatHandlerUtil {
-
- def getScanForSegments(
- @transient relation: HadoopFsRelation,
- output: Seq[Attribute],
- outputSchema: StructType,
- partitionFilters: Seq[Expression],
- dataFilters: Seq[Expression],
- tableIdentifier: Option[TableIdentifier]
- ): FileSourceScanExec = {
- val pushedDownFilters = dataFilters.flatMap(DataSourceStrategy.translateFilter)
- FileSourceScanExec(
- relation,
- output,
- outputSchema,
- partitionFilters,
- pushedDownFilters,
- tableIdentifier)
- }
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/catalog/HiveTableRelation.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/catalog/HiveTableRelation.scala
deleted file mode 100644
index eb3e88d..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/catalog/HiveTableRelation.scala
+++ /dev/null
@@ -1,56 +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.sql.catalyst.catalog
-
-import com.google.common.base.Objects
-import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
-import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan, Statistics}
-import org.apache.spark.sql.internal.SQLConf
-
-/**
- * A `LogicalPlan` that represents a hive table.
- *
- * TODO: remove this after we completely make hive as a data source.
- */
-case class HiveTableRelation(
- tableMeta: CatalogTable,
- dataCols: Seq[AttributeReference],
- partitionCols: Seq[AttributeReference]) extends LeafNode with MultiInstanceRelation {
- assert(tableMeta.identifier.database.isDefined)
- assert(tableMeta.partitionSchema.sameType(partitionCols.toStructType))
- assert(tableMeta.schema.sameType(dataCols.toStructType))
-
- // The partition column should always appear after data columns.
- override def output: Seq[AttributeReference] = dataCols ++ partitionCols
-
- def isPartitioned: Boolean = partitionCols.nonEmpty
-
- override def equals(relation: Any): Boolean = relation match {
- case other: HiveTableRelation => tableMeta == other.tableMeta && output == other.output
- case _ => false
- }
-
- override def hashCode(): Int = {
- Objects.hashCode(tableMeta.identifier, output)
- }
-
- override def newInstance(): HiveTableRelation = copy(
- dataCols = dataCols.map(_.newInstance()),
- partitionCols = partitionCols.map(_.newInstance()))
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/optimizer/MigrateOptimizer.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/optimizer/MigrateOptimizer.scala
deleted file mode 100644
index 9a88255..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/catalyst/optimizer/MigrateOptimizer.scala
+++ /dev/null
@@ -1,129 +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.sql.catalyst.optimizer
-
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, CaseKeyWhen, CreateArray, CreateMap, CreateNamedStructLike, GetArrayItem, GetArrayStructFields, GetMapValue, GetStructField, IntegerLiteral, Literal}
-import org.apache.spark.sql.catalyst.expressions.aggregate.First
-import org.apache.spark.sql.catalyst.expressions.objects.{LambdaVariable, MapObjects}
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Project, UnaryNode}
-import org.apache.spark.sql.catalyst.rules.Rule
-
-class MigrateOptimizer {
-
-}
-
-/**
- * Replaces logical [[Deduplicate]] operator with an [[Aggregate]] operator.
- */
-object ReplaceDeduplicateWithAggregate extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case Deduplicate(keys, child, streaming) if !streaming =>
- val keyExprIds = keys.map(_.exprId)
- val aggCols = child.output.map { attr =>
- if (keyExprIds.contains(attr.exprId)) {
- attr
- } else {
- Alias(new First(attr).toAggregateExpression(), attr.name)(attr.exprId)
- }
- }
- Aggregate(keys, aggCols, child)
- }
-}
-
-/** A logical plan for `dropDuplicates`. */
-case class Deduplicate(
- keys: Seq[Attribute],
- child: LogicalPlan,
- streaming: Boolean) extends UnaryNode {
-
- override def output: Seq[Attribute] = child.output
-}
-
-/**
- * Remove projections from the query plan that do not make any modifications.
- */
-object RemoveRedundantProject extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case p @ Project(_, child) if p.output == child.output => child
- }
-}
-
-/**
- * push down operations into [[CreateNamedStructLike]].
- */
-object SimplifyCreateStructOps extends Rule[LogicalPlan] {
- override def apply(plan: LogicalPlan): LogicalPlan = {
- plan.transformExpressionsUp {
- // push down field extraction
- case GetStructField(createNamedStructLike: CreateNamedStructLike, ordinal, _) =>
- createNamedStructLike.valExprs(ordinal)
- }
- }
-}
-
-/**
- * push down operations into [[CreateArray]].
- */
-object SimplifyCreateArrayOps extends Rule[LogicalPlan] {
- override def apply(plan: LogicalPlan): LogicalPlan = {
- plan.transformExpressionsUp {
- // push down field selection (array of structs)
- case GetArrayStructFields(CreateArray(elems), field, ordinal, numFields, containsNull) =>
- // instead f selecting the field on the entire array,
- // select it from each member of the array.
- // pushing down the operation this way open other optimizations opportunities
- // (i.e. struct(...,x,...).x)
- CreateArray(elems.map(GetStructField(_, ordinal, Some(field.name))))
- // push down item selection.
- case ga @ GetArrayItem(CreateArray(elems), IntegerLiteral(idx)) =>
- // instead of creating the array and then selecting one row,
- // remove array creation altgether.
- if (idx >= 0 && idx < elems.size) {
- // valid index
- elems(idx)
- } else {
- // out of bounds, mimic the runtime behavior and return null
- Literal(null, ga.dataType)
- }
- }
- }
-}
-
-/**
- * push down operations into [[CreateMap]].
- */
-object SimplifyCreateMapOps extends Rule[LogicalPlan] {
- override def apply(plan: LogicalPlan): LogicalPlan = {
- plan.transformExpressionsUp {
- case GetMapValue(CreateMap(elems), key) => CaseKeyWhen(key, elems)
- }
- }
-}
-
-
-/**
- * Removes MapObjects when the following conditions are satisfied
- * 1. Mapobject(... lambdavariable(..., false) ...), which means types for input and output
- * are primitive types with non-nullable
- * 2. no custom collection class specified representation of data item.
- */
-object EliminateMapObjects extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
- case MapObjects(_, _, _, LambdaVariable(_, _, _), inputData) => inputData
- }
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSQLConf.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSQLConf.scala
deleted file mode 100644
index 8eb05fc..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSQLConf.scala
+++ /dev/null
@@ -1,132 +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.sql.hive
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.internal.SQLConf.SQLConfigBuilder
-
-import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants}
-import org.apache.carbondata.core.util.CarbonProperties
-
-/**
- * To initialize dynamic values default param
- */
-class CarbonSQLConf(sparkSession: SparkSession) {
-
- val carbonProperties = CarbonProperties.getInstance()
-
- /**
- * To initialize dynamic param defaults along with usage docs
- */
- def addDefaultCarbonParams(): Unit = {
- val ENABLE_UNSAFE_SORT =
- SQLConfigBuilder(CarbonCommonConstants.ENABLE_UNSAFE_SORT)
- .doc("To enable/ disable unsafe sort.")
- .booleanConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
- CarbonCommonConstants.ENABLE_UNSAFE_SORT_DEFAULT).toBoolean)
- val CARBON_CUSTOM_BLOCK_DISTRIBUTION =
- SQLConfigBuilder(CarbonCommonConstants.CARBON_CUSTOM_BLOCK_DISTRIBUTION)
- .doc("To set carbon task distribution.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.CARBON_TASK_DISTRIBUTION,
- CarbonCommonConstants.CARBON_TASK_DISTRIBUTION_DEFAULT))
- val BAD_RECORDS_LOGGER_ENABLE =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE)
- .doc("To enable/ disable carbon bad record logger.")
- .booleanConf
- .createWithDefault(CarbonLoadOptionConstants
- .CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE_DEFAULT.toBoolean)
- val BAD_RECORDS_ACTION =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION)
- .doc("To configure the bad records action.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION,
- CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION_DEFAULT))
- val IS_EMPTY_DATA_BAD_RECORD =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD)
- .doc("Property to decide weather empty data to be considered bad/ good record.")
- .booleanConf
- .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD_DEFAULT
- .toBoolean)
- val SORT_SCOPE =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_SORT_SCOPE)
- .doc("Property to specify sort scope.")
- .stringConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.LOAD_SORT_SCOPE,
- CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT))
- val BAD_RECORD_PATH =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH)
- .doc("Property to configure the bad record location.")
- .stringConf
- .createWithDefault(carbonProperties.getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC,
- CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL))
- val GLOBAL_SORT_PARTITIONS =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS)
- .doc("Property to configure the global sort partitions.")
- .stringConf
- .createWithDefault(carbonProperties
- .getProperty(CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS,
- CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS_DEFAULT))
- val DATEFORMAT =
- SQLConfigBuilder(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT)
- .doc("Property to configure data format for date type columns.")
- .stringConf
- .createWithDefault(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT_DEFAULT)
- val CARBON_INPUT_SEGMENTS = SQLConfigBuilder(
- "carbon.input.segments.<database_name>.<table_name>")
- .doc("Property to configure the list of segments to query.").stringConf
- .createWithDefault(carbonProperties
- .getProperty("carbon.input.segments.<database_name>.<table_name>", "*"))
- }
- /**
- * to set the dynamic properties default values
- */
- def addDefaultCarbonSessionParams(): Unit = {
- sparkSession.conf.set(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
- carbonProperties.getProperty(CarbonCommonConstants.ENABLE_UNSAFE_SORT,
- CarbonCommonConstants.ENABLE_UNSAFE_SORT_DEFAULT).toBoolean)
- sparkSession.conf.set(CarbonCommonConstants.CARBON_CUSTOM_BLOCK_DISTRIBUTION,
- carbonProperties
- .getProperty(CarbonCommonConstants.CARBON_TASK_DISTRIBUTION,
- CarbonCommonConstants.CARBON_TASK_DISTRIBUTION_DEFAULT))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE,
- CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_LOGGER_ENABLE_DEFAULT.toBoolean)
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION,
- carbonProperties.getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION,
- CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION_DEFAULT))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD,
- CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD_DEFAULT.toBoolean)
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_SORT_SCOPE,
- carbonProperties.getProperty(CarbonCommonConstants.LOAD_SORT_SCOPE,
- CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH,
- carbonProperties.getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC,
- CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORD_PATH,
- carbonProperties.getProperty(CarbonCommonConstants.CARBON_BADRECORDS_LOC,
- CarbonCommonConstants.CARBON_BADRECORDS_LOC_DEFAULT_VAL))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_GLOBAL_SORT_PARTITIONS,
- carbonProperties.getProperty(CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS,
- CarbonCommonConstants.LOAD_GLOBAL_SORT_PARTITIONS_DEFAULT))
- sparkSession.conf.set(CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT,
- CarbonLoadOptionConstants.CARBON_OPTIONS_DATEFORMAT_DEFAULT)
- }
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
deleted file mode 100644
index 6b94806..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CarbonSessionState.scala
+++ /dev/null
@@ -1,379 +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.sql.hive
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
-import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry}
-import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTablePartition, FunctionResourceLoader, GlobalTempViewManager, SessionCatalog}
-import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference, BoundReference, Expression, InterpretedPredicate, PredicateSubquery, ScalarSubquery}
-import org.apache.spark.sql.catalyst.optimizer.Optimizer
-import org.apache.spark.sql.catalyst.parser.{ParserInterface, SqlBaseParser}
-import org.apache.spark.sql.catalyst.parser.ParserUtils.{string, _}
-import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{CreateTableContext, ShowTablesContext}
-import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, SubqueryAlias}
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.{CatalystConf, TableIdentifier}
-import org.apache.spark.sql.execution.command.table.{CarbonExplainCommand, CarbonShowTablesCommand}
-import org.apache.spark.sql.execution.datasources._
-import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
-import org.apache.spark.sql.execution.{SparkOptimizer, SparkSqlAstBuilder}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonLateDecodeRule, CarbonUDFTransformRule}
-import org.apache.spark.sql.parser.{CarbonHelperSqlAstBuilder, CarbonSpark2SqlParser, CarbonSparkSqlParser, CarbonSparkSqlParserUtil}
-import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, ExperimentalMethods, SparkSession, Strategy}
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datamap.DataMapStoreManager
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
-import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.spark.util.CarbonScalaUtil
-import org.apache.spark.util.CarbonReflectionUtils
-
-/**
- * This class will have carbon catalog and refresh the relation from cache if the carbontable in
- * carbon catalog is not same as cached carbon relation's carbon table
- *
- * @param externalCatalog
- * @param globalTempViewManager
- * @param sparkSession
- * @param functionResourceLoader
- * @param functionRegistry
- * @param conf
- * @param hadoopConf
- */
-class CarbonHiveSessionCatalog(
- externalCatalog: HiveExternalCatalog,
- globalTempViewManager: GlobalTempViewManager,
- sparkSession: SparkSession,
- functionResourceLoader: FunctionResourceLoader,
- functionRegistry: FunctionRegistry,
- conf: SQLConf,
- hadoopConf: Configuration)
- extends HiveSessionCatalog(
- externalCatalog,
- globalTempViewManager,
- sparkSession,
- functionResourceLoader,
- functionRegistry,
- conf,
- hadoopConf) with CarbonSessionCatalog {
-
- private lazy val carbonEnv = {
- val env = new CarbonEnv
- env.init(sparkSession)
- env
- }
- /**
- * return's the carbonEnv instance
- * @return
- */
- override def getCarbonEnv() : CarbonEnv = {
- carbonEnv
- }
-
- def alterAddColumns(tableIdentifier: TableIdentifier,
- schemaParts: String,
- cols: Option[Seq[ColumnSchema]])
- : Unit = {
- alterTable(tableIdentifier, schemaParts, cols)
- }
-
- def alterDropColumns(tableIdentifier: TableIdentifier,
- schemaParts: String,
- cols: Option[Seq[ColumnSchema]])
- : Unit = {
- alterTable(tableIdentifier, schemaParts, cols)
- }
-
- def alterColumnChangeDataTypeOrRename(tableIdentifier: TableIdentifier,
- schemaParts: String,
- cols: Option[Seq[ColumnSchema]])
- : Unit = {
- alterTable(tableIdentifier, schemaParts, cols)
- }
-
- // Initialize all listeners to the Operation bus.
- CarbonEnv.init
-
- /**
- * This method will invalidate carbonrelation from cache if carbon table is updated in
- * carbon catalog
- *
- * @param name
- * @param alias
- * @return
- */
- override def lookupRelation(name: TableIdentifier,
- alias: Option[String]): LogicalPlan = {
- val rtnRelation = super.lookupRelation(name, alias)
- var toRefreshRelation = false
- rtnRelation match {
- case SubqueryAlias(_,
- LogicalRelation(carbonDatasourceHadoopRelation: CarbonDatasourceHadoopRelation, _, _), _) =>
- toRefreshRelation = CarbonEnv.isRefreshRequired(name)(sparkSession)
- case LogicalRelation(_: CarbonDatasourceHadoopRelation, _, _) =>
- toRefreshRelation = CarbonEnv.isRefreshRequired(name)(sparkSession)
- case _ =>
- }
-
- if (toRefreshRelation) {
- super.lookupRelation(name, alias)
- } else {
- rtnRelation
- }
- }
-
- /**
- * returns hive client from session state
- *
- * @return
- */
- override def getClient(): org.apache.spark.sql.hive.client.HiveClient = {
- sparkSession.sessionState.asInstanceOf[CarbonSessionState].metadataHive
- }
-
- override def createPartitions(
- tableName: TableIdentifier,
- parts: Seq[CatalogTablePartition],
- ignoreIfExists: Boolean): Unit = {
- try {
- val table = CarbonEnv.getCarbonTable(tableName)(sparkSession)
- val updatedParts = CarbonScalaUtil.updatePartitions(parts, table)
- super.createPartitions(tableName, updatedParts, ignoreIfExists)
- } catch {
- case e: Exception =>
- super.createPartitions(tableName, parts, ignoreIfExists)
- }
- }
-
- /**
- * This is alternate way of getting partition information. It first fetches all partitions from
- * hive and then apply filter instead of querying hive along with filters.
- * @param partitionFilters
- * @param sparkSession
- * @param identifier
- * @return
- */
- def getPartitionsAlternate(
- partitionFilters: Seq[Expression],
- sparkSession: SparkSession,
- identifier: TableIdentifier) = {
- val allPartitions = sparkSession.sessionState.catalog.listPartitions(identifier)
- val catalogTable = sparkSession.sessionState.catalog.getTableMetadata(identifier)
- val partitionSchema = catalogTable.partitionSchema
- if (partitionFilters.nonEmpty) {
- val boundPredicate =
- InterpretedPredicate.create(partitionFilters.reduce(And).transform {
- case att: AttributeReference =>
- val index = partitionSchema.indexWhere(_.name == att.name)
- BoundReference(index, partitionSchema(index).dataType, nullable = true)
- })
- allPartitions.filter { p => boundPredicate(p.toRow(partitionSchema)) }
- } else {
- allPartitions
- }
- }
-
- /**
- * Update the storageformat with new location information
- */
- override def updateStorageLocation(
- path: Path,
- storage: CatalogStorageFormat,
- newTableName: String,
- dbName: String): CatalogStorageFormat = {
- storage.copy(locationUri = Some(path.toString))
- }
-}
-
-/**
- * Session state implementation to override sql parser and adding strategies
- * @param sparkSession
- */
-class CarbonSessionState(sparkSession: SparkSession) extends HiveSessionState(sparkSession) {
-
- override lazy val sqlParser: ParserInterface = new CarbonSparkSqlParser(conf, sparkSession)
-
- experimentalMethods.extraStrategies = extraStrategies
-
- experimentalMethods.extraOptimizations = extraOptimizations
-
- def extraStrategies: Seq[Strategy] = {
- Seq(
- new StreamingTableStrategy(sparkSession),
- new CarbonLateDecodeStrategy,
- new DDLStrategy(sparkSession)
- )
- }
-
- def extraOptimizations: Seq[Rule[LogicalPlan]] = {
- Seq(new CarbonIUDRule,
- new CarbonUDFTransformRule,
- new CarbonLateDecodeRule)
- }
-
- override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)
-
- def extendedAnalyzerRules: Seq[Rule[LogicalPlan]] = Nil
- def internalAnalyzerRules: Seq[Rule[LogicalPlan]] = {
- catalog.ParquetConversions ::
- catalog.OrcConversions ::
- CarbonPreInsertionCasts(sparkSession) ::
- CarbonIUDAnalysisRule(sparkSession) ::
- AnalyzeCreateTable(sparkSession) ::
- PreprocessTableInsertion(conf) ::
- DataSourceAnalysis(conf) ::
- (if (conf.runSQLonFile) {
- new ResolveDataSource(sparkSession) :: Nil
- } else { Nil })
- }
-
- override lazy val analyzer: Analyzer =
- new CarbonAnalyzer(catalog, conf, sparkSession,
- new Analyzer(catalog, conf) {
- override val extendedResolutionRules =
- if (extendedAnalyzerRules.nonEmpty) {
- extendedAnalyzerRules ++ internalAnalyzerRules
- } else {
- internalAnalyzerRules
- }
- override val extendedCheckRules = Seq(
- PreWriteCheck(conf, catalog))
- }
- )
-
- /**
- * Internal catalog for managing table and database states.
- */
- override lazy val catalog = {
- new CarbonHiveSessionCatalog(
- sparkSession.sharedState.externalCatalog.asInstanceOf[HiveExternalCatalog],
- sparkSession.sharedState.globalTempViewManager,
- sparkSession,
- functionResourceLoader,
- functionRegistry,
- conf,
- newHadoopConf())
- }
-}
-
-class CarbonAnalyzer(catalog: SessionCatalog,
- conf: CatalystConf,
- sparkSession: SparkSession,
- analyzer: Analyzer) extends Analyzer(catalog, conf) {
-
- val mvPlan = try {
- CarbonReflectionUtils.createObject(
- "org.apache.carbondata.mv.datamap.MVAnalyzerRule",
- sparkSession)._1.asInstanceOf[Rule[LogicalPlan]]
- } catch {
- case e: Exception =>
- null
- }
-
- override def execute(plan: LogicalPlan): LogicalPlan = {
- val logicalPlan = analyzer.execute(plan)
- if (mvPlan != null) {
- mvPlan.apply(logicalPlan)
- } else {
- logicalPlan
- }
- }
-}
-
-class CarbonOptimizer(
- catalog: SessionCatalog,
- conf: SQLConf,
- experimentalMethods: ExperimentalMethods)
- extends SparkOptimizer(catalog, conf, experimentalMethods) {
-
- override def execute(plan: LogicalPlan): LogicalPlan = {
- val transFormedPlan: LogicalPlan = CarbonOptimizerUtil.transformForScalarSubQuery(plan)
- super.execute(transFormedPlan)
- }
-}
-
-object CarbonOptimizerUtil {
- def transformForScalarSubQuery(plan: LogicalPlan) : LogicalPlan = {
- // In case scalar subquery add flag in relation to skip the decoder plan in optimizer rule,
- // And optimize whole plan at once.
- val transFormedPlan = plan.transform {
- case filter: Filter =>
- filter.transformExpressions {
- case s: ScalarSubquery =>
- val tPlan = s.plan.transform {
- case lr: LogicalRelation
- if lr.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
- lr.relation.asInstanceOf[CarbonDatasourceHadoopRelation].isSubquery += true
- lr
- }
- ScalarSubquery(tPlan, s.children, s.exprId)
- case p: PredicateSubquery =>
- val tPlan = p.plan.transform {
- case lr: LogicalRelation
- if lr.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
- lr.relation.asInstanceOf[CarbonDatasourceHadoopRelation].isSubquery += true
- lr
- }
- PredicateSubquery(tPlan, p.children, p.nullAware, p.exprId)
- }
- }
- transFormedPlan
- }
-}
-
-class CarbonSqlAstBuilder(conf: SQLConf, parser: CarbonSpark2SqlParser, sparkSession: SparkSession)
- extends SparkSqlAstBuilder(conf) {
-
- val helper = new CarbonHelperSqlAstBuilder(conf, parser, sparkSession)
-
- override def visitCreateTable(ctx: CreateTableContext): LogicalPlan = {
- val fileStorage = CarbonSparkSqlParserUtil.getFileStorage(ctx.createFileFormat)
-
- if (fileStorage.equalsIgnoreCase("'carbondata'") ||
- fileStorage.equalsIgnoreCase("carbondata") ||
- fileStorage.equalsIgnoreCase("'carbonfile'") ||
- fileStorage.equalsIgnoreCase("'org.apache.carbondata.format'")) {
- val createTableTuple = (ctx.createTableHeader, ctx.skewSpec, ctx.bucketSpec,
- ctx.partitionColumns, ctx.columns, ctx.tablePropertyList, ctx.locationSpec(),
- Option(ctx.STRING()).map(string),
- ctx.AS, ctx.query, fileStorage)
- helper.createCarbonTable(createTableTuple)
- } else {
- super.visitCreateTable(ctx)
- }
- }
-
- override def visitShowTables(ctx: ShowTablesContext): LogicalPlan = {
- withOrigin(ctx) {
- if (CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.CARBON_SHOW_DATAMAPS,
- CarbonCommonConstants.CARBON_SHOW_DATAMAPS_DEFAULT).toBoolean) {
- super.visitShowTables(ctx)
- } else {
- CarbonShowTablesCommand(
- Option(ctx.db).map(_.getText),
- Option(ctx.pattern).map(string))
- }
- }
- }
-
- override def visitExplain(ctx: SqlBaseParser.ExplainContext): LogicalPlan = {
- CarbonExplainCommand(super.visitExplain(ctx))
- }
-}
diff --git a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala b/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
deleted file mode 100644
index dd690e4..0000000
--- a/integration/spark2/src/main/spark2.1/org/apache/spark/sql/hive/CreateCarbonSourceTableAsSelectCommand.scala
+++ /dev/null
@@ -1,165 +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.sql.hive
-
-import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
-import org.apache.spark.sql.catalyst.catalog.{
- CatalogRelation, CatalogTable, CatalogTableType,
- SimpleCatalogRelation
-}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.command.{
- AlterTableRecoverPartitionsCommand, DDLUtils,
- RunnableCommand
-}
-import org.apache.spark.sql.execution.datasources.{DataSource, HadoopFsRelation, LogicalRelation}
-import org.apache.spark.sql.sources.InsertableRelation
-import org.apache.spark.sql.types.StructType
-
-/**
- * Create table 'using carbondata' and insert the query result into it.
- * @param table the Catalog Table
- * @param mode SaveMode:Ignore,OverWrite,ErrorIfExists,Append
- * @param query the query whose result will be insert into the new relation
- */
-
-case class CreateCarbonSourceTableAsSelectCommand(
- table: CatalogTable,
- mode: SaveMode,
- query: LogicalPlan)
- extends RunnableCommand {
-
- override protected def innerChildren: Seq[LogicalPlan] = Seq(query)
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- assert(table.tableType != CatalogTableType.VIEW)
- assert(table.provider.isDefined)
- assert(table.schema.isEmpty)
-
- val provider = table.provider.get
- val sessionState = sparkSession.sessionState
- val db = table.identifier.database.getOrElse(sessionState.catalog.getCurrentDatabase)
- val tableIdentWithDB = table.identifier.copy(database = Some(db))
- val tableName = tableIdentWithDB.unquotedString
-
- var createMetastoreTable = false
- var existingSchema = Option.empty[StructType]
- if (sparkSession.sessionState.catalog.tableExists(tableIdentWithDB)) {
- // Check if we need to throw an exception or just return.
- mode match {
- case SaveMode.ErrorIfExists =>
- throw new AnalysisException(s"Table $tableName already exists. " +
- s"If you are using saveAsTable, you can set SaveMode to " +
- s"SaveMode.Append to " +
- s"insert data into the table or set SaveMode to SaveMode" +
- s".Overwrite to overwrite" +
- s"the existing data. " +
- s"Or, if you are using SQL CREATE TABLE, you need to drop " +
- s"$tableName first.")
- case SaveMode.Ignore =>
- // Since the table already exists and the save mode is Ignore, we will just return.
- return Seq.empty[Row]
- case SaveMode.Append =>
- // Check if the specified data source match the data source of the existing table.
- val existingProvider = DataSource.lookupDataSource(provider)
- // TODO: Check that options from the resolved relation match the relation that we are
- // inserting into (i.e. using the same compression).
-
- // Pass a table identifier with database part, so that `lookupRelation` won't get temp
- // views unexpectedly.
- EliminateSubqueryAliases(sessionState.catalog.lookupRelation(tableIdentWithDB)) match {
- case l@LogicalRelation(_: InsertableRelation | _: HadoopFsRelation, _, _) =>
- // check if the file formats match
- l.relation match {
- case r: HadoopFsRelation if r.fileFormat.getClass != existingProvider =>
- throw new AnalysisException(
- s"The file format of the existing table $tableName is " +
- s"`${ r.fileFormat.getClass.getName }`. It doesn't match the specified " +
- s"format `$provider`")
- case _ =>
- }
- if (query.schema.size != l.schema.size) {
- throw new AnalysisException(
- s"The column number of the existing schema[${ l.schema }] " +
- s"doesn't match the data schema[${ query.schema }]'s")
- }
- existingSchema = Some(l.schema)
- case s: SimpleCatalogRelation if DDLUtils.isDatasourceTable(s.metadata) =>
- existingSchema = Some(s.metadata.schema)
- case c: CatalogRelation if c.catalogTable.provider == Some(DDLUtils.HIVE_PROVIDER) =>
- throw new AnalysisException("Saving data in the Hive serde table " +
- s"${ c.catalogTable.identifier } is not supported yet. " +
- s"Please use the insertInto() API as an alternative..")
- case o =>
- throw new AnalysisException(s"Saving data in ${ o.toString } is not supported.")
- }
- case SaveMode.Overwrite =>
- sessionState.catalog.dropTable(tableIdentWithDB, ignoreIfNotExists = true, purge = false)
- // Need to create the table again.
- createMetastoreTable = true
- }
- } else {
- // The table does not exist. We need to create it in metastore.
- createMetastoreTable = true
- }
-
- val data = Dataset.ofRows(sparkSession, query)
- val df = existingSchema match {
- // If we are inserting into an existing table, just use the existing schema.
- case Some(s) => data.selectExpr(s.fieldNames: _*)
- case None => data
- }
-
- val tableLocation = if (table.tableType == CatalogTableType.MANAGED) {
- Some(sessionState.catalog.defaultTablePath(table.identifier))
- } else {
- table.storage.locationUri
- }
-
- // Create the relation based on the data of df.
- val pathOption = tableLocation.map("path" -> _)
- val dataSource = DataSource(
- sparkSession,
- className = provider,
- partitionColumns = table.partitionColumnNames,
- bucketSpec = table.bucketSpec,
- options = table.storage.properties ++ pathOption,
- catalogTable = Some(table))
-
- val result = try {
- dataSource.write(mode, df)
- } catch {
- case ex: AnalysisException =>
- logError(s"Failed to write to table $tableName in $mode mode", ex)
- throw ex
- }
- result match {
- case fs: HadoopFsRelation if table.partitionColumnNames.nonEmpty &&
- sparkSession.sqlContext.conf.manageFilesourcePartitions =>
- // Need to recover partitions into the metastore so our saved data is visible.
- sparkSession.sessionState.executePlan(
- AlterTableRecoverPartitionsCommand(table.identifier)).toRdd
- case _ =>
- }
-
- // Refresh the cache of the table in the catalog.
- sessionState.catalog.refreshTable(tableIdentWithDB)
- Seq.empty[Row]
- }
-}
diff --git a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/CarbonToSparkAdapter.scala b/integration/spark2/src/main/spark2.2/org/apache/spark/sql/CarbonToSparkAdapter.scala
deleted file mode 100644
index 446b5a5..0000000
--- a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/CarbonToSparkAdapter.scala
+++ /dev/null
@@ -1,90 +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.sql
-
-import java.net.URI
-
-import org.apache.spark.SparkContext
-import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
-import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSet, ExprId, Expression, ExpressionSet, NamedExpression}
-import org.apache.spark.sql.catalyst.optimizer.OptimizeCodegen
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.command.ExplainCommand
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{DataType, Metadata}
-
-object CarbonToSparkAdapter {
-
- def addSparkListener(sparkContext: SparkContext) = {
- sparkContext.addSparkListener(new SparkListener {
- override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
- SparkSession.setDefaultSession(null)
- SparkSession.sqlListener.set(null)
- }
- })
- }
-
- def createAttributeReference(name: String, dataType: DataType, nullable: Boolean,
- metadata: Metadata,exprId: ExprId, qualifier: Option[String],
- attrRef : NamedExpression): AttributeReference = {
- AttributeReference(
- name,
- dataType,
- nullable,
- metadata)(exprId, qualifier,attrRef.isGenerated)
- }
-
- def createAliasRef(child: Expression,
- name: String,
- exprId: ExprId = NamedExpression.newExprId,
- qualifier: Option[String] = None,
- explicitMetadata: Option[Metadata] = None,
- namedExpr: Option[NamedExpression] = None): Alias = {
- val isGenerated:Boolean = if (namedExpr.isDefined) {
- namedExpr.get.isGenerated
- } else {
- false
- }
- Alias(child, name)(exprId, qualifier, explicitMetadata,isGenerated)
- }
-
- def getExplainCommandObj() : ExplainCommand = {
- ExplainCommand(OneRowRelation)
- }
-
- def getPartitionKeyFilter(
- partitionSet: AttributeSet,
- filterPredicates: Seq[Expression]): ExpressionSet = {
- ExpressionSet(
- ExpressionSet(filterPredicates)
- .filter(_.references.subsetOf(partitionSet)))
- }
-
- def getOptimizeCodegenRule(conf :SQLConf): Seq[Rule[LogicalPlan]] = {
- Seq(OptimizeCodegen(conf))
- }
-
- def getUpdatedStorageFormat(storageFormat: CatalogStorageFormat,
- map: Map[String, String],
- tablePath: String): CatalogStorageFormat = {
- storageFormat.copy(properties = map, locationUri = Some(new URI(tablePath)))
- }
-}
diff --git a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala b/integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
deleted file mode 100644
index 7177f1a..0000000
--- a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonSqlAstBuilder.scala
+++ /dev/null
@@ -1,52 +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.sql.hive
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.parser.ParserUtils.string
-import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{AddTableColumnsContext, CreateHiveTableContext}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.SparkSqlAstBuilder
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.parser.{CarbonHelperSqlAstBuilder, CarbonSpark2SqlParser, CarbonSparkSqlParserUtil}
-
-class CarbonSqlAstBuilder(conf: SQLConf, parser: CarbonSpark2SqlParser, sparkSession: SparkSession)
- extends SparkSqlAstBuilder(conf) with SqlAstBuilderHelper {
-
- val helper = new CarbonHelperSqlAstBuilder(conf, parser, sparkSession)
-
- override def visitCreateHiveTable(ctx: CreateHiveTableContext): LogicalPlan = {
- val fileStorage = CarbonSparkSqlParserUtil.getFileStorage(ctx.createFileFormat)
-
- if (fileStorage.equalsIgnoreCase("'carbondata'") ||
- fileStorage.equalsIgnoreCase("carbondata") ||
- fileStorage.equalsIgnoreCase("'carbonfile'") ||
- fileStorage.equalsIgnoreCase("'org.apache.carbondata.format'")) {
- val createTableTuple = (ctx.createTableHeader, ctx.skewSpec,
- ctx.bucketSpec, ctx.partitionColumns, ctx.columns, ctx.tablePropertyList,ctx.locationSpec(),
- Option(ctx.STRING()).map(string), ctx.AS, ctx.query, fileStorage)
- helper.createCarbonTable(createTableTuple)
- } else {
- super.visitCreateHiveTable(ctx)
- }
- }
-
- override def visitAddTableColumns(ctx: AddTableColumnsContext): LogicalPlan = {
- visitAddTableColumns(parser,ctx)
- }
-}
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonBoundReference.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonBoundReference.scala
new file mode 100644
index 0000000..d134de1
--- /dev/null
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonBoundReference.scala
@@ -0,0 +1,28 @@
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Attribute, ExprId, LeafExpression, NamedExpression}
+import org.apache.spark.sql.types.DataType
+
+import org.apache.carbondata.core.scan.expression.ColumnExpression
+
+case class CarbonBoundReference(colExp: ColumnExpression, dataType: DataType, nullable: Boolean)
+ extends LeafExpression with NamedExpression with CodegenFallback {
+
+ type EvaluatedType = Any
+
+ override def toString: String = s"input[" + colExp.getColIndex + "]"
+
+ override def eval(input: InternalRow): Any = input.get(colExp.getColIndex, dataType)
+
+ override def name: String = colExp.getColumnName
+
+ override def toAttribute: Attribute = throw new UnsupportedOperationException
+
+ override def exprId: ExprId = throw new UnsupportedOperationException
+
+ override def qualifier: Option[String] = null
+
+ override def newInstance(): NamedExpression = throw new UnsupportedOperationException
+}
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonToSparkAdapater.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonToSparkAdapater.scala
index 9094dfe..7003c26 100644
--- a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonToSparkAdapater.scala
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/CarbonToSparkAdapater.scala
@@ -23,11 +23,11 @@
import org.apache.spark.SparkContext
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat
-import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSet, ExprId, Expression, ExpressionSet, NamedExpression, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.ExprCode
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSet, ExprId, Expression, ExpressionSet, NamedExpression, ScalaUDF, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command.ExplainCommand
-import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, Metadata}
object CarbonToSparkAdapter {
@@ -41,8 +41,8 @@
}
def createAttributeReference(name: String, dataType: DataType, nullable: Boolean,
- metadata: Metadata, exprId: ExprId, qualifier: Option[String],
- attrRef : NamedExpression): AttributeReference = {
+ metadata: Metadata, exprId: ExprId, qualifier: Option[String],
+ attrRef : NamedExpression = null): AttributeReference = {
AttributeReference(
name,
dataType,
@@ -50,14 +50,23 @@
metadata)(exprId, qualifier)
}
- def createAliasRef(child: Expression,
- name: String,
- exprId: ExprId = NamedExpression.newExprId,
- qualifier: Option[String] = None,
- explicitMetadata: Option[Metadata] = None,
- namedExpr : Option[NamedExpression] = None ) : Alias = {
+ def createScalaUDF(s: ScalaUDF, reference: AttributeReference): ScalaUDF = {
+ ScalaUDF(s.function, s.dataType, Seq(reference), s.inputTypes)
+ }
- Alias(child, name)(exprId, qualifier, explicitMetadata)
+ def createExprCode(code: String, isNull: String, value: String, dataType: DataType = null
+ ): ExprCode = {
+ ExprCode(code, isNull, value)
+ }
+
+ def createAliasRef(child: Expression,
+ name: String,
+ exprId: ExprId = NamedExpression.newExprId,
+ qualifier: Option[String] = None,
+ explicitMetadata: Option[Metadata] = None,
+ namedExpr : Option[NamedExpression] = None ) : Alias = {
+
+ Alias(child, name)(exprId, qualifier, explicitMetadata)
}
def getExplainCommandObj() : ExplainCommand = {
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/MixedFormatHandlerUtil.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/MixedFormatHandlerUtil.scala
new file mode 100644
index 0000000..bffb900
--- /dev/null
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/MixedFormatHandlerUtil.scala
@@ -0,0 +1,27 @@
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.execution.FileSourceScanExec
+import org.apache.spark.sql.execution.datasources.HadoopFsRelation
+import org.apache.spark.sql.types.StructType
+
+object MixedFormatHandlerUtil {
+
+ def getScanForSegments(
+ @transient relation: HadoopFsRelation,
+ output: Seq[Attribute],
+ outputSchema: StructType,
+ partitionFilters: Seq[Expression],
+ dataFilters: Seq[Expression],
+ tableIdentifier: Option[TableIdentifier]
+ ): FileSourceScanExec = {
+ FileSourceScanExec(
+ relation,
+ output,
+ outputSchema,
+ partitionFilters,
+ dataFilters,
+ tableIdentifier)
+ }
+}
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
index 5435f04..25d5543 100644
--- a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
@@ -17,9 +17,9 @@
package org.apache.spark.sql.execution.strategy
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala
similarity index 97%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala
rename to integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala
index 36f166d..4ad3b11 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/CarbonInMemorySessionState.scala
@@ -21,21 +21,20 @@
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil
-import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.{Analyzer, FunctionRegistry}
import org.apache.spark.sql.catalyst.catalog._
import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.optimizer.Optimizer
import org.apache.spark.sql.catalyst.parser.ParserInterface
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.execution.strategy.{CarbonLateDecodeStrategy, DDLStrategy, StreamingTableStrategy}
import org.apache.spark.sql.internal.{SQLConf, SessionResourceLoader, SessionState, SessionStateBuilder}
import org.apache.spark.sql.optimizer.{CarbonIUDRule, CarbonLateDecodeRule, CarbonUDFTransformRule}
import org.apache.spark.sql.parser.CarbonSparkSqlParser
import org.apache.spark.sql.types.{StructField, StructType}
-import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
+import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.util.CarbonUtil
diff --git a/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/SparkAdapter.scala b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/SparkAdapter.scala
new file mode 100644
index 0000000..61959c2
--- /dev/null
+++ b/integration/spark2/src/main/spark2.3/org/apache/spark/sql/hive/SparkAdapter.scala
@@ -0,0 +1,14 @@
+package org.apache.spark.sql.hive
+
+import org.apache.spark.sql.catalyst.catalog.{ExternalCatalog, GlobalTempViewManager}
+import org.apache.spark.sql.hive.client.HiveClient
+
+object SparkAdapter {
+ def getExternalCatalogCatalog(catalog: HiveExternalCatalog) = catalog
+
+ def getGlobalTempViewManager(manager: GlobalTempViewManager) = manager
+
+ def getHiveClient(client: HiveClient) = client
+
+ def getHiveExternalCatalog(catalog: ExternalCatalog) = catalog.asInstanceOf[HiveExternalCatalog]
+}
diff --git a/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonBoundReference.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonBoundReference.scala
new file mode 100644
index 0000000..ee4c9ce
--- /dev/null
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonBoundReference.scala
@@ -0,0 +1,44 @@
+package org.apache.spark.sql
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, ExprId, LeafExpression, NamedExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.types.DataType
+
+import org.apache.carbondata.core.scan.expression.ColumnExpression
+/*
+ * 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.
+ */
+
+case class CarbonBoundReference(colExp: ColumnExpression, dataType: DataType, nullable: Boolean)
+ extends LeafExpression with NamedExpression with CodegenFallback {
+
+ type EvaluatedType = Any
+
+ override def toString: String = s"input[" + colExp.getColIndex + "]"
+
+ override def eval(input: InternalRow): Any = input.get(colExp.getColIndex, dataType)
+
+ override def name: String = colExp.getColumnName
+
+ override def toAttribute: Attribute = throw new UnsupportedOperationException
+
+ override def exprId: ExprId = throw new UnsupportedOperationException
+
+ override def qualifier: Seq[String] = null
+
+ override def newInstance(): NamedExpression = throw new UnsupportedOperationException
+}
\ No newline at end of file
diff --git a/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonToSparkAdapter.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonToSparkAdapter.scala
new file mode 100644
index 0000000..082d1ec
--- /dev/null
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/CarbonToSparkAdapter.scala
@@ -0,0 +1,134 @@
+/*
+ * 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.sql
+
+import java.net.URI
+
+import org.apache.spark.SparkContext
+import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
+import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, AttributeSet, ExprId, Expression, ExpressionSet, NamedExpression, ScalaUDF, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.command.ExplainCommand
+import org.apache.spark.sql.types.{DataType, Metadata}
+
+object CarbonToSparkAdapter {
+
+ def addSparkListener(sparkContext: SparkContext) = {
+ sparkContext.addSparkListener(new SparkListener {
+ override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = {
+ SparkSession.setDefaultSession(null)
+ }
+ })
+ }
+
+ def createAttributeReference(
+ name: String,
+ dataType: DataType,
+ nullable: Boolean,
+ metadata: Metadata,
+ exprId: ExprId,
+ qualifier: Option[String],
+ attrRef : NamedExpression = null): AttributeReference = {
+ val qf = if (qualifier.nonEmpty) Seq(qualifier.get) else Seq.empty
+ AttributeReference(
+ name,
+ dataType,
+ nullable,
+ metadata)(exprId, qf)
+ }
+
+ def createAttributeReference(
+ name: String,
+ dataType: DataType,
+ nullable: Boolean,
+ metadata: Metadata,
+ exprId: ExprId,
+ qualifier: Seq[String]): AttributeReference = {
+ AttributeReference(
+ name,
+ dataType,
+ nullable,
+ metadata)(exprId, qualifier)
+ }
+
+ def createScalaUDF(s: ScalaUDF, reference: AttributeReference) = {
+ ScalaUDF(s.function, s.dataType, Seq(reference), s.inputsNullSafe, s.inputTypes)
+ }
+
+ def createExprCode(code: String, isNull: String, value: String, dataType: DataType) = {
+ ExprCode(
+ code"$code",
+ JavaCode.isNullVariable(isNull),
+ JavaCode.variable(value, dataType))
+ }
+
+ def createAliasRef(
+ child: Expression,
+ name: String,
+ exprId: ExprId = NamedExpression.newExprId,
+ qualifier: Seq[String] = Seq.empty,
+ explicitMetadata: Option[Metadata] = None,
+ namedExpr: Option[NamedExpression] = None) : Alias = {
+ Alias(child, name)(exprId, qualifier, explicitMetadata)
+ }
+
+ def createAliasRef(
+ child: Expression,
+ name: String,
+ exprId: ExprId,
+ qualifier: Option[String]) : Alias = {
+ Alias(child, name)(exprId,
+ if (qualifier.isEmpty) Seq.empty else Seq(qualifier.get),
+ None)
+ }
+
+ def getExplainCommandObj() : ExplainCommand = {
+ ExplainCommand(OneRowRelation())
+ }
+
+ /**
+ * As a part of SPARK-24085 Hive tables supports scala subquery for
+ * parition tables, so Carbon also needs to supports
+ * @param partitionSet
+ * @param filterPredicates
+ * @return
+ */
+ def getPartitionKeyFilter(
+ partitionSet: AttributeSet,
+ filterPredicates: Seq[Expression]): ExpressionSet = {
+ ExpressionSet(
+ ExpressionSet(filterPredicates)
+ .filterNot(SubqueryExpression.hasSubquery)
+ .filter(_.references.subsetOf(partitionSet)))
+ }
+
+ // As per SPARK-22520 OptimizeCodegen is removed in 2.3.1
+ def getOptimizeCodegenRule(conf :SQLConf): Seq[Rule[LogicalPlan]] = {
+ Seq.empty
+ }
+
+ def getUpdatedStorageFormat(storageFormat: CatalogStorageFormat,
+ map: Map[String, String],
+ tablePath: String): CatalogStorageFormat = {
+ storageFormat.copy(properties = map, locationUri = Some(new URI(tablePath)))
+ }
+}
diff --git a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/MixedFomatHandlerUtil.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/MixedFormatHandlerUtil.scala
similarity index 98%
rename from integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/MixedFomatHandlerUtil.scala
rename to integration/spark2/src/main/spark2.4/org/apache/spark/sql/MixedFormatHandlerUtil.scala
index d180cd3..7b20c06 100644
--- a/integration/spark2/src/main/commonTo2.2And2.3/org/apache/spark/sql/MixedFomatHandlerUtil.scala
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/MixedFormatHandlerUtil.scala
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.spark.sql
import org.apache.spark.sql.catalyst.TableIdentifier
@@ -37,6 +38,7 @@
output,
outputSchema,
partitionFilters,
+ None,
dataFilters,
tableIdentifier)
}
diff --git a/integration/spark2/src/main/commonTo2.1And2.2/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
similarity index 81%
rename from integration/spark2/src/main/commonTo2.1And2.2/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
rename to integration/spark2/src/main/spark2.4/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
index 7605574..60ee7ea 100644
--- a/integration/spark2/src/main/commonTo2.1And2.2/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/execution/strategy/CarbonDataSourceScan.scala
@@ -17,9 +17,9 @@
package org.apache.spark.sql.execution.strategy
import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
+import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
@@ -32,7 +32,7 @@
val rdd: RDD[InternalRow],
@transient override val relation: HadoopFsRelation,
val partitioning: Partitioning,
- override val metadata: Map[String, String],
+ val md: Map[String, String],
identifier: Option[TableIdentifier],
@transient private val logicalRelation: LogicalRelation)
extends FileSourceScanExec(
@@ -40,14 +40,20 @@
output,
relation.dataSchema,
Seq.empty,
+ None,
Seq.empty,
identifier) {
- override val supportsBatch: Boolean = true
+ // added lazy since spark 2.3.2 version (SPARK-PR#21815)
+ override lazy val supportsBatch: Boolean = true
- override val (outputPartitioning, outputOrdering): (Partitioning, Seq[SortOrder]) =
+ // added lazy since spark 2.3.2 version (SPARK-PR#21815)
+ override lazy val (outputPartitioning, outputOrdering): (Partitioning, Seq[SortOrder]) =
(partitioning, Nil)
+ // added lazy since spark 2.3.2 version (SPARK-PR#21815)
+ override lazy val metadata: Map[String, String] = md
+
override def inputRDDs(): Seq[RDD[InternalRow]] = rdd :: Nil
}
diff --git a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonOptimizer.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/CarbonOptimizer.scala
similarity index 95%
rename from integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonOptimizer.scala
rename to integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/CarbonOptimizer.scala
index 2f9ad3e..808099d 100644
--- a/integration/spark2/src/main/spark2.2/org/apache/spark/sql/hive/CarbonOptimizer.scala
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/CarbonOptimizer.scala
@@ -23,15 +23,14 @@
import org.apache.spark.sql.execution.SparkOptimizer
import org.apache.spark.sql.internal.SQLConf
-
class CarbonOptimizer(
catalog: SessionCatalog,
conf: SQLConf,
experimentalMethods: ExperimentalMethods)
- extends SparkOptimizer(catalog, conf, experimentalMethods) {
+ extends SparkOptimizer(catalog, experimentalMethods) {
override def execute(plan: LogicalPlan): LogicalPlan = {
val transFormedPlan: LogicalPlan = CarbonOptimizerUtil.transformForScalarSubQuery(plan)
super.execute(transFormedPlan)
}
-}
+}
\ No newline at end of file
diff --git a/integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/SparkAdapter.scala b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/SparkAdapter.scala
new file mode 100644
index 0000000..8132188
--- /dev/null
+++ b/integration/spark2/src/main/spark2.4/org/apache/spark/sql/hive/SparkAdapter.scala
@@ -0,0 +1,35 @@
+/*
+ * 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.sql.hive
+
+import org.apache.spark.sql.catalyst.catalog.{ExternalCatalog, ExternalCatalogWithListener, GlobalTempViewManager}
+import org.apache.spark.sql.hive.client.HiveClient
+
+object SparkAdapter {
+ def getExternalCatalogCatalog(catalog: HiveExternalCatalog) =
+ () => catalog
+
+ def getGlobalTempViewManager(manager: GlobalTempViewManager) =
+ () => manager
+
+ def getHiveClient(client: HiveClient) =
+ () => client
+
+ def getHiveExternalCatalog(catalog: ExternalCatalogWithListener) =
+ catalog.unwrapped.asInstanceOf[HiveExternalCatalog]
+}
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFunctionSuite.scala b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFunctionSuite.scala
index c1a5862..e7ca31b 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFunctionSuite.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFunctionSuite.scala
@@ -24,6 +24,7 @@
val dataMapName = "bloom_dm"
override protected def beforeAll(): Unit = {
+ sqlContext.sparkContext.setLogLevel("info")
deleteFile(bigFile)
new File(CarbonProperties.getInstance().getSystemFolderLocation).delete()
createFile(bigFile, line = 2000)
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
deleted file mode 100644
index e69de29..0000000
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
+++ /dev/null
diff --git a/pom.xml b/pom.xml
index 66214df..f74a11b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -487,7 +487,7 @@
<profiles>
<profile>
- <!--This profile does not build spark module, so user should explicitly give spark profile also like spark-2.1 -->
+ <!--This profile does not build spark module, so user should explicitly give spark profile -->
<id>build-with-format</id>
<modules>
<module>format</module>
@@ -501,103 +501,6 @@
</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.eluder.coveralls</groupId>
- <artifactId>coveralls-maven-plugin</artifactId>
- <version>4.3.0</version>
- <configuration>
- <repoToken>opPwqWW41vYppv6KISea3u1TJvE1ugJ5Y</repoToken>
- <sourceEncoding>UTF-8</sourceEncoding>
- <jacocoReports>
- <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml
- </jacocoReport>
- </jacocoReports>
- <sourceDirectories>
- <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/spark2.1</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.1And2.2</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/streaming/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/streaming/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/store/sdk/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/datamap/bloom/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/datamap/lucene/src/main/java</sourceDirectory>
- </sourceDirectories>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
- <id>spark-2.2</id>
- <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.eluder.coveralls</groupId>
- <artifactId>coveralls-maven-plugin</artifactId>
- <version>4.3.0</version>
- <configuration>
- <repoToken>opPwqWW41vYppv6KISea3u1TJvE1ugJ5Y</repoToken>
- <sourceEncoding>UTF-8</sourceEncoding>
- <jacocoReports>
- <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml
- </jacocoReport>
- </jacocoReports>
- <sourceDirectories>
- <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/spark2.2</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.1And2.2</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2And2.3</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/streaming/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/streaming/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/store/sdk/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/datamap/bloom/src/main/java</sourceDirectory>
- <sourceDirectory>${basedir}/datamap/lucene/src/main/java</sourceDirectory>
- </sourceDirectories>
- </configuration>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
<id>spark-2.3</id>
<activation>
<activeByDefault>true</activeByDefault>
@@ -626,11 +529,64 @@
<sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory>
<sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory>
- <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2And2.3</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2AndAbove</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.3And2.4</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/below2.4</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark2/src/main/spark2.3</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common/src/main/below2.4</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/hive/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/presto/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/presto/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/streaming/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/streaming/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/store/sdk/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/datamap/bloom/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/datamap/lucene/src/main/java</sourceDirectory>
+ </sourceDirectories>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>spark-2.4</id>
+ <properties>
+ <spark.version>2.4.4</spark.version>
+ <scala.binary.version>2.11</scala.binary.version>
+ <scala.version>2.11.8</scala.version>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.eluder.coveralls</groupId>
+ <artifactId>coveralls-maven-plugin</artifactId>
+ <version>4.3.0</version>
+ <configuration>
+ <repoToken>opPwqWW41vYppv6KISea3u1TJvE1ugJ5Y</repoToken>
+ <sourceEncoding>UTF-8</sourceEncoding>
+ <jacocoReports>
+ <jacocoReport>${basedir}/target/carbondata-coverage-report/carbondata-coverage-report.xml
+ </jacocoReport>
+ </jacocoReports>
+ <sourceDirectories>
+ <sourceDirectory>${basedir}/common/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/core/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/processing/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/hadoop/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.2AndAbove</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/commonTo2.3And2.4</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/spark2.4</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark2/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common/src/main/scala</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common/src/main/java</sourceDirectory>
+ <sourceDirectory>${basedir}/integration/spark-common/src/main/spark2.4</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark-common-test/src/main/scala</sourceDirectory>
<sourceDirectory>${basedir}/integration/spark-common-test/src/main/java</sourceDirectory>
<sourceDirectory>${basedir}/integration/hive/src/main/scala</sourceDirectory>
