You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by jb...@apache.org on 2016/06/23 14:16:03 UTC
[15/56] [abbrv] incubator-carbondata git commit: Refactored core
package and fixed all testcases (#684)
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index 600519f..4c7c460 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.errors.attachTree
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.execution.{SparkPlan, UnaryNode}
-import org.apache.spark.sql.hive.CarbonMetastoreCatalog
+import org.apache.spark.sql.hive.{CarbonMetastoreCatalog, CarbonMetastoreTypes}
import org.apache.spark.sql.optimizer.{CarbonAliasDecoderRelation, CarbonDecoderRelation}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String
@@ -54,20 +54,19 @@ case class CarbonDictionaryDecoder(
child.output.map { a =>
val attr = aliasMap.getOrElse(a, a)
val relation = relations.find(p => p.contains(attr))
- if(relation.isDefined) {
+ if(relation.isDefined && canBeDecoded(attr)) {
val carbonTable = relation.get.carbonRelation.carbonRelation.metaData.carbonTable
val carbonDimension = carbonTable
.getDimensionByName(carbonTable.getFactTableName, attr.name)
if (carbonDimension != null &&
carbonDimension.hasEncoding(Encoding.DICTIONARY) &&
- !carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY) &&
- canBeDecoded(attr)) {
+ !carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
val newAttr = AttributeReference(a.name,
- convertCarbonToSparkDataType(carbonDimension),
+ convertCarbonToSparkDataType(carbonDimension,
+ relation.get.carbonRelation.carbonRelation),
a.nullable,
a.metadata)(a.exprId,
a.qualifiers).asInstanceOf[Attribute]
- newAttr.resolved
newAttr
} else {
a
@@ -89,15 +88,29 @@ case class CarbonDictionaryDecoder(
}
}
- def convertCarbonToSparkDataType(carbonDimension: CarbonDimension): types.DataType = {
+ def convertCarbonToSparkDataType(carbonDimension: CarbonDimension,
+ relation: CarbonRelation): types.DataType = {
carbonDimension.getDataType match {
case DataType.STRING => StringType
case DataType.INT => IntegerType
case DataType.LONG => LongType
case DataType.DOUBLE => DoubleType
case DataType.BOOLEAN => BooleanType
- case DataType.DECIMAL => DecimalType.DoubleDecimal
+ case DataType.DECIMAL =>
+ val scale: Int = carbonDimension.getColumnSchema.getScale
+ val precision: Int = carbonDimension.getColumnSchema.getPrecision
+ if (scale > 0 && precision > 0) {
+ DecimalType(scale, precision)
+ } else {
+ DecimalType(18, 2)
+ }
case DataType.TIMESTAMP => TimestampType
+ case DataType.STRUCT =>
+ CarbonMetastoreTypes
+ .toDataType(s"struct<${ relation.getStructChildren(carbonDimension.getColName) }>")
+ case DataType.ARRAY =>
+ CarbonMetastoreTypes
+ .toDataType(s"array<${ relation.getArrayChildren(carbonDimension.getColName) }>")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
index cb20246..ba4c37e 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonOperators.scala
@@ -26,24 +26,20 @@ import org.apache.hadoop.conf.Configuration
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.execution.LeafNode
import org.apache.spark.sql.hive.CarbonMetastoreCatalog
-import org.apache.spark.sql.types.{DataType, Decimal}
import org.apache.spark.unsafe.types.UTF8String
import org.carbondata.core.constants.CarbonCommonConstants
import org.carbondata.core.util.CarbonProperties
import org.carbondata.query.carbon.model._
-import org.carbondata.spark.{CarbonFilters, RawKey, RawKeyImpl}
+import org.carbondata.spark.{CarbonFilters, RawValue, RawValueImpl}
import org.carbondata.spark.rdd.CarbonScanRDD
case class CarbonScan(
var attributesRaw: Seq[Attribute],
relationRaw: CarbonRelation,
- dimensionPredicatesRaw: Seq[Expression],
- aggExprsRaw: Option[Seq[Expression]],
- useBinaryAggregator: Boolean)(@transient val ocRaw: SQLContext) extends LeafNode {
+ dimensionPredicatesRaw: Seq[Expression])(@transient val ocRaw: SQLContext) extends LeafNode {
val carbonTable = relationRaw.metaData.carbonTable
val selectedDims = scala.collection.mutable.MutableList[QueryDimension]()
val selectedMsrs = scala.collection.mutable.MutableList[QueryMeasure]()
@@ -95,24 +91,6 @@ case class CarbonScan(
}
}
}
- // Just find out that any aggregation functions are present on dimensions.
- aggExprsRaw match {
- case Some(aggExprs) =>
- aggExprs.foreach {
- case Alias(agg: AggregateExpression, name) =>
- agg.collect {
- case attr: AttributeReference =>
- val dims = selectedDims.filter(m => m.getColumnName.equalsIgnoreCase(attr.name))
- if(dims.nonEmpty) {
- plan.addAggDimAggInfo(dims.head.getColumnName,
- dims.head.getAggregateFunction,
- dims.head.getQueryOrder)
- }
- }
- case _ =>
- }
- case _ =>
- }
// Fill the selected dimensions & measures obtained from
// attributes to query plan for detailed query
@@ -179,14 +157,13 @@ case class CarbonScan(
}
- def inputRdd: CarbonScanRDD[Array[Any], Any] = {
+ def inputRdd: CarbonScanRDD[Array[Any]] = {
val conf = new Configuration()
val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
- buildCarbonPlan.getDimAggregatorInfos.clear()
val model = QueryModel.createModel(
absoluteTableIdentifier, buildCarbonPlan, carbonTable)
- val kv: RawKey[Array[Any], Any] = new RawKeyImpl()
+ val kv: RawValue[Array[Any]] = new RawValueImpl
// setting queryid
buildCarbonPlan.setQueryId(ocRaw.getConf("queryId", System.nanoTime() + ""))
@@ -224,9 +201,9 @@ case class CarbonScan(
override def next(): InternalRow =
if (outUnsafeRows) {
- unsafeProjection(new GenericMutableRow(iter.next()._1.map(toType)))
+ unsafeProjection(new GenericMutableRow(iter.next().map(toType)))
} else {
- new GenericMutableRow(iter.next()._1.map(toType))
+ new GenericMutableRow(iter.next().map(toType))
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSQLConf.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSQLConf.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSQLConf.scala
index 96a86a7..6ed8c0d 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSQLConf.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSQLConf.scala
@@ -20,13 +20,6 @@ package org.apache.spark.sql
import org.apache.spark.sql.SQLConf.SQLConfEntry
import org.apache.spark.sql.hive.CarbonSQLDialect
-object CarbonSQLConf {
-
- val PUSH_COMPUTATION = SQLConfEntry.booleanConf("spark.sql.carbon.push.computation",
- defaultValue = Some(true))
-
-}
-
/**
* A trait that enables the setting and getting of mutable config parameters/hints.
*
@@ -40,8 +33,4 @@ class CarbonSQLConf extends SQLConf {
override def caseSensitiveAnalysis: Boolean = getConf(SQLConf.CASE_SENSITIVE, false)
- import CarbonSQLConf._
-
- private[sql] def pushComputation: Boolean = getConf(PUSH_COMPUTATION)
-
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
index a72aaa1..9239cec 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
@@ -170,7 +170,7 @@ class CarbonSqlParser()
initLexical
phrase(start)(new lexical.Scanner(input)) match {
case Success(plan, _) => plan match {
- case x: LoadCube =>
+ case x: LoadTable =>
x.inputSqlString = input
x
case logicalPlan => logicalPlan
@@ -969,7 +969,7 @@ class CarbonSqlParser()
}
val patitionOptionsMap = partionDataOptions.toMap
- LoadCube(schema, cubename, filePath, dimFolderPath.getOrElse(Seq()),
+ LoadTable(schema, cubename, filePath, dimFolderPath.getOrElse(Seq()),
patitionOptionsMap, false)
}
@@ -985,7 +985,7 @@ class CarbonSqlParser()
validateOptions(partionDataOptions)
}
val patitionOptionsMap = partionDataOptions.getOrElse(List.empty[(String, String)]).toMap
- LoadCube(schema, cubename, filePath, Seq(), patitionOptionsMap, isOverwrite.isDefined)
+ LoadTable(schema, cubename, filePath, Seq(), patitionOptionsMap, isOverwrite.isDefined)
}
private def validateOptions(partionDataOptions: Option[List[(String, String)]]): Unit = {
@@ -1079,9 +1079,7 @@ class CarbonSqlParser()
opt(";") ^^ {
case tabletype ~ exists ~ schemaName ~ resourceName =>
tabletype match {
- case agg ~ table =>
- DropAggregateTableCommand(exists.isDefined, schemaName, resourceName.toLowerCase())
- case _ => DropCubeCommand(exists.isDefined, schemaName, resourceName.toLowerCase())
+ case _ => DropTableCommand(exists.isDefined, schemaName, resourceName.toLowerCase())
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index fcc2e8a..e36d148 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -1381,7 +1381,7 @@ private[sql] case class DeleteLoadsByLoadDate(
}
-private[sql] case class LoadCube(
+private[sql] case class LoadTable(
schemaNameOp: Option[String],
tableName: String,
factPathFromUser: String,
@@ -1693,7 +1693,7 @@ private[sql] case class MergeTable(dbName: String, cubeName: String, tableName:
}
}
-private[sql] case class DropCubeCommand(ifExistsSet: Boolean, schemaNameOp: Option[String],
+private[sql] case class DropTableCommand(ifExistsSet: Boolean, schemaNameOp: Option[String],
tableName: String)
extends RunnableCommand {
@@ -1755,7 +1755,7 @@ private[sql] case class DropCubeCommand(ifExistsSet: Boolean, schemaNameOp: Opti
Some(relation.cubeMeta.carbonTableIdentifier.getDatabaseName))
)(sqlContext)
CarbonDataRDDFactory
- .dropCube(sqlContext.sparkContext, dbName, tableName,
+ .dropTable(sqlContext.sparkContext, dbName, tableName,
relation.cubeMeta.partitioner)
QueryPartitionHelper.getInstance().removePartition(dbName, tableName)
@@ -1789,33 +1789,6 @@ private[sql] case class DropCubeCommand(ifExistsSet: Boolean, schemaNameOp: Opti
}
}
-private[sql] case class DropAggregateTableCommand(ifExistsSet: Boolean,
- schemaNameOp: Option[String],
- tableName: String) extends RunnableCommand {
-
- def run(sqlContext: SQLContext): Seq[Row] = {
- val dbName = getDB.getDatabaseName(schemaNameOp, sqlContext)
- val identifier = TableIdentifier(tableName, Option(dbName))
- val relation = CarbonEnv.getInstance(sqlContext).carbonCatalog
- .lookupRelation1(identifier)(sqlContext).asInstanceOf[CarbonRelation]
-
- if (relation == null) {
- if (!ifExistsSet) {
- sys.error(s"Aggregate Table $dbName.$tableName does not exist")
- }
- }
- else {
- CarbonDataRDDFactory.dropAggregateTable(
- sqlContext.sparkContext,
- dbName,
- tableName,
- relation.cubeMeta.partitioner)
- }
-
- Seq.empty
- }
-}
-
private[sql] case class ShowLoads(
schemaNameOp: Option[String],
tableName: String,
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
index 829c487..76edc11 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategies.scala
@@ -33,7 +33,6 @@ import org.apache.spark.sql.hive.execution.{DescribeHiveTableCommand, DropTable,
import org.apache.spark.sql.optimizer.{CarbonAliasDecoderRelation, CarbonDecoderRelation}
import org.carbondata.common.logging.LogServiceFactory
-import org.carbondata.core.carbon.metadata.schema.table.CarbonTable
import org.carbondata.spark.exception.MalformedCarbonCommandException
@@ -64,10 +63,7 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
carbonRawScan(projectList,
predicates,
carbonRelation,
- l,
- None,
- detailQuery = true,
- useBinaryAggregation = false)(sqlContext)._1 :: Nil
+ l)(sqlContext) :: Nil
}
case CarbonDictionaryCatalystDecoder(relations, profile, aliasMap, _, child) =>
CarbonDictionaryDecoder(relations,
@@ -85,10 +81,7 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
private def carbonRawScan(projectList: Seq[NamedExpression],
predicates: Seq[Expression],
relation: CarbonDatasourceRelation,
- logicalRelation: LogicalRelation,
- groupExprs: Option[Seq[Expression]],
- detailQuery: Boolean,
- useBinaryAggregation: Boolean)(sc: SQLContext): (SparkPlan, Boolean) = {
+ logicalRelation: LogicalRelation)(sc: SQLContext): SparkPlan = {
val tableName: String =
relation.carbonRelation.metaData.carbonTable.getFactTableName.toLowerCase
@@ -97,49 +90,32 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
val projectSet = AttributeSet(projectList.flatMap(_.references))
val scan = CarbonScan(projectSet.toSeq,
relation.carbonRelation,
- predicates,
- groupExprs,
- useBinaryAggregation)(sqlContext)
- val dimAggrsPresence: Boolean = scan.buildCarbonPlan.getDimAggregatorInfos.size() > 0
+ predicates)(sqlContext)
projectList.map {
case attr: AttributeReference =>
case Alias(attr: AttributeReference, _) =>
case others =>
- others.references
- .map(f => scan.attributesNeedToDecode.add(f.asInstanceOf[AttributeReference]))
- }
- if (!detailQuery) {
- if (scan.attributesNeedToDecode.size > 0) {
- val decoder = getCarbonDecoder(logicalRelation,
- sc,
- tableName,
- scan.attributesNeedToDecode.asScala.toSeq,
- scan)
- if (scan.unprocessedExprs.nonEmpty) {
- val filterCondToAdd = scan.unprocessedExprs.reduceLeftOption(expressions.And)
- (Project(projectList, filterCondToAdd.map(Filter(_, decoder)).getOrElse(decoder)), true)
- } else {
- (Project(projectList, decoder), true)
+ others.references.map{f =>
+ val dictionary = relation.carbonRelation.metaData.dictionaryMap.get(f.name)
+ if (dictionary.isDefined && dictionary.get) {
+ scan.attributesNeedToDecode.add(f.asInstanceOf[AttributeReference])
+ }
}
+ }
+ if (scan.attributesNeedToDecode.size() > 0) {
+ val decoder = getCarbonDecoder(logicalRelation,
+ sc,
+ tableName,
+ scan.attributesNeedToDecode.asScala.toSeq,
+ scan)
+ if (scan.unprocessedExprs.nonEmpty) {
+ val filterCondToAdd = scan.unprocessedExprs.reduceLeftOption(expressions.And)
+ Project(projectList, filterCondToAdd.map(Filter(_, decoder)).getOrElse(decoder))
} else {
- (scan, dimAggrsPresence)
+ Project(projectList, decoder)
}
} else {
- if (scan.attributesNeedToDecode.size() > 0) {
- val decoder = getCarbonDecoder(logicalRelation,
- sc,
- tableName,
- scan.attributesNeedToDecode.asScala.toSeq,
- scan)
- if (scan.unprocessedExprs.nonEmpty) {
- val filterCondToAdd = scan.unprocessedExprs.reduceLeftOption(expressions.And)
- (Project(projectList, filterCondToAdd.map(Filter(_, decoder)).getOrElse(decoder)), true)
- } else {
- (Project(projectList, decoder), true)
- }
- } else {
- (Project(projectList, scan), dimAggrsPresence)
- }
+ Project(projectList, scan)
}
}
@@ -158,9 +134,7 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
val projectExprsNeedToDecode = new java.util.HashSet[Attribute]()
val scan = CarbonScan(projectList.map(_.toAttribute),
relation.carbonRelation,
- predicates,
- None,
- useBinaryAggregator = false)(sqlContext)
+ predicates)(sqlContext)
projectExprsNeedToDecode.addAll(scan.attributesNeedToDecode)
if (projectExprsNeedToDecode.size() > 0) {
val decoder = getCarbonDecoder(logicalRelation,
@@ -206,18 +180,6 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
case _ => false
}
}
-
- private def isGroupByPresentOnMeasures(groupingExpressions: Seq[Expression],
- carbonTable: CarbonTable): Boolean = {
- groupingExpressions.map { g =>
- g.collect {
- case attr: AttributeReference
- if carbonTable.getMeasureByName(carbonTable.getFactTableName, attr.name) != null =>
- return true
- }
- }
- false
- }
}
object DDLStrategies extends Strategy {
@@ -233,17 +195,17 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
case DropTable(tableName, ifNotExists)
if CarbonEnv.getInstance(sqlContext).carbonCatalog
.tableExists(TableIdentifier(tableName.toLowerCase))(sqlContext) =>
- ExecutedCommand(DropCubeCommand(ifNotExists, None, tableName.toLowerCase)) :: Nil
+ ExecutedCommand(DropTableCommand(ifNotExists, None, tableName.toLowerCase)) :: Nil
case ShowAggregateTablesCommand(schemaName) =>
ExecutedCommand(ShowAggregateTables(schemaName, plan.output)) :: Nil
case ShowLoadsCommand(schemaName, cube, limit) =>
ExecutedCommand(ShowLoads(schemaName, cube, limit, plan.output)) :: Nil
- case LoadCube(schemaNameOp, cubeName, factPathFromUser, dimFilesPath,
+ case LoadTable(schemaNameOp, cubeName, factPathFromUser, dimFilesPath,
partionValues, isOverwriteExist, inputSqlString) =>
val isCarbonTable = CarbonEnv.getInstance(sqlContext).carbonCatalog
.tableExists(TableIdentifier(cubeName, schemaNameOp))(sqlContext)
if (isCarbonTable || partionValues.nonEmpty) {
- ExecutedCommand(LoadCube(schemaNameOp, cubeName, factPathFromUser,
+ ExecutedCommand(LoadTable(schemaNameOp, cubeName, factPathFromUser,
dimFilesPath, partionValues, isOverwriteExist, inputSqlString)) :: Nil
} else {
ExecutedCommand(HiveNativeCommand(inputSqlString)) :: Nil
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
index 03bb23e..0f583d0 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonDecoderOptimizerHelper.scala
@@ -55,17 +55,13 @@ class CarbonDecoderProcessor {
case cd: CarbonDictionaryTempDecoder =>
nodeList.add(Node(cd))
process(cd.child, nodeList)
- case j: Join =>
+ case j: BinaryNode =>
val leftList = new util.ArrayList[AbstractNode]
val rightList = new util.ArrayList[AbstractNode]
nodeList.add(JoinNode(leftList, rightList))
process(j.left, leftList)
process(j.right, rightList)
- case p: Project => process(p.child, nodeList)
- case f: Filter => process(f.child, nodeList)
- case s: Sort => process(s.child, nodeList)
- case a: Aggregate => process(a.child, nodeList)
- case l: Limit => process(l.child, nodeList)
+ case e: UnaryNode => process(e.child, nodeList)
case _ =>
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
index d80c065..73cb3d5 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/optimizer/CarbonOptimizer.scala
@@ -44,13 +44,9 @@ class CarbonOptimizer(optimizer: Optimizer, conf: CatalystConf)
override def execute(plan: LogicalPlan): LogicalPlan = {
val executedPlan: LogicalPlan = optimizer.execute(plan)
- if (!conf.asInstanceOf[CarbonSQLConf].pushComputation) {
- val relations = collectCarbonRelation(plan)
- if (relations.nonEmpty) {
- new ResolveCarbonFunctions(relations)(executedPlan)
- } else {
- executedPlan
- }
+ val relations = collectCarbonRelation(plan)
+ if (relations.nonEmpty) {
+ new ResolveCarbonFunctions(relations)(executedPlan)
} else {
executedPlan
}
@@ -111,19 +107,28 @@ class CarbonOptimizer(optimizer: Optimizer, conf: CatalystConf)
case agg: Aggregate if !agg.child.isInstanceOf[CarbonDictionaryTempDecoder] =>
val attrsOndimAggs = new util.HashSet[Attribute]
- agg.aggregateExpressions.map { aggExp =>
- aggExp.transform {
- case aggExp: AggregateExpression =>
- collectDimensionAggregates(aggExp, attrsOndimAggs, aliasMap)
- aggExp
- case a@Alias(attr: Attribute, name) =>
- aliasMap.put(a.toAttribute, attr)
- a
- }
+ agg.aggregateExpressions.map {
+ case attr: AttributeReference =>
+ case a@Alias(attr: AttributeReference, name) => aliasMap.put(a.toAttribute, attr)
+ case aggExp: AggregateExpression =>
+ aggExp.transform {
+ case aggExp: AggregateExpression =>
+ collectDimensionAggregates(aggExp, attrsOndimAggs, aliasMap)
+ aggExp
+ case a@Alias(attr: Attribute, name) =>
+ aliasMap.put(a.toAttribute, attr)
+ a
+ }
+ case others =>
+ others.collect {
+ case attr: AttributeReference
+ if isDictionaryEncoded(attr, relations, aliasMap) =>
+ attrsOndimAggs.add(aliasMap.getOrElse(attr, attr))
+ }
}
var child = agg.child
// Incase if the child also aggregate then push down decoder to child
- if (attrsOndimAggs.size() > 0 && !(child.equals(agg))) {
+ if (attrsOndimAggs.size() > 0 && !child.equals(agg)) {
child = CarbonDictionaryTempDecoder(attrsOndimAggs,
new util.HashSet[Attribute](),
agg.child)
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/CarbonFilters.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/CarbonFilters.scala b/integration/spark/src/main/scala/org/carbondata/spark/CarbonFilters.scala
index aee375a..cf31a7b 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/CarbonFilters.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/CarbonFilters.scala
@@ -105,14 +105,19 @@ object CarbonFilters {
def selectFilters(filters: Seq[Expression],
attrList: java.util.HashSet[Attribute],
aliasMap: CarbonAliasDecoderRelation): Unit = {
- def translate(expr: Expression): Option[sources.Filter] = {
+ def translate(expr: Expression, or: Boolean = false): Option[sources.Filter] = {
expr match {
- case Or(left, right) =>
- for {
- leftFilter <- translate(left)
- rightFilter <- translate(right)
- } yield {
- sources.Or(leftFilter, rightFilter)
+ case or@ Or(left, right) =>
+
+ val leftFilter = translate(left, true)
+ val rightFilter = translate(right, true)
+ if (leftFilter.isDefined && rightFilter.isDefined) {
+ Some( sources.Or(leftFilter.get, rightFilter.get))
+ } else {
+ or.collect {
+ case attr: AttributeReference => attrList.add(aliasMap.getOrElse(attr, attr))
+ }
+ None
}
case And(left, right) =>
@@ -151,28 +156,35 @@ object CarbonFilters {
Some(sources.In(a.name, hSet.toArray))
case others =>
- others.collect {
- case attr: AttributeReference =>
- attrList.add(aliasMap.getOrElse(attr, attr))
+ if (!or) {
+ others.collect {
+ case attr: AttributeReference =>
+ attrList.add(aliasMap.getOrElse(attr, attr))
+ }
}
None
}
}
- filters.flatMap(translate).toArray
+ filters.flatMap(translate(_, false)).toArray
}
def processExpression(exprs: Seq[Expression],
attributesNeedToDecode: java.util.HashSet[AttributeReference],
unprocessedExprs: ArrayBuffer[Expression],
carbonTable: CarbonTable): Option[CarbonExpression] = {
- def transformExpression(expr: Expression): Option[CarbonExpression] = {
+ def transformExpression(expr: Expression, or: Boolean = false): Option[CarbonExpression] = {
expr match {
- case Or(left, right) =>
- for {
- leftFilter <- transformExpression(left)
- rightFilter <- transformExpression(right)
- } yield {
- new OrExpression(leftFilter, rightFilter)
+ case or@ Or(left, right) =>
+ val leftFilter = transformExpression(left, true)
+ val rightFilter = transformExpression(right, true)
+ if (leftFilter.isDefined && rightFilter.isDefined) {
+ Some(new OrExpression(leftFilter.get, rightFilter.get))
+ } else {
+ or.collect {
+ case attr: AttributeReference => attributesNeedToDecode.add(attr)
+ }
+ unprocessedExprs += or
+ None
}
case And(left, right) =>
@@ -220,14 +232,16 @@ object CarbonFilters {
CarbonLiteralExpression(name, CarbonScalaUtil.convertSparkToCarbonDataType(dataType)))
case Cast(left, right) if !left.isInstanceOf[Literal] => transformExpression(left)
case others =>
- others.collect {
- case attr: AttributeReference => attributesNeedToDecode.add(attr)
+ if (!or) {
+ others.collect {
+ case attr: AttributeReference => attributesNeedToDecode.add(attr)
+ }
+ unprocessedExprs += others
}
- unprocessedExprs += others
None
}
}
- exprs.flatMap(transformExpression).reduceOption(new AndExpression(_, _))
+ exprs.flatMap(transformExpression(_, false)).reduceOption(new AndExpression(_, _))
}
private def getActualCarbonDataType(column: String, carbonTable: CarbonTable) = {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/KeyVal.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/KeyVal.scala b/integration/spark/src/main/scala/org/carbondata/spark/KeyVal.scala
index 6ee882b..cb87818 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/KeyVal.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/KeyVal.scala
@@ -26,35 +26,23 @@
package org.carbondata.spark
import org.carbondata.core.load.LoadMetadataDetails
-import org.carbondata.query.carbon.result.BatchRawResult
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
-trait KeyVal[K, V] extends Serializable {
- def getKey(key: CarbonKey, value: CarbonValue): (K, V)
-
-}
-
-class KeyValImpl extends KeyVal[CarbonKey, CarbonValue] {
- override def getKey(key: CarbonKey, value: CarbonValue): (CarbonKey, CarbonValue) = (key, value)
+trait Value[V] extends Serializable {
+ def getValue(value: Array[Object]): V
}
-trait RawKeyVal[K, V] extends Serializable {
- def getKey(key: BatchRawResult, value: Any): (K, V)
-
+class ValueImpl extends Value[Array[Object]] {
+ override def getValue(value: Array[Object]): Array[Object] = value
}
-class RawKeyValImpl extends RawKeyVal[BatchRawResult, Any] {
- override def getKey(key: BatchRawResult, value: Any): (BatchRawResult, Any) = (key, value)
+trait RawValue[V] extends Serializable {
+ def getValue(value: Array[Any]): V
}
-trait RawKey[K, V] extends Serializable {
- def getKey(key: Array[Any], value: Any): (K, V)
-
+class RawValueImpl extends RawValue[Array[Any]] {
+ override def getValue(value: Array[Any]): Array[Any] = value
}
-class RawKeyImpl extends RawKey[Array[Any], Any] {
- override def getKey(key: Array[Any], value: Any): (Array[Any], Any) = (key, value)
-}
trait Result[K, V] extends Serializable {
def getKey(key: Int, value: LoadMetadataDetails): (K, V)
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonCleanFilesRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonCleanFilesRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonCleanFilesRDD.scala
index 4e71be1..17d83b2 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonCleanFilesRDD.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonCleanFilesRDD.scala
@@ -18,23 +18,23 @@
package org.carbondata.spark.rdd
import scala.collection.JavaConverters._
+import scala.reflect.ClassTag
import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.execution.command.Partitioner
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
-import org.carbondata.spark.KeyVal
+import org.carbondata.spark.Value
import org.carbondata.spark.util.CarbonQueryUtil
-class CarbonCleanFilesRDD[K, V](
+class CarbonCleanFilesRDD[V: ClassTag](
sc: SparkContext,
- keyClass: KeyVal[K, V],
+ valueClass: Value[V],
schemaName: String,
cubeName: String,
partitioner: Partitioner)
- extends RDD[(K, V)](sc, Nil) with Logging {
+ extends RDD[V](sc, Nil) with Logging {
sc.setLocalProperty("spark.scheduler.pool", "DDL")
@@ -43,8 +43,8 @@ class CarbonCleanFilesRDD[K, V](
splits.zipWithIndex.map(s => new CarbonLoadPartition(id, s._2, s._1))
}
- override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
- val iter = new Iterator[(K, V)] {
+ override def compute(theSplit: Partition, context: TaskContext): Iterator[V] = {
+ val iter = new Iterator[(V)] {
val split = theSplit.asInstanceOf[CarbonLoadPartition]
logInfo("Input split: " + split.serializableHadoopSplit.value)
// TODO call CARBON delete API
@@ -61,14 +61,12 @@ class CarbonCleanFilesRDD[K, V](
!finished
}
- override def next(): (K, V) = {
+ override def next(): V = {
if (!hasNext) {
throw new java.util.NoSuchElementException("End of stream")
}
havePair = false
- val row = new CarbonKey(null)
- val value = new CarbonValue(null)
- keyClass.getKey(row, value)
+ valueClass.getValue(null)
}
}
@@ -78,7 +76,7 @@ class CarbonCleanFilesRDD[K, V](
override def getPreferredLocations(split: Partition): Seq[String] = {
val theSplit = split.asInstanceOf[CarbonLoadPartition]
val s = theSplit.serializableHadoopSplit.value.getLocations.asScala
- logInfo("Host Name : " + s(0) + s.length)
+ logInfo("Host Name : " + s.head + s.length)
s
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 0672281..af2271f 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -29,23 +29,22 @@ import org.apache.hadoop.conf.{Configurable, Configuration}
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileSplit
import org.apache.spark.{Logging, Partition, SparkContext}
-import org.apache.spark.sql.{CarbonEnv, CarbonRelation, SQLContext}
+import org.apache.spark.sql.{CarbonEnv, SQLContext}
import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, Partitioner}
import org.apache.spark.util.{FileUtils, SplitUtils}
import org.carbondata.common.logging.LogServiceFactory
-import org.carbondata.core.carbon.{AbsoluteTableIdentifier, CarbonDataLoadSchema, CarbonTableIdentifier}
+import org.carbondata.core.carbon.CarbonDataLoadSchema
import org.carbondata.core.carbon.datastore.block.TableBlockInfo
import org.carbondata.core.carbon.metadata.CarbonMetadata
import org.carbondata.core.carbon.metadata.schema.table.CarbonTable
import org.carbondata.core.constants.CarbonCommonConstants
import org.carbondata.core.load.{BlockDetails, LoadMetadataDetails}
import org.carbondata.core.locks.{CarbonLockFactory, LockUsage}
-import org.carbondata.core.util.{CarbonProperties, CarbonUtil}
+import org.carbondata.core.util.CarbonUtil
import org.carbondata.integration.spark.merger.CompactionType
import org.carbondata.lcm.status.SegmentStatusManager
import org.carbondata.processing.util.CarbonDataProcessorUtil
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
import org.carbondata.spark._
import org.carbondata.spark.load._
import org.carbondata.spark.merger.CarbonDataMergerUtil
@@ -129,7 +128,7 @@ object CarbonDataRDDFactory extends Logging {
if (-1 == currentRestructNumber) {
currentRestructNumber = 0
}
- var segmentStatusManager = new SegmentStatusManager(cube.getAbsoluteTableIdentifier)
+ val segmentStatusManager = new SegmentStatusManager(cube.getAbsoluteTableIdentifier)
val loadMetadataDetailsArray = segmentStatusManager.readLoadMetadata(cube.getMetaDataFilepath())
.toList
val resultMap = new CarbonDeleteLoadByDateRDD(
@@ -408,9 +407,8 @@ object CarbonDataRDDFactory extends Logging {
// Check if any load need to be deleted before loading new data
deleteLoadsAndUpdateMetadata(carbonLoadModel, carbonTable, partitioner, hdfsStoreLocation,
- false,
- currentRestructNumber
- )
+ isForceDeletion = false,
+ currentRestructNumber)
if (null == carbonLoadModel.getLoadMetadataDetails) {
readLoadMetadataDetails(carbonLoadModel, hdfsStoreLocation)
}
@@ -668,9 +666,9 @@ object CarbonDataRDDFactory extends Logging {
def readLoadMetadataDetails(model: CarbonLoadModel, hdfsStoreLocation: String): Unit = {
val metadataPath = model.getCarbonDataLoadSchema.getCarbonTable.getMetaDataFilepath
- var segmentStatusManager = new SegmentStatusManager(model.getCarbonDataLoadSchema.getCarbonTable
- .
- getAbsoluteTableIdentifier)
+ val segmentStatusManager =
+ new SegmentStatusManager(
+ model.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier)
val details = segmentStatusManager.readLoadMetadata(metadataPath)
model.setLoadMetadataDetails(details.toList.asJava)
}
@@ -704,22 +702,13 @@ object CarbonDataRDDFactory extends Logging {
}
}
- def dropAggregateTable(
+ def dropTable(
sc: SparkContext,
schema: String,
cube: String,
partitioner: Partitioner) {
- val kv: KeyVal[CarbonKey, CarbonValue] = new KeyValImpl()
- new CarbonDropAggregateTableRDD(sc, kv, schema, cube, partitioner).collect
- }
-
- def dropCube(
- sc: SparkContext,
- schema: String,
- cube: String,
- partitioner: Partitioner) {
- val kv: KeyVal[CarbonKey, CarbonValue] = new KeyValImpl()
- new CarbonDropCubeRDD(sc, kv, schema, cube, partitioner).collect
+ val v: Value[Array[Object]] = new ValueImpl()
+ new CarbonDropTableRDD(sc, v, schema, cube, partitioner).collect
}
def cleanFiles(
@@ -735,7 +724,7 @@ object CarbonDataRDDFactory extends Logging {
if (-1 == currentRestructNumber) {
currentRestructNumber = 0
}
- var carbonLock = CarbonLockFactory
+ val carbonLock = CarbonLockFactory
.getCarbonLockObj(cube.getMetaDataFilepath, LockUsage.METADATA_LOCK)
try {
if (carbonLock.lockWithRetries()) {
@@ -744,8 +733,7 @@ object CarbonDataRDDFactory extends Logging {
partitioner,
hdfsStoreLocation,
isForceDeletion = true,
- currentRestructNumber
- )
+ currentRestructNumber)
}
}
finally {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDeleteLoadRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDeleteLoadRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDeleteLoadRDD.scala
index 9fc5f9e..4616ca9 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDeleteLoadRDD.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDeleteLoadRDD.scala
@@ -18,36 +18,34 @@
package org.carbondata.spark.rdd
import scala.collection.JavaConverters._
+import scala.reflect.ClassTag
import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.execution.command.Partitioner
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
-import org.carbondata.spark.KeyVal
+import org.carbondata.spark.Value
import org.carbondata.spark.util.CarbonQueryUtil
-class CarbonDeleteLoadRDD[K, V](
- sc: SparkContext,
- keyClass: KeyVal[K, V],
- loadId: Int,
- schemaName: String,
- cubeName: String,
- partitioner: Partitioner)
- extends RDD[(K, V)](sc, Nil) with Logging {
+class CarbonDeleteLoadRDD[V: ClassTag](
+ sc: SparkContext,
+ valueClass: Value[V],
+ loadId: Int,
+ schemaName: String,
+ cubeName: String,
+ partitioner: Partitioner)
+ extends RDD[V](sc, Nil) with Logging {
sc.setLocalProperty("spark.scheduler.pool", "DDL")
override def getPartitions: Array[Partition] = {
val splits = CarbonQueryUtil.getTableSplits(schemaName, cubeName, null, partitioner)
- val result = new Array[Partition](splits.length)
- for (i <- 0 until result.length) {
- result(i) = new CarbonLoadPartition(id, i, splits(i))
+ splits.zipWithIndex.map {f =>
+ new CarbonLoadPartition(id, f._2, f._1)
}
- result
}
- override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
- val iter = new Iterator[(K, V)] {
+ override def compute(theSplit: Partition, context: TaskContext): Iterator[V] = {
+ val iter = new Iterator[V] {
val split = theSplit.asInstanceOf[CarbonLoadPartition]
logInfo("Input split: " + split.serializableHadoopSplit.value)
// TODO call CARBON delete API
@@ -57,20 +55,18 @@ class CarbonDeleteLoadRDD[K, V](
override def hasNext: Boolean = {
if (!finished && !havePair) {
- finished = !false
+ finished = true
havePair = !finished
}
!finished
}
- override def next(): (K, V) = {
+ override def next(): V = {
if (!hasNext) {
throw new java.util.NoSuchElementException("End of stream")
}
havePair = false
- val row = new CarbonKey(null)
- val value = new CarbonValue(null)
- keyClass.getKey(row, value)
+ valueClass.getValue(null)
}
}
@@ -81,7 +77,7 @@ class CarbonDeleteLoadRDD[K, V](
override def getPreferredLocations(split: Partition): Seq[String] = {
val theSplit = split.asInstanceOf[CarbonLoadPartition]
val s = theSplit.serializableHadoopSplit.value.getLocations.asScala
- logInfo("Host Name : " + s(0) + s.length)
+ logInfo("Host Name : " + s.head + s.length)
s
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropAggregateTableRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropAggregateTableRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropAggregateTableRDD.scala
deleted file mode 100644
index 1b57890..0000000
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropAggregateTableRDD.scala
+++ /dev/null
@@ -1,85 +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.carbondata.spark.rdd
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.execution.command.Partitioner
-
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
-import org.carbondata.spark.KeyVal
-import org.carbondata.spark.util.CarbonQueryUtil
-
-
-class CarbonDropAggregateTableRDD[K, V](
- sc: SparkContext,
- keyClass: KeyVal[K, V],
- schemaName: String,
- cubeName: String,
- partitioner: Partitioner)
- extends RDD[(K, V)](sc, Nil) with Logging {
-
- sc.setLocalProperty("spark.scheduler.pool", "DDL")
-
- override def getPartitions: Array[Partition] = {
- val splits = CarbonQueryUtil.getTableSplits(schemaName, cubeName, null, partitioner)
- splits.zipWithIndex.map { s =>
- new CarbonLoadPartition(id, s._2, s._1)
- }
- }
-
- override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
- val iter = new Iterator[(K, V)] {
- val split = theSplit.asInstanceOf[CarbonLoadPartition]
- logInfo("Input split: " + split.serializableHadoopSplit.value)
- // TODO call CARBON delete API
-
- var havePair = false
- var finished = false
-
- override def hasNext: Boolean = {
- if (!finished && !havePair) {
- finished = true
- havePair = !finished
- }
- !finished
- }
-
- override def next(): (K, V) = {
- if (!hasNext) {
- throw new java.util.NoSuchElementException("End of stream")
- }
- havePair = false
- val row = new CarbonKey(null)
- val value = new CarbonValue(null)
- keyClass.getKey(row, value)
- }
- }
- iter
- }
-
- override def getPreferredLocations(split: Partition): Seq[String] = {
- val theSplit = split.asInstanceOf[CarbonLoadPartition]
- val s = theSplit.serializableHadoopSplit.value.getLocations.asScala
- logInfo("Host Name : " + s.head + s.length)
- s
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropCubeRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropCubeRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropCubeRDD.scala
deleted file mode 100644
index d0bc5d1..0000000
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropCubeRDD.scala
+++ /dev/null
@@ -1,77 +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.carbondata.spark.rdd
-
-import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.execution.command.Partitioner
-
-import org.carbondata.query.scanner.impl.{CarbonKey, CarbonValue}
-import org.carbondata.spark.KeyVal
-import org.carbondata.spark.util.CarbonQueryUtil
-
-class CarbonDropCubeRDD[K, V](
- sc: SparkContext,
- keyClass: KeyVal[K, V],
- schemaName: String,
- cubeName: String,
- partitioner: Partitioner)
- extends RDD[(K, V)](sc, Nil) with Logging {
-
- sc.setLocalProperty("spark.scheduler.pool", "DDL")
-
- override def getPartitions: Array[Partition] = {
- val splits = CarbonQueryUtil.getTableSplits(schemaName, cubeName, null, partitioner)
- splits.zipWithIndex.map { s =>
- new CarbonLoadPartition(id, s._2, s._1)
- }
- }
-
- override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
-
- val iter = new Iterator[(K, V)] {
- val split = theSplit.asInstanceOf[CarbonLoadPartition]
-
- val partitionCount = partitioner.partitionCount
- // TODO: Clear Btree from memory
-
- var havePair = false
- var finished = false
-
- override def hasNext: Boolean = {
- if (!finished && !havePair) {
- finished = true
- havePair = !finished
- }
- !finished
- }
-
- override def next(): (K, V) = {
- if (!hasNext) {
- throw new java.util.NoSuchElementException("End of stream")
- }
- havePair = false
- val row = new CarbonKey(null)
- val value = new CarbonValue(null)
- keyClass.getKey(row, value)
- }
- }
- iter
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropTableRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropTableRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropTableRDD.scala
new file mode 100644
index 0000000..513916c
--- /dev/null
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonDropTableRDD.scala
@@ -0,0 +1,76 @@
+/*
+ * 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.carbondata.spark.rdd
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.execution.command.Partitioner
+
+import org.carbondata.spark.Value
+import org.carbondata.spark.util.CarbonQueryUtil
+
+class CarbonDropTableRDD[V: ClassTag](
+ sc: SparkContext,
+ valueClass: Value[V],
+ schemaName: String,
+ cubeName: String,
+ partitioner: Partitioner)
+ extends RDD[V](sc, Nil) with Logging {
+
+ sc.setLocalProperty("spark.scheduler.pool", "DDL")
+
+ override def getPartitions: Array[Partition] = {
+ val splits = CarbonQueryUtil.getTableSplits(schemaName, cubeName, null, partitioner)
+ splits.zipWithIndex.map { s =>
+ new CarbonLoadPartition(id, s._2, s._1)
+ }
+ }
+
+ override def compute(theSplit: Partition, context: TaskContext): Iterator[V] = {
+
+ val iter = new Iterator[V] {
+ val split = theSplit.asInstanceOf[CarbonLoadPartition]
+
+ val partitionCount = partitioner.partitionCount
+ // TODO: Clear Btree from memory
+
+ var havePair = false
+ var finished = false
+
+ override def hasNext: Boolean = {
+ if (!finished && !havePair) {
+ finished = true
+ havePair = !finished
+ }
+ !finished
+ }
+
+ override def next(): V = {
+ if (!hasNext) {
+ throw new java.util.NoSuchElementException("End of stream")
+ }
+ havePair = false
+ valueClass.getValue(null)
+ }
+ }
+ iter
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
index 5cb33fa..dd2a10a 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonMergerRDD.scala
@@ -36,7 +36,6 @@ import org.carbondata.core.util.CarbonProperties
import org.carbondata.hadoop.{CarbonInputFormat, CarbonInputSplit}
import org.carbondata.integration.spark.merger.{CarbonCompactionExecutor, CarbonCompactionUtil,
RowResultMerger}
-import org.carbondata.query.carbon.result.{RowResult}
import org.carbondata.query.carbon.result.iterator.RawResultIterator
import org.carbondata.spark.MergeResult
import org.carbondata.spark.load.{CarbonLoaderUtil, CarbonLoadModel}
@@ -179,7 +178,7 @@ class CarbonMergerRDD[K, V](
val absoluteTableIdentifier: AbsoluteTableIdentifier = new AbsoluteTableIdentifier(
hdfsStoreLocation, new CarbonTableIdentifier(schemaName, factTableName, tableId)
)
- val (carbonInputFormat: CarbonInputFormat[RowResult], job: Job) =
+ val (carbonInputFormat: CarbonInputFormat[Array[Object]], job: Job) =
QueryPlanUtil.createCarbonInputFormat(absoluteTableIdentifier)
val result = new util.ArrayList[Partition](defaultParallelism)
val mapsOfNodeBlockMapping: util.List[util.Map[String, util.List[TableBlockInfo]]] = new
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala
index a95ae27..6693108 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -21,6 +21,7 @@ package org.carbondata.spark.rdd
import java.util
import scala.collection.JavaConverters._
+import scala.reflect.ClassTag
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.Job
@@ -33,10 +34,10 @@ import org.carbondata.core.iterator.CarbonIterator
import org.carbondata.hadoop.{CarbonInputFormat, CarbonInputSplit}
import org.carbondata.query.carbon.executor.QueryExecutorFactory
import org.carbondata.query.carbon.model.QueryModel
-import org.carbondata.query.carbon.result.{BatchRawResult, RowResult}
-import org.carbondata.query.carbon.result.iterator.ChunkRawRowIterartor
+import org.carbondata.query.carbon.result.BatchResult
+import org.carbondata.query.carbon.result.iterator.ChunkRowIterator
import org.carbondata.query.expression.Expression
-import org.carbondata.spark.RawKey
+import org.carbondata.spark.RawValue
import org.carbondata.spark.load.CarbonLoaderUtil
import org.carbondata.spark.util.QueryPlanUtil
@@ -58,29 +59,29 @@ class CarbonSparkPartition(rddId: Int, val idx: Int,
* CarbonData file, this RDD will leverage CarbonData's index information to do CarbonData file
* level filtering in driver side.
*/
-class CarbonScanRDD[K, V](
+class CarbonScanRDD[V: ClassTag](
sc: SparkContext,
queryModel: QueryModel,
filterExpression: Expression,
- keyClass: RawKey[K, V],
+ keyClass: RawValue[V],
@transient conf: Configuration,
cubeCreationTime: Long,
schemaLastUpdatedTime: Long,
baseStoreLocation: String)
- extends RDD[(K, V)](sc, Nil) with Logging {
+ extends RDD[V](sc, Nil) with Logging {
val defaultParallelism = sc.defaultParallelism
override def getPartitions: Array[Partition] = {
val startTime = System.currentTimeMillis()
- val (carbonInputFormat: CarbonInputFormat[RowResult], job: Job) =
+ val (carbonInputFormat: CarbonInputFormat[Array[Object]], job: Job) =
QueryPlanUtil.createCarbonInputFormat(queryModel.getAbsoluteTableIdentifier)
val result = new util.ArrayList[Partition](defaultParallelism)
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
// set filter resolver tree
try {
- var filterResolver = carbonInputFormat
+ val filterResolver = carbonInputFormat
.getResolvedFilter(job.getConfiguration, filterExpression)
CarbonInputFormat.setFilterPredicates(job.getConfiguration, filterResolver)
@@ -147,15 +148,15 @@ class CarbonScanRDD[K, V](
result.toArray(new Array[Partition](result.size()))
}
- override def compute(thepartition: Partition, context: TaskContext): Iterator[(K, V)] = {
- val LOGGER = LogServiceFactory.getLogService(this.getClass().getName());
- val iter = new Iterator[(K, V)] {
+ override def compute(thepartition: Partition, context: TaskContext): Iterator[V] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ val iter = new Iterator[V] {
var rowIterator: CarbonIterator[Array[Any]] = _
var queryStartTime: Long = 0
try {
val carbonSparkPartition = thepartition.asInstanceOf[CarbonSparkPartition]
if(!carbonSparkPartition.tableBlockInfos.isEmpty) {
- queryModel.setQueryId(queryModel.getQueryId() + "_" + carbonSparkPartition.idx)
+ queryModel.setQueryId(queryModel.getQueryId + "_" + carbonSparkPartition.idx)
// fill table block info
queryModel.setTableBlockInfos(carbonSparkPartition.tableBlockInfos)
queryStartTime = System.currentTimeMillis
@@ -164,13 +165,13 @@ class CarbonScanRDD[K, V](
logInfo("*************************" + carbonPropertiesFilePath)
if (null == carbonPropertiesFilePath) {
System.setProperty("carbon.properties.filepath",
- System.getProperty("user.dir") + '/' + "conf" + '/' + "carbon.properties");
+ System.getProperty("user.dir") + '/' + "conf" + '/' + "carbon.properties")
}
// execute query
- rowIterator = new ChunkRawRowIterartor(
- QueryExecutorFactory.getQueryExecutor(queryModel).execute(queryModel)
- .asInstanceOf[CarbonIterator[BatchRawResult]])
- .asInstanceOf[CarbonIterator[Array[Any]]]
+ rowIterator = new ChunkRowIterator(
+ QueryExecutorFactory.getQueryExecutor(queryModel).execute(queryModel).
+ asInstanceOf[CarbonIterator[BatchResult]]).asInstanceOf[CarbonIterator[Array[Any]]]
+
}
} catch {
case e: Exception =>
@@ -187,19 +188,18 @@ class CarbonScanRDD[K, V](
override def hasNext: Boolean = {
if (!finished && !havePair) {
- finished = (null == rowIterator) || (!rowIterator.hasNext())
+ finished = (null == rowIterator) || (!rowIterator.hasNext)
havePair = !finished
}
!finished
}
- override def next(): (K, V) = {
+ override def next(): V = {
if (!hasNext) {
throw new java.util.NoSuchElementException("End of stream")
}
havePair = false
- val row = rowIterator.next()
- keyClass.getKey(row, null)
+ keyClass.getValue(rowIterator.next())
}
logInfo("********************** Total Time Taken to execute the query in Carbon Side: " +
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/main/scala/org/carbondata/spark/util/QueryPlanUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/util/QueryPlanUtil.scala b/integration/spark/src/main/scala/org/carbondata/spark/util/QueryPlanUtil.scala
index 81b6feb..321e0f8 100644
--- a/integration/spark/src/main/scala/org/carbondata/spark/util/QueryPlanUtil.scala
+++ b/integration/spark/src/main/scala/org/carbondata/spark/util/QueryPlanUtil.scala
@@ -25,7 +25,6 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.carbondata.core.carbon.AbsoluteTableIdentifier
import org.carbondata.hadoop.CarbonInputFormat
-import org.carbondata.query.carbon.result.RowResult
/**
* All the utility functions for carbon plan creation
@@ -36,8 +35,8 @@ object QueryPlanUtil {
* createCarbonInputFormat from query model
*/
def createCarbonInputFormat(absoluteTableIdentifier: AbsoluteTableIdentifier) :
- (CarbonInputFormat[RowResult], Job) = {
- val carbonInputFormat = new CarbonInputFormat[RowResult]()
+ (CarbonInputFormat[Array[Object]], Job) = {
+ val carbonInputFormat = new CarbonInputFormat[Array[Object]]()
val jobConf: JobConf = new JobConf(new Configuration)
val job: Job = new Job(jobConf)
FileInputFormat.addInputPath(job, new Path(absoluteTableIdentifier.getStorePath))
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/integration/spark/src/test/scala/org/carbondata/spark/testsuite/allqueries/AllDataTypesTestCaseAggregate.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/carbondata/spark/testsuite/allqueries/AllDataTypesTestCaseAggregate.scala b/integration/spark/src/test/scala/org/carbondata/spark/testsuite/allqueries/AllDataTypesTestCaseAggregate.scala
index 4edfb7a..c2a2277 100644
--- a/integration/spark/src/test/scala/org/carbondata/spark/testsuite/allqueries/AllDataTypesTestCaseAggregate.scala
+++ b/integration/spark/src/test/scala/org/carbondata/spark/testsuite/allqueries/AllDataTypesTestCaseAggregate.scala
@@ -64,6 +64,7 @@ class AllDataTypesTestCaseAggregate extends QueryTest with BeforeAndAfterAll {
sql("select channelsId, Latest_DAY from Carbon_automation_test where count(channelsId) = 1").collect
} catch {
case ce: UnsupportedOperationException => ce.getMessage
+ case ce: Exception => ce.getMessage
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6288ec71/processing/src/test/java/org/carbondata/processing/store/colgroup/ColGroupMinMaxTest.java
----------------------------------------------------------------------
diff --git a/processing/src/test/java/org/carbondata/processing/store/colgroup/ColGroupMinMaxTest.java b/processing/src/test/java/org/carbondata/processing/store/colgroup/ColGroupMinMaxTest.java
index 18b2bfe..641d18a 100644
--- a/processing/src/test/java/org/carbondata/processing/store/colgroup/ColGroupMinMaxTest.java
+++ b/processing/src/test/java/org/carbondata/processing/store/colgroup/ColGroupMinMaxTest.java
@@ -84,7 +84,7 @@ public class ColGroupMinMaxTest {
}
setMinData(data[i]);
setMaxData(data[i]);
- System.out.println(Arrays.toString(data[i]));
+// System.out.println(Arrays.toString(data[i]));
}
mdkeyData = new byte[1000][];
for (int i = 0; i < 1000; i++) {