| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.carbondata.spark.rdd |
| |
| import java.util.concurrent.ExecutorService |
| |
| import scala.collection.JavaConverters._ |
| |
| import org.apache.spark.sql.{CarbonSession, SQLContext} |
| import org.apache.spark.sql.execution.command.CompactionModel |
| import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand |
| import org.apache.spark.sql.execution.command.preaaggregate.{CommitPreAggregateListener, PreAggregateUtil} |
| |
| import org.apache.carbondata.core.constants.CarbonCommonConstants |
| import org.apache.carbondata.core.datamap.Segment |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable |
| import org.apache.carbondata.core.mutate.CarbonUpdateUtil |
| import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager} |
| import org.apache.carbondata.core.util.path.CarbonTablePath |
| import org.apache.carbondata.events.OperationContext |
| import org.apache.carbondata.processing.loading.model.CarbonLoadModel |
| import org.apache.carbondata.processing.merger.{CarbonDataMergerUtil, CompactionType} |
| |
| /** |
| * Used to perform compaction on Aggregate data map. |
| */ |
| class AggregateDataMapCompactor(carbonLoadModel: CarbonLoadModel, |
| compactionModel: CompactionModel, |
| executor: ExecutorService, |
| sqlContext: SQLContext, |
| storeLocation: String, |
| operationContext: OperationContext) |
| extends Compactor(carbonLoadModel, compactionModel, executor, sqlContext, storeLocation) { |
| |
| override def executeCompaction(): Unit = { |
| val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable |
| val loadMetaDataDetails = identifySegmentsToBeMerged() |
| // If segmentFile name is specified in load details then segment is for partition table |
| // therefore the segment file name should be loadName#segmentFileName.segment |
| val segments = loadMetaDataDetails.asScala.map { |
| loadDetail => |
| new Segment(loadDetail.getLoadName, loadDetail.getSegmentFile, null).toString |
| } |
| |
| if (segments.nonEmpty) { |
| val mergedLoadName = CarbonDataMergerUtil.getMergedLoadName(loadMetaDataDetails).split("_")(1) |
| CarbonSession.threadSet( |
| CarbonCommonConstants.CARBON_INPUT_SEGMENTS + |
| carbonLoadModel.getDatabaseName + "." + |
| carbonLoadModel.getTableName, |
| segments.mkString(",")) |
| CarbonSession.threadSet( |
| CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS + |
| carbonLoadModel.getDatabaseName + "." + |
| carbonLoadModel.getTableName, "false") |
| val loadCommand = operationContext.getProperty(carbonTable.getTableName + "_Compaction") |
| .asInstanceOf[CarbonLoadDataCommand] |
| val uuid = Option(loadCommand.operationContext.getProperty("uuid")).getOrElse("").toString |
| try { |
| val newInternalOptions = loadCommand.internalOptions ++ |
| Map("mergedSegmentName" -> mergedLoadName) |
| loadCommand.internalOptions = newInternalOptions |
| loadCommand.dataFrame = |
| Some(PreAggregateUtil.getDataFrame( |
| sqlContext.sparkSession, loadCommand.logicalPlan.get)) |
| CarbonSession.threadSet(CarbonCommonConstants.SUPPORT_DIRECT_QUERY_ON_DATAMAP, |
| "true") |
| loadCommand.processData(sqlContext.sparkSession) |
| // After load is completed for child table the UUID table status will have 0.1 as success |
| // and the table status file will have 0,1,2,3 as Success and 0.1 as In Progress. |
| // Therefore we will read the table status and write 0,1,2,3 as Compacted as the commit |
| // listener will take care of merging the UUID and the table status files. |
| val newMetadataDetails = SegmentStatusManager.readLoadMetadata( |
| carbonTable.getMetadataPath, uuid) |
| val mergedContent = loadMetaDataDetails.asScala.map { |
| segment => segment.setSegmentStatus(SegmentStatus.COMPACTED) |
| segment.setMergedLoadName(mergedLoadName) |
| segment.setModificationOrdeletionTimesStamp(CarbonUpdateUtil.readCurrentTime) |
| segment |
| } ++ newMetadataDetails |
| SegmentStatusManager.writeLoadDetailsIntoFile( |
| CarbonTablePath.getTableStatusFilePathWithUUID(carbonTable.getTablePath, uuid), |
| mergedContent.toArray) |
| carbonLoadModel.setLoadMetadataDetails((carbonLoadModel.getLoadMetadataDetails.asScala ++ |
| newMetadataDetails).asJava) |
| // If isCompaction is true then it means that the compaction on aggregate table was |
| // triggered by the maintable thus no need to commit the tablestatus file but if the |
| // compaction was triggered directly for aggregate table then commit has to be fired as |
| // the commit listener would not be called. |
| val directAggregateCompactionCall = Option(operationContext |
| .getProperty("isCompaction")).getOrElse("false").toString.toBoolean |
| if (!directAggregateCompactionCall) { |
| commitAggregateTableStatus(carbonTable, uuid) |
| } |
| } finally { |
| // check if any other segments needs compaction on in case of MINOR_COMPACTION. |
| // For example: after 8.1 creation 0.1, 4.1, 8.1 have to be merged to 0.2 if threshhold |
| // allows it. |
| // Also as the load which will be fired for 2nd level compaction will read the |
| // tablestatus file and not the tablestatus_UUID therefore we have to commit the |
| // intermediate tablestatus file for 2nd level compaction to be successful. |
| // This is required because: |
| // 1. after doing 12 loads and a compaction after every 4 loads the table status file will |
| // have 0.1, 4.1, 8, 9, 10, 11 as Success segments. While tablestatus_UUID will have |
| // 0.1, 4.1, 8.1. |
| // 2. Now for 2nd level compaction 0.1, 8.1, 4.1 have to be merged to 0.2. therefore we |
| // need to read the tablestatus_UUID. But load flow should always read tablestatus file |
| // because it contains the actual In-Process status for the segments. |
| // 3. If we read the tablestatus then 8, 9, 10, 11 will keep getting compacted into 8.1. |
| // 4. Therefore tablestatus file will be committed in between multiple commits. |
| if (!compactionModel.compactionType.equals(CompactionType.MAJOR) && |
| !compactionModel.compactionType.equals(CompactionType.CUSTOM)) { |
| if (!identifySegmentsToBeMerged().isEmpty) { |
| commitAggregateTableStatus(carbonTable, uuid) |
| executeCompaction() |
| } |
| } |
| CarbonSession |
| .threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS + |
| carbonLoadModel.getDatabaseName + "." + |
| carbonLoadModel.getTableName) |
| CarbonSession.threadUnset(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS + |
| carbonLoadModel.getDatabaseName + "." + |
| carbonLoadModel.getTableName) |
| LOGGER |
| .info(s"Compaction request for datamap ${ carbonTable.getTableUniqueName } is successful") |
| } |
| } |
| } |
| |
| /** |
| * Used to merge the contents of tablestatus and tablestatus_uuid files and write the new |
| * details to tablestatus file. For Example:- |
| * tablestatus contents are = 0(Success), 1(Success),2(Success),3(Success), 0.1(In Progress) |
| * tablestatus_uuid contents are = 0(Compacted), 1(Compacted),2(Compacted),3(Compacted), 0.1 |
| * (Success). |
| * |
| * So after merging the tablestatus file will have: 0(Compacted), 1(Compacted),2(Compacted), |
| * 3(Compacted), 0.1(Success). |
| * |
| * NOTE: This method will be called when direct compaction is fired on child aggregate table or |
| * when there are anymore segments to be compacted and the intermediate state of the |
| * tablestatus has to be committed for further compaction to pick other segments. |
| */ |
| private def commitAggregateTableStatus(carbonTable: CarbonTable, uuid: String) { |
| if (!CommitPreAggregateListener.mergeTableStatusContents(carbonTable, CarbonTablePath |
| .getTableStatusFilePathWithUUID(carbonTable.getTablePath, uuid), CarbonTablePath |
| .getTableStatusFilePathWithUUID(carbonTable.getTablePath, ""))) { |
| throw new RuntimeException("Unable to acquire lock for table status updation") |
| } |
| } |
| |
| } |
