Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1464#discussion_r149001761
--- Diff:
integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
---
@@ -49,70 +50,109 @@ object PreAggregateUtil {
def getParentCarbonTable(plan: LogicalPlan): CarbonTable = {
plan match {
- case Aggregate(_, aExp, SubqueryAlias(_, l: LogicalRelation, _))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
-
l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.metaData.carbonTable
+ case Aggregate(_, _, SubqueryAlias(_, logicalRelation:
LogicalRelation, _))
+ if
logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
+
logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].
+ carbonRelation.metaData.carbonTable
+ case Aggregate(_, _, logicalRelation: LogicalRelation)
+ if
logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
+
logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].
+ carbonRelation.metaData.carbonTable
case _ => throw new MalformedCarbonCommandException("table does not
exist")
}
}
/**
* Below method will be used to validate the select plan
* and get the required fields from select plan
- * Currently only aggregate query is support any other type of query will
- * fail
+ * Currently only aggregate query is support any other type of query
will fail
+ *
* @param plan
* @param selectStmt
* @return list of fields
*/
def validateActualSelectPlanAndGetAttrubites(plan: LogicalPlan,
selectStmt: String): scala.collection.mutable.LinkedHashMap[Field,
DataMapField] = {
- val fieldToDataMapFieldMap =
scala.collection.mutable.LinkedHashMap.empty[Field, DataMapField]
plan match {
- case Aggregate(_, aExp, SubqueryAlias(_, l: LogicalRelation, _))
- if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
- val carbonTable =
l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation
- .metaData.carbonTable
- val parentTableName =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
- .getTableName
- val parentDatabaseName =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
- .getDatabaseName
- val parentTableId =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
- .getTableId
- if
(!carbonTable.getTableInfo.getParentRelationIdentifiers.isEmpty) {
+ case Aggregate(groupByExp, aggExp, SubqueryAlias(_, logicalRelation:
LogicalRelation, _)) =>
+ getFieldsFromPlan(groupByExp, aggExp, logicalRelation, selectStmt)
+ case Aggregate(groupByExp, aggExp, logicalRelation: LogicalRelation)
=>
+ getFieldsFromPlan(groupByExp, aggExp, logicalRelation, selectStmt)
+ case _ =>
+ throw new MalformedCarbonCommandException(s"Unsupported Select
Statement:${ selectStmt } ")
+ }
+ }
+
+ /**
+ * Below method will be used to get the fields from expressions
+ * @param groupByExp
+ * grouping expression
+ * @param aggExp
+ * aggregate expression
+ * @param logicalRelation
+ * logical relation
+ * @param selectStmt
+ * select statement
+ * @return fields from expressions
+ */
+ def getFieldsFromPlan(groupByExp: Seq[Expression],
+ aggExp: Seq[NamedExpression], logicalRelation: LogicalRelation,
selectStmt: String):
+ scala.collection.mutable.LinkedHashMap[Field, DataMapField] = {
+ val fieldToDataMapFieldMap =
scala.collection.mutable.LinkedHashMap.empty[Field, DataMapField]
+ if
(!logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+ throw new MalformedCarbonCommandException("Un-supported table")
+ }
+ val carbonTable = logicalRelation.relation.
+ asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation
+ .metaData.carbonTable
+ val parentTableName =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
+ .getTableName
+ val parentDatabaseName =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
+ .getDatabaseName
+ val parentTableId =
carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
+ .getTableId
+ if (!carbonTable.getTableInfo.getParentRelationIdentifiers.isEmpty) {
+ throw new MalformedCarbonCommandException(
+ "Pre Aggregation is not supported on Pre-Aggregated Table")
+ }
+ groupByExp.map {
+ case attr: AttributeReference =>
+ fieldToDataMapFieldMap += getField(attr.name,
+ attr.dataType,
+ parentColumnId = carbonTable.getColumnByName(parentTableName,
attr.name).getColumnId,
+ parentTableName = parentTableName,
+ parentDatabaseName = parentDatabaseName, parentTableId =
parentTableId)
+ case _ =>
+ throw new MalformedCarbonCommandException(s"Unsupported Select
Statement:${
--- End diff --
it is unsupported function exception
---
