You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@carbondata.apache.org by kunal642 <gi...@git.apache.org> on 2017/11/16 12:35:26 UTC
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
GitHub user kunal642 opened a pull request:
https://github.com/apache/carbondata/pull/1508
[CARBONDATA-1738] Block direct insert/load on pre-aggregate table
Block load/insert on pre-aggregate table
- [X] Any interfaces changed?
No
- [X] Any backward compatibility impacted?
No
- [X] Document update required?
No
- [X] Testing done
Added test case
- [X] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kunal642/carbondata block_load_on_preagg
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/carbondata/pull/1508.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1508
----
commit 9d48d35a93116ac53951ad6b9a3dc6e110075d36
Author: kunal642 <ku...@gmail.com>
Date: 2017-11-16T12:28:50Z
blocked direct insert/load on pre-aggregate table
----
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1498/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1288/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1210/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1850/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153723385
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -49,13 +50,54 @@ object LoadPostAggregateListener extends OperationEventListener {
carbonLoadModel.getTableName, "false")
val childTableName = dataMapSchema.getRelationIdentifier.getTableName
val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
- val selectQuery = dataMapSchema.getProperties.get("CHILD_SELECT QUERY")
- sparkSession.sql(s"insert into $childDatabaseName.$childTableName $selectQuery")
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
+ .addPreAggLoadFunction(s"${ dataMapSchema.getProperties.get("CHILD_SELECT QUERY") } "))
+ .drop("preAggLoad")
+ val headers = dataMapSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
+ .mkString(",")
+ try {
+ LoadTableCommand(Some(childDatabaseName),
+ childTableName,
+ null,
+ Nil,
+ Map("fileheader" -> headers),
+ isOverwriteTable = false,
+ dataFrame = Some(childDataFrame),
+ internalOptions = Map(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true"))
+ .run(sparkSession)
+ } finally {
+ CarbonSession.threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
+ CarbonSession.threadUnset(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
+ }
}
}
}
}
+object LoadPreAggregateTablePreListener extends OperationEventListener {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override def onEvent(event: Event, operationContext: OperationContext): Unit = {
+ val loadEvent = event.asInstanceOf[LoadTablePreExecutionEvent]
+ val carbonLoadModel = loadEvent.carbonLoadModel
+ val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val isInternalLoadCall = carbonLoadModel.isAggLoadRequest
+ if (table.isChildDataMap && !isInternalLoadCall) {
+ throw new UnsupportedOperationException(
+ "Cannot insert/load data directly into pre-aggregate table")
+ }
+
--- End diff --
remove line
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r152597828
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/execution/command/CarbonHiveCommands.scala ---
@@ -95,8 +95,14 @@ object CarbonSetCommand {
}
} else if (key.startsWith(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS)) {
sessionParams.addProperty(key.toLowerCase(), value)
+ } else if (key.startsWith(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL)) {
--- End diff --
I don't think it is required to use `set command` for this internal call. we are not going to give the option to load the aggregate table as it may corrupt the table.
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153722006
--- Diff: integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/DataLoadingUtil.scala ---
@@ -221,7 +221,6 @@ object DataLoadingUtil {
ValidateUtil.validateDateTimeFormat(timestampformat, "TimestampFormat")
ValidateUtil.validateDateTimeFormat(dateFormat, "DateFormat")
ValidateUtil.validateSortScope(table, sort_scope)
-
--- End diff --
don 't change file unnecessarly
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1308/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 1.6, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/392/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2053/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1326/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1324/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1802/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153068092
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -49,13 +50,52 @@ object LoadPostAggregateListener extends OperationEventListener {
carbonLoadModel.getTableName, "false")
val childTableName = dataMapSchema.getRelationIdentifier.getTableName
val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
- val selectQuery = dataMapSchema.getProperties.get("CHILD_SELECT QUERY")
- sparkSession.sql(s"insert into $childDatabaseName.$childTableName $selectQuery")
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
+ .addPreAggLoadFunction(s"${dataMapSchema.getProperties.get("CHILD_SELECT QUERY")} ")).drop("preAggLoad")
+ val headers = dataMapSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
+ .mkString(",")
+ try {
+ LoadTableCommand(Some(childDatabaseName),
+ childTableName,
+ null,
+ Nil,
+ Map("fileheader" -> headers),
+ isOverwriteTable = false,
+ dataFrame = Some(childDataFrame),
+ internalOptions = Map(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true")).run(sparkSession)
+ } finally {
+ CarbonSession.threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
+ CarbonSession.threadUnset(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
--- End diff --
Indentation is wrong, format properly
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153724471
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala ---
@@ -570,6 +570,14 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
}
}
+ def addPreAggLoadFunction(sql: String): String = {
+ addPreAggLoad(new lexical.Scanner(sql.toLowerCase)) match {
+ case Success(query, _) => query
+ case _ => throw new MalformedCarbonCommandException(
--- End diff --
move down line
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153067591
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala ---
@@ -94,10 +95,22 @@ case class CreatePreAggregateTableCommand(
dmproperties.foreach(f => childSchema.getProperties.put(f._1, f._2))
// updating the parent table about child table
PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
- val loadAvailable = PreAggregateUtil.checkMainTableLoad(parentTable)
- if (loadAvailable) {
- sparkSession.sql(
- s"insert into ${ tableModel.databaseName }.${ tableModel.tableName } $queryString")
+ val availableLoads = PreAggregateUtil.checkMainTableLoad(parentTable)
+ if (availableLoads) {
+ val headers = childSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
+ .mkString(",")
+ val childDataFrame = Dataset.ofRows(sparkSession, new CarbonSpark2SqlParser()
+ .parse(s"insert into ${ tableModel.databaseName }.${
+ tableModel.tableName} $queryString"))
--- End diff --
why it is not just `queryString`? why insertinto required here as you are already using load command
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1801/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1182/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1670/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
retest sdv please
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1482/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1819/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
Posted by gvramana <gi...@git.apache.org>.
Github user gvramana commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r151408491
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -51,11 +51,39 @@ object LoadPostAggregateListener extends OperationEventListener {
val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
val selectQuery = dataMapSchema.getProperties.get("CHILD_SELECT QUERY")
sparkSession.sql(s"insert into $childDatabaseName.$childTableName $selectQuery")
+ CarbonSession.threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
--- End diff --
add this unset to finally
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153067934
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala ---
@@ -736,6 +739,45 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
}
+object CarbonPreAggregateDataLoadingRules extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+
+ plan transform {
+ case aggregate@Aggregate(_, aExp, _) =>
+ val isLoadPlan = aExp.exists(_.name.equalsIgnoreCase("preAggLoad"))
+ if (aExp.exists(_.name.equalsIgnoreCase("PreAgg"))) {
--- End diff --
move this `if` condition to the `case` block
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1485/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153722722
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala ---
@@ -94,10 +95,21 @@ case class CreatePreAggregateTableCommand(
dmproperties.foreach(f => childSchema.getProperties.put(f._1, f._2))
// updating the parent table about child table
PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
- val loadAvailable = PreAggregateUtil.checkMainTableLoad(parentTable)
- if (loadAvailable) {
- sparkSession.sql(
- s"insert into ${ tableModel.databaseName }.${ tableModel.tableName } $queryString")
+ val availableLoads = PreAggregateUtil.checkMainTableLoad(parentTable)
+ if (availableLoads) {
+ val headers = childSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
+ .mkString(",")
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
--- End diff --
Move down after (
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1405/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2057/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2031/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by kunal642 <gi...@git.apache.org>.
Github user kunal642 commented on the issue:
https://github.com/apache/carbondata/pull/1508
retest this please
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153723895
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala ---
@@ -751,6 +754,58 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
}
+object CarbonPreAggregateDataLoadingRules extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+
+ plan transform {
+
+ case aggregate@Aggregate(_, aExp, _) if validateAggregateExpressions(aExp) =>
+ val newExpressions = aExp.flatMap {
+ case alias@Alias(attrExpression: AggregateExpression, _) =>
+ attrExpression.aggregateFunction match {
+ case Average(attr: AttributeReference) =>
+ Seq(Alias(attrExpression
+ .copy(aggregateFunction = Sum(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_sum")(),
+ Alias(attrExpression
+ .copy(aggregateFunction = Count(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_count")())
+ case Average(cast@Cast(attr: AttributeReference, _)) =>
+ Seq(Alias(attrExpression
+ .copy(aggregateFunction = Sum(cast),
+ resultId = NamedExpression.newExprId),
+ attr.name + "_sum")(),
+ Alias(attrExpression
+ .copy(aggregateFunction = Count(cast),
+ resultId = NamedExpression.newExprId), attr.name + "_count")())
--- End diff --
Please format it properly
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1887/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1466/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1479/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1900/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1289/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1469/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
retest sdv please
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
LGTM
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
Posted by gvramana <gi...@git.apache.org>.
Github user gvramana commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r151410067
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -51,11 +51,39 @@ object LoadPostAggregateListener extends OperationEventListener {
val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
val selectQuery = dataMapSchema.getProperties.get("CHILD_SELECT QUERY")
sparkSession.sql(s"insert into $childDatabaseName.$childTableName $selectQuery")
+ CarbonSession.threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
+ CarbonSession.threadUnset(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
+ carbonLoadModel.getDatabaseName + "." +
+ carbonLoadModel.getTableName)
}
}
}
}
+object LoadPreAggregateTablePreListener extends OperationEventListener {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override def onEvent(event: Event, operationContext: OperationContext): Unit = {
+ val loadEvent = event.asInstanceOf[LoadTablePreExecutionEvent]
+ val carbonLoadModel = loadEvent.carbonLoadModel
+ val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
+ val isInternalLoadCall = carbonSessionInfo.getSessionParams.getAll.keySet().asScala
--- End diff --
Use specifc parameter instead of using this VALIDATE_CARBON_INPUT_SEGMENTS
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
Posted by jackylk <gi...@git.apache.org>.
Github user jackylk commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r151637632
--- Diff: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java ---
@@ -67,6 +67,12 @@
public static final String VALIDATE_CARBON_INPUT_SEGMENTS = "validate.carbon.input.segments.";
/**
+ * Whether load/insert command is fired internally or by the user.
+ * Used to block load/insert on pre-aggregate if fired by user
+ */
+ public static final String IS_INTERNAL_LOAD_CALL = "is.internal.load.call";
--- End diff --
Seems no testcase for this option. And the option name should start with `carbon`
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1215/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153067988
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala ---
@@ -736,6 +739,45 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
}
+object CarbonPreAggregateDataLoadingRules extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+
+ plan transform {
+ case aggregate@Aggregate(_, aExp, _) =>
+ val isLoadPlan = aExp.exists(_.name.equalsIgnoreCase("preAggLoad"))
--- End diff --
Even this condition also move to `case ` block
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153722923
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -49,13 +50,54 @@ object LoadPostAggregateListener extends OperationEventListener {
carbonLoadModel.getTableName, "false")
val childTableName = dataMapSchema.getRelationIdentifier.getTableName
val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
- val selectQuery = dataMapSchema.getProperties.get("CHILD_SELECT QUERY")
- sparkSession.sql(s"insert into $childDatabaseName.$childTableName $selectQuery")
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
+ .addPreAggLoadFunction(s"${ dataMapSchema.getProperties.get("CHILD_SELECT QUERY") } "))
+ .drop("preAggLoad")
+ val headers = dataMapSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
--- End diff --
move down
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
Posted by kunal642 <gi...@git.apache.org>.
Github user kunal642 commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r151638473
--- Diff: core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java ---
@@ -67,6 +67,12 @@
public static final String VALIDATE_CARBON_INPUT_SEGMENTS = "validate.carbon.input.segments.";
/**
+ * Whether load/insert command is fired internally or by the user.
+ * Used to block load/insert on pre-aggregate if fired by user
+ */
+ public static final String IS_INTERNAL_LOAD_CALL = "is.internal.load.call";
--- End diff --
This option/property will not be exposed to the user. It will be set by the post load listener to know whether the load is fired by the user or is it an internal call.
Test case is added in TestPreAggregateLoad
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1668/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
retest sdv please
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1784/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1477/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153724030
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala ---
@@ -751,6 +754,58 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule
}
}
+object CarbonPreAggregateDataLoadingRules extends Rule[LogicalPlan] {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+
+ plan transform {
+
+ case aggregate@Aggregate(_, aExp, _) if validateAggregateExpressions(aExp) =>
+ val newExpressions = aExp.flatMap {
+ case alias@Alias(attrExpression: AggregateExpression, _) =>
+ attrExpression.aggregateFunction match {
+ case Average(attr: AttributeReference) =>
+ Seq(Alias(attrExpression
+ .copy(aggregateFunction = Sum(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_sum")(),
+ Alias(attrExpression
+ .copy(aggregateFunction = Count(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_count")())
+ case Average(cast@Cast(attr: AttributeReference, _)) =>
+ Seq(Alias(attrExpression
+ .copy(aggregateFunction = Sum(cast),
+ resultId = NamedExpression.newExprId),
+ attr.name + "_sum")(),
+ Alias(attrExpression
+ .copy(aggregateFunction = Count(cast),
+ resultId = NamedExpression.newExprId), attr.name + "_count")())
+ case _ => Seq(alias)
+ }
+ case namedExpr: NamedExpression => Seq(namedExpr)
+ }
+ aggregate.copy(aggregateExpressions = newExpressions.asInstanceOf[Seq[NamedExpression]])
+ case plan: LogicalPlan => plan
+ }
+ }
+
+ /**
+ * Called by PreAggregateLoadingRules to validate if plan is valid for applying rules or not.
+ * If the plan has PreAggLoad i.e Loading UDF and does not have PreAgg i.e Query UDF then it is
+ * valid.
+ *
+ * @param namedExpression
+ * @return
+ */
+ private def validateAggregateExpressions(namedExpression: Seq[NamedExpression]): Boolean = {
+ val filteredExpressions = namedExpression.filterNot(_.isInstanceOf[UnresolvedAlias])
+ filteredExpressions
+ .exists {
+ expr => !expr.name.equalsIgnoreCase("PreAgg") &&
+ expr.name.equalsIgnoreCase("preAggLoad")
--- End diff --
format properly
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] Block direct insert/load on...
Posted by gvramana <gi...@git.apache.org>.
Github user gvramana commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r151407441
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala ---
@@ -212,6 +212,18 @@ object CarbonSession {
ThreadLocalSessionInfo.setCarbonSessionInfo(currentThreadSessionInfo)
}
+ def threadUnset(key: String): Unit = {
+ var currentThreadSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
+ if (currentThreadSessionInfo == null) {
--- End diff --
If null then not required to unset
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1240/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by kunal642 <gi...@git.apache.org>.
Github user kunal642 commented on the issue:
https://github.com/apache/carbondata/pull/1508
@ravipesala Handled all the review comments
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1664/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1884/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1883/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1924/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153068009
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala ---
@@ -55,6 +55,7 @@ class CarbonEnv {
// added for handling preaggregate table creation. when user will fire create ddl for
// create table we are adding a udf so no need to apply PreAggregate rules.
sparkSession.udf.register("preAgg", () => "")
+ sparkSession.udf.register("preAggLoad", () => "")
--- End diff --
Add comment about usage of this udf
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by kunal642 <gi...@git.apache.org>.
Github user kunal642 commented on the issue:
https://github.com/apache/carbondata/pull/1508
@ravipesala Please review
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
LGTM
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1903/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.2, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/400/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2055/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1639/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1915/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/carbondata/pull/1508
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1508
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/1896/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153722623
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala ---
@@ -94,10 +95,21 @@ case class CreatePreAggregateTableCommand(
dmproperties.foreach(f => childSchema.getProperties.put(f._1, f._2))
// updating the parent table about child table
PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
- val loadAvailable = PreAggregateUtil.checkMainTableLoad(parentTable)
- if (loadAvailable) {
- sparkSession.sql(
- s"insert into ${ tableModel.databaseName }.${ tableModel.tableName } $queryString")
+ val availableLoads = PreAggregateUtil.checkMainTableLoad(parentTable)
+ if (availableLoads) {
+ val headers = childSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
--- End diff --
Move line done
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1522/
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] [PreAgg] Block direct insert/load ...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1641/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153068121
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala ---
@@ -94,10 +95,22 @@ case class CreatePreAggregateTableCommand(
dmproperties.foreach(f => childSchema.getProperties.put(f._1, f._2))
// updating the parent table about child table
PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
- val loadAvailable = PreAggregateUtil.checkMainTableLoad(parentTable)
- if (loadAvailable) {
- sparkSession.sql(
- s"insert into ${ tableModel.databaseName }.${ tableModel.tableName } $queryString")
+ val availableLoads = PreAggregateUtil.checkMainTableLoad(parentTable)
+ if (availableLoads) {
+ val headers = childSchema.getChildSchema.getListOfColumns.asScala.map(_.getColumnName)
+ .mkString(",")
+ val childDataFrame = Dataset.ofRows(sparkSession, new CarbonSpark2SqlParser()
+ .parse(s"insert into ${ tableModel.databaseName }.${
+ tableModel.tableName} $queryString"))
--- End diff --
Here why don't use preAggLoad UDF?
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by kunal642 <gi...@git.apache.org>.
Github user kunal642 commented on the issue:
https://github.com/apache/carbondata/pull/1508
retest this please
---
[GitHub] carbondata issue #1508: [CARBONDATA-1738] Block direct insert/load on pre-ag...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1508
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1313/
---
[GitHub] carbondata pull request #1508: [CARBONDATA-1738] [PreAgg] Block direct inser...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1508#discussion_r153723567
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala ---
@@ -253,8 +260,8 @@ object PreAggregateUtil {
carbonTable.getColumnByName(parentTableName, attr.name).getColumnId,
parentTableName,
parentDatabaseName, parentTableId = parentTableId)
- case _ =>
- throw new MalformedCarbonCommandException("Un-Supported Aggregation Type")
+ case a@_ =>
--- End diff --
Keep as case others
---