| /* |
| * 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.view |
| |
| import java._ |
| import java.io.IOException |
| |
| import scala.collection.JavaConverters._ |
| |
| import com.google.gson.Gson |
| import org.apache.log4j.Logger |
| import org.apache.spark.sql.{CarbonUtils, SparkSession} |
| import org.apache.spark.sql.execution.command.management.CarbonInsertIntoCommand |
| import org.apache.spark.sql.parser.MVQueryParser |
| |
| import org.apache.carbondata.common.exceptions.sql.NoSuchMVException |
| import org.apache.carbondata.common.logging.LogServiceFactory |
| import org.apache.carbondata.core.constants.CarbonCommonConstants |
| import org.apache.carbondata.core.locks.ICarbonLock |
| import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, RelationIdentifier} |
| import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager} |
| import org.apache.carbondata.core.util.path.CarbonTablePath |
| import org.apache.carbondata.core.view.{MVSchema, MVStatus} |
| import org.apache.carbondata.processing.util.CarbonLoaderUtil |
| |
| class MVRefresher{ |
| |
| } |
| |
| object MVRefresher { |
| |
| private val LOGGER: Logger = LogServiceFactory.getLogService( |
| classOf[MVRefresher].getCanonicalName) |
| |
| /** |
| * Refresh the mv by loading all existing data from related table |
| * This is called when refreshing the mv when |
| * 1. after mv creation and no "WITH DEFERRED REBUILD" defined |
| * 2. user manually trigger REFRESH MATERIALIZED VIEW command |
| */ |
| @throws[IOException] |
| @throws[NoSuchMVException] |
| def refresh(viewSchema: MVSchema, session: SparkSession): Boolean = { |
| var newLoadName: String = "" |
| var segmentMap: String = "" |
| val viewTable: CarbonTable = CarbonTable.buildFromTablePath( |
| viewSchema.getIdentifier.getTableName, |
| viewSchema.getIdentifier.getDatabaseName, |
| viewSchema.getIdentifier.getTablePath, |
| viewSchema.getIdentifier.getTableId) |
| val viewIdentifier = viewSchema.getIdentifier |
| val viewTableIdentifier = viewTable.getAbsoluteTableIdentifier |
| // Clean up the old invalid segment data before creating a new entry for new load. |
| SegmentStatusManager.deleteLoadsAndUpdateMetadata(viewTable, false, null) |
| val segmentStatusManager: SegmentStatusManager = new SegmentStatusManager(viewTableIdentifier) |
| // Acquire table status lock to handle concurrent data loading |
| val lock: ICarbonLock = segmentStatusManager.getTableStatusLock |
| val segmentMapping: util.Map[String, util.List[String]] = |
| new util.HashMap[String, util.List[String]] |
| val viewManager = MVManagerInSpark.get(session) |
| try if (lock.lockWithRetries) { |
| LOGGER.info("Acquired lock for mv " + viewIdentifier + " for table status update") |
| val viewTableMetadataPath: String = |
| CarbonTablePath.getMetadataPath(viewIdentifier.getTablePath) |
| val loadMetadataDetails = SegmentStatusManager.readLoadMetadata(viewTableMetadataPath) |
| val loadMetadataDetailList: util.List[LoadMetadataDetails] = |
| new util.ArrayList[LoadMetadataDetails](CarbonCommonConstants.DEFAULT_COLLECTION_SIZE) |
| // Mark for delete all stale loadMetataDetail |
| for (loadMetadataDetail <- loadMetadataDetails) { |
| if (((loadMetadataDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS) || |
| (loadMetadataDetail.getSegmentStatus eq SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS)) && |
| loadMetadataDetail.getVisibility.equalsIgnoreCase("false")) { |
| loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE) |
| } |
| loadMetadataDetailList.add(loadMetadataDetail) |
| } |
| if (viewSchema.isRefreshOnManual) { |
| // check if rebuild to mv is already in progress and throw exception |
| if (loadMetadataDetails.nonEmpty) { |
| for (loadMetaDetail <- loadMetadataDetails) { |
| if (((loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS) || |
| (loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS)) && |
| SegmentStatusManager.isLoadInProgress(viewTableIdentifier, |
| loadMetaDetail.getLoadName)) { |
| throw new RuntimeException( |
| "Rebuild to materialized view " + viewSchema.getIdentifier.getTableName + |
| " is already in progress") |
| } |
| } |
| } |
| } |
| if (viewSchema.isRefreshIncremental) { |
| if (!getSpecificSegmentsTobeLoaded(viewSchema, segmentMapping, loadMetadataDetailList)) { |
| return false |
| } |
| } else { |
| // set segment mapping only for carbondata table |
| val relatedTableIds = |
| viewSchema.getRelatedTables.asScala.filter(_.isCarbonDataTable) |
| for (relatedTableId <- relatedTableIds) { |
| val relatedTableSegmentList: util.List[String] = |
| SegmentStatusManager.getValidSegmentList(relatedTableId) |
| if (relatedTableSegmentList.isEmpty) { |
| return false |
| } |
| segmentMapping.put(relatedTableId.toString, relatedTableSegmentList) |
| } |
| } |
| segmentMap = new Gson().toJson(segmentMapping) |
| // To handle concurrent data loading to mv, create new loadMetaEntry and |
| // set segmentMap to new loadMetaEntry and pass new segmentId with load command |
| val loadMetadataDetail: LoadMetadataDetails = new LoadMetadataDetails |
| val segmentId: String = String.valueOf( |
| SegmentStatusManager.createNewSegmentId(loadMetadataDetails)) |
| loadMetadataDetail.setLoadName(segmentId) |
| loadMetadataDetail.setSegmentStatus(SegmentStatus.INSERT_IN_PROGRESS) |
| loadMetadataDetail.setExtraInfo(segmentMap) |
| loadMetadataDetailList.add(loadMetadataDetail) |
| newLoadName = segmentId |
| SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath.getTableStatusFilePath( |
| viewSchema.getIdentifier.getTablePath), |
| loadMetadataDetailList.toArray(new Array[LoadMetadataDetails](loadMetadataDetailList |
| .size))) |
| } else { |
| LOGGER.error("Not able to acquire the lock for table status update for table " + |
| viewSchema.getIdentifier.getDatabaseName + "." + |
| viewSchema.getIdentifier.getTableName) |
| viewManager.setStatus(viewSchema.getIdentifier, MVStatus.DISABLED) |
| return false |
| } finally { |
| if (lock.unlock) { |
| LOGGER.info("Table unlocked successfully after table status update" + |
| viewSchema.getIdentifier.getDatabaseName + "." + |
| viewSchema.getIdentifier.getTableName) |
| } else { |
| LOGGER.error("Unable to unlock Table lock for table" + |
| viewSchema.getIdentifier.getDatabaseName + "." + |
| viewSchema.getIdentifier.getTableName + |
| " during table status update") |
| } |
| } |
| refreshInternal(viewManager, viewSchema, viewTable, newLoadName, segmentMapping, session) |
| } |
| |
| @throws[IOException] |
| private def refreshInternal( |
| viewManager: MVManagerInSpark, |
| viewSchema: MVSchema, |
| viewTable: CarbonTable, |
| newLoadName: String, |
| segmentMap: java.util.Map[String, java.util.List[String]], |
| session: SparkSession): Boolean = { |
| val query = viewSchema.getQuery |
| if (query != null) { |
| val viewIdentifier = viewSchema.getIdentifier |
| val updatedQuery = MVQueryParser.getQuery(query, session) |
| val isFullRefresh = !viewSchema.isRefreshIncremental |
| // Set specified segments for incremental load |
| val segmentMapIterator = segmentMap.entrySet().iterator() |
| while (segmentMapIterator.hasNext) { |
| val entry = segmentMapIterator.next() |
| setInputSegments(entry.getKey, entry.getValue) |
| } |
| val header = viewTable.getTableInfo.getFactTable.getListOfColumns.asScala |
| .filter { column => |
| !column.getColumnName |
| .equalsIgnoreCase(CarbonCommonConstants.DEFAULT_INVISIBLE_DUMMY_MEASURE) |
| }.sortBy(_.getSchemaOrdinal).map(_.getColumnName).mkString(",") |
| val insertIntoCommand = CarbonInsertIntoCommand( |
| databaseNameOp = Some(viewIdentifier.getDatabaseName), |
| tableName = viewIdentifier.getTableName, |
| options = scala.collection.immutable.Map("fileheader" -> header), |
| isFullRefresh, |
| logicalPlan = updatedQuery.queryExecution.analyzed, |
| tableInfo = viewTable.getTableInfo, |
| internalOptions = Map("mergedSegmentName" -> newLoadName, |
| CarbonCommonConstants.IS_INTERNAL_LOAD_CALL -> "true"), |
| partition = Map.empty) |
| try { |
| insertIntoCommand.run(session) |
| } catch { |
| case exception: Exception => |
| // If load to mv table fails, disable the mv and if newLoad is still |
| // in INSERT_IN_PROGRESS state, mark for delete the newLoad and update table status file |
| viewManager.setStatus(viewSchema.getIdentifier, MVStatus.DISABLED) |
| LOGGER.error("Data Load failed for mv: ", exception) |
| CarbonLoaderUtil.updateTableStatusInCaseOfFailure( |
| newLoadName, viewTable, SegmentStatus.INSERT_IN_PROGRESS) |
| throw exception |
| } finally { |
| unsetInputSegments(viewSchema) |
| } |
| } |
| true |
| } |
| |
| /** |
| * This method will compare main table and mv table segment List and loads only newly added |
| * segment from main table to mv table. |
| * In case if mainTable is compacted, then based on mv to main tables segmentMapping, mv |
| * will be loaded |
| * Eg: |
| * case 1: Consider mainTableSegmentList: {0, 1, 2}, mvToMainTable segmentMap: |
| * { 0 -> 0, 1-> 1,2}. If (1, 2) segments of main table are compacted to 1.1 and new segment (3) |
| * is loaded to main table, then mainTableSegmentList will be updated to{0, 1.1, 3}. |
| * In this case, segment (1) of mv table will be marked for delete, and new segment |
| * {2 -> 1.1, 3} will be loaded to mv table |
| * case 2: Consider mainTableSegmentList: {0, 1, 2, 3}, mvToMainTable segmentMap: |
| * { 0 -> 0,1,2, 1-> 3}. If (1, 2) segments of main table are compacted to 1.1 and new segment |
| * (4) is loaded to main table, then mainTableSegmentList will be updated to {0, 1.1, 3, 4}. |
| * In this case, segment (0) of mv table will be marked for delete and segment (0) of |
| * main table will be added to validSegmentList which needs to be loaded again. Now, new mv |
| * table segment (2) with main table segmentList{2 -> 1.1, 4, 0} will be loaded to mv table. |
| * mvToMainTable segmentMap will be updated to {1 -> 3, 2 -> 1.1, 4, 0} after rebuild |
| */ |
| @throws[IOException] |
| private def getSpecificSegmentsTobeLoaded(schema: MVSchema, |
| segmentMapping: util.Map[String, util.List[String]], |
| listOfLoadFolderDetails: util.List[LoadMetadataDetails]): Boolean = { |
| val relationIdentifiers: util.List[RelationIdentifier] = schema.getRelatedTables |
| // invalidSegmentList holds segment list which needs to be marked for delete |
| val invalidSegmentList: util.HashSet[String] = new util.HashSet[String] |
| if (listOfLoadFolderDetails.isEmpty) { |
| // If segment Map is empty, load all valid segments from main tables to mv |
| for (relationIdentifier <- relationIdentifiers.asScala) { |
| val mainTableSegmentList: util.List[String] = SegmentStatusManager.getValidSegmentList( |
| relationIdentifier) |
| // If mainTableSegmentList is empty, no need to trigger load command |
| // TODO: handle in case of multiple tables load to mv table |
| if (mainTableSegmentList.isEmpty) return false |
| segmentMapping.put(relationIdentifier.getDatabaseName + CarbonCommonConstants.POINT + |
| relationIdentifier.getTableName, mainTableSegmentList) |
| } |
| } |
| else { |
| for (relationIdentifier <- relationIdentifiers.asScala) { |
| val segmentList: util.List[String] = new util.ArrayList[String] |
| // Get all segments for parent relationIdentifier |
| val mainTableSegmentList: util.List[String] = SegmentStatusManager.getValidSegmentList( |
| relationIdentifier) |
| var ifTableStatusUpdateRequired: Boolean = false |
| for (loadMetaDetail <- listOfLoadFolderDetails.asScala) { |
| if ((loadMetaDetail.getSegmentStatus eq SegmentStatus.SUCCESS) || |
| (loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS)) { |
| val segmentMaps: util.Map[String, util.List[String]] = |
| new Gson().fromJson(loadMetaDetail.getExtraInfo, classOf[util.Map[_, _]]) |
| val mainTableMetaDataPath: String = CarbonTablePath.getMetadataPath(relationIdentifier |
| .getTablePath) |
| val parentTableLoadMetaDataDetails: Array[LoadMetadataDetails] = SegmentStatusManager |
| .readLoadMetadata( |
| mainTableMetaDataPath) |
| val table: String = relationIdentifier.getDatabaseName + CarbonCommonConstants.POINT + |
| relationIdentifier.getTableName |
| for (segmentId <- mainTableSegmentList.asScala) { |
| // In case if mv segment(0) is mapped |
| // to mainTable segments{0,1,2} and if |
| // {0,1,2} segments of mainTable are compacted to 0.1. Then, |
| // on next rebuild/load to mv, no need to load segment(0.1) again. Update the |
| // segmentMapping of mv segment from {0,1,2} to {0.1} |
| if (!checkIfSegmentsToBeReloaded(parentTableLoadMetaDataDetails, |
| segmentMaps.get(table), |
| segmentId)) { |
| ifTableStatusUpdateRequired = true |
| // Update loadMetaDetail with updated segment info and clear old segmentMap |
| val updatedSegmentMap: util.Map[String, util.List[String]] = |
| new util.HashMap[String, util.List[String]] |
| val segmentIdList: util.List[String] = new util.ArrayList[String] |
| segmentIdList.add(segmentId) |
| updatedSegmentMap.put(table, segmentIdList) |
| segmentList.add(segmentId) |
| loadMetaDetail.setExtraInfo(new Gson().toJson(updatedSegmentMap)) |
| segmentMaps.get(table).clear() |
| } |
| } |
| segmentList.addAll(segmentMaps.get(table)) |
| } |
| } |
| val originSegmentList: util.List[String] = new util.ArrayList[String](segmentList) |
| segmentList.removeAll(mainTableSegmentList) |
| mainTableSegmentList.removeAll(originSegmentList) |
| if (ifTableStatusUpdateRequired && mainTableSegmentList.isEmpty) { |
| SegmentStatusManager.writeLoadDetailsIntoFile(CarbonTablePath.getTableStatusFilePath( |
| schema.getIdentifier.getTablePath), |
| listOfLoadFolderDetails.toArray(new Array[LoadMetadataDetails](listOfLoadFolderDetails |
| .size))) |
| return false |
| } else if (mainTableSegmentList.isEmpty) { |
| return false |
| } |
| if (!segmentList.isEmpty) { |
| val invalidMainTableSegmentList: util.List[String] = new util.ArrayList[String] |
| // validMainTableSegmentList holds segment list which needs to be loaded again |
| val validMainTableSegmentList: util.HashSet[String] = new util.HashSet[String] |
| // For mv segments which are not in main table segment list(if main table |
| // is compacted), iterate over those segments and get mv segments which needs to |
| // be marked for delete and main table segments which needs to be loaded again |
| for (segmentId <- segmentList.asScala) { |
| for (loadMetaDetail <- listOfLoadFolderDetails.asScala) { |
| if ((loadMetaDetail.getSegmentStatus eq SegmentStatus.SUCCESS) || |
| (loadMetaDetail.getSegmentStatus eq SegmentStatus.INSERT_IN_PROGRESS)) { |
| val segmentMaps: util.Map[String, util.List[String]] = |
| new Gson().fromJson(loadMetaDetail.getExtraInfo, classOf[util.Map[_, _]]) |
| val segmentIds: util.List[String] = segmentMaps.get( |
| relationIdentifier.getDatabaseName + CarbonCommonConstants.POINT + |
| relationIdentifier.getTableName) |
| if (segmentIds.contains(segmentId)) { |
| segmentIds.remove(segmentId) |
| validMainTableSegmentList.addAll(segmentIds) |
| invalidMainTableSegmentList.add(segmentId) |
| invalidSegmentList.add(loadMetaDetail.getLoadName) |
| } |
| } |
| } |
| } |
| // remove invalid segment from validMainTableSegmentList if present |
| validMainTableSegmentList.removeAll(invalidMainTableSegmentList) |
| // Add all valid segments of main table which needs to be loaded again |
| mainTableSegmentList.addAll(validMainTableSegmentList) |
| segmentMapping.put(relationIdentifier.getDatabaseName + CarbonCommonConstants.POINT + |
| relationIdentifier.getTableName, mainTableSegmentList) |
| } |
| else segmentMapping.put(relationIdentifier.getDatabaseName + CarbonCommonConstants.POINT + |
| relationIdentifier.getTableName, mainTableSegmentList) |
| } |
| } |
| // Remove invalid mv segments |
| if (!invalidSegmentList.isEmpty) { |
| for (loadMetadataDetail <- listOfLoadFolderDetails.asScala) { |
| if (invalidSegmentList.contains(loadMetadataDetail.getLoadName)) { |
| loadMetadataDetail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE) |
| } |
| } |
| } |
| true |
| } |
| |
| /** |
| * This method checks if mv table segment has to be reloaded again or not |
| */ |
| private def checkIfSegmentsToBeReloaded(loadMetaDataDetails: Array[LoadMetadataDetails], |
| segmentIds: util.List[String], |
| segmentId: String): Boolean = { |
| var isToBeLoadedAgain: Boolean = true |
| val mergedSegments: util.List[String] = new util.ArrayList[String] |
| for (loadMetadataDetail <- loadMetaDataDetails) { |
| if (null != loadMetadataDetail.getMergedLoadName && |
| loadMetadataDetail.getMergedLoadName.equalsIgnoreCase(segmentId)) { |
| mergedSegments.add(loadMetadataDetail.getLoadName) |
| } |
| } |
| if (!mergedSegments.isEmpty && segmentIds.containsAll(mergedSegments)) { |
| isToBeLoadedAgain = false |
| } |
| isToBeLoadedAgain |
| } |
| |
| /** |
| * This method will set main table segments which needs to be loaded to mv |
| */ |
| private def setInputSegments(tableUniqueName: String, |
| mainTableSegmentList: java.util.List[String]): Unit = { |
| CarbonUtils |
| .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS + |
| tableUniqueName, mainTableSegmentList.asScala.mkString(",")) |
| } |
| |
| private def unsetInputSegments(schema: MVSchema): Unit = { |
| val relatedTableIdentifiers = schema.getRelatedTables |
| for (relationIdentifier <- relatedTableIdentifiers.asScala) { |
| CarbonUtils |
| .threadUnset(CarbonCommonConstants.CARBON_INPUT_SEGMENTS + |
| relationIdentifier.getDatabaseName + "." + |
| relationIdentifier.getTableName) |
| } |
| } |
| |
| } |
