[CARBONDATA-1526] [PreAgg] Added support to compact segments in pre-agg table #1605
Conversation
|
@jackylk @ravipesala Can you please start first level review. |
|
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2051/ |
|
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1661/ |
kunal642
force-pushed
the
preagg_compaction_support
branch
from
cc58e9d
to
0558db5
Compare
|
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2054/ |
|
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1667/ |
kunal642
force-pushed
the
preagg_compaction_support
branch
from
0558db5
to
5ef3e18
Compare
kunal642
changed the title
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/417/ |
|
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2075/ |
|
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1689/ |
|
@kunal642 Please add the description of the work done in this PR. |
|
@ravipesala description added |
| @@ -133,25 +133,25 @@ case class AlterTableRenameAbortEvent(carbonTable: CarbonTable, | |||
| /** | |||
| * | |||
| * @param carbonTable | |||
| * @param carbonLoadModel | |||
| * @param carbonMergerMapping | |||
There was a problem hiding this comment.
Add description of all these parameters
| * @param mergedLoadName | ||
| * @param sQLContext | ||
| */ | ||
| case class AlterTableCompactionPreEvent(sparkSession: SparkSession, carbonTable: CarbonTable, | ||
| carbonLoadModel: CarbonLoadModel, | ||
| case class AlterTableCompactionPreEvent(carbonTable: CarbonTable, |
There was a problem hiding this comment.
move carbonTable to next line, same for all events
| mergedLoadName: String, | ||
| sQLContext: SQLContext) extends Event with AlterTableCompactionEventInfo | ||
|
|
||
|
|
||
| /** | ||
| * | ||
| * @param carbonTable | ||
| * @param carbonLoadModel | ||
| * @param carbonMergerMapping |
There was a problem hiding this comment.
Add description of all these parameters, same for all related events
| * Class for handling operations after data load completion and before final commit of load | ||
| * operation. Example usage: For loading pre-aggregate tables | ||
| */ | ||
| case class LoadTablePreStatusUpdateEvent(sparkSession: SparkSession, |
There was a problem hiding this comment.
The usage of this is not clear, why is it needed
There was a problem hiding this comment.
This event will be used to perform some task just before updating the carbontable status.
For example: When loading data into parent table we need to start load for all child datamaps so that if any of the load fails then parent table status file would not be written.
| carbonLoadModel, | ||
| mergedLoadName, | ||
| sc) | ||
| AlterTableCompactionPreEvent(carbonTable, carbonMergerMapping, mergedLoadName, sc) |
There was a problem hiding this comment.
please modify line 55, it is not storePath, it is tablePath. change CarbonMergerMapping definition also
| @@ -492,6 +492,10 @@ object CarbonDataRDDFactory { | |||
| throw new Exception("No Data to load") | |||
| } | |||
| writeDictionary(carbonLoadModel, result, writeAll = false) | |||
| val loadTablePreStatusUpdateEvent = LoadTablePreStatusUpdateEvent(sqlContext.sparkSession, | |||
There was a problem hiding this comment.
Used to do some operation before commiting table status of parent
| @@ -159,7 +161,14 @@ case class CarbonAlterTableCompactionCommand( | |||
| // if system level compaction is enabled then only one compaction can run in the system | |||
| // if any other request comes at this time then it will create a compaction request file. | |||
| // so that this will be taken up by the compaction process which is executing. | |||
| if (!isConcurrentCompactionAllowed) { | |||
| if (carbonTable.isChildDataMap) { | |||
There was a problem hiding this comment.
please modify the comment start from line 161, put them in correct place and add comment for this newly added if condition
| @@ -159,7 +161,14 @@ case class CarbonAlterTableCompactionCommand( | |||
| // if system level compaction is enabled then only one compaction can run in the system | |||
| // if any other request comes at this time then it will create a compaction request file. | |||
| // so that this will be taken up by the compaction process which is executing. | |||
| if (!isConcurrentCompactionAllowed) { | |||
| if (carbonTable.isChildDataMap) { | |||
| carbonLoadModel.setCompactionType(alterTableModel.compactionType.toUpperCase match { | |||
There was a problem hiding this comment.
change setCompactionType to accept enum instead of String
| val segments = loadMetaDataDetails.asScala.map(_.getLoadName) | ||
| if (segments.nonEmpty) { | ||
| CarbonSession | ||
| .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS + |
There was a problem hiding this comment.
Do not pass parameters in this way, it is not good for maintenance. Pass the parameter to the function you need
There was a problem hiding this comment.
This is used to set the segments to access to load incremental data into the child table. There is no other way to set segments in CarbonScanRDD other than this
| private def startCompactionForDataMap(carbonLoadModel: CarbonLoadModel, | ||
| sparkSession: SparkSession): Unit = { | ||
| val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable | ||
| val loadMetaDataDetails = CarbonDataMergerUtil |
There was a problem hiding this comment.
correct the coding style in this function, please follow this coding style in future:
val a = foo(
paramA,
paramB)
| carbonLoadModel.getTableName, "false") | ||
| val headers = carbonTable.getTableInfo.getFactTable.getListOfColumns.asScala | ||
| .map(_.getColumnName).mkString(",") | ||
| val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser() |
There was a problem hiding this comment.
What does this do? Can you make it more readable
| .addPreAggLoadFunction(PreAggregateUtil | ||
| .createChildSelectQuery(carbonTable.getTableInfo.getFactTable))).drop("preAggLoad") | ||
| try { | ||
| CarbonLoadDataCommand(Some(carbonTable.getDatabaseName), |
There was a problem hiding this comment.
When invoking command, follow this code style:
CarbonXXXCommand(
paramA = valueA,
paramB = valueB,
...
).run(sparkSession)
Do add parameter name for each line
| // allows it. | ||
| if (!carbonLoadModel.getCompactionType.equals(CompactionType.MAJOR)) { | ||
| CommonUtil.readLoadMetadataDetails(carbonLoadModel) | ||
| startCompactionForDataMap(carbonLoadModel, sparkSession) |
There was a problem hiding this comment.
We should avoid the recursive call, you can invoke it in caller of this function.
| @@ -130,6 +131,9 @@ case class CarbonLoadDataCommand( | |||
| carbonLoadModel.setFactFilePath(factPath) | |||
| carbonLoadModel.setAggLoadRequest(internalOptions | |||
| .getOrElse(CarbonCommonConstants.IS_INTERNAL_LOAD_CALL, "false").toBoolean) | |||
| carbonLoadModel | |||
There was a problem hiding this comment.
following the coding style
a.foo(
paramA,
paramB)
| * @param event | ||
| * @param operationContext | ||
| */ | ||
| override def onEvent(event: Event, operationContext: OperationContext): Unit = { |
There was a problem hiding this comment.
What does the listener do? please add comment
|
|
||
| } | ||
|
|
||
| object AlterPreAggregateTableCompactionPostEvent extends OperationEventListener { |
| val compactionType = compactionEvent.carbonMergerMapping.campactionType | ||
| val sparkSession = compactionEvent.sQLContext.sparkSession | ||
| if (carbonTable.hasDataMapSchema) { | ||
| for (dataMapSchema: DataMapSchema <- carbonTable.getTableInfo.getDataMapSchemaList |
There was a problem hiding this comment.
use carbonTable.getTableInfo.getDataMapSchemaList.foreach instead
kunal642
force-pushed
the
preagg_compaction_support
branch
from
21ae844
to
be9c72b
Compare
|
Build Failed with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/531/ |
kunal642
force-pushed
the
preagg_compaction_support
branch
from
be9c72b
to
f5c9d09
Compare
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/533/ |
|
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2158/ |
|
SDV Build Fail , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2161/ |
|
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2162/ |
|
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1791/ |
|
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2164/ |
| import org.apache.carbondata.processing.merger.CarbonDataMergerUtil | ||
| import org.apache.carbondata.processing.util.CarbonLoaderUtil | ||
|
|
||
| abstract class Compactable(carbonLoadModel: CarbonLoadModel, |
There was a problem hiding this comment.
name this class as Compactor, and other Compactor class can be moved to CarbonTableCompactor
There was a problem hiding this comment.
Changed the class name to Compactor
| /** | ||
| * This class is used to perform compaction on carbon table. | ||
| */ | ||
| class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel, |
There was a problem hiding this comment.
Move all the compactor interfaces and impls to spark-common
There was a problem hiding this comment.
AggregateDataMapCompactor requires CarbonSession which is not available in spark-common
kunal642
force-pushed
the
preagg_compaction_support
branch
from
f5c9d09
to
9a6ace9
Compare
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/553/ |
| @@ -541,4 +541,20 @@ object PreAggregateUtil { | |||
| } | |||
| } | |||
|
|
|||
| def createChildSelectQuery(tableSchema: TableSchema): String = { | |||
| val aggregateColumns = scala.collection.mutable.ArrayBuffer.empty[String] | |||
| val groupingExpressions = scala.collection.mutable.ArrayBuffer.empty[String] | |||
There was a problem hiding this comment.
please use ArrayBuffer.empty[String]
| } | ||
| } | ||
| s"select ${ groupingExpressions.mkString(",") },${ aggregateColumns.mkString(",") | ||
| } from ${ tableSchema.getTableName } group by ${ groupingExpressions.mkString(",") }" |
There was a problem hiding this comment.
why not require the database name
| // adding the aggregation load UDF | ||
| .addPreAggLoadFunction(PreAggregateUtil | ||
| // creating the select query on the bases on table schema | ||
| .createChildSelectQuery(carbonTable.getTableInfo.getFactTable))).drop("preAggLoad") |
| val testData = s"$resourcesPath/sample.csv" | ||
|
|
||
| override def beforeEach(): Unit = { | ||
| sql("DROP TABLE IF EXISTS maintable") |
There was a problem hiding this comment.
suggest using a new database, not default database
kunal642
force-pushed
the
preagg_compaction_support
branch
from
9a6ace9
to
f32cebf
Compare
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/560/ |
|
Build Failed with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1809/ |
|
Retest this please |
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/571/ |
kunal642
force-pushed
the
preagg_compaction_support
branch
from
f32cebf
to
205ac43
Compare
|
Build Success with Spark 2.2.0, Please check CI http://88.99.58.216:8080/job/ApacheCarbonPRBuilder/583/ |
|
Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder1/1817/ |
|
retest sdv please |
|
SDV Build Success , Please check CI http://144.76.159.231:8080/job/ApacheSDVTests/2189/ |
|
LGTM |
…gg table This PR will add to compact the pre-aggregate tables. A pre-aggregate table can be compacted using the alter command i.e alter table table_name compact 'minor/major'. If a table with some pre-aggregate table is compacted, then all the pre-aggregate tables are also compacted with the parent table This closes apache#1605
…gg table This PR will add to compact the pre-aggregate tables. A pre-aggregate table can be compacted using the alter command i.e alter table table_name compact 'minor/major'. If a table with some pre-aggregate table is compacted, then all the pre-aggregate tables are also compacted with the parent table This closes apache#1605
This PR will add to compact the pre-aggregate tables.
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.