| /* |
| * 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.processing.merger; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.apache.carbondata.common.Strings; |
| import org.apache.carbondata.common.logging.LogServiceFactory; |
| import org.apache.carbondata.core.constants.CarbonCommonConstants; |
| import org.apache.carbondata.core.datastore.block.TableBlockInfo; |
| import org.apache.carbondata.core.datastore.block.TaskBlockInfo; |
| import org.apache.carbondata.core.datastore.filesystem.CarbonFile; |
| import org.apache.carbondata.core.datastore.impl.FileFactory; |
| import org.apache.carbondata.core.indexstore.BlockletDetailInfo; |
| import org.apache.carbondata.core.indexstore.blockletindex.SegmentIndexFileStore; |
| import org.apache.carbondata.core.metadata.CarbonTableIdentifier; |
| import org.apache.carbondata.core.metadata.blocklet.DataFileFooter; |
| import org.apache.carbondata.core.metadata.datatype.DataType; |
| import org.apache.carbondata.core.metadata.datatype.DataTypes; |
| import org.apache.carbondata.core.metadata.schema.table.CarbonTable; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension; |
| import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure; |
| import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema; |
| import org.apache.carbondata.core.scan.executor.util.QueryUtil; |
| import org.apache.carbondata.core.scan.executor.util.RestructureUtil; |
| import org.apache.carbondata.core.scan.expression.ColumnExpression; |
| import org.apache.carbondata.core.scan.expression.Expression; |
| import org.apache.carbondata.core.scan.expression.LiteralExpression; |
| import org.apache.carbondata.core.scan.expression.conditional.EqualToExpression; |
| import org.apache.carbondata.core.scan.expression.conditional.GreaterThanExpression; |
| import org.apache.carbondata.core.scan.expression.conditional.LessThanEqualToExpression; |
| import org.apache.carbondata.core.scan.expression.logical.AndExpression; |
| import org.apache.carbondata.core.scan.expression.logical.OrExpression; |
| import org.apache.carbondata.core.statusmanager.LoadMetadataDetails; |
| import org.apache.carbondata.core.util.ByteUtil; |
| import org.apache.carbondata.core.util.CarbonUtil; |
| import org.apache.carbondata.core.util.DataTypeUtil; |
| import org.apache.carbondata.core.util.path.CarbonTablePath; |
| import org.apache.carbondata.format.IndexHeader; |
| import org.apache.carbondata.hadoop.CarbonInputSplit; |
| |
| import org.apache.log4j.Logger; |
| |
| /** |
| * Utility Class for the Compaction Flow. |
| */ |
| public class CarbonCompactionUtil { |
| |
| private static final Logger LOGGER = |
| LogServiceFactory.getLogService(CarbonCompactionUtil.class.getName()); |
| |
| |
| /** |
| * The processTableBlocks returns a Map<String, List<RawResultIterator>>. |
| * |
| * This Map has two K,V Pairs. The element with Key=UNSORTED_IDX |
| * is the list of all the iterators which yield unsorted data. |
| * |
| * This Map has two K,V Pairs. The element with Key=SORTED_IDX |
| * is the list of all the iterators which yield sorted data. |
| */ |
| public static final String UNSORTED_IDX = "UNSORTED_IDX"; |
| public static final String SORTED_IDX = "SORTED_IDX"; |
| |
| /** |
| * To create a mapping of Segment Id and TableBlockInfo. |
| * |
| * @param tableBlockInfoList |
| * @return |
| */ |
| public static Map<String, TaskBlockInfo> createMappingForSegments( |
| List<TableBlockInfo> tableBlockInfoList) { |
| |
| // stores taskBlockInfo of each segment |
| Map<String, TaskBlockInfo> segmentBlockInfoMapping = |
| new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| |
| |
| for (TableBlockInfo info : tableBlockInfoList) { |
| String segId = info.getSegmentId(); |
| // check if segId is already present in map |
| TaskBlockInfo taskBlockInfoMapping = segmentBlockInfoMapping.get(segId); |
| // extract task ID from file Path. |
| String taskNo = CarbonTablePath.DataFileUtil.getTaskNo(info.getFilePath()); |
| // if taskBlockInfo is not there, then create and add |
| if (null == taskBlockInfoMapping) { |
| taskBlockInfoMapping = new TaskBlockInfo(); |
| groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo); |
| // put the taskBlockInfo with respective segment id |
| segmentBlockInfoMapping.put(segId, taskBlockInfoMapping); |
| } else { |
| groupCorrespodingInfoBasedOnTask(info, taskBlockInfoMapping, taskNo); |
| } |
| } |
| return segmentBlockInfoMapping; |
| |
| } |
| |
| /** |
| * Grouping the taskNumber and list of TableBlockInfo. |
| * @param info |
| * @param taskBlockMapping |
| * @param taskNo |
| */ |
| private static void groupCorrespodingInfoBasedOnTask(TableBlockInfo info, |
| TaskBlockInfo taskBlockMapping, String taskNo) { |
| // get the corresponding list from task mapping. |
| List<TableBlockInfo> blockLists = taskBlockMapping.getTableBlockInfoList(taskNo); |
| if (null != blockLists) { |
| blockLists.add(info); |
| } else { |
| blockLists = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); |
| blockLists.add(info); |
| taskBlockMapping.addTableBlockInfoList(taskNo, blockLists); |
| } |
| } |
| |
| /** |
| * To create a mapping of Segment Id and DataFileFooter. |
| * |
| * @param tableBlockInfoList |
| * @return |
| */ |
| public static Map<String, List<DataFileFooter>> createDataFileFooterMappingForSegments( |
| List<TableBlockInfo> tableBlockInfoList, boolean isSortedTable) throws IOException { |
| Map<String, List<DataFileFooter>> segmentBlockInfoMapping = new HashMap<>(); |
| for (TableBlockInfo blockInfo : tableBlockInfoList) { |
| List<DataFileFooter> eachSegmentBlocks = new ArrayList<>(); |
| String segId = blockInfo.getSegmentId(); |
| DataFileFooter dataFileMatadata = null; |
| // check if segId is already present in map |
| List<DataFileFooter> metadataList = segmentBlockInfoMapping.get(segId); |
| // check to decide whether to read file footer of carbondata file forcefully. This will help |
| // in getting the schema last updated time based on which compaction flow is decided that |
| // whether it will go to restructure compaction flow or normal compaction flow. |
| // This decision will impact the compaction performance so it needs to be decided carefully |
| BlockletDetailInfo blockletDetailInfo = blockInfo.getDetailInfo(); |
| if (null == blockletDetailInfo || blockletDetailInfo.getBlockletInfo() == null || |
| blockInfo.getDetailInfo().getSchemaUpdatedTimeStamp() == 0L |
| || null == blockletDetailInfo.getBlockletInfo().isSorted() || !blockletDetailInfo |
| .getBlockletInfo().isSorted()) { |
| dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo, true); |
| if (blockletDetailInfo == null) { |
| blockletDetailInfo = QueryUtil.getBlockletDetailInfo(dataFileMatadata, blockInfo); |
| blockInfo.setDetailInfo(blockletDetailInfo); |
| } |
| if (null == dataFileMatadata.isSorted()) { |
| dataFileMatadata.setSorted(isSortedTable); |
| } |
| } else { |
| dataFileMatadata = CarbonUtil.readMetadataFile(blockInfo); |
| } |
| blockInfo.setDataFileFooter(dataFileMatadata); |
| if (null == metadataList) { |
| // if it is not present |
| eachSegmentBlocks.add(dataFileMatadata); |
| segmentBlockInfoMapping.put(segId, eachSegmentBlocks); |
| } else { |
| // if its already present then update the list. |
| metadataList.add(dataFileMatadata); |
| } |
| } |
| return segmentBlockInfoMapping; |
| } |
| |
| /** |
| * Check whether the file to indicate the compaction is present or not. |
| * @param metaFolderPath |
| * @return |
| */ |
| public static boolean isCompactionRequiredForTable(String metaFolderPath) { |
| String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.minorCompactionRequiredFile; |
| |
| String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.majorCompactionRequiredFile; |
| try { |
| if (FileFactory.isFileExist(minorCompactionStatusFile) || FileFactory |
| .isFileExist(majorCompactionStatusFile)) { |
| return true; |
| } |
| } catch (IOException e) { |
| LOGGER.error("Exception in isFileExist compaction request file " + e.getMessage(), e); |
| } |
| return false; |
| } |
| |
| /** |
| * Determine the type of the compaction received. |
| * @param metaFolderPath |
| * @return |
| */ |
| public static CompactionType determineCompactionType(String metaFolderPath) { |
| String minorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.minorCompactionRequiredFile; |
| |
| String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.majorCompactionRequiredFile; |
| try { |
| if (FileFactory.isFileExist(minorCompactionStatusFile)) { |
| return CompactionType.MINOR; |
| } |
| if (FileFactory.isFileExist(majorCompactionStatusFile)) { |
| return CompactionType.MAJOR; |
| } |
| |
| } catch (IOException e) { |
| LOGGER.error("Exception in determining the compaction request file " + e.getMessage(), e); |
| } |
| return CompactionType.MINOR; |
| } |
| |
| /** |
| * Delete the compation request file once the compaction is done. |
| * @param metaFolderPath |
| * @param compactionType |
| * @return |
| */ |
| public static boolean deleteCompactionRequiredFile(String metaFolderPath, |
| CompactionType compactionType) { |
| String compactionRequiredFile; |
| if (compactionType.equals(CompactionType.MINOR)) { |
| compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.minorCompactionRequiredFile; |
| } else { |
| compactionRequiredFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.majorCompactionRequiredFile; |
| } |
| try { |
| if (FileFactory |
| .isFileExist(compactionRequiredFile)) { |
| if (FileFactory |
| .getCarbonFile(compactionRequiredFile) |
| .delete()) { |
| LOGGER.info("Deleted the compaction request file " + compactionRequiredFile); |
| return true; |
| } else { |
| LOGGER.error("Unable to delete the compaction request file " + compactionRequiredFile); |
| } |
| } else { |
| LOGGER.info("Compaction request file is not present. file is : " + compactionRequiredFile); |
| } |
| } catch (IOException e) { |
| LOGGER.error("Exception in deleting the compaction request file " + e.getMessage(), e); |
| } |
| return false; |
| } |
| |
| /** |
| * Creation of the compaction request if someother compaction is in progress. |
| * @param metaFolderPath |
| * @param compactionType |
| * @return |
| */ |
| public static boolean createCompactionRequiredFile(String metaFolderPath, |
| CompactionType compactionType) { |
| String statusFile; |
| if (CompactionType.MINOR == compactionType) { |
| statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.minorCompactionRequiredFile; |
| } else { |
| statusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR |
| + CarbonCommonConstants.majorCompactionRequiredFile; |
| } |
| try { |
| if (!FileFactory.isFileExist(statusFile)) { |
| if (FileFactory.createNewFile(statusFile)) { |
| LOGGER.info("successfully created a compaction required file - " + statusFile); |
| return true; |
| } else { |
| LOGGER.error("Not able to create a compaction required file - " + statusFile); |
| return false; |
| } |
| } else { |
| LOGGER.info("Compaction request file : " + statusFile + " already exist."); |
| } |
| } catch (IOException e) { |
| LOGGER.error("Exception in creating the compaction request file " + e.getMessage(), e); |
| } |
| return false; |
| } |
| |
| /** |
| * This will check if any compaction request has been received for any table. |
| * |
| * @param carbonTables |
| * @return |
| */ |
| public static CarbonTable getNextTableToCompact(CarbonTable[] carbonTables, |
| List<CarbonTableIdentifier> skipList) { |
| for (CarbonTable ctable : carbonTables) { |
| String metadataPath = ctable.getMetadataPath(); |
| // check for the compaction required file and at the same time exclude the tables which are |
| // present in the skip list. |
| if (CarbonCompactionUtil.isCompactionRequiredForTable(metadataPath) && !skipList |
| .contains(ctable.getCarbonTableIdentifier())) { |
| return ctable; |
| } |
| } |
| return null; |
| } |
| |
| /** |
| * This method will update the {@code updatedColumnSchemaList} according to the master schema |
| */ |
| public static void updateColumnSchema( |
| CarbonTable carbonTable, List<ColumnSchema> updatedColumnSchemaList) { |
| List<CarbonDimension> masterDimensions = carbonTable.getVisibleDimensions(); |
| for (CarbonDimension dimension : masterDimensions) { |
| updatedColumnSchemaList.add(dimension.getColumnSchema()); |
| |
| if (dimension.getNumberOfChild() > 0) { |
| fillColumnSchemaListForComplexDims(dimension.getListOfChildDimensions(), |
| updatedColumnSchemaList); |
| } |
| } |
| // add measures to the column schema list |
| List<CarbonMeasure> masterSchemaMeasures = carbonTable.getVisibleMeasures(); |
| for (CarbonMeasure measure : masterSchemaMeasures) { |
| updatedColumnSchemaList.add(measure.getColumnSchema()); |
| } |
| } |
| |
| /** |
| * This method is to get the chile dimensions of the complex dimension and |
| * update the cardinality for all complex dimensions |
| */ |
| private static void fillColumnSchemaListForComplexDims(List<CarbonDimension> carbonDimensionsList, |
| List<ColumnSchema> updatedColumnSchemaList) { |
| for (CarbonDimension carbonDimension : carbonDimensionsList) { |
| updatedColumnSchemaList.add(carbonDimension.getColumnSchema()); |
| List<CarbonDimension> childDims = carbonDimension.getListOfChildDimensions(); |
| if (null != childDims && childDims.size() > 0) { |
| fillColumnSchemaListForComplexDims(childDims, updatedColumnSchemaList); |
| } |
| } |
| } |
| |
| /** |
| * This method will check for any restructured block in the blocks selected for compaction |
| * |
| * @param segmentMapping |
| * @param dataFileMetadataSegMapping |
| * @param tableLastUpdatedTime |
| * @return |
| */ |
| public static boolean checkIfAnyRestructuredBlockExists( |
| Map<String, TaskBlockInfo> segmentMapping, |
| Map<String, List<DataFileFooter>> dataFileMetadataSegMapping, |
| long tableLastUpdatedTime) { |
| |
| for (Map.Entry<String, TaskBlockInfo> segmentEntry : segmentMapping.entrySet()) { |
| String segmentId = segmentEntry.getKey(); |
| List<DataFileFooter> listMetadata = dataFileMetadataSegMapping.get(segmentId); |
| |
| if (isRestructured(listMetadata, tableLastUpdatedTime)) { |
| return true; |
| } |
| } |
| |
| return false; |
| } |
| |
| /** |
| * Returns if any element in the list of DataFileFooter |
| * is restructured or not. |
| * |
| * @param listMetadata |
| * @param tableLastUpdatedTime |
| * @return |
| */ |
| public static boolean isRestructured(List<DataFileFooter> listMetadata, |
| long tableLastUpdatedTime) { |
| /* |
| * TODO: only in case of add and drop this variable should be true |
| */ |
| for (DataFileFooter dataFileFooter : listMetadata) { |
| // if schema modified timestamp is greater than footer stored schema timestamp, |
| // it indicates it is a restructured block |
| if (tableLastUpdatedTime > dataFileFooter.getSchemaUpdatedTimeStamp()) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| // This method will return an Expression(And/Or) for each range based on the datatype |
| // This Expression will be passed to each task as a Filter Query to get the data |
| public static Expression getFilterExpressionForRange(CarbonColumn rangeColumn, Object minVal, |
| Object maxVal, DataType dataType) { |
| Expression finalExpr; |
| Expression exp1, exp2; |
| String colName = rangeColumn.getColName(); |
| |
| // In case of null values create an OrFilter expression and |
| // for other cases create and AndFilter Expression |
| if (null == minVal) { |
| // First task |
| exp1 = new EqualToExpression(new ColumnExpression(colName, dataType), |
| new LiteralExpression(null, dataType), true); |
| if (null == maxVal) { |
| // If both the min/max values are null, that means, if data contains only |
| // null value then pass only one expression as a filter expression |
| finalExpr = exp1; |
| } else { |
| exp2 = new LessThanEqualToExpression(new ColumnExpression(colName, dataType), |
| new LiteralExpression(maxVal, dataType)); |
| if (rangeColumn.getDataType() == DataTypes.DATE) { |
| exp2.setAlreadyResolved(true); |
| } |
| finalExpr = new OrExpression(exp1, exp2); |
| } |
| } else if (null == maxVal) { |
| // Last task |
| finalExpr = new GreaterThanExpression(new ColumnExpression(colName, dataType), |
| new LiteralExpression(minVal, dataType)); |
| if (rangeColumn.getDataType() == DataTypes.DATE) { |
| finalExpr.setAlreadyResolved(true); |
| } |
| } else { |
| // Remaining all intermediate ranges |
| exp1 = new GreaterThanExpression(new ColumnExpression(colName, dataType), |
| new LiteralExpression(minVal, dataType)); |
| exp2 = new LessThanEqualToExpression(new ColumnExpression(colName, dataType), |
| new LiteralExpression(maxVal, dataType)); |
| if (rangeColumn.getDataType() == DataTypes.DATE) { |
| exp2.setAlreadyResolved(true); |
| exp1.setAlreadyResolved(true); |
| } |
| finalExpr = new AndExpression(exp1, exp2); |
| } |
| return finalExpr; |
| } |
| |
| public static Object[] getOverallMinMax(CarbonInputSplit[] carbonInputSplits, |
| CarbonColumn rangeCol, boolean isSortCol) { |
| byte[] minVal = null; |
| byte[] maxVal = null; |
| int dictMinVal = Integer.MAX_VALUE; |
| int dictMaxVal = Integer.MIN_VALUE; |
| int idx = -1; |
| DataType dataType = rangeCol.getDataType(); |
| Object[] minMaxVals = new Object[2]; |
| boolean isDictEncode = rangeCol.getDataType() == DataTypes.DATE; |
| try { |
| for (CarbonInputSplit split : carbonInputSplits) { |
| DataFileFooter dataFileFooter = null; |
| dataFileFooter = |
| CarbonUtil.readMetadataFile(CarbonInputSplit.getTableBlockInfo(split), true); |
| |
| if (-1 == idx) { |
| List<ColumnSchema> allColumns = dataFileFooter.getColumnInTable(); |
| for (int i = 0; i < allColumns.size(); i++) { |
| if (allColumns.get(i).getColumnName().equalsIgnoreCase(rangeCol.getColName())) { |
| idx = i; |
| break; |
| } |
| } |
| } |
| if (isDictEncode) { |
| byte[] tempMin = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMinValues()[idx]; |
| int tempMinVal = CarbonUtil.getSurrogateInternal(tempMin, 0, tempMin.length); |
| byte[] tempMax = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMaxValues()[idx]; |
| int tempMaxVal = CarbonUtil.getSurrogateInternal(tempMax, 0, tempMax.length); |
| if (dictMinVal > tempMinVal) { |
| dictMinVal = tempMinVal; |
| } |
| if (dictMaxVal < tempMaxVal) { |
| dictMaxVal = tempMaxVal; |
| } |
| } else { |
| if (null == minVal) { |
| minVal = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMinValues()[idx]; |
| maxVal = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMaxValues()[idx]; |
| } else { |
| byte[] tempMin = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMinValues()[idx]; |
| byte[] tempMax = dataFileFooter.getBlockletIndex().getMinMaxIndex().getMaxValues()[idx]; |
| if (ByteUtil.compare(tempMin, minVal) <= 0) { |
| minVal = tempMin; |
| } |
| if (ByteUtil.compare(tempMax, maxVal) >= 0) { |
| maxVal = tempMax; |
| } |
| } |
| } |
| } |
| |
| // Based on how min/max value is stored in the footer we change the data |
| if (isDictEncode) { |
| minMaxVals[0] = dictMinVal; |
| minMaxVals[1] = dictMaxVal; |
| } else { |
| if (!isSortCol && (dataType == DataTypes.INT || dataType == DataTypes.LONG)) { |
| minMaxVals[0] = ByteUtil.toLong(minVal, 0, minVal.length); |
| minMaxVals[1] = ByteUtil.toLong(maxVal, 0, maxVal.length); |
| } else if (dataType == DataTypes.DOUBLE) { |
| minMaxVals[0] = ByteUtil.toDouble(minVal, 0, minVal.length); |
| minMaxVals[1] = ByteUtil.toDouble(maxVal, 0, maxVal.length); |
| } else { |
| minMaxVals[0] = |
| DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(minVal, dataType, true); |
| minMaxVals[1] = |
| DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(maxVal, dataType, true); |
| } |
| } |
| |
| } catch (IOException e) { |
| LOGGER.error(e.getMessage()); |
| } |
| return minMaxVals; |
| } |
| |
| public static int getTaskCountForSegment(CarbonInputSplit[] splits) { |
| Set<String> taskIdSet = new HashSet<>(); |
| for (CarbonInputSplit split : splits) { |
| String taskId = split.taskId; |
| taskIdSet.add(taskId); |
| } |
| return taskIdSet.size(); |
| } |
| |
| private static boolean compareSortColumns(CarbonTable table, List<ColumnSchema> fileColumns) { |
| // When sort_columns is modified, it will be consider as no_sort also. |
| List<CarbonDimension> sortColumnsOfSegment = new ArrayList<>(); |
| for (ColumnSchema column : fileColumns) { |
| if (column.isDimensionColumn() && column.isSortColumn()) { |
| sortColumnsOfSegment.add(new CarbonDimension(column, -1, -1, -1)); |
| } |
| } |
| if (sortColumnsOfSegment.size() < table.getNumberOfSortColumns()) { |
| return false; |
| } |
| List<CarbonDimension> sortColumnsOfTable = new ArrayList<>(); |
| for (CarbonDimension dimension : table.getVisibleDimensions()) { |
| if (dimension.isSortColumn()) { |
| sortColumnsOfTable.add(dimension); |
| } |
| } |
| int sortColumnNums = sortColumnsOfTable.size(); |
| if (sortColumnsOfSegment.size() < sortColumnNums) { |
| return false; |
| } |
| // compare sort_columns |
| for (int i = 0; i < sortColumnNums; i++) { |
| if (!RestructureUtil.isColumnMatches(table.isTransactionalTable(), sortColumnsOfTable.get(i), |
| sortColumnsOfSegment.get(i))) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * Returns if the DataFileFooter containing carbondata file contains |
| * sorted data or not. |
| * |
| * @param table |
| * @param footer |
| * @return |
| */ |
| public static boolean isSortedByCurrentSortColumns(CarbonTable table, DataFileFooter footer) { |
| if (footer.isSorted()) { |
| return compareSortColumns(table, footer.getColumnInTable()); |
| } else { |
| return false; |
| } |
| } |
| |
| public static boolean isSortedByCurrentSortColumns(CarbonTable table, LoadMetadataDetails load) { |
| List<String> sortColumnList = table.getSortColumns(); |
| if (sortColumnList.isEmpty()) { |
| return false; |
| } |
| // table sort_columns |
| String sortColumns = Strings.mkString( |
| sortColumnList.toArray(new String[sortColumnList.size()]), ","); |
| String segmentPath = |
| CarbonTablePath.getSegmentPath(table.getTablePath(), load.getLoadName()); |
| // segment sort_columns |
| String segmentSortColumns = getSortColumnsOfSegment(segmentPath); |
| if (segmentSortColumns == null) { |
| return false; |
| } else { |
| return segmentSortColumns.equalsIgnoreCase(sortColumns); |
| } |
| } |
| |
| private static String mkSortColumnsString( |
| List<org.apache.carbondata.format.ColumnSchema> columnList) { |
| StringBuilder builder = new StringBuilder(); |
| for (org.apache.carbondata.format.ColumnSchema column : columnList) { |
| if (column.isDimension()) { |
| Map<String, String> properties = column.getColumnProperties(); |
| if (properties != null) { |
| if (properties.get(CarbonCommonConstants.SORT_COLUMNS) != null) { |
| builder.append(column.column_name).append(","); |
| } |
| } |
| } |
| } |
| if (builder.length() > 1) { |
| return builder.substring(0, builder.length() - 1); |
| } else { |
| return null; |
| } |
| } |
| |
| public static String getSortColumnsOfSegment(String segmentFolder) { |
| CarbonFile[] files = SegmentIndexFileStore.getCarbonIndexFiles( |
| segmentFolder, FileFactory.getConfiguration()); |
| Set<Boolean> isSortSet = new HashSet<>(); |
| Set<String> sortColumnsSet = new HashSet<>(); |
| if (files != null) { |
| for (CarbonFile file : files) { |
| IndexHeader indexHeader = SegmentIndexFileStore.readIndexHeader( |
| file.getCanonicalPath(), FileFactory.getConfiguration()); |
| if (indexHeader != null) { |
| if (indexHeader.isSetIs_sort()) { |
| isSortSet.add(indexHeader.is_sort); |
| if (indexHeader.is_sort) { |
| sortColumnsSet.add(mkSortColumnsString(indexHeader.getTable_columns())); |
| } |
| } else { |
| // if is_sort is not set, it will be old store and consider as local_sort by default. |
| sortColumnsSet.add(mkSortColumnsString(indexHeader.getTable_columns())); |
| } |
| } |
| if (isSortSet.size() >= 2 || sortColumnsSet.size() >= 2) { |
| break; |
| } |
| } |
| } |
| // for all index files, sort_columns should be same |
| if (isSortSet.size() <= 1 && sortColumnsSet.size() == 1) { |
| return sortColumnsSet.iterator().next(); |
| } else { |
| return null; |
| } |
| } |
| |
| } |
