| /* |
| * 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.spark.sql.secondaryindex.load; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| |
| 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.AbsoluteTableIdentifier; |
| import org.apache.carbondata.core.metadata.SegmentFileStore; |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable; |
| import org.apache.carbondata.core.mutate.CarbonUpdateUtil; |
| import org.apache.carbondata.core.statusmanager.LoadMetadataDetails; |
| import org.apache.carbondata.core.statusmanager.SegmentStatus; |
| import org.apache.carbondata.core.statusmanager.SegmentStatusManager; |
| import org.apache.carbondata.core.util.path.CarbonTablePath; |
| import org.apache.carbondata.processing.loading.model.CarbonLoadModel; |
| import org.apache.carbondata.processing.util.CarbonLoaderUtil; |
| |
| import org.apache.log4j.Logger; |
| import org.apache.spark.sql.index.CarbonIndexUtil; |
| |
| public class CarbonInternalLoaderUtil { |
| |
| private static final Logger LOGGER = |
| LogServiceFactory.getLogService(CarbonInternalLoaderUtil.class.getName()); |
| |
| public static List<String> getListOfValidSlices(LoadMetadataDetails[] details) { |
| List<String> activeSlices = |
| new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| for (LoadMetadataDetails oneLoad : details) { |
| if (SegmentStatus.SUCCESS.equals(oneLoad.getSegmentStatus()) |
| || SegmentStatus.LOAD_PARTIAL_SUCCESS.equals(oneLoad.getSegmentStatus()) |
| || SegmentStatus.MARKED_FOR_UPDATE.equals(oneLoad.getSegmentStatus())) { |
| activeSlices.add(oneLoad.getLoadName()); |
| } |
| } |
| return activeSlices; |
| } |
| |
| /** |
| * This method will return the mapping of valid segments to segment laod start time |
| * |
| */ |
| public static Map<String, Long> getSegmentToLoadStartTimeMapping(LoadMetadataDetails[] details) { |
| Map<String, Long> segmentToLoadStartTimeMap = new HashMap<>(details.length); |
| for (LoadMetadataDetails oneLoad : details) { |
| // valid segments will only have Success status |
| if (SegmentStatus.SUCCESS.equals(oneLoad.getSegmentStatus()) |
| || SegmentStatus.LOAD_PARTIAL_SUCCESS.equals(oneLoad.getSegmentStatus())) { |
| segmentToLoadStartTimeMap.put(oneLoad.getLoadName(), oneLoad.getLoadStartTime()); |
| } |
| } |
| return segmentToLoadStartTimeMap; |
| } |
| |
| /** |
| * This API will write the load level metadata for the loadmanagement module inorder to |
| * manage the load and query execution management smoothly. |
| * |
| * @return boolean which determines whether status update is done or not. |
| */ |
| public static boolean recordLoadMetadata(List<LoadMetadataDetails> newLoadMetadataDetails, |
| List<String> validSegments, CarbonTable carbonTable, List<CarbonTable> indexCarbonTables, |
| String databaseName, String tableName) { |
| boolean status = false; |
| String metaDataFilepath = carbonTable.getMetadataPath(); |
| AbsoluteTableIdentifier absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier(); |
| SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier); |
| ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock(); |
| try { |
| if (carbonLock.lockWithRetries()) { |
| LOGGER.info("Acquired lock for table" + databaseName + "." + tableName |
| + " for table status updation"); |
| |
| if (isSegmentsAlreadyCompactedForNewMetaDataDetails(indexCarbonTables, tableName, |
| newLoadMetadataDetails)) { |
| return false; |
| } |
| |
| LoadMetadataDetails[] currentLoadMetadataDetails = |
| SegmentStatusManager.readLoadMetadata(metaDataFilepath); |
| |
| List<LoadMetadataDetails> updatedLoadMetadataDetails = |
| new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| |
| // check which load needs to be overwritten which are in in progress state |
| boolean found = false; |
| for (int i = 0; i < currentLoadMetadataDetails.length; i++) { |
| for (LoadMetadataDetails newLoadMetadataDetail : newLoadMetadataDetails) { |
| if (currentLoadMetadataDetails[i].getLoadName() |
| .equals(newLoadMetadataDetail.getLoadName())) { |
| currentLoadMetadataDetails[i] = newLoadMetadataDetail; |
| found = true; |
| break; |
| } |
| } |
| updatedLoadMetadataDetails.add(currentLoadMetadataDetails[i]); |
| } |
| |
| // check if newLoadMetadataDetail has segments which are not in currentLoadMetaDetails |
| // and add them to the updatedLoadMetadataDetails |
| boolean foundNext = false; |
| for (int i = 0; i < newLoadMetadataDetails.size(); i++) { |
| foundNext = false; |
| for (int j = 0; j < currentLoadMetadataDetails.length; j++) { |
| if (newLoadMetadataDetails.get(i).getLoadName().equals(currentLoadMetadataDetails[j].getLoadName())) { |
| foundNext = true; |
| break; |
| } |
| if (j == currentLoadMetadataDetails.length - 1 && !foundNext) { |
| // if not found in the list then add it |
| updatedLoadMetadataDetails.add(newLoadMetadataDetails.get(i)); |
| found = true; |
| } |
| } |
| } |
| |
| // when data load is done for first time, add all the details |
| if (currentLoadMetadataDetails.length == 0 || !found) { |
| updatedLoadMetadataDetails.addAll(newLoadMetadataDetails); |
| } |
| |
| List<String> indexTables = CarbonIndexUtil.getSecondaryIndexes(carbonTable); |
| if (!indexTables.isEmpty()) { |
| List<LoadMetadataDetails> newSegmentDetailsListForIndexTable = |
| new ArrayList<>(validSegments.size()); |
| for (String segmentId : validSegments) { |
| LoadMetadataDetails newSegmentDetailsObject = new LoadMetadataDetails(); |
| newSegmentDetailsObject.setLoadName(segmentId); |
| newSegmentDetailsListForIndexTable.add(newSegmentDetailsObject); |
| } |
| for (CarbonTable indexTable : indexCarbonTables) { |
| List<LoadMetadataDetails> indexTableDetailsList = CarbonIndexUtil |
| .getTableStatusDetailsForIndexTable(updatedLoadMetadataDetails, indexTable, |
| newSegmentDetailsListForIndexTable); |
| |
| SegmentStatusManager.writeLoadDetailsIntoFile( |
| CarbonTablePath.getTableStatusFilePath(indexTable.getTablePath()), |
| indexTableDetailsList |
| .toArray(new LoadMetadataDetails[0])); |
| } |
| } else if (carbonTable.isIndexTable()) { |
| SegmentStatusManager.writeLoadDetailsIntoFile( |
| metaDataFilepath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonTablePath.TABLE_STATUS_FILE, updatedLoadMetadataDetails |
| .toArray(new LoadMetadataDetails[0])); |
| } |
| status = true; |
| } else { |
| LOGGER.error( |
| "Not able to acquire the lock for Table status updation for table " + databaseName + "." |
| + tableName); |
| } |
| } catch (IOException e) { |
| LOGGER.error( |
| "Not able to acquire the lock for Table status updation for table " + databaseName + "." |
| + tableName); |
| } |
| finally { |
| if (carbonLock.unlock()) { |
| LOGGER.info("Table unlocked successfully after table status updation" + databaseName + "." |
| + tableName); |
| } else { |
| LOGGER.error("Unable to unlock Table lock for table" + databaseName + "." + tableName |
| + " during table status updation"); |
| } |
| } |
| return status; |
| } |
| |
| /** |
| * This method read the details of SI table and check whether new metadatadetails are already |
| * compacted, if it is, then already compaction for SI is completed and updating with new segment |
| * status is useless, this can happen in case of updating the status of index while loading |
| * segments for failed segments, so do not update anything, just exit gracefully |
| */ |
| private static boolean isSegmentsAlreadyCompactedForNewMetaDataDetails( |
| List<CarbonTable> indexTables, String indexTableName, |
| List<LoadMetadataDetails> newLoadMetadataDetails) { |
| CarbonTable currentIndexTable = null; |
| for (CarbonTable indexTable : indexTables) { |
| if (indexTable.getTableName().equalsIgnoreCase(indexTableName)) { |
| currentIndexTable = indexTable; |
| break; |
| } |
| } |
| boolean isIndexTableSegmentsCompacted = false; |
| if (null != currentIndexTable) { |
| LoadMetadataDetails[] existingLoadMetaDataDetails = |
| SegmentStatusManager.readLoadMetadata(currentIndexTable.getMetadataPath()); |
| for (LoadMetadataDetails existingLoadMetaDataDetail : existingLoadMetaDataDetails) { |
| for (LoadMetadataDetails newLoadMetadataDetail : newLoadMetadataDetails) { |
| if (existingLoadMetaDataDetail.getLoadName() |
| .equalsIgnoreCase(newLoadMetadataDetail.getLoadName()) |
| && existingLoadMetaDataDetail.getSegmentStatus() == SegmentStatus.COMPACTED) { |
| isIndexTableSegmentsCompacted = true; |
| break; |
| } |
| } |
| if (isIndexTableSegmentsCompacted) { |
| break; |
| } |
| } |
| return isIndexTableSegmentsCompacted; |
| } else { |
| return false; |
| } |
| } |
| |
| /** |
| * method to update table status in case of IUD Update Delta Compaction. |
| * |
| */ |
| public static boolean updateLoadMetadataWithMergeStatus(CarbonTable indexCarbonTable, |
| String[] loadsToMerge, String mergedLoadNumber, CarbonLoadModel carbonLoadModel, |
| Map<String, String> segmentToLoadStartTimeMap, long mergeLoadStartTime, |
| SegmentStatus segmentStatus, long newLoadStartTime, List<String> rebuiltSegments) |
| throws IOException { |
| boolean tableStatusUpdationStatus = false; |
| List<String> loadMergeList = new ArrayList<>(Arrays.asList(loadsToMerge)); |
| AbsoluteTableIdentifier absoluteTableIdentifier = |
| carbonLoadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier(); |
| SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier); |
| |
| ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock(); |
| |
| try { |
| if (carbonLock.lockWithRetries()) { |
| LOGGER.info("Acquired lock for the table " + carbonLoadModel.getDatabaseName() + "." |
| + carbonLoadModel.getTableName() + " for table status updation "); |
| LoadMetadataDetails[] loadDetails = |
| SegmentStatusManager.readLoadMetadata(indexCarbonTable.getMetadataPath()); |
| |
| long modificationOrDeletionTimeStamp = CarbonUpdateUtil.readCurrentTime(); |
| for (LoadMetadataDetails loadDetail : loadDetails) { |
| // check if this segment is merged. |
| if (loadMergeList.contains(loadDetail.getLoadName()) || loadMergeList |
| .contains(loadDetail.getMergedLoadName())) { |
| // if the compacted load is deleted after the start of the compaction process, |
| // then need to discard the compaction process and treat it as failed compaction. |
| if (loadDetail.getSegmentStatus() == SegmentStatus.MARKED_FOR_DELETE) { |
| LOGGER.error("Compaction is aborted as the segment " + loadDetail.getLoadName() |
| + " is deleted after the compaction is started."); |
| return false; |
| } |
| loadDetail.setSegmentStatus(SegmentStatus.COMPACTED); |
| loadDetail.setModificationOrDeletionTimestamp(modificationOrDeletionTimeStamp); |
| loadDetail.setMergedLoadName(mergedLoadNumber); |
| } |
| } |
| |
| // create entry for merged one. |
| LoadMetadataDetails loadMetadataDetails = new LoadMetadataDetails(); |
| loadMetadataDetails.setSegmentStatus(segmentStatus); |
| long loadEnddate = CarbonUpdateUtil.readCurrentTime(); |
| loadMetadataDetails.setLoadEndTime(loadEnddate); |
| loadMetadataDetails.setLoadName(mergedLoadNumber); |
| loadMetadataDetails.setSegmentFile(SegmentFileStore.genSegmentFileName(mergedLoadNumber, |
| String.valueOf(segmentToLoadStartTimeMap.get(mergedLoadNumber))) |
| + CarbonTablePath.SEGMENT_EXT); |
| CarbonLoaderUtil |
| .addDataIndexSizeIntoMetaEntry(loadMetadataDetails, mergedLoadNumber, indexCarbonTable); |
| if (rebuiltSegments.contains(loadMetadataDetails.getLoadName())) { |
| loadMetadataDetails.setLoadStartTime(newLoadStartTime); |
| } else { |
| loadMetadataDetails.setLoadStartTime(mergeLoadStartTime); |
| } |
| |
| // put the merged folder entry |
| for (int i = 0; i < loadDetails.length; i++) { |
| if (loadDetails[i].getLoadName().equals(loadMetadataDetails.getLoadName())) { |
| loadDetails[i] = loadMetadataDetails; |
| } |
| } |
| |
| // if this is a major compaction then set the segment as major compaction. |
| List<LoadMetadataDetails> updatedDetailsList = new ArrayList<>(Arrays.asList(loadDetails)); |
| |
| SegmentStatusManager.writeLoadDetailsIntoFile( |
| CarbonTablePath.getTableStatusFilePath(indexCarbonTable.getTablePath()), |
| updatedDetailsList.toArray(new LoadMetadataDetails[0])); |
| tableStatusUpdationStatus = true; |
| } else { |
| LOGGER.error( |
| "Could not able to obtain lock for table" + carbonLoadModel.getDatabaseName() + "." |
| + carbonLoadModel.getTableName() + "for table status updation"); |
| } |
| } finally { |
| if (carbonLock.unlock()) { |
| LOGGER.info("Table unlocked successfully after table status updation" + carbonLoadModel |
| .getDatabaseName() + "." + carbonLoadModel.getTableName()); |
| } else { |
| LOGGER.error( |
| "Unable to unlock Table lock for table" + carbonLoadModel.getDatabaseName() + "." |
| + carbonLoadModel.getTableName() + " during table status updation"); |
| } |
| } |
| return tableStatusUpdationStatus; |
| } |
| |
| /** |
| * Method to check if main table and SI have same number of valid segments or not |
| */ |
| public static boolean checkMainTableSegEqualToSISeg(LoadMetadataDetails[] mainTableLoadMetadataDetails, |
| LoadMetadataDetails[] siTableLoadMetadataDetails) { |
| List<String> mainTableSegmentsList = |
| getListOfValidSlices(mainTableLoadMetadataDetails); |
| List<String> indexList = |
| getListOfValidSlices(siTableLoadMetadataDetails); |
| Collections.sort(mainTableSegmentsList); |
| Collections.sort(indexList); |
| // In the case when number of SI segments are more than the maintable segments do nothing |
| // and proceed to process the segments. Return False in case if maintable segments are more |
| // than SI Segments |
| if (indexList.size() < mainTableSegmentsList.size()) { |
| return false; |
| } |
| // There can be cases when the number of segments in the main table are less than the index |
| // table. In this case mapping all the segments in main table to SI table. |
| // Return False if a segment in maintable is not in indextable |
| HashSet<String> indexTableSet = new HashSet<String>(); |
| for (int i = 0; i < indexList.size(); i++) { |
| indexTableSet.add(indexList.get(i)); |
| } |
| for (int i = 0; i < mainTableSegmentsList.size(); i++) { |
| if (!indexTableSet.contains(mainTableSegmentsList.get(i))) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * Method to check if main table has in progress load and same segment not present in SI |
| */ |
| public static boolean checkInProgLoadInMainTableAndSI(CarbonTable carbonTable, |
| LoadMetadataDetails[] mainTableLoadMetadataDetails, |
| LoadMetadataDetails[] siTableLoadMetadataDetails) { |
| List<String> allSiSlices = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| for (LoadMetadataDetails oneLoad : siTableLoadMetadataDetails) { |
| allSiSlices.add(oneLoad.getLoadName()); |
| } |
| |
| if (mainTableLoadMetadataDetails.length != 0) { |
| for (LoadMetadataDetails loadDetail : mainTableLoadMetadataDetails) { |
| // if load in progress and check if same load is present in SI. |
| if (SegmentStatusManager.isLoadInProgress(carbonTable.getAbsoluteTableIdentifier(), loadDetail.getLoadName())) { |
| if (!allSiSlices.contains(loadDetail.getLoadName())) { |
| return false; |
| } |
| } |
| } |
| } |
| return true; |
| } |
| } |
