content: /*
   * 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;
      }
    }
  
  }
  