| /* |
| * 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.core.mutate; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| |
| import org.apache.carbondata.common.logging.LogServiceFactory; |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| import org.apache.carbondata.core.constants.CarbonCommonConstantsInternal; |
| import org.apache.carbondata.core.datastore.filesystem.CarbonFile; |
| import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter; |
| import org.apache.carbondata.core.datastore.impl.FileFactory; |
| import org.apache.carbondata.core.index.Segment; |
| 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.data.BlockMappingVO; |
| import org.apache.carbondata.core.mutate.data.RowCountDetailsVO; |
| 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.statusmanager.SegmentUpdateStatusManager; |
| import org.apache.carbondata.core.util.CarbonProperties; |
| import org.apache.carbondata.core.util.CarbonUtil; |
| import org.apache.carbondata.core.util.path.CarbonTablePath; |
| |
| import org.apache.commons.collections.CollectionUtils; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.log4j.Logger; |
| |
| /** |
| * This class contains all update utility methods |
| */ |
| public class CarbonUpdateUtil { |
| |
| private static final Logger LOGGER = |
| LogServiceFactory.getLogService(CarbonUpdateUtil.class.getName()); |
| |
| /** |
| * returns required filed from tuple id |
| * |
| */ |
| public static String getRequiredFieldFromTID(String Tid, int index) { |
| return Tid.split(CarbonCommonConstants.FILE_SEPARATOR)[index]; |
| } |
| |
| /** |
| * returns required filed from tuple id |
| * |
| * @param Tid |
| * @param tid |
| * @return |
| */ |
| public static String getRequiredFieldFromTID(String Tid, TupleIdEnum tid) { |
| return Tid.split("/")[tid.getTupleIdIndex()]; |
| } |
| |
| /** |
| * returns segment along with block id |
| * @param Tid |
| * @return |
| */ |
| public static String getSegmentWithBlockFromTID(String Tid, boolean isPartitionTable) { |
| if (isPartitionTable) { |
| return getRequiredFieldFromTID(Tid, TupleIdEnum.PARTITION_SEGMENT_ID); |
| } |
| // this case is to check for add segment case, as now the segment id is present at first index, |
| // in add segment case, it will be in second index as the blockletID is generated by adding the |
| // complete external path |
| // this is in case of the external segment, where the tuple id has external path with # |
| // here no need to check for any path present in metadta details, as # can come in tuple id in |
| // case of multiple partitions, so partition check is already present above. |
| if (Tid.contains("#/")) { |
| return getRequiredFieldFromTID(Tid, TupleIdEnum.EXTERNAL_SEGMENT_ID) |
| + CarbonCommonConstants.FILE_SEPARATOR + getRequiredFieldFromTID(Tid, |
| TupleIdEnum.EXTERNAL_BLOCK_ID); |
| } |
| return getRequiredFieldFromTID(Tid, TupleIdEnum.SEGMENT_ID) |
| + CarbonCommonConstants.FILE_SEPARATOR + getRequiredFieldFromTID(Tid, TupleIdEnum.BLOCK_ID); |
| } |
| |
| /** |
| * Returns block path from tuple id |
| */ |
| public static String getTableBlockPath(String tid, String tablePath, boolean isStandardTable, |
| boolean isPartitionTable) { |
| String partField = "0"; |
| // If it has segment file then part field can be appended directly to table path |
| if (!isStandardTable) { |
| if (isPartitionTable) { |
| partField = getRequiredFieldFromTID(tid, TupleIdEnum.PARTITION_PART_ID); |
| return tablePath + CarbonCommonConstants.FILE_SEPARATOR + partField.replace("#", "/"); |
| } else { |
| return tablePath; |
| } |
| } |
| String part = CarbonTablePath.addPartPrefix(partField); |
| String segment = |
| CarbonTablePath.addSegmentPrefix(getRequiredFieldFromTID(tid, TupleIdEnum.SEGMENT_ID)); |
| return CarbonTablePath.getFactDir(tablePath) + CarbonCommonConstants.FILE_SEPARATOR + part |
| + CarbonCommonConstants.FILE_SEPARATOR + segment; |
| } |
| |
| /** |
| * returns delete delta file path |
| * |
| * @param blockPath |
| * @param blockPath |
| * @param timestamp |
| * @return |
| */ |
| public static String getDeleteDeltaFilePath(String blockPath, String blockName, |
| String timestamp) { |
| return blockPath + CarbonCommonConstants.FILE_SEPARATOR + blockName |
| + CarbonCommonConstants.HYPHEN + timestamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT; |
| |
| } |
| |
| /** |
| * @param updateDetailsList |
| * @param table |
| * @param updateStatusFileIdentifier |
| * @return |
| */ |
| public static boolean updateSegmentStatus(List<SegmentUpdateDetails> updateDetailsList, |
| CarbonTable table, String updateStatusFileIdentifier, boolean isCompaction, |
| boolean isForceWrite) { |
| boolean status = false; |
| SegmentUpdateStatusManager segmentUpdateStatusManager = new SegmentUpdateStatusManager(table); |
| if (isForceWrite && !CollectionUtils.isEmpty(updateDetailsList)) { |
| String segId = String.valueOf(SegmentStatusManager |
| .createNewSegmentId(segmentUpdateStatusManager.getLoadMetadataDetails())); |
| for (SegmentUpdateDetails detail : updateDetailsList) { |
| detail.setSegmentName(segId); |
| } |
| } |
| ICarbonLock updateLock = segmentUpdateStatusManager.getTableUpdateStatusLock(); |
| boolean lockStatus = false; |
| |
| try { |
| lockStatus = updateLock.lockWithRetries(); |
| if (lockStatus) { |
| |
| // read the existing file if present and update the same. |
| SegmentUpdateDetails[] oldDetails = segmentUpdateStatusManager |
| .getUpdateStatusDetails(); |
| |
| List<SegmentUpdateDetails> oldList = new ArrayList(Arrays.asList(oldDetails)); |
| |
| for (SegmentUpdateDetails newBlockEntry : updateDetailsList) { |
| mergeSegmentUpdate(isCompaction, oldList, newBlockEntry); |
| } |
| |
| List<SegmentUpdateDetails> updateDetailsValidSeg = new ArrayList<>(); |
| Set<String> loadDetailsSet = new HashSet<>(); |
| for (LoadMetadataDetails details : segmentUpdateStatusManager.getLoadMetadataDetails()) { |
| loadDetailsSet.add(details.getLoadName()); |
| } |
| for (SegmentUpdateDetails updateDetails : oldList) { |
| if (loadDetailsSet.contains(updateDetails.getSegmentName())) { |
| // we should only keep the update info of segments in table status, especially after |
| // compaction and clean files some compacted segments will be removed. It can keep |
| // tableupdatestatus file in small size which is good for performance. |
| updateDetailsValidSeg.add(updateDetails); |
| } |
| } |
| // In case of ForceWrite, write the segmentUpdateDetails to the tableUpdateStatus file |
| // without any validation of segments. |
| if (isForceWrite) { |
| segmentUpdateStatusManager.writeLoadDetailsIntoFile(oldList, updateStatusFileIdentifier); |
| } else { |
| segmentUpdateStatusManager |
| .writeLoadDetailsIntoFile(updateDetailsValidSeg, updateStatusFileIdentifier); |
| } |
| status = true; |
| } else { |
| LOGGER.error("Not able to acquire the segment update lock."); |
| status = false; |
| } |
| } catch (IOException e) { |
| status = false; |
| } finally { |
| if (lockStatus) { |
| if (updateLock.unlock()) { |
| LOGGER.info("Unlock the segment update lock successful."); |
| } else { |
| LOGGER.error("Not able to unlock the segment update lock."); |
| } |
| } |
| } |
| return status; |
| } |
| |
| public static void mergeSegmentUpdate(boolean isCompaction, List<SegmentUpdateDetails> oldList, |
| SegmentUpdateDetails newBlockEntry) { |
| int index = oldList.indexOf(newBlockEntry); |
| if (index != -1) { |
| // update the element in existing list. |
| SegmentUpdateDetails blockDetail = oldList.get(index); |
| if (blockDetail.getDeleteDeltaStartTimestamp().isEmpty() || isCompaction) { |
| blockDetail |
| .setDeleteDeltaStartTimestamp(newBlockEntry.getDeleteDeltaStartTimestamp()); |
| } |
| blockDetail.setDeleteDeltaEndTimestamp(newBlockEntry.getDeleteDeltaEndTimestamp()); |
| blockDetail.setSegmentStatus(newBlockEntry.getSegmentStatus()); |
| blockDetail.setDeletedRowsInBlock(newBlockEntry.getDeletedRowsInBlock()); |
| // If the start and end time is different then the delta is there in multiple files so |
| // add them to the list to get the delta files easily with out listing. |
| if (!blockDetail.getDeleteDeltaStartTimestamp() |
| .equals(blockDetail.getDeleteDeltaEndTimestamp())) { |
| blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaStartTimestamp()); |
| blockDetail.addDeltaFileStamp(blockDetail.getDeleteDeltaEndTimestamp()); |
| } else { |
| blockDetail.setDeltaFileStamps(null); |
| } |
| } else { |
| // add the new details to the list. |
| oldList.add(newBlockEntry); |
| } |
| } |
| |
| /** |
| * Update table status |
| * @param updatedSegmentsList |
| * @param table |
| * @param updatedTimeStamp |
| * @param isTimestampUpdateRequired |
| * @param segmentsToBeDeleted |
| * @return |
| */ |
| public static Map<String, String> updateTableMetadataStatus(Set<Segment> updatedSegmentsList, |
| CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired, |
| boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted, |
| String tblStatusWriteVersion) { |
| return updateTableMetadataStatus(updatedSegmentsList, table, updatedTimeStamp, |
| isTimestampUpdateRequired, isUpdateStatusFileUpdateRequired, |
| segmentsToBeDeleted, new ArrayList<Segment>(), tblStatusWriteVersion); |
| } |
| |
| /** |
| * |
| * @param updatedSegmentsList |
| * @param table |
| * @param updatedTimeStamp |
| * @param isTimestampUpdateRequired |
| * @param segmentsToBeDeleted |
| * @return |
| */ |
| public static Map<String, String> updateTableMetadataStatus(Set<Segment> updatedSegmentsList, |
| CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdateRequired, |
| boolean isUpdateStatusFileUpdateRequired, List<Segment> segmentsToBeDeleted, |
| List<Segment> segmentFilesTobeUpdated, String uuid) { |
| |
| boolean status = false; |
| Map<String, String> tuple = new HashMap<>(); |
| String metaDataFilepath = table.getMetadataPath(); |
| AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier(); |
| String tableStatusPath = |
| CarbonTablePath.getTableStatusFilePath(identifier.getTablePath(), uuid); |
| SegmentStatusManager segmentStatusManager = |
| new SegmentStatusManager(identifier, table.getTableStatusVersion()); |
| |
| ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock(); |
| boolean lockStatus = false; |
| try { |
| lockStatus = carbonLock.lockWithRetries(); |
| if (lockStatus) { |
| LOGGER.info("Acquired lock for table" + table.getDatabaseName() + "." + table.getTableName() |
| + " for table status update"); |
| |
| if (uuid.isEmpty() && CarbonProperties.isTableStatusMultiVersionEnabled()) { |
| String tblStatusWriteVersion = String.valueOf(System.currentTimeMillis()); |
| tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath(), |
| tblStatusWriteVersion); |
| tuple.put("tblStatusWriteVersion", tblStatusWriteVersion); |
| } |
| LoadMetadataDetails[] listOfLoadFolderDetailsArray = |
| SegmentStatusManager.readLoadMetadata(metaDataFilepath, table.getTableStatusVersion()); |
| // to update table status only when required. |
| boolean isUpdateRequired = false; |
| |
| for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) { |
| // we are storing the link between the 2 status files in the segment 0 only. |
| if (isUpdateStatusFileUpdateRequired && |
| loadMetadata.getLoadName().equalsIgnoreCase("0")) { |
| loadMetadata.setUpdateStatusFileName( |
| CarbonUpdateUtil.getUpdateStatusFileName(updatedTimeStamp)); |
| } |
| |
| if (isTimestampUpdateRequired) { |
| // if the segments is in the list of marked for delete then update the status. |
| if (segmentsToBeDeleted.contains(new Segment(loadMetadata.getLoadName()))) { |
| loadMetadata.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE); |
| loadMetadata.setModificationOrDeletionTimestamp(Long.parseLong(updatedTimeStamp)); |
| isUpdateRequired = true; |
| } |
| } |
| for (Segment segName : updatedSegmentsList) { |
| if (loadMetadata.getLoadName().equalsIgnoreCase(segName.getSegmentNo())) { |
| // if this call is coming from the delete delta flow then the time stamp |
| // String will come empty then no need to write into table status file. |
| if (isTimestampUpdateRequired) { |
| // if in case of update flow. |
| if (loadMetadata.getUpdateDeltaStartTimestamp().isEmpty()) { |
| // this means for first time it is getting updated . |
| loadMetadata.setUpdateDeltaStartTimestamp(updatedTimeStamp); |
| } |
| // update end timestamp for each time. |
| loadMetadata.setUpdateDeltaEndTimestamp(updatedTimeStamp); |
| isUpdateRequired = true; |
| } |
| if (segmentFilesTobeUpdated |
| .contains(Segment.toSegment(loadMetadata.getLoadName(), null))) { |
| loadMetadata.setSegmentFile(loadMetadata.getLoadName() + "_" + updatedTimeStamp |
| + CarbonTablePath.SEGMENT_EXT); |
| isUpdateRequired = true; |
| } |
| } |
| } |
| } |
| |
| try { |
| if (isUpdateRequired || isUpdateStatusFileUpdateRequired) { |
| SegmentStatusManager |
| .writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray); |
| } |
| } catch (IOException e) { |
| tuple.put("status", "false"); |
| return tuple; |
| } |
| |
| status = true; |
| } else { |
| LOGGER.error("Not able to acquire the lock for Table status update for table " + table |
| .getDatabaseName() + "." + table.getTableName()); |
| } |
| } finally { |
| if (lockStatus) { |
| if (carbonLock.unlock()) { |
| LOGGER.info( |
| "Table unlocked successfully after table status update" + table.getDatabaseName() |
| + "." + table.getTableName()); |
| } else { |
| LOGGER.error( |
| "Unable to unlock Table lock for table" + table.getDatabaseName() + "." + table |
| .getTableName() + " during table status update"); |
| } |
| } |
| } |
| tuple.put("status", String.valueOf(status)); |
| return tuple; |
| |
| } |
| |
| /** |
| * gets the file name of the update status file. by appending the latest timestamp to it. |
| * |
| * @param updatedTimeStamp |
| * @return |
| */ |
| public static String getUpdateStatusFileName(String updatedTimeStamp) { |
| return CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME + CarbonCommonConstants.HYPHEN |
| + updatedTimeStamp; |
| } |
| |
| /** |
| * This will handle the clean up cases if the update fails. |
| * |
| * @param table |
| * @param timeStamp |
| */ |
| public static void cleanStaleDeltaFiles(CarbonTable table, final String timeStamp) { |
| |
| AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier(); |
| String partitionDir = CarbonTablePath.getPartitionDir(identifier.getTablePath()); |
| CarbonFile file = |
| FileFactory.getCarbonFile(partitionDir); |
| if (!file.exists()) { |
| return; |
| } |
| for (CarbonFile eachDir : file.listFiles()) { |
| // for each dir check if the file with the delta timestamp is present or not. |
| CarbonFile[] toBeDeleted = eachDir.listFiles(new CarbonFileFilter() { |
| @Override |
| public boolean accept(CarbonFile file) { |
| String fileName = file.getName(); |
| return fileName.endsWith(timeStamp + CarbonCommonConstants.DELETE_DELTA_FILE_EXT); |
| } |
| }); |
| // deleting the files of a segment. |
| try { |
| CarbonUtil.deleteFoldersAndFilesSilent(toBeDeleted); |
| } catch (IOException | InterruptedException e) { |
| LOGGER.error("Exception in deleting the delta files." + e); |
| } |
| } |
| } |
| |
| /** |
| * returns timestamp as long value |
| * |
| * @param timestamp |
| * @return |
| */ |
| public static Long getTimeStampAsLong(String timestamp) { |
| try { |
| return Long.parseLong(timestamp); |
| } catch (NumberFormatException nfe) { |
| String errorMsg = "Invalid timestamp : " + timestamp; |
| LOGGER.error(errorMsg); |
| return null; |
| } |
| } |
| |
| /** |
| * returns integer value from given string |
| * |
| * @param value |
| * @return |
| * @throws Exception |
| */ |
| public static Integer getIntegerValue(String value) throws Exception { |
| try { |
| return Integer.parseInt(value); |
| } catch (NumberFormatException nfe) { |
| LOGGER.error("Invalid row : " + value + nfe.getLocalizedMessage()); |
| throw new Exception("Invalid row : " + nfe.getLocalizedMessage()); |
| } |
| } |
| |
| /** |
| * return only block name from completeBlockName |
| * |
| * @param completeBlockName |
| * @return |
| */ |
| public static String getBlockName(String completeBlockName) { |
| return completeBlockName |
| .substring(0, completeBlockName.lastIndexOf(CarbonCommonConstants.HYPHEN)); |
| } |
| |
| /** |
| * returns segment id from segment name |
| * |
| * @param segmentName |
| * @return |
| */ |
| public static String getSegmentId(String segmentName) { |
| return segmentName.split(CarbonCommonConstants.UNDERSCORE)[1]; |
| } |
| |
| public static long getLatestTaskIdForSegment(Segment segment, String tablePath) |
| throws IOException { |
| long max = 0; |
| List<String> dataFiles = new ArrayList<>(); |
| if (segment.getSegmentFileName() != null) { |
| SegmentFileStore fileStore = new SegmentFileStore(tablePath, segment.getSegmentFileName()); |
| fileStore.readIndexFiles(FileFactory.getConfiguration()); |
| Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap(); |
| List<String> dataFilePaths = new ArrayList<>(); |
| for (List<String> paths : indexFilesMap.values()) { |
| dataFilePaths.addAll(paths); |
| } |
| for (String dataFilePath : dataFilePaths) { |
| dataFiles.add(new Path(dataFilePath).getName()); |
| } |
| |
| } else { |
| String segmentDirPath = CarbonTablePath.getSegmentPath(tablePath, segment.getSegmentNo()); |
| // scan all the carbondata files and get the latest task ID. |
| CarbonFile segmentDir = |
| FileFactory.getCarbonFile(segmentDirPath); |
| CarbonFile[] carbonDataFiles = segmentDir.listFiles(new CarbonFileFilter() { |
| @Override |
| public boolean accept(CarbonFile file) { |
| return file.getName().endsWith(CarbonCommonConstants.FACT_FILE_EXT); |
| } |
| }); |
| for (CarbonFile carbonDataFile : carbonDataFiles) { |
| dataFiles.add(carbonDataFile.getName()); |
| } |
| } |
| for (String name : dataFiles) { |
| long taskNumber = |
| Long.parseLong(CarbonTablePath.DataFileUtil.getTaskNo(name).split("_")[0]); |
| if (taskNumber > max) { |
| max = taskNumber; |
| } |
| } |
| // return max task No |
| return max; |
| |
| } |
| |
| /** |
| * This will tell whether the max query timeout has been expired or not. |
| * @param fileTimestamp |
| * @return |
| */ |
| public static boolean isMaxQueryTimeoutExceeded(long fileTimestamp) { |
| // record current time. |
| long currentTime = CarbonUpdateUtil.readCurrentTime(); |
| int maxTime; |
| try { |
| maxTime = Integer.parseInt(CarbonProperties.getInstance() |
| .getProperty(CarbonCommonConstants.MAX_QUERY_EXECUTION_TIME, |
| Integer.toString(CarbonCommonConstants.DEFAULT_MAX_QUERY_EXECUTION_TIME))); |
| } catch (NumberFormatException e) { |
| maxTime = CarbonCommonConstants.DEFAULT_MAX_QUERY_EXECUTION_TIME; |
| } |
| |
| long difference = currentTime - fileTimestamp; |
| |
| long minutesElapsed = (difference / (1000 * 60)); |
| |
| return minutesElapsed >= maxTime; |
| |
| } |
| |
| private static boolean deleteInvalidFiles(CarbonFile invalidFile) { |
| boolean isDeleted; |
| try { |
| LOGGER.info("deleting the invalid file : " + invalidFile.getName()); |
| CarbonUtil.deleteFoldersAndFiles(invalidFile); |
| isDeleted = true; |
| } catch (IOException | InterruptedException e) { |
| LOGGER.error("error in clean up of invalid files." + e.getMessage(), e); |
| isDeleted = false; |
| } |
| return isDeleted; |
| } |
| |
| public static boolean isBlockInvalid(SegmentStatus blockStatus) { |
| return blockStatus == SegmentStatus.COMPACTED || blockStatus == SegmentStatus.MARKED_FOR_DELETE; |
| } |
| |
| /** |
| * This will return the current time in millis. |
| * @return |
| */ |
| public static long readCurrentTime() { |
| return System.currentTimeMillis(); |
| } |
| |
| /** |
| * |
| * @param details |
| * @param segmentBlockCount |
| */ |
| public static void decrementDeletedBlockCount(SegmentUpdateDetails details, |
| Map<String, Long> segmentBlockCount) { |
| |
| String segId = details.getSegmentName(); |
| |
| segmentBlockCount.put(details.getSegmentName(), segmentBlockCount.get(segId) - 1); |
| |
| } |
| |
| /** |
| * |
| * @param segmentBlockCount |
| * @return |
| */ |
| public static List<Segment> getListOfSegmentsToMarkDeleted(Map<String, Long> segmentBlockCount) { |
| List<Segment> segmentsToBeDeleted = |
| new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| |
| for (Map.Entry<String, Long> eachSeg : segmentBlockCount.entrySet()) { |
| |
| if (eachSeg.getValue() == 0) { |
| segmentsToBeDeleted.add(new Segment(eachSeg.getKey(), (String) null)); |
| } |
| |
| } |
| return segmentsToBeDeleted; |
| } |
| |
| /** |
| * Return row count of input block |
| */ |
| public static long getRowCount(BlockMappingVO blockMappingVO, CarbonTable carbonTable) { |
| if (blockMappingVO.getBlockRowCountMapping().size() == 1 |
| && blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT) |
| != null) { |
| return blockMappingVO.getBlockRowCountMapping().get(CarbonCommonConstantsInternal.ROW_COUNT); |
| } |
| SegmentUpdateStatusManager updateStatusManager = |
| new SegmentUpdateStatusManager(carbonTable); |
| long rowCount = 0; |
| Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping(); |
| for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) { |
| String key = blockRowEntry.getKey(); |
| long alreadyDeletedCount = 0; |
| SegmentUpdateDetails detail = updateStatusManager.getDetailsForABlock(key); |
| if (detail != null) { |
| alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock()); |
| } |
| rowCount += (blockRowEntry.getValue() - alreadyDeletedCount); |
| } |
| return rowCount; |
| } |
| |
| /** |
| * |
| * @param blockMappingVO |
| * @param segmentUpdateStatusManager |
| */ |
| public static void createBlockDetailsMap(BlockMappingVO blockMappingVO, |
| SegmentUpdateStatusManager segmentUpdateStatusManager) { |
| |
| Map<String, Long> blockRowCountMap = blockMappingVO.getBlockRowCountMapping(); |
| |
| Map<String, RowCountDetailsVO> outputMap = |
| new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| |
| for (Map.Entry<String, Long> blockRowEntry : blockRowCountMap.entrySet()) { |
| String key = blockRowEntry.getKey(); |
| long alreadyDeletedCount = 0; |
| |
| SegmentUpdateDetails detail = segmentUpdateStatusManager.getDetailsForABlock(key); |
| |
| if (null != detail) { |
| |
| alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock()); |
| |
| } |
| |
| RowCountDetailsVO rowCountDetailsVO = |
| new RowCountDetailsVO(blockRowEntry.getValue(), alreadyDeletedCount); |
| outputMap.put(key, rowCountDetailsVO); |
| |
| } |
| |
| blockMappingVO.setCompleteBlockRowDetailVO(outputMap); |
| |
| } |
| |
| /** |
| * |
| * @param segID |
| * @param blockName |
| * @return |
| */ |
| public static String getSegmentBlockNameKey(String segID, String blockName, |
| boolean isPartitionTable) { |
| String blockNameWithOutPartAndBatchNo = blockName |
| .substring(blockName.indexOf(CarbonCommonConstants.HYPHEN) + 1, |
| blockName.lastIndexOf(CarbonTablePath.getCarbonDataExtension())) |
| .replace(CarbonTablePath.BATCH_PREFIX, CarbonCommonConstants.UNDERSCORE); |
| // to remove compressor name |
| int index = blockNameWithOutPartAndBatchNo.lastIndexOf(CarbonCommonConstants.POINT); |
| if (index != -1) { |
| blockNameWithOutPartAndBatchNo = blockNameWithOutPartAndBatchNo |
| .replace(blockNameWithOutPartAndBatchNo.substring(index), ""); |
| } |
| if (isPartitionTable) { |
| return blockNameWithOutPartAndBatchNo; |
| } else { |
| return segID + CarbonCommonConstants.FILE_SEPARATOR + blockNameWithOutPartAndBatchNo; |
| } |
| } |
| |
| /** |
| * Below method will be used to get the latest delete delta file timestamp |
| * @param deleteDeltaFiles |
| * @return latest delete delta file time stamp |
| */ |
| public static long getLatestDeleteDeltaTimestamp(String[] deleteDeltaFiles) { |
| long latestTimestamp = 0; |
| for (int i = 0; i < deleteDeltaFiles.length; i++) { |
| long convertTimeStampToLong = Long.parseLong( |
| CarbonTablePath.DataFileUtil.getTimeStampFromDeleteDeltaFile(deleteDeltaFiles[i])); |
| if (latestTimestamp < convertTimeStampToLong) { |
| latestTimestamp = convertTimeStampToLong; |
| } |
| } |
| return latestTimestamp; |
| } |
| |
| |
| /** |
| * Handling of the clean up of old carbondata files, index files , delete delta, |
| * update status files. |
| * |
| * @param table clean up will be handled on this table. |
| * @param isDryRun if clean files dryRun selected, then only size will be shown, |
| * files will not be deleted. |
| */ |
| public static long cleanUpDeltaFiles(CarbonTable table, boolean isDryRun) throws IOException { |
| |
| SegmentStatusManager ssm = |
| new SegmentStatusManager(table.getAbsoluteTableIdentifier(), table.getTableStatusVersion()); |
| LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(table.getMetadataPath(), |
| table.getTableStatusVersion()); |
| long totalSizeDeleted = 0; |
| ArrayList<CarbonFile> filesToBeDeleted = new ArrayList<>(); |
| SegmentUpdateStatusManager updateStatusManager = new SegmentUpdateStatusManager(table); |
| SegmentUpdateDetails[] segmentUpdateDetails = updateStatusManager.getUpdateStatusDetails(); |
| // hold all the segments updated so that wen can check the delta files in them, ne need to |
| // check the others. |
| Set<String> updatedSegments = new HashSet<>(); |
| for (SegmentUpdateDetails updateDetails : segmentUpdateDetails) { |
| updatedSegments.add(updateDetails.getSegmentName()); |
| } |
| boolean isInvalidFile = false; |
| // take the update status file name from 0th segment. |
| String validUpdateStatusFile = ssm.getUpdateStatusFileName(details); |
| // scan through each segment. |
| |
| if (table.isHivePartitionTable()) { |
| List<CarbonFile> partitionList = Arrays.stream(FileFactory.getCarbonFile(table |
| .getTablePath()).listFiles()).filter(partitionName -> partitionName.getName() |
| .contains("=")).collect(Collectors.toList()); |
| |
| List<CarbonFile> totalDeltaFiles = new ArrayList<>(); |
| |
| for (CarbonFile carbonFile : partitionList) { |
| totalDeltaFiles.addAll(carbonFile.listFiles(true).stream().filter(fileName -> fileName |
| .getName().endsWith(CarbonCommonConstants.DELETE_DELTA_FILE_EXT)) |
| .collect(Collectors.toList())); |
| } |
| SegmentUpdateDetails[] updateDetails = updateStatusManager.readLoadMetadata(); |
| |
| // Case 1: When deleteDeltaStartTimestamp = deleteDeltaEndTimestamp. in this case only 1 |
| // delta file is present and deltaFileStamps is NULL |
| // Case 2: When deleteDeltaStartTimestamp != deleteDeltaEndTimestamp. in this case more |
| // than 1 delta files are present, then can blindly read deltaFilesStamps variable |
| Arrays.stream(updateDetails).forEach(block -> { |
| if (block.getDeleteDeltaStartTimestamp().equals(block.getDeleteDeltaEndTimestamp())) { |
| totalDeltaFiles.removeIf(deltaFile -> deltaFile.getName().endsWith(block |
| .getDeleteDeltaEndTimestamp() + CarbonCommonConstants.DELETE_DELTA_FILE_EXT)); |
| } else { |
| block.getDeltaFileStamps().stream().forEach(fileName -> totalDeltaFiles |
| .removeIf(deltaFile -> deltaFile.getName().endsWith(fileName + |
| CarbonCommonConstants.DELETE_DELTA_FILE_EXT))); |
| } |
| }); |
| for (CarbonFile invalidFile: totalDeltaFiles) { |
| totalSizeDeleted += invalidFile.getSize(); |
| filesToBeDeleted.add(invalidFile); |
| } |
| } else { |
| for (LoadMetadataDetails segment : details) { |
| // if this segment is valid then only we will go for delta file deletion. |
| // if the segment is mark for delete or compacted then any way it will get deleted. |
| if (segment.getSegmentStatus() == SegmentStatus.SUCCESS |
| || segment.getSegmentStatus() == SegmentStatus.LOAD_PARTIAL_SUCCESS) { |
| // when there is no update operations done on table, then no need to go ahead. So |
| // just check the update delta start timestamp and proceed if not empty |
| if (!segment.getUpdateDeltaStartTimestamp().isEmpty() |
| || updatedSegments.contains(segment.getLoadName())) { |
| // take the list of files from this segment. |
| String segmentPath = CarbonTablePath.getSegmentPath( |
| table.getAbsoluteTableIdentifier().getTablePath(), segment.getLoadName()); |
| CarbonFile segDir = |
| FileFactory.getCarbonFile(segmentPath); |
| CarbonFile[] allSegmentFiles = segDir.listFiles(); |
| SegmentUpdateDetails[] updateDetails = updateStatusManager.readLoadMetadata(); |
| for (SegmentUpdateDetails block : updateDetails) { |
| CarbonFile[] invalidDeleteDeltaFiles; |
| if (!block.getSegmentName().equalsIgnoreCase(segment.getLoadName())) { |
| continue; |
| } |
| invalidDeleteDeltaFiles = updateStatusManager.getDeleteDeltaInvalidFilesList(block, |
| false, allSegmentFiles, isInvalidFile); |
| for (CarbonFile invalidFile : invalidDeleteDeltaFiles) { |
| totalSizeDeleted += invalidFile.getSize(); |
| filesToBeDeleted.add(invalidFile); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| // delete the update table status files which are old. |
| if (null != validUpdateStatusFile && !validUpdateStatusFile.isEmpty()) { |
| |
| final String updateStatusTimestamp = validUpdateStatusFile |
| .substring(validUpdateStatusFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1); |
| |
| String tablePath = table.getAbsoluteTableIdentifier().getTablePath(); |
| CarbonFile metaFolder = FileFactory.getCarbonFile( |
| CarbonTablePath.getMetadataPath(tablePath)); |
| |
| CarbonFile[] invalidUpdateStatusFiles = metaFolder.listFiles(new CarbonFileFilter() { |
| @Override |
| public boolean accept(CarbonFile file) { |
| if (file.getName().startsWith(CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME)) { |
| // CHECK if this is valid or not. |
| // we only send invalid ones to delete. |
| return !file.getName().endsWith(updateStatusTimestamp); |
| } |
| return false; |
| } |
| }); |
| |
| for (CarbonFile invalidFile : invalidUpdateStatusFiles) { |
| totalSizeDeleted += invalidFile.getSize(); |
| filesToBeDeleted.add(invalidFile); |
| } |
| } |
| if (!isDryRun) { |
| for (CarbonFile invalidFile : filesToBeDeleted) { |
| invalidFile.deleteFile(); |
| } |
| } |
| return totalSizeDeleted; |
| } |
| } |
| |
