You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@carbondata.apache.org by kumarvishal09 <gi...@git.apache.org> on 2018/01/19 13:52:05 UTC
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
GitHub user kumarvishal09 opened a pull request:
https://github.com/apache/carbondata/pull/1837
[WIP] Refactored code segregated process meta and process data in load command
Be sure to do all of the following checklist to help us incorporate
your contribution quickly and easily:
- [ ] Any interfaces changed?
- [ ] Any backward compatibility impacted?
- [ ] Document update required?
- [ ] Testing done
Please provide details on
- Whether new unit test cases have been added or why no new tests are required?
- How it is tested? Please attach test report.
- Is it a performance related change? Please attach the performance test report.
- Any additional information to help reviewers in testing this change.
- [ ] 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/kumarvishal09/incubator-carbondata CodeRefactor
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/carbondata/pull/1837.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 #1837
----
commit f8e6fc8b036a5fde537d2db135d8860fd2c50c95
Author: kumarvishal <ku...@...>
Date: 2018-01-19T11:52:28Z
Refactor code segregated load from metadata
----
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1881/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163760694
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonCreateTableAsSelectCommand.scala ---
@@ -42,15 +42,11 @@ case class CarbonCreateTableAsSelectCommand(
ifNotExistsSet: Boolean = false,
tableLocation: Option[String] = None) extends AtomicRunnableCommand {
- /**
- * variable to be used for insert into command for checking whether the
- * table is created newly or already existed
- */
- var isTableCreated: Boolean = false
-
+ var loadCommand: CarbonInsertIntoCommand = _
--- End diff --
enter new line
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3097/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3015/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3007/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3092/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3001/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1865/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1777/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by kumarvishal09 <gi...@git.apache.org>.
Github user kumarvishal09 commented on the issue:
https://github.com/apache/carbondata/pull/1837
@ravipesala I am working on auto merger after that i will update the PR
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3093/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3113/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1884/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3079/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Success with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1888/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3016/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
LGTM
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3008/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163758021
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -20,18 +20,141 @@ package org.apache.spark.sql.execution.command.preaaggregate
import scala.collection.JavaConverters._
import scala.collection.mutable
-import org.apache.spark.sql.CarbonEnv
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.AlterTableModel
-import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
+import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonLoadDataCommand}
+import org.apache.spark.sql.parser.CarbonSpark2SqlParser
-import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.schema.table.AggregationDataMapSchema
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.events._
-import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+object CompactionProcessMetaListener extends OperationEventListener {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override protected def onEvent(event: Event,
+ operationContext: OperationContext): Unit = {
+ val sparkSession = SparkSession.getActiveSession.get
+ val tableEvent = event.asInstanceOf[LoadMetadataEvent]
+ val table = tableEvent.getCarbonTable
+ if (!table.isChildDataMap && CarbonUtil.hasAggregationDataMap(table)) {
+ val aggregationDataMapList = table.getTableInfo.getDataMapSchemaList.asScala
+ .filter(_.isInstanceOf[AggregationDataMapSchema])
+ .asInstanceOf[mutable.ArrayBuffer[AggregationDataMapSchema]]
+ for (dataMapSchema: AggregationDataMapSchema <- aggregationDataMapList) {
+ val childTableName = dataMapSchema.getRelationIdentifier.getTableName
+ val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
+ // Creating a new query string to insert data into pre-aggregate table from that same table.
+ // For example: To compact preaggtable1 we can fire a query like insert into preaggtable1
+ // select * from preaggtable1
+ // The following code will generate the select query with a load UDF that will be used to
+ // apply DataLoadingRules
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
+ // adding the aggregation load UDF
+ .addPreAggLoadFunction(
+ // creating the select query on the bases on table schema
+ PreAggregateUtil.createChildSelectQuery(
+ dataMapSchema.getChildSchema, table.getDatabaseName))).drop("preAggLoad")
+ val loadCommand = PreAggregateUtil.createLoadCommandForChild(
+ dataMapSchema.getChildSchema.getListOfColumns,
+ TableIdentifier(childTableName, Some(childDatabaseName)),
+ childDataFrame,
+ false,
+ sparkSession)
+ operationContext
+ .setProperty(dataMapSchema.getChildSchema.getTableName + "_Compaction", loadCommand)
+ }
+ } else {
+ val childTableName = table.getTableName
+ val childDatabaseName = table.getDatabaseName
+ // Creating a new query string to insert data into pre-aggregate table from that same table.
+ // For example: To compact preaggtable1 we can fire a query like insert into preaggtable1
+ // select * from preaggtable1
+ // The following code will generate the select query with a load UDF that will be used to
+ // apply DataLoadingRules
+ val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
+ // adding the aggregation load UDF
+ .addPreAggLoadFunction(
+ // creating the select query on the bases on table schema
+ PreAggregateUtil.createChildSelectQuery(
+ table.getTableInfo.getFactTable, table.getDatabaseName))).drop("preAggLoad")
+ val loadCommand = PreAggregateUtil.createLoadCommandForChild(
+ table.getTableInfo.getFactTable.getListOfColumns,
+ TableIdentifier(childTableName, Some(childDatabaseName)),
+ childDataFrame,
+ false,
+ sparkSession)
+ operationContext.setProperty(table.getTableName + "_Compaction", loadCommand)
+ }
+ }
+}
+object LoadProcessMetaListener extends OperationEventListener {
--- End diff --
Add comment
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
@kumarvishal09 Please remove WIP and add jira id if you are done with this PR
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Success with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1782/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3012/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Success with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1778/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1781/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163757666
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala ---
@@ -649,6 +659,7 @@ case class CarbonLoadDataCommand(
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_SERIALIZATION_NULL_FORMAT)
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION)
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD)
+ CarbonSession.threadUnset("partition.opeartioncontext")
--- End diff --
Please take this to CarbonCommonConstants
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3012/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3111/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163751453
--- Diff: integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala ---
@@ -57,30 +58,16 @@ class AggregateDataMapCompactor(carbonLoadModel: CarbonLoadModel,
CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
carbonLoadModel.getDatabaseName + "." +
carbonLoadModel.getTableName, "false")
- val headers = carbonTable.getTableInfo.getFactTable.getListOfColumns.asScala
- .map(_.getColumnName).mkString(",")
- // Creating a new query string to insert data into pre-aggregate table from that same table.
- // For example: To compact preaggtable1 we can fire a query like insert into preaggtable1
- // select * from preaggtable1
- // The following code will generate the select query with a load UDF that will be used to
- // apply DataLoadingRules
- val childDataFrame = sqlContext.sparkSession.sql(new CarbonSpark2SqlParser()
- // adding the aggregation load UDF
- .addPreAggLoadFunction(
- // creating the select query on the bases on table schema
- PreAggregateUtil.createChildSelectQuery(
- carbonTable.getTableInfo.getFactTable, carbonTable.getDatabaseName))).drop("preAggLoad")
+ CarbonSession.updateSessionInfoToCurrentThread(sqlContext.sparkSession)
+ val loadCommand = operationContext.getProperty(carbonTable.getTableName + "_Compaction")
+ .asInstanceOf[CarbonLoadDataCommand]
try {
- CarbonLoadDataCommand(
- Some(carbonTable.getDatabaseName),
- carbonTable.getTableName,
- null,
- Nil,
- Map("fileheader" -> headers),
- isOverwriteTable = false,
- dataFrame = Some(childDataFrame),
- internalOptions = Map(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true",
- "mergedSegmentName" -> mergedLoadName)).run(sqlContext.sparkSession)
+ val newInternalOptions = loadCommand.internalOptions ++
+ Map("mergedSegmentName" -> mergedLoadName)
+ loadCommand.internalOptions = newInternalOptions
+ loadCommand.dataFrame = Some(PreAggregateUtil
--- End diff --
Please correct the format
```
loadCommand.dataFrame =
Some(PreAggregateUtil.getDataFrame(sqlContext.sparkSession, loadCommand.logicalPlan.get))
```
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163757600
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala ---
@@ -87,12 +84,41 @@ case class CarbonLoadDataCommand(
options: scala.collection.immutable.Map[String, String],
isOverwriteTable: Boolean,
var inputSqlString: String = null,
- dataFrame: Option[DataFrame] = None,
+ var dataFrame: Option[DataFrame] = None,
updateModel: Option[UpdateTableModel] = None,
var tableInfoOp: Option[TableInfo] = None,
- internalOptions: Map[String, String] = Map.empty,
- partition: Map[String, Option[String]] = Map.empty) extends DataCommand {
+ var internalOptions: Map[String, String] = Map.empty,
+ partition: Map[String, Option[String]] = Map.empty,
+ logicalPlan: Option[LogicalPlan] = None,
+ var operationContext: OperationContext = new OperationContext) extends AtomicRunnableCommand {
+ var table: CarbonTable = _
+
+ override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+ val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+ val dbName = CarbonEnv.getDatabaseName(databaseNameOp)(sparkSession)
+ table = if (tableInfoOp.isDefined) {
+ CarbonTable.buildFromTableInfo(tableInfoOp.get)
+ } else {
+ val relation = CarbonEnv.getInstance(sparkSession).carbonMetastore
+ .lookupRelation(Option(dbName), tableName)(sparkSession).asInstanceOf[CarbonRelation]
+ if (relation == null) {
+ throw new NoSuchTableException(dbName, tableName)
+ }
+ if (null == relation.carbonTable) {
+ LOGGER.error(s"Data loading failed. table not found: $dbName.$tableName")
+ LOGGER.audit(s"Data loading failed. table not found: $dbName.$tableName")
+ throw new NoSuchTableException(dbName, tableName)
+ }
+ relation.carbonTable
+ }
+ operationContext.setProperty("isOverwrite", isOverwriteTable)
+ if (CarbonUtil.hasAggregationDataMap(table)) {
+ val loadMetadataEvent = new LoadMetadataEvent(table, false)
+ OperationListenerBus.getInstance().fireEvent(loadMetadataEvent, operationContext)
--- End diff --
Always fire the event here, handle if condition inside the registered listner
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163760282
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala ---
@@ -900,4 +876,33 @@ object PreAggregateUtil {
aggDataMapSchema.getProperties.get("CHILD_SELECT QUERY").replace("&", "=")),
CarbonCommonConstants.DEFAULT_CHARSET)
}
+
+ /**
+ * This method will start load process on the data map
+ */
+ def createLoadCommandForChild(
+ columns: java.util.List[ColumnSchema],
+ dataMapIdentifier: TableIdentifier,
+ dataFrame: DataFrame,
+ isOverwrite: Boolean,
+ sparkSession: SparkSession): CarbonLoadDataCommand = {
+ val headers = columns.asScala.filter { column =>
+ !column.getColumnName.equalsIgnoreCase(CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE)
+ }.sortBy(_.getSchemaOrdinal).map(_.getColumnName).mkString(",")
+ val loadCommand = CarbonLoadDataCommand(dataMapIdentifier.database,
+ dataMapIdentifier.table,
+ null,
+ Nil,
+ Map("fileheader" -> headers),
+ isOverwriteTable = isOverwrite,
+ dataFrame = None,
+ internalOptions = Map(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true"),
+ logicalPlan = Some(dataFrame.queryExecution.logical))
+ loadCommand.processMetadata(sparkSession)
--- End diff --
Better don't process here, just return the command and call `processMetadata(sparkSession)` in caller for more meaningful as method name says `createLoadCommandForChild`
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3078/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3090/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by jackylk <gi...@git.apache.org>.
Github user jackylk commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r162627861
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala ---
@@ -45,10 +47,10 @@ case class CarbonInsertIntoCommand(
updateModel = None,
tableInfoOp = None,
internalOptions = Map.empty,
- partition = partition).run(sparkSession)
- // updating relation metadata. This is in case of auto detect high cardinality
- relation.carbonRelation.metaData =
- CarbonSparkUtil.createSparkMeta(relation.carbonRelation.carbonTable)
- load
+ partition = partition)
+ loadCommand.processMetadata(sparkSession)
+ }
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ loadCommand.processData(sparkSession)
--- End diff --
add `if (loadCommand != null)` check
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1867/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3098/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163761030
--- Diff: processing/src/main/java/org/apache/carbondata/processing/loading/events/LoadEvents.java ---
@@ -128,6 +129,20 @@ public CarbonTableIdentifier getCarbonTableIdentifier() {
}
}
+ public static class LoadMetadataEvent extends Event {
--- End diff --
Please add comment
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3120/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3095/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3014/
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3116/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163757369
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala ---
@@ -53,42 +54,50 @@ import org.apache.carbondata.streaming.segment.StreamSegment
*/
case class CarbonAlterTableCompactionCommand(
alterTableModel: AlterTableModel,
- tableInfoOp: Option[TableInfo] = None)
- extends DataCommand {
+ tableInfoOp: Option[TableInfo] = None,
+ val operationContext: OperationContext = new OperationContext )
+ extends AtomicRunnableCommand {
+ var table: CarbonTable = _
- override def processData(sparkSession: SparkSession): Seq[Row] = {
- val LOGGER: LogService =
- LogServiceFactory.getLogService(this.getClass.getName)
+ override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+ val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
val tableName = alterTableModel.tableName.toLowerCase
- val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
-
- val table = if (tableInfoOp.isDefined) {
- val tableInfo = tableInfoOp.get
- // To DO: CarbonEnv.updateStorePath
- CarbonTable.buildFromTableInfo(tableInfo)
+ val dbName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
+ table = if (tableInfoOp.isDefined) {
+ CarbonTable.buildFromTableInfo(tableInfoOp.get)
} else {
- val relation =
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .lookupRelation(Option(databaseName), tableName)(sparkSession)
- .asInstanceOf[CarbonRelation]
+ val relation = CarbonEnv.getInstance(sparkSession).carbonMetastore
+ .lookupRelation(Option(dbName), tableName)(sparkSession).asInstanceOf[CarbonRelation]
if (relation == null) {
- throw new NoSuchTableException(databaseName, tableName)
+ throw new NoSuchTableException(dbName, tableName)
}
if (null == relation.carbonTable) {
- LOGGER.error(s"alter table failed. table not found: $databaseName.$tableName")
- throw new NoSuchTableException(databaseName, tableName)
+ LOGGER.error(s"Data loading failed. table not found: $dbName.$tableName")
+ LOGGER.audit(s"Data loading failed. table not found: $dbName.$tableName")
+ throw new NoSuchTableException(dbName, tableName)
}
relation.carbonTable
}
+ if (CarbonUtil.hasAggregationDataMap(table) ||
+ (table.isChildDataMap && null == operationContext.getProperty(table.getTableName))) {
--- End diff --
Irrespective of this condition I guess we always need to fire the event. If the listener does not want to execute then move this if condition to the listner instead of here
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1879/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163757962
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala ---
@@ -20,18 +20,141 @@ package org.apache.spark.sql.execution.command.preaaggregate
import scala.collection.JavaConverters._
import scala.collection.mutable
-import org.apache.spark.sql.CarbonEnv
-import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.{SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.AlterTableModel
-import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
+import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonLoadDataCommand}
+import org.apache.spark.sql.parser.CarbonSpark2SqlParser
-import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.schema.table.AggregationDataMapSchema
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.events._
-import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+object CompactionProcessMetaListener extends OperationEventListener {
--- End diff --
Add the comment for usage
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3096/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163753419
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala ---
@@ -53,42 +54,50 @@ import org.apache.carbondata.streaming.segment.StreamSegment
*/
case class CarbonAlterTableCompactionCommand(
alterTableModel: AlterTableModel,
- tableInfoOp: Option[TableInfo] = None)
- extends DataCommand {
+ tableInfoOp: Option[TableInfo] = None,
+ val operationContext: OperationContext = new OperationContext )
+ extends AtomicRunnableCommand {
+ var table: CarbonTable = _
- override def processData(sparkSession: SparkSession): Seq[Row] = {
- val LOGGER: LogService =
- LogServiceFactory.getLogService(this.getClass.getName)
+ override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
+ val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
val tableName = alterTableModel.tableName.toLowerCase
- val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
-
- val table = if (tableInfoOp.isDefined) {
- val tableInfo = tableInfoOp.get
- // To DO: CarbonEnv.updateStorePath
- CarbonTable.buildFromTableInfo(tableInfo)
+ val dbName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
+ table = if (tableInfoOp.isDefined) {
+ CarbonTable.buildFromTableInfo(tableInfoOp.get)
} else {
- val relation =
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .lookupRelation(Option(databaseName), tableName)(sparkSession)
- .asInstanceOf[CarbonRelation]
+ val relation = CarbonEnv.getInstance(sparkSession).carbonMetastore
+ .lookupRelation(Option(dbName), tableName)(sparkSession).asInstanceOf[CarbonRelation]
if (relation == null) {
- throw new NoSuchTableException(databaseName, tableName)
+ throw new NoSuchTableException(dbName, tableName)
}
if (null == relation.carbonTable) {
- LOGGER.error(s"alter table failed. table not found: $databaseName.$tableName")
- throw new NoSuchTableException(databaseName, tableName)
+ LOGGER.error(s"Data loading failed. table not found: $dbName.$tableName")
+ LOGGER.audit(s"Data loading failed. table not found: $dbName.$tableName")
--- End diff --
remove the duplicate audit log
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3013/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/3011/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r162910788
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala ---
@@ -649,6 +659,7 @@ case class CarbonLoadDataCommand(
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_SERIALIZATION_NULL_FORMAT)
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_BAD_RECORDS_ACTION)
CarbonSession.threadUnset(CarbonLoadOptionConstants.CARBON_OPTIONS_IS_EMPTY_DATA_BAD_RECORD)
+ CarbonSession.threadUnset("partition.opeartioncontext")
--- End diff --
typo mistake `operationcontext` , better take it to CarbonCommonCOnstants
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163760950
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala ---
@@ -215,8 +214,19 @@ case class CarbonRelation(
carbonTable.getCarbonTableIdentifier).getPath
val fileType = FileFactory.getFileType(tablePath)
if (FileFactory.isFileExist(tablePath, fileType)) {
+ // get the valid segments
+ val segments = new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier)
+ .getValidAndInvalidSegments.getValidSegments.asScala
+ var size = 0L
+ // for each segment calculate the size
+ segments.foreach {validSeg =>
+ size = size +
+ FileFactory.getDirectorySize(CarbonTablePath.getSegmentPath(tablePath, validSeg))
--- End diff --
Please format as below
```
size = size + FileFactory.getDirectorySize(
CarbonTablePath.getSegmentPath(tablePath, validSeg))
```
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1771/
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/2994/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by jackylk <gi...@git.apache.org>.
Github user jackylk commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r162626990
--- Diff: core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDataMapIndexStore.java ---
@@ -136,7 +136,8 @@ public BlockletDataMap get(TableBlockIndexUniqueIdentifier identifier)
partitionFileStore.readAllPartitionsOfSegment(carbonFiles, segmentPath);
partitionFileStoreMap.put(identifier.getSegmentId(), partitionFileStore);
for (CarbonFile file : carbonFiles) {
- locationMap.put(file.getAbsolutePath(), file.getLocations());
+ locationMap
+ .put(FileFactory.getUpdatedFilePath(file.getAbsolutePath()), file.getLocations());
--- End diff --
move .put to previous line
---
[GitHub] carbondata issue #1837: [CARBONDATA-2076] Refactored code segregated process...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on the issue:
https://github.com/apache/carbondata/pull/1837
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/3099/
---
[GitHub] carbondata pull request #1837: [CARBONDATA-2076] Refactored code segregated ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/carbondata/pull/1837
---
[GitHub] carbondata issue #1837: [WIP] Refactored code segregated process meta and pr...
Posted by CarbonDataQA <gi...@git.apache.org>.
Github user CarbonDataQA commented on the issue:
https://github.com/apache/carbondata/pull/1837
Build Failed with Spark 2.2.1, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/1764/
---
[GitHub] carbondata pull request #1837: [WIP] Refactored code segregated process meta...
Posted by ravipesala <gi...@git.apache.org>.
Github user ravipesala commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1837#discussion_r163753157
--- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala ---
@@ -53,42 +54,50 @@ import org.apache.carbondata.streaming.segment.StreamSegment
*/
case class CarbonAlterTableCompactionCommand(
alterTableModel: AlterTableModel,
- tableInfoOp: Option[TableInfo] = None)
- extends DataCommand {
+ tableInfoOp: Option[TableInfo] = None,
+ val operationContext: OperationContext = new OperationContext )
+ extends AtomicRunnableCommand {
--- End diff --
Enter new line here.
---