Indhumathi27 commented on a change in pull request #3661: [WIP] Support materialized view URL: https://github.com/apache/carbondata/pull/3661#discussion_r396354903
########## File path: integration/spark/src/main/scala/org/apache/carbondata/view/MaterializedViewRefresher.scala ########## @@ -0,0 +1,398 @@ +/* + * 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.MaterializedViewQueryParser + +import org.apache.carbondata.common.exceptions.sql.NoSuchMaterializedViewException +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.{MaterializedViewSchema, MaterializedViewStatus} +import org.apache.carbondata.processing.util.CarbonLoaderUtil + +class MaterializedViewRefresher{ + +} + +object MaterializedViewRefresher { + + private val LOGGER: Logger = LogServiceFactory.getLogService( + classOf[MaterializedViewRefresher].getCanonicalName) + + /** + * Refresh the mv by loading all existing data from associated 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[NoSuchMaterializedViewException] + def refresh(viewSchema: MaterializedViewSchema, 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 dataloading + val lock: ICarbonLock = segmentStatusManager.getTableStatusLock + val segmentMapping: util.Map[String, util.List[String]] = + new util.HashMap[String, util.List[String]] + val viewManager = MaterializedViewManagerInSpark.get(session) + try if (lock.lockWithRetries) { + LOGGER.info("Acquired lock for mv " + viewIdentifier + " for table status updation") + 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 loadMetadetail + 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 associatedTableIds = + viewSchema.getAssociatedTables.asScala.filter(_.isCarbonDataTable) + for (associatedTableId <- associatedTableIds) { + val associatedTableSegmentList: util.List[String] = + SegmentStatusManager.getValidSegmentList(associatedTableId) + if (associatedTableSegmentList.isEmpty) { + return false + } + segmentMapping.put(associatedTableId.toString, associatedTableSegmentList) + } + } + segmentMap = new Gson().toJson(segmentMapping) + // To handle concurrent dataloading 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 updation for table " + + viewSchema.getIdentifier.getDatabaseName + "." + + viewSchema.getIdentifier.getTableName) + viewManager.setStatus(viewSchema.getIdentifier, MaterializedViewStatus.DISABLED) + return false + } finally { + if (lock.unlock) { + LOGGER.info("Table unlocked successfully after table status updation" + + 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 updation") + } + } + refreshInternal(viewManager, viewSchema, viewTable, newLoadName, segmentMapping, session) + } + + @throws[IOException] + private def refreshInternal( + viewManager: MaterializedViewManagerInSpark, + viewSchema: MaterializedViewSchema, + 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 = MaterializedViewQueryParser.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 dataMap table fails, disable the dataMap and if newLoad is still + // in INSERT_IN_PROGRESS state, mark for delete the newLoad and update table status file + viewManager.setStatus(viewSchema.getIdentifier, MaterializedViewStatus.DISABLED) + LOGGER.error("Data Load failed for DataMap: ", exception) + CarbonLoaderUtil.updateTableStatusInCaseOfFailure( + newLoadName, + viewTable.getAbsoluteTableIdentifier, + viewTable.getTableName, + viewTable.getDatabaseName, + viewTable.getTablePath, + viewTable.getMetadataPath) + throw exception + } finally { + unsetInputSegments(viewSchema) + } + } + true + } + + /** + * This method will compare mainTable and dataMapTable segment List and loads only newly added + * segment from main table to dataMap table. + * In case if mainTable is compacted, then based on dataMap to mainTables segmentMapping, dataMap + * will be loaded + * Eg: + * case 1: Consider mainTableSegmentList: {0, 1, 2}, dataMapToMainTable 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 dataMap table will be marked for delete, and new segment + * {2 -> 1.1, 3} will be loaded to dataMap table + * case 2: Consider mainTableSegmentList: {0, 1, 2, 3}, dataMapToMainTable 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 dataMap 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 dataMap + * table segment (2) with main table segmentList{2 -> 1.1, 4, 0} will be loaded to dataMap table. + * dataMapToMainTable segmentMap will be updated to {1 -> 3, 2 -> 1.1, 4, 0} after rebuild + */ + @throws[IOException] + private def getSpecificSegmentsTobeLoaded(schema: MaterializedViewSchema, + segmentMapping: util.Map[String, util.List[String]], + listOfLoadFolderDetails: util.List[LoadMetadataDetails]): Boolean = { + val relationIdentifiers: util.List[RelationIdentifier] = schema.getAssociatedTables + // invalidDataMapSegmentList holds segment list which needs to be marked for delete + val invalidDataMapSegmentList: util.HashSet[String] = new util.HashSet[String] Review comment: Can check and rename all variables to mv ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
