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:15:58 UTC
[10/56] [abbrv] incubator-carbondata git commit: Update SQL planning
in carbon-spark (#682)
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/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 56deb8a..829c487 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
@@ -15,38 +15,27 @@
* limitations under the License.
*/
-
package org.apache.spark.sql.hive
-import scala.math.BigInt.int2bigInt
+import scala.collection.JavaConverters._
import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
-import org.apache.spark.sql.catalyst.planning.{ExtractEquiJoinKeys, PhysicalOperation, QueryPlanner}
-import org.apache.spark.sql.catalyst.plans.Inner
-import org.apache.spark.sql.catalyst.plans.logical.{BroadcastHint, Limit, LogicalPlan, Sort}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.expressions
+import org.apache.spark.sql.catalyst.expressions.{AttributeSet, _}
+import org.apache.spark.sql.catalyst.planning.{PhysicalOperation, QueryPlanner}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter => LogicalFilter, LogicalPlan}
import org.apache.spark.sql.execution.{DescribeCommand => RunnableDescribeCommand, ExecutedCommand, Filter, Project, SparkPlan}
-import org.apache.spark.sql.execution.aggregate.Utils
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.datasources.{DescribeCommand => LogicalDescribeCommand, LogicalRelation}
-import org.apache.spark.sql.execution.joins.{BroadCastFilterPushJoin, BuildLeft, BuildRight}
import org.apache.spark.sql.hive.execution.{DescribeHiveTableCommand, DropTable, HiveNativeCommand}
+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
-object CarbonHiveSyntax {
-
- @transient
- protected val sqlParser = new CarbonSqlParser
-
- def parse(sqlText: String): LogicalPlan = {
- sqlParser.parse(sqlText)
- }
-}
class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
@@ -55,342 +44,179 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
val LOGGER = LogServiceFactory.getLogService("CarbonStrategies")
def getStrategies: Seq[Strategy] = {
- val total = sqlContext.planner.strategies :+ CarbonTableScans :+ DDLStrategies
+ val total = sqlContext.planner.strategies :+ CarbonTableScan
total
}
/**
- * Carbon strategies for Carbon cube scanning
+ * Carbon strategies for performing late materizlization (decoding dictionary key
+ * as late as possbile)
*/
- private[sql] object CarbonTableScans extends Strategy {
-
- def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
- case PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)) =>
- carbonScan(projectList,
- predicates,
- carbonRelation.carbonRelation,
- None,
- None,
- None,
- isGroupByPresent = false,
- detailQuery = true) :: Nil
-
- case Limit(IntegerLiteral(limit),
- Sort(order, _,
- p@CarbonAggregation(groupingExpressions,
- aggregateExpressions,
- PhysicalOperation(
- projectList,
- predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _))))) =>
- val aggPlan = handleAggregation(plan, p, projectList, predicates, carbonRelation,
- groupingExpressions, aggregateExpressions)
- org.apache.spark.sql.execution.TakeOrderedAndProject(limit,
- order,
- None,
- aggPlan.head) :: Nil
-
- case Limit(IntegerLiteral(limit), p@CarbonAggregation(
- groupingExpressions,
- aggregateExpressions,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)))) =>
- val aggPlan = handleAggregation(plan, p, projectList, predicates, carbonRelation,
- groupingExpressions, aggregateExpressions)
- org.apache.spark.sql.execution.Limit(limit, aggPlan.head) :: Nil
-
- case CarbonAggregation(
- groupingExpressions,
- aggregateExpressions,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _))) =>
- handleAggregation(plan, plan, projectList, predicates, carbonRelation,
- groupingExpressions, aggregateExpressions)
-
- case Limit(IntegerLiteral(limit),
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _))) =>
- val (_, _, _, _, groupExprs, substitutesortExprs, limitExpr) = extractPlan(plan)
- val s = carbonScan(projectList, predicates, carbonRelation.carbonRelation, groupExprs,
- substitutesortExprs, limitExpr, isGroupByPresent = false, detailQuery = true)
- org.apache.spark.sql.execution.Limit(limit, s) :: Nil
-
- case Limit(IntegerLiteral(limit),
- Sort(order, _,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)))) =>
- val (_, _, _, _, groupExprs, substitutesortExprs, limitExpr) = extractPlan(plan)
- val s = carbonScan(projectList, predicates, carbonRelation.carbonRelation, groupExprs,
- substitutesortExprs, limitExpr, isGroupByPresent = false, detailQuery = true)
- org.apache.spark.sql.execution.TakeOrderedAndProject(limit,
- order,
- None,
- s) :: Nil
-
- case ExtractEquiJoinKeys(Inner, leftKeys, rightKeys, condition,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)), right)
- if canPushDownJoin(right, condition) =>
- LOGGER.info(s"pushing down for ExtractEquiJoinKeys:right")
- val carbon = carbonScan(projectList,
- predicates,
- carbonRelation.carbonRelation,
- None,
- None,
- None,
- isGroupByPresent = false,
- detailQuery = true)
- val pushedDownJoin = BroadCastFilterPushJoin(
- leftKeys: Seq[Expression],
- rightKeys: Seq[Expression],
- BuildRight,
- carbon,
- planLater(right),
- condition)
-
- condition.map(Filter(_, pushedDownJoin)).getOrElse(pushedDownJoin) :: Nil
-
- case ExtractEquiJoinKeys(Inner, leftKeys, rightKeys, condition, left,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)))
- if canPushDownJoin(left, condition) =>
- LOGGER.info(s"pushing down for ExtractEquiJoinKeys:left")
- val carbon = carbonScan(projectList,
- predicates,
- carbonRelation.carbonRelation,
- None,
- None,
- None,
- isGroupByPresent = false,
- detailQuery = true)
-
- val pushedDownJoin = BroadCastFilterPushJoin(
- leftKeys: Seq[Expression],
- rightKeys: Seq[Expression],
- BuildLeft,
- planLater(left),
- carbon,
- condition)
- condition.map(Filter(_, pushedDownJoin)).getOrElse(pushedDownJoin) :: Nil
-
- case _ => Nil
- }
-
- def handleAggregation(plan: LogicalPlan,
- aggPlan: LogicalPlan,
- projectList: Seq[NamedExpression],
- predicates: Seq[Expression],
- carbonRelation: CarbonDatasourceRelation,
- groupingExpressions: Seq[Expression],
- namedGroupingAttributes: Seq[NamedExpression]):
- Seq[SparkPlan] = {
- val (_, _, _, _, groupExprs, substitutesortExprs, limitExpr) = extractPlan(plan)
-
- val s =
- try {
- carbonScan(projectList, predicates, carbonRelation.carbonRelation,
- Some(namedGroupingAttributes), substitutesortExprs,
- limitExpr, groupingExpressions.nonEmpty)
- } catch {
- case e: Exception => null
- }
-
- if (s != null) {
- aggregatePlan(groupingExpressions, namedGroupingAttributes, s)
-
- } else {
- (aggPlan, true) match {
- case CarbonAggregation(
- groupingExpressions,
- namedGroupingAttributes,
- PhysicalOperation(projectList, predicates,
- l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _))) =>
- val (_, _, _, _, groupExprs, substitutesortExprs, limitExpr) = extractPlan(plan)
-
-
- val s = carbonScan(projectList, predicates, carbonRelation.carbonRelation,
- Some(namedGroupingAttributes), substitutesortExprs, limitExpr,
- groupingExpressions.nonEmpty, detailQuery = true)
-
- aggregatePlan(groupingExpressions, namedGroupingAttributes, s)
- }
+ private[sql] object CarbonTableScan extends Strategy {
+
+ def apply(plan: LogicalPlan): Seq[SparkPlan] = {
+ plan match {
+ case PhysicalOperation(projectList, predicates,
+ l@LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)) =>
+ if (isStarQuery(plan)) {
+ carbonRawScanForStarQuery(projectList, predicates, carbonRelation, l)(sqlContext) :: Nil
+ } else {
+ carbonRawScan(projectList,
+ predicates,
+ carbonRelation,
+ l,
+ None,
+ detailQuery = true,
+ useBinaryAggregation = false)(sqlContext)._1 :: Nil
+ }
+ case CarbonDictionaryCatalystDecoder(relations, profile, aliasMap, _, child) =>
+ CarbonDictionaryDecoder(relations,
+ profile,
+ aliasMap,
+ planLater(child))(sqlContext) :: Nil
+ case _ =>
+ Nil
}
}
- // TODO: It is duplicated code from spark. Need to find a way
- private def aggregatePlan(groupingExpressions: Seq[Expression],
- resultExpressions: Seq[NamedExpression],
- child: SparkPlan) = {
- // A single aggregate expression might appear multiple times in resultExpressions.
- // In order to avoid evaluating an individual aggregate function multiple times, we'll
- // build a set of the distinct aggregate expressions and build a function which can
- // be used to re-write expressions so that they reference the single copy of the
- // aggregate function which actually gets computed.
- val aggregateExpressions = resultExpressions.flatMap { expr =>
- expr.collect {
- case agg: AggregateExpression => agg
- }
- }.distinct
- // For those distinct aggregate expressions, we create a map from the
- // aggregate function to the corresponding attribute of the function.
- val aggregateFunctionToAttribute = aggregateExpressions.map { agg =>
- val aggregateFunction = agg.aggregateFunction
- val attribute = Alias(aggregateFunction, aggregateFunction.toString)().toAttribute
- (aggregateFunction, agg.isDistinct) -> attribute
- }.toMap
-
- val (functionsWithDistinct, functionsWithoutDistinct) =
- aggregateExpressions.partition(_.isDistinct)
- if (functionsWithDistinct.map(_.aggregateFunction.children).distinct.length > 1) {
- // This is a sanity check. We should not reach here when we have multiple distinct
- // column sets. Our MultipleDistinctRewriter should take care this case.
- sys.error("You hit a query analyzer bug. Please report your query to " +
- "Spark user mailing list.")
- }
-
- val namedGroupingExpressions = groupingExpressions.map {
- case ne: NamedExpression => ne -> ne
- // If the expression is not a NamedExpressions, we add an alias.
- // So, when we generate the result of the operator, the Aggregate Operator
- // can directly get the Seq of attributes representing the grouping expressions.
- case other =>
- val withAlias = Alias(other, other.toString)()
- other -> withAlias
- }
- val groupExpressionMap = namedGroupingExpressions.toMap
-
- // The original `resultExpressions` are a set of expressions which may reference
- // aggregate expressions, grouping column values, and constants. When aggregate operator
- // emits output rows, we will use `resultExpressions` to generate an output projection
- // which takes the grouping columns and final aggregate result buffer as input.
- // Thus, we must re-write the result expressions so that their attributes match up with
- // the attributes of the final result projection's input row:
- val rewrittenResultExpressions = resultExpressions.map { expr =>
- expr.transformDown {
- case AggregateExpression(aggregateFunction, _, isDistinct) =>
- // The final aggregation buffer's attributes will be `finalAggregationAttributes`,
- // so replace each aggregate expression by its corresponding attribute in the set:
- aggregateFunctionToAttribute(aggregateFunction, isDistinct)
- case expression =>
- // Since we're using `namedGroupingAttributes` to extract the grouping key
- // columns, we need to replace grouping key expressions with their corresponding
- // attributes. We do not rely on the equality check at here since attributes may
- // differ cosmetically. Instead, we use semanticEquals.
- groupExpressionMap.collectFirst {
- case (expr, ne) if expr semanticEquals expression => ne.toAttribute
- }.getOrElse(expression)
- }.asInstanceOf[NamedExpression]
+ /**
+ * Create carbon scan
+ */
+ 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) = {
+
+ val tableName: String =
+ relation.carbonRelation.metaData.carbonTable.getFactTableName.toLowerCase
+ // Check out any expressions are there in project list. if they are present then we need to
+ // decode them as well.
+ 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
+ projectList.map {
+ case attr: AttributeReference =>
+ case Alias(attr: AttributeReference, _) =>
+ case others =>
+ others.references
+ .map(f => scan.attributesNeedToDecode.add(f.asInstanceOf[AttributeReference]))
}
-
- val aggregateOperator =
- if (aggregateExpressions.map(_.aggregateFunction).exists(!_.supportsPartial)) {
- if (functionsWithDistinct.nonEmpty) {
- sys.error("Distinct columns cannot exist in Aggregate operator containing " +
- "aggregate functions which don't support partial aggregation.")
+ 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 {
- Utils.planAggregateWithoutPartial(
- namedGroupingExpressions.map(_._2),
- aggregateExpressions,
- aggregateFunctionToAttribute,
- rewrittenResultExpressions,
- child)
+ (Project(projectList, decoder), true)
}
- } else if (functionsWithDistinct.isEmpty) {
- Utils.planAggregateWithoutDistinct(
- namedGroupingExpressions.map(_._2),
- aggregateExpressions,
- aggregateFunctionToAttribute,
- rewrittenResultExpressions,
- child)
} else {
- Utils.planAggregateWithOneDistinct(
- namedGroupingExpressions.map(_._2),
- functionsWithDistinct,
- functionsWithoutDistinct,
- aggregateFunctionToAttribute,
- rewrittenResultExpressions,
- child)
+ (scan, dimAggrsPresence)
+ }
+ } 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)
}
-
- aggregateOperator
- }
-
- private def canPushDownJoin(otherRDDPlan: LogicalPlan,
- joinCondition: Option[Expression]): Boolean = {
- val pushdowmJoinEnabled = sqlContext.sparkContext.conf
- .getBoolean("spark.carbon.pushdown.join.as.filter", defaultValue = true)
-
- if (!pushdowmJoinEnabled) {
- return false
- }
-
- otherRDDPlan match {
- case BroadcastHint(p) => true
- case p if sqlContext.conf.autoBroadcastJoinThreshold > 0 &&
- p.statistics.sizeInBytes <= sqlContext.conf.autoBroadcastJoinThreshold =>
- LOGGER.info("canPushDownJoin statistics:" + p.statistics.sizeInBytes)
- true
- case _ => false
}
}
/**
- * Create carbon scan
+ * Create carbon scan for star query
*/
- private def carbonScan(projectList: Seq[NamedExpression],
- predicates: Seq[Expression],
- relation: CarbonRelation,
- groupExprs: Option[Seq[Expression]],
- substitutesortExprs: Option[Seq[SortOrder]],
- limitExpr: Option[Expression],
- isGroupByPresent: Boolean,
- detailQuery: Boolean = false) = {
+ private def carbonRawScanForStarQuery(projectList: Seq[NamedExpression],
+ predicates: Seq[Expression],
+ relation: CarbonDatasourceRelation,
+ logicalRelation: LogicalRelation)(sc: SQLContext): SparkPlan = {
+
+ val tableName: String =
+ relation.carbonRelation.metaData.carbonTable.getFactTableName.toLowerCase
+ // Check out any expressions are there in project list. if they are present then we need to
+ // decode them as well.
+ val projectExprsNeedToDecode = new java.util.HashSet[Attribute]()
+ val scan = CarbonScan(projectList.map(_.toAttribute),
+ relation.carbonRelation,
+ predicates,
+ None,
+ useBinaryAggregator = false)(sqlContext)
+ projectExprsNeedToDecode.addAll(scan.attributesNeedToDecode)
+ if (projectExprsNeedToDecode.size() > 0) {
+ val decoder = getCarbonDecoder(logicalRelation,
+ sc,
+ tableName,
+ projectExprsNeedToDecode.asScala.toSeq,
+ scan)
+ if (scan.unprocessedExprs.nonEmpty) {
+ val filterCondToAdd = scan.unprocessedExprs.reduceLeftOption(expressions.And)
+ filterCondToAdd.map(Filter(_, decoder)).getOrElse(decoder)
+ } else {
+ decoder
+ }
+ } else {
+ scan
+ }
+ }
- if (!detailQuery) {
- val projectSet = AttributeSet(projectList.flatMap(_.references))
- CarbonTableScan(
- projectSet.toSeq,
- relation,
- predicates,
- groupExprs,
- substitutesortExprs,
- limitExpr,
- isGroupByPresent,
- detailQuery)(sqlContext)
+ def getCarbonDecoder(logicalRelation: LogicalRelation,
+ sc: SQLContext,
+ tableName: String,
+ projectExprsNeedToDecode: Seq[Attribute],
+ scan: CarbonScan): CarbonDictionaryDecoder = {
+ val relation = CarbonDecoderRelation(logicalRelation.attributeMap,
+ logicalRelation.relation.asInstanceOf[CarbonDatasourceRelation])
+ val attrs = projectExprsNeedToDecode.map { attr =>
+ val newAttr = AttributeReference(attr.name,
+ attr.dataType,
+ attr.nullable,
+ attr.metadata)(attr.exprId, Seq(tableName))
+ relation.addAttribute(newAttr)
+ newAttr
}
- else {
- val projectSet = AttributeSet(projectList.flatMap(_.references))
- Project(projectList,
- CarbonTableScan(projectSet.toSeq,
- relation,
- predicates,
- groupExprs,
- substitutesortExprs,
- limitExpr,
- isGroupByPresent,
- detailQuery)(sqlContext))
+ CarbonDictionaryDecoder(Seq(relation), IncludeProfile(attrs),
+ CarbonAliasDecoderRelation(), scan)(sc)
+ }
+ private def isStarQuery(plan: LogicalPlan) = {
+ plan match {
+ case LogicalFilter(condition,
+ LogicalRelation(carbonRelation: CarbonDatasourceRelation, _)) => true
+ case LogicalRelation(carbonRelation: CarbonDatasourceRelation, _) => true
+ case _ => false
}
}
- private def extractPlan(plan: LogicalPlan) = {
- val (a, b, c, aliases, groupExprs, sortExprs, limitExpr) =
- PhysicalOperation1.collectProjectsAndFilters(plan)
- val substitutesortExprs = sortExprs match {
- case Some(sort) =>
- Some(sort.map {
- case SortOrder(a: Alias, direction) =>
- val ref = aliases.getOrElse(a.toAttribute, a) match {
- case Alias(reference, name) => reference
- case others => others
- }
- SortOrder(ref, direction)
- case others => others
- })
- case others => others
+ 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
+ }
}
- (a, b, c, aliases, groupExprs, substitutesortExprs, limitExpr)
+ false
}
}
@@ -406,7 +232,7 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
ExecutedCommand(ShowAllTablesDetail(schemaName, plan.output)) :: Nil
case DropTable(tableName, ifNotExists)
if CarbonEnv.getInstance(sqlContext).carbonCatalog
- .tableExists(TableIdentifier(tableName.toLowerCase))(sqlContext) =>
+ .tableExists(TableIdentifier(tableName.toLowerCase))(sqlContext) =>
ExecutedCommand(DropCubeCommand(ifNotExists, None, tableName.toLowerCase)) :: Nil
case ShowAggregateTablesCommand(schemaName) =>
ExecutedCommand(ShowAggregateTables(schemaName, plan.output)) :: Nil
@@ -415,7 +241,7 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
case LoadCube(schemaNameOp, cubeName, factPathFromUser, dimFilesPath,
partionValues, isOverwriteExist, inputSqlString) =>
val isCarbonTable = CarbonEnv.getInstance(sqlContext).carbonCatalog
- .tableExists(TableIdentifier(cubeName, schemaNameOp))(sqlContext)
+ .tableExists(TableIdentifier(cubeName, schemaNameOp))(sqlContext)
if (isCarbonTable || partionValues.nonEmpty) {
ExecutedCommand(LoadCube(schemaNameOp, cubeName, factPathFromUser,
dimFilesPath, partionValues, isOverwriteExist, inputSqlString)) :: Nil
@@ -433,15 +259,14 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
}
case DescribeFormattedCommand(sql, tblIdentifier) =>
val isCube = CarbonEnv.getInstance(sqlContext).carbonCatalog
- .tableExists(tblIdentifier)(sqlContext)
+ .tableExists(tblIdentifier)(sqlContext)
if (isCube) {
val describe =
LogicalDescribeCommand(UnresolvedRelation(tblIdentifier, None), isExtended = false)
val resolvedTable = sqlContext.executePlan(describe.table).analyzed
val resultPlan = sqlContext.executePlan(resolvedTable).executedPlan
ExecutedCommand(DescribeCommandFormatted(resultPlan, plan.output, tblIdentifier)) :: Nil
- }
- else {
+ } else {
ExecutedCommand(DescribeNativeCommand(sql, plan.output)) :: Nil
}
case describe@LogicalDescribeCommand(table, isExtended) =>
@@ -461,3 +286,13 @@ class CarbonStrategies(sqlContext: SQLContext) extends QueryPlanner[SparkPlan] {
}
}
+
+object CarbonHiveSyntax {
+
+ @transient
+ protected val sqlParser = new CarbonSqlParser
+
+ def parse(sqlText: String): LogicalPlan = {
+ sqlParser.parse(sqlText)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategy.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategy.scala b/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategy.scala
deleted file mode 100644
index 58c02ff..0000000
--- a/integration/spark/src/main/scala/org/apache/spark/sql/hive/CarbonStrategy.scala
+++ /dev/null
@@ -1,54 +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.plans.logical.LogicalPlan
-
-import org.carbondata.spark.exception.MalformedCarbonCommandException
-
-private[sql] object CarbonStrategy {
- def getStrategy(context: SQLContext): Seq[Strategy] = {
- val carbonStrategy = new CarbonStrategies(context)
- if (context.conf.asInstanceOf[CarbonSQLConf].pushComputation) {
- Seq(carbonStrategy.CarbonTableScans, carbonStrategy.DDLStrategies)
- } else {
- // TODO: need to remove duplicate code in strategies.
- Seq(new CarbonRawStrategies(context).CarbonRawTableScans, carbonStrategy.DDLStrategies)
- }
- }
-}
-
-private[spark] class CarbonSQLDialect(context: HiveContext) extends HiveQLDialect(context) {
-
- @transient
- protected val sqlParser = new CarbonSqlParser
-
- override def parse(sqlText: String): LogicalPlan = {
-
- try {
- sqlParser.parse(sqlText)
- } catch {
- // MalformedCarbonCommandException need to throw directly
- // because hive can no parse carbon command
- case ce: MalformedCarbonCommandException =>
- throw ce
- case _ => super.parse(sqlText)
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonQueryRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonQueryRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonQueryRDD.scala
deleted file mode 100644
index 60c6a78..0000000
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonQueryRDD.scala
+++ /dev/null
@@ -1,241 +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 java.util
-
-import scala.collection.JavaConverters._
-
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.mapreduce.Job
-import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
-import org.apache.spark.rdd.RDD
-
-import org.carbondata.common.logging.LogServiceFactory
-import org.carbondata.core.cache.dictionary.Dictionary
-import org.carbondata.core.carbon.datastore.block.TableBlockInfo
-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.RowResult
-import org.carbondata.query.expression.Expression
-import org.carbondata.query.filter.resolver.FilterResolverIntf
-import org.carbondata.spark.KeyVal
-import org.carbondata.spark.load.CarbonLoaderUtil
-import org.carbondata.spark.util.QueryPlanUtil
-
-class CarbonSparkPartition(rddId: Int, val idx: Int,
- val locations: Array[String],
- val tableBlockInfos: util.List[TableBlockInfo])
- extends Partition {
-
- override val index: Int = idx
-
- // val serializableHadoopSplit = new SerializableWritable[Array[String]](locations)
- override def hashCode(): Int = {
- 41 * (41 + rddId) + idx
- }
-}
-
-
- /**
- * This RDD is used to perform query.
- */
-class CarbonQueryRDD[K, V](
- sc: SparkContext,
- queryModel: QueryModel,
- filterExpression: Expression,
- keyClass: KeyVal[K, V],
- @transient conf: Configuration,
- cubeCreationTime: Long,
- schemaLastUpdatedTime: Long,
- baseStoreLocation: String)
- extends RDD[(K, V)](sc, Nil) with Logging {
-
- val defaultParallelism = sc.defaultParallelism
-
- override def getPartitions: Array[Partition] = {
- val startTime = System.currentTimeMillis()
- val (carbonInputFormat: CarbonInputFormat[RowResult], 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
- .getResolvedFilter(job.getConfiguration, filterExpression)
-
- CarbonInputFormat.setFilterPredicates(job.getConfiguration, filterResolver)
- queryModel.setFilterExpressionResolverTree(filterResolver)
- }
- catch {
- case e: Exception =>
- LOGGER.error(e)
- sys.error("Exception occurred in query execution :: " + e.getMessage)
- }
- // get splits
- val splits = carbonInputFormat.getSplits(job)
- if (!splits.isEmpty) {
- val carbonInputSplits = splits.asScala.map(_.asInstanceOf[CarbonInputSplit])
-
- val blockList = carbonInputSplits.map(inputSplit =>
- new TableBlockInfo(inputSplit.getPath.toString,
- inputSplit.getStart, inputSplit.getSegmentId,
- inputSplit.getLocations, inputSplit.getLength
- )
- )
- if (blockList.nonEmpty) {
- // group blocks to nodes, tasks
- val nodeBlockMapping =
- CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1, defaultParallelism)
-
- var i = 0
- // Create Spark Partition for each task and assign blocks
- nodeBlockMapping.asScala.foreach { entry =>
- entry._2.asScala.foreach { blocksPerTask =>
- if (blocksPerTask.size() != 0) {
- result.add(new CarbonSparkPartition(id, i, Seq(entry._1).toArray, blocksPerTask))
- i += 1
- }
- }
- }
- val noOfBlocks = blockList.size
- val noOfNodes = nodeBlockMapping.size
- val noOfTasks = result.size()
- logInfo(s"Identified no.of.Blocks: $noOfBlocks,"
- + s"parallelism: $defaultParallelism , " +
- s"no.of.nodes: $noOfNodes, no.of.tasks: $noOfTasks"
- )
- logInfo("Time taken to identify Blocks to scan : " +
- (System.currentTimeMillis() - startTime)
- )
- result.asScala.foreach { r =>
- val cp = r.asInstanceOf[CarbonSparkPartition]
- logInfo(s"Node : " + cp.locations.toSeq.mkString(",")
- + ", No.Of Blocks : " + cp.tableBlockInfos.size()
- )
- }
- } else {
- logInfo("No blocks identified to scan")
- val nodesPerBlock = new util.ArrayList[TableBlockInfo]()
- result.add(new CarbonSparkPartition(id, 0, Seq("").toArray, nodesPerBlock))
- }
- }
- else {
- logInfo("No valid segments found to scan")
- val nodesPerBlock = new util.ArrayList[TableBlockInfo]()
- result.add(new CarbonSparkPartition(id, 0, Seq("").toArray, nodesPerBlock))
- }
- 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)] {
- var rowIterator: CarbonIterator[_] = _
- var queryStartTime: Long = 0
- try {
- val carbonSparkPartition = thepartition.asInstanceOf[CarbonSparkPartition]
- if (!carbonSparkPartition.tableBlockInfos.isEmpty) {
- queryModel.setQueryId(queryModel.getQueryId + "_" + carbonSparkPartition.idx)
- // fill table block info
- queryModel.setTableBlockInfos(carbonSparkPartition.tableBlockInfos)
- queryStartTime = System.currentTimeMillis
-
- val carbonPropertiesFilePath = System.getProperty("carbon.properties.filepath", null)
- logInfo("*************************" + carbonPropertiesFilePath)
- if (null == carbonPropertiesFilePath) {
- System.setProperty("carbon.properties.filepath",
- System.getProperty("user.dir") + '/' + "conf" + '/' + "carbon.properties"
- )
- }
- // execute query
- rowIterator = QueryExecutorFactory.getQueryExecutor(queryModel).execute(queryModel)
- .asInstanceOf[CarbonIterator[RowResult]]
- }
- // TODOi
- // : CarbonQueryUtil.isQuickFilter quick filter from dictionary needs to support
- } catch {
- case e: Throwable =>
- clearDictionaryCache(queryModel.getColumnToDictionaryMapping)
- LOGGER.error(e)
- // updateCubeAndLevelCacheStatus(levelCacheKeys)
- if (null != e.getMessage) {
- sys.error("Exception occurred in query execution :: " + e.getMessage)
- } else {
- sys.error("Exception occurred in query execution.Please check logs.")
- }
- }
-
- var havePair = false
- var finished = false
- var recordCount = 0
-
- override def hasNext: Boolean = {
- if (!finished && !havePair) {
- finished = (null == rowIterator) || (!rowIterator.hasNext)
- havePair = !finished
- }
- if (finished) {
- clearDictionaryCache(queryModel.getColumnToDictionaryMapping)
- }
- !finished
- }
-
- override def next(): (K, V) = {
- if (!hasNext) {
- throw new java.util.NoSuchElementException("End of stream")
- }
- havePair = false
- val row = rowIterator.next()
- val key = row.asInstanceOf[RowResult].getKey()
- val value = row.asInstanceOf[RowResult].getValue()
- recordCount += 1
- if (queryModel.getLimit != -1 && recordCount >= queryModel.getLimit) {
- clearDictionaryCache(queryModel.getColumnToDictionaryMapping)
- }
- keyClass.getKey(key, value)
- }
-
- def clearDictionaryCache(columnToDictionaryMap: java.util.Map[String, Dictionary]) = {
- if (null != columnToDictionaryMap) {
- org.carbondata.spark.util.CarbonQueryUtil
- .clearColumnDictionaryCache(columnToDictionaryMap)
- }
- }
-
- logInfo("*************************** Total Time Taken to execute the query in Carbon Side: " +
- (System.currentTimeMillis - queryStartTime)
- )
- }
- iter
- }
-
-
- /**
- * Get the preferred locations where to launch this task.
- */
- override def getPreferredLocations(partition: Partition): Seq[String] = {
- val theSplit = partition.asInstanceOf[CarbonSparkPartition]
- theSplit.locations.filter(_ != "localhost")
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonRawQueryRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonRawQueryRDD.scala b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonRawQueryRDD.scala
deleted file mode 100644
index 5993677..0000000
--- a/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonRawQueryRDD.scala
+++ /dev/null
@@ -1,128 +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.hadoop.conf.Configuration
-import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
-
-import org.carbondata.common.logging.LogServiceFactory
-import org.carbondata.core.iterator.CarbonIterator
-import org.carbondata.query.carbon.executor.QueryExecutorFactory
-import org.carbondata.query.carbon.model.QueryModel
-import org.carbondata.query.carbon.result.BatchRawResult
-import org.carbondata.query.carbon.result.iterator.ChunkRawRowIterartor
-import org.carbondata.query.expression.Expression
-import org.carbondata.spark.{RawKey, RawKeyVal}
-
-
-/**
- * This RDD is used to perform query with raw data, it means it doesn't convert dictionary values
- * to actual data.
- *
- * @param sc
- * @param queryModel
- * @param filterExpression
- * @param keyClass
- * @param conf
- * @param cubeCreationTime
- * @param schemaLastUpdatedTime
- * @param baseStoreLocation
- * @tparam K
- * @tparam V
- */
-class CarbonRawQueryRDD[K, V](
- sc: SparkContext,
- queryModel: QueryModel,
- filterExpression: Expression,
- keyClass: RawKey[K, V],
- @transient conf: Configuration,
- cubeCreationTime: Long,
- schemaLastUpdatedTime: Long,
- baseStoreLocation: String)
- extends CarbonQueryRDD[K, V](sc,
- queryModel,
- filterExpression,
- null,
- conf,
- cubeCreationTime,
- schemaLastUpdatedTime,
- baseStoreLocation) with Logging {
-
-
- override def compute(thepartition: Partition, context: TaskContext): Iterator[(K, V)] = {
- val LOGGER = LogServiceFactory.getLogService(this.getClass().getName());
- val iter = new Iterator[(K, 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)
- // fill table block info
- queryModel.setTableBlockInfos(carbonSparkPartition.tableBlockInfos)
- queryStartTime = System.currentTimeMillis
-
- val carbonPropertiesFilePath = System.getProperty("carbon.properties.filepath", null)
- logInfo("*************************" + carbonPropertiesFilePath)
- if (null == carbonPropertiesFilePath) {
- System.setProperty("carbon.properties.filepath",
- System.getProperty("user.dir") + '/' + "conf" + '/' + "carbon.properties");
- }
- // execute query
- rowIterator = new ChunkRawRowIterartor(
- QueryExecutorFactory.getQueryExecutor(queryModel).execute(queryModel)
- .asInstanceOf[CarbonIterator[BatchRawResult]]).asInstanceOf[CarbonIterator[Array[Any]]]
- }
- } catch {
- case e: Exception =>
- LOGGER.error(e)
- // updateCubeAndLevelCacheStatus(levelCacheKeys)
- if (null != e.getMessage) {
- sys.error("Exception occurred in query execution :: " + e.getMessage)
- } else {
- sys.error("Exception occurred in query execution.Please check logs.")
- }
- }
-
- var havePair = false
- var finished = false
-
- override def hasNext: Boolean = {
- if (!finished && !havePair) {
- finished = (null == rowIterator) || (!rowIterator.hasNext())
- havePair = !finished
- }
- !finished
- }
-
- override def next(): (K, V) = {
- if (!hasNext) {
- throw new java.util.NoSuchElementException("End of stream")
- }
- havePair = false
- val row = rowIterator.next()
- keyClass.getKey(row, null)
- }
-
- logInfo("*************************** Total Time Taken to execute the query in Carbon Side: " +
- (System.currentTimeMillis - queryStartTime)
- )
- }
- iter
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/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
new file mode 100644
index 0000000..a95ae27
--- /dev/null
+++ b/integration/spark/src/main/scala/org/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -0,0 +1,219 @@
+/*
+ * 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 java.util
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapreduce.Job
+import org.apache.spark.{Logging, Partition, SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+
+import org.carbondata.common.logging.LogServiceFactory
+import org.carbondata.core.carbon.datastore.block.TableBlockInfo
+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.expression.Expression
+import org.carbondata.spark.RawKey
+import org.carbondata.spark.load.CarbonLoaderUtil
+import org.carbondata.spark.util.QueryPlanUtil
+
+class CarbonSparkPartition(rddId: Int, val idx: Int,
+ val locations: Array[String],
+ val tableBlockInfos: util.List[TableBlockInfo])
+ extends Partition {
+
+ override val index: Int = idx
+
+ // val serializableHadoopSplit = new SerializableWritable[Array[String]](locations)
+ override def hashCode(): Int = {
+ 41 * (41 + rddId) + idx
+ }
+}
+
+ /**
+ * This RDD is used to perform query on CarbonData file. Before sending tasks to scan
+ * CarbonData file, this RDD will leverage CarbonData's index information to do CarbonData file
+ * level filtering in driver side.
+ */
+class CarbonScanRDD[K, V](
+ sc: SparkContext,
+ queryModel: QueryModel,
+ filterExpression: Expression,
+ keyClass: RawKey[K, V],
+ @transient conf: Configuration,
+ cubeCreationTime: Long,
+ schemaLastUpdatedTime: Long,
+ baseStoreLocation: String)
+ extends RDD[(K, V)](sc, Nil) with Logging {
+
+ val defaultParallelism = sc.defaultParallelism
+
+ override def getPartitions: Array[Partition] = {
+ val startTime = System.currentTimeMillis()
+ val (carbonInputFormat: CarbonInputFormat[RowResult], 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
+ .getResolvedFilter(job.getConfiguration, filterExpression)
+
+ CarbonInputFormat.setFilterPredicates(job.getConfiguration, filterResolver)
+ queryModel.setFilterExpressionResolverTree(filterResolver)
+ }
+ catch {
+ case e: Exception =>
+ LOGGER.error(e)
+ sys.error("Exception occurred in query execution :: " + e.getMessage)
+ }
+ // get splits
+ val splits = carbonInputFormat.getSplits(job)
+ if (!splits.isEmpty) {
+ val carbonInputSplits = splits.asScala.map(_.asInstanceOf[CarbonInputSplit])
+
+ val blockList = carbonInputSplits.map(inputSplit =>
+ new TableBlockInfo(inputSplit.getPath.toString,
+ inputSplit.getStart, inputSplit.getSegmentId,
+ inputSplit.getLocations, inputSplit.getLength
+ )
+ )
+ if (blockList.nonEmpty) {
+ // group blocks to nodes, tasks
+ val nodeBlockMapping =
+ CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1, defaultParallelism)
+
+ var i = 0
+ // Create Spark Partition for each task and assign blocks
+ nodeBlockMapping.asScala.foreach { entry =>
+ entry._2.asScala.foreach { blocksPerTask =>
+ if (blocksPerTask.size() != 0) {
+ result.add(new CarbonSparkPartition(id, i, Seq(entry._1).toArray, blocksPerTask))
+ i += 1
+ }
+ }
+ }
+ val noOfBlocks = blockList.size
+ val noOfNodes = nodeBlockMapping.size
+ val noOfTasks = result.size()
+ logInfo(s"Identified no.of.Blocks: $noOfBlocks,"
+ + s"parallelism: $defaultParallelism , " +
+ s"no.of.nodes: $noOfNodes, no.of.tasks: $noOfTasks"
+ )
+ logInfo("Time taken to identify Blocks to scan : " +
+ (System.currentTimeMillis() - startTime)
+ )
+ result.asScala.foreach { r =>
+ val cp = r.asInstanceOf[CarbonSparkPartition]
+ logInfo(s"Node : " + cp.locations.toSeq.mkString(",")
+ + ", No.Of Blocks : " + cp.tableBlockInfos.size()
+ )
+ }
+ } else {
+ logInfo("No blocks identified to scan")
+ val nodesPerBlock = new util.ArrayList[TableBlockInfo]()
+ result.add(new CarbonSparkPartition(id, 0, Seq("").toArray, nodesPerBlock))
+ }
+ }
+ else {
+ logInfo("No valid segments found to scan")
+ val nodesPerBlock = new util.ArrayList[TableBlockInfo]()
+ result.add(new CarbonSparkPartition(id, 0, Seq("").toArray, nodesPerBlock))
+ }
+ 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)] {
+ 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)
+ // fill table block info
+ queryModel.setTableBlockInfos(carbonSparkPartition.tableBlockInfos)
+ queryStartTime = System.currentTimeMillis
+
+ val carbonPropertiesFilePath = System.getProperty("carbon.properties.filepath", null)
+ logInfo("*************************" + carbonPropertiesFilePath)
+ if (null == carbonPropertiesFilePath) {
+ System.setProperty("carbon.properties.filepath",
+ System.getProperty("user.dir") + '/' + "conf" + '/' + "carbon.properties");
+ }
+ // execute query
+ rowIterator = new ChunkRawRowIterartor(
+ QueryExecutorFactory.getQueryExecutor(queryModel).execute(queryModel)
+ .asInstanceOf[CarbonIterator[BatchRawResult]])
+ .asInstanceOf[CarbonIterator[Array[Any]]]
+ }
+ } catch {
+ case e: Exception =>
+ LOGGER.error(e)
+ if (null != e.getMessage) {
+ sys.error("Exception occurred in query execution :: " + e.getMessage)
+ } else {
+ sys.error("Exception occurred in query execution.Please check logs.")
+ }
+ }
+
+ var havePair = false
+ var finished = false
+
+ override def hasNext: Boolean = {
+ if (!finished && !havePair) {
+ finished = (null == rowIterator) || (!rowIterator.hasNext())
+ havePair = !finished
+ }
+ !finished
+ }
+
+ override def next(): (K, V) = {
+ if (!hasNext) {
+ throw new java.util.NoSuchElementException("End of stream")
+ }
+ havePair = false
+ val row = rowIterator.next()
+ keyClass.getKey(row, null)
+ }
+
+ logInfo("********************** Total Time Taken to execute the query in Carbon Side: " +
+ (System.currentTimeMillis - queryStartTime)
+ )
+ }
+ iter
+ }
+
+ /**
+ * Get the preferred locations where to launch this task.
+ */
+ override def getPreferredLocations(partition: Partition): Seq[String] = {
+ val theSplit = partition.asInstanceOf[CarbonSparkPartition]
+ theSplit.locations.filter(_ != "localhost")
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/a83dba34/integration/spark/src/test/scala/org/carbondata/spark/testsuite/joinquery/EquiJoinTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/test/scala/org/carbondata/spark/testsuite/joinquery/EquiJoinTestCase.scala b/integration/spark/src/test/scala/org/carbondata/spark/testsuite/joinquery/EquiJoinTestCase.scala
deleted file mode 100644
index 3943352..0000000
--- a/integration/spark/src/test/scala/org/carbondata/spark/testsuite/joinquery/EquiJoinTestCase.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-package org.carbondata.spark.testsuite.joinquery
-
-import org.apache.spark.sql.common.util.CarbonHiveContext._
-import org.apache.spark.sql.common.util.QueryTest
-import org.scalatest.BeforeAndAfterAll
-import org.apache.spark.sql.execution.joins.BroadCastFilterPushJoin
-
-class EquiJoinTestCase extends QueryTest with BeforeAndAfterAll {
- override def beforeAll {
- //loading to hive table
- sql("create table employee_hive (empid string,empname string,mobilename string,mobilecolor string,salary int)row format delimited fields terminated by ','")
- sql("create table mobile_hive (mobileid string,mobilename string, mobilecolor string, sales int)row format delimited fields terminated by ','");
- sql("LOAD DATA LOCAL INPATH './src/test/resources/join/employee.csv' into table employee_hive")
- sql("LOAD DATA LOCAL INPATH './src/test/resources/join/mobile.csv' into table mobile_hive")
- //loading to carbon table
- sql("create table employee (empid string,empname string,mobilename string,mobilecolor string,salary int) stored by 'org.apache.carbondata.format'")
- sql("create table mobile (mobileid string,mobilename string, mobilecolor string, sales int) stored by 'org.apache.carbondata.format'");
- sql("LOAD DATA LOCAL INPATH './src/test/resources/join/employee.csv' into table employee options('FILEHEADER'='empid,empname,mobilename,mobilecolor,salary')")
- sql("LOAD DATA LOCAL INPATH './src/test/resources/join/mobile.csv' into table mobile options('FILEHEADER'='mobileid,mobilename,mobilecolor,sales')")
- }
-
- test("test equijoin query") {
- val df = sql("select employee.empname,mobile.mobilename from employee,mobile where employee.mobilename = mobile.mobilename")
- var broadcastJoinExists = false
- df.queryExecution.sparkPlan.collect {
- case bcf: BroadCastFilterPushJoin =>
- broadcastJoinExists = true
- }
- if (!broadcastJoinExists) {
- assert(false)
- }
- checkAnswer(df,
- sql("select employee_hive.empname,mobile_hive.mobilename from employee_hive,mobile_hive where employee_hive.mobilename = mobile_hive.mobilename"))
- }
- override def afterAll {
- sql("drop table employee_hive")
- sql("drop table mobile_hive")
- sql("drop table employee")
- sql("drop table mobile")
- }
-}
\ No newline at end of file