/*
 * 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.index;

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 java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.carbondata.common.annotations.InterfaceAudience;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.index.dev.BlockletSerializer;
import org.apache.carbondata.core.index.dev.Index;
import org.apache.carbondata.core.index.dev.IndexFactory;
import org.apache.carbondata.core.index.dev.cgindex.CoarseGrainIndex;
import org.apache.carbondata.core.index.dev.expr.IndexInputSplitWrapper;
import org.apache.carbondata.core.index.dev.fgindex.FineGrainBlocklet;
import org.apache.carbondata.core.indexstore.Blocklet;
import org.apache.carbondata.core.indexstore.BlockletDetailsFetcher;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.indexstore.SegmentPropertiesFetcher;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.IndexSchema;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterUtil;
import org.apache.carbondata.core.scan.filter.executer.FilterExecutor;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.events.Event;
import org.apache.carbondata.events.OperationContext;
import org.apache.carbondata.events.OperationEventListener;

import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * Index at the table level, user can add any number of Index for one table, by
 * {@code
 *   CREATE INDEX index ON table
 *   USING 'IndexProvider'
 * }
 * Depends on the filter condition it can prune the data (blocklet or row level).
 */
@InterfaceAudience.Internal
public final class TableIndex extends OperationEventListener {

  private CarbonTable table;

  private AbsoluteTableIdentifier identifier;

  private IndexSchema indexSchema;

  private IndexFactory indexFactory;

  private BlockletDetailsFetcher blockletDetailsFetcher;

  private SegmentPropertiesFetcher segmentPropertiesFetcher;

  private static final Logger LOG =
      LogServiceFactory.getLogService(TableIndex.class.getName());

  /**
   * It is called to initialize and load the required table index metadata.
   */
  TableIndex(CarbonTable table, IndexSchema indexSchema,
      IndexFactory indexFactory, BlockletDetailsFetcher blockletDetailsFetcher,
      SegmentPropertiesFetcher segmentPropertiesFetcher) {
    this.identifier = table.getAbsoluteTableIdentifier();
    this.table = table;
    this.indexSchema = indexSchema;
    this.indexFactory = indexFactory;
    this.blockletDetailsFetcher = blockletDetailsFetcher;
    this.segmentPropertiesFetcher = segmentPropertiesFetcher;
  }

  public BlockletDetailsFetcher getBlockletDetailsFetcher() {
    return blockletDetailsFetcher;
  }

  public CarbonTable getTable() {
    return table;
  }

  /**
   * Pass the valid segments and prune the index using filter expression
   *
   * @param allSegments
   * @param filter
   * @return
   */
  public List<ExtendedBlocklet> prune(List<Segment> allSegments, final IndexFilter filter,
      final List<PartitionSpec> partitions) throws IOException {
    final List<ExtendedBlocklet> blocklets = new ArrayList<>();
    List<Segment> segments = getCarbonSegments(allSegments);
    final Map<Segment, List<Index>> indexes;
    boolean isFilterPresent = filter != null && !filter.isEmpty();
    Set<Path> partitionLocations = getPartitionLocations(partitions);
    if (table.isHivePartitionTable() && isFilterPresent && !partitionLocations.isEmpty()) {
      indexes = indexFactory.getIndexes(segments, partitionLocations, filter);
    } else {
      indexes = indexFactory.getIndexes(segments, filter);
    }

    if (indexes.isEmpty()) {
      return blocklets;
    }
    // for non-filter queries
    // for filter queries
    int totalFiles = 0;
    int indexesCount = 0;
    // In case if filter is present, then update the segments with index's segment list
    // based on segment or partition pruning
    if (isFilterPresent) {
      segments = new ArrayList<>(indexes.keySet());
    }
    for (Segment segment : segments) {
      for (Index index : indexes.get(segment)) {
        totalFiles += index.getNumberOfEntries();
        indexesCount++;
      }
    }
    int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();
    int carbonDriverPruningMultiThreadEnableFilesCount =
        CarbonProperties.getDriverPruningMultiThreadEnableFilesCount();
    if (numOfThreadsForPruning == 1 || indexesCount < numOfThreadsForPruning || totalFiles
            < carbonDriverPruningMultiThreadEnableFilesCount) {
      // use multi-thread, only if the files are more than 0.1 million.
      // As 0.1 million files block pruning can take only 1 second.
      // Doing multi-thread for smaller values is not recommended as
      // driver should have minimum threads opened to support multiple concurrent queries.
      if (filter == null || filter.isEmpty()) {
        // if filter is not passed, then return all the blocklets.
        return pruneWithoutFilter(segments, partitionLocations, blocklets);
      }
      return pruneWithFilter(segments, filter, partitionLocations, blocklets, indexes);
    }
    // handle by multi-thread
    List<ExtendedBlocklet> extendedBlocklets = pruneMultiThread(
        segments, filter, blocklets, indexes, totalFiles);
    return extendedBlocklets;
  }

  private List<Segment> getCarbonSegments(List<Segment> allSegments) {
    List<Segment> segments = new ArrayList<>();
    for (Segment segment : allSegments) {
      if (segment.isCarbonSegment()) {
        segments.add(segment);
      }
    }
    return segments;
  }

  private List<ExtendedBlocklet> pruneWithoutFilter(List<Segment> segments,
      Set<Path> partitionLocations, List<ExtendedBlocklet> blocklets) throws IOException {
    for (Segment segment : segments) {
      List<Blocklet> allBlocklets =
          blockletDetailsFetcher.getAllBlocklets(segment, partitionLocations);
      blocklets.addAll(
          addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(allBlocklets, segment),
              segment));
    }
    return blocklets;
  }

  private Set<Path> getPartitionLocations(List<PartitionSpec> partitionSpecs) {
    Set<Path> partitionsLocations = new HashSet<>();
    if (null != partitionSpecs) {
      for (PartitionSpec partitionSpec : partitionSpecs) {
        partitionsLocations.add(partitionSpec.getLocation());
      }
    }
    return partitionsLocations;
  }

  private List<ExtendedBlocklet> pruneWithFilter(List<Segment> segments, IndexFilter filter,
      Set<Path> partitionLocations, List<ExtendedBlocklet> blocklets,
      Map<Segment, List<Index>> indexes) throws IOException {
    for (Segment segment : segments) {
      if (segment == null ||
          indexes.get(segment) == null || indexes.get(segment).isEmpty()) {
        continue;
      }
      boolean isExternalSegment = segment.getSegmentPath() != null;
      List<Blocklet> pruneBlocklets = new ArrayList<>();
      SegmentProperties segmentProperties =
          segmentPropertiesFetcher.getSegmentProperties(segment, partitionLocations);
      if (filter.isResolvedOnSegment(segmentProperties)) {
        FilterExecutor filterExecutor;
        if (!isExternalSegment) {
          filterExecutor = FilterUtil
              .getFilterExecutorTree(filter.getResolver(), segmentProperties, null,
                  table.getMinMaxCacheColumns(segmentProperties), false);
        } else {
          filterExecutor = FilterUtil
              .getFilterExecutorTree(filter.getExternalSegmentResolver(), segmentProperties, null,
                  table.getMinMaxCacheColumns(segmentProperties), false);
        }
        for (Index index : indexes.get(segment)) {
          if (!isExternalSegment) {
            pruneBlocklets.addAll(index
                .prune(filter.getResolver(), segmentProperties, filterExecutor, this.table));
          } else {
            pruneBlocklets.addAll(index
                .prune(filter.getExternalSegmentResolver(), segmentProperties, filterExecutor,
                    this.table));
          }
        }
      } else {
        FilterExecutor filterExecutor;
        Expression expression = filter.getExpression();
        if (!isExternalSegment) {
          filterExecutor = FilterUtil.getFilterExecutorTree(
              new IndexFilter(segmentProperties, table, expression).getResolver(),
              segmentProperties, null, table.getMinMaxCacheColumns(segmentProperties), false);
        } else {
          filterExecutor = FilterUtil.getFilterExecutorTree(
              new IndexFilter(segmentProperties, table, expression).getExternalSegmentResolver(),
              segmentProperties, null, table.getMinMaxCacheColumns(segmentProperties), false);
        }
        for (Index index : indexes.get(segment)) {
          if (!isExternalSegment) {
            pruneBlocklets.addAll(index.prune(
                filter.getExpression(), segmentProperties, table, filterExecutor));
          } else {
            pruneBlocklets.addAll(index.prune(
                filter.getExternalSegmentFilter(), segmentProperties, table, filterExecutor));
          }
        }
      }
      blocklets.addAll(
          addSegmentId(blockletDetailsFetcher.getExtendedBlocklets(pruneBlocklets, segment),
              segment));
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> pruneMultiThread(List<Segment> segments,
      final IndexFilter filter, List<ExtendedBlocklet> blocklets,
      final Map<Segment, List<Index>> indexes, int totalFiles) {
    /*
     *********************************************************************************
     * Below is the example of how this part of code works.
     * consider a scenario of having 5 segments, 10 indexes in each segment,
     * and each index has one record. So total 50 records.
     *
     * indexes in each segment looks like below.
     * s0 [0-9], s1 [0-9], s2 [0-9], s3[0-9], s4[0-9]
     *
     * If number of threads are 4. so filesPerEachThread = 50/4 = 12 files per each thread.
     *
     * SegmentIndexGroup look like below: [SegmentId, fromIndex, toIndex]
     * In each segment only those indexes are processed between fromIndex and toIndex.
     *
     * Final result will be: (4 list created as numOfThreadsForPruning is 4)
     * Thread1 list: s0 [0-9], s1 [0-1]  : 12 files
     * Thread2 list: s1 [2-9], s2 [0-3]  : 12 files
     * Thread3 list: s2 [4-9], s3 [0-5]  : 12 files
     * Thread4 list: s3 [6-9], s4 [0-9]  : 14 files
     * so each thread will process almost equal number of records.
     *
     *********************************************************************************
     */

    int numOfThreadsForPruning = CarbonProperties.getNumOfThreadsForPruning();
    int filesPerEachThread = totalFiles / numOfThreadsForPruning;
    int prev;
    int filesCount = 0;
    int processedFileCount = 0;
    List<List<SegmentIndexGroup>> indexListForEachThread =
        new ArrayList<>(numOfThreadsForPruning);
    List<SegmentIndexGroup> segmentIndexGroupList = new ArrayList<>();
    for (Segment segment : segments) {
      List<Index> eachSegmentIndexList = indexes.get(segment);
      prev = 0;
      for (int i = 0; i < eachSegmentIndexList.size(); i++) {
        Index index = eachSegmentIndexList.get(i);
        filesCount += index.getNumberOfEntries();
        if (filesCount >= filesPerEachThread) {
          if (indexListForEachThread.size() != numOfThreadsForPruning - 1) {
            // not the last segmentList
            segmentIndexGroupList.add(new SegmentIndexGroup(segment, prev, i));
            // save the last value to process in next thread
            prev = i + 1;
            indexListForEachThread.add(segmentIndexGroupList);
            segmentIndexGroupList = new ArrayList<>();
            processedFileCount += filesCount;
            filesCount = 0;
          } else {
            // add remaining in the end
            processedFileCount += filesCount;
            filesCount = 0;
          }
        }
      }
      if (prev == 0 || prev != eachSegmentIndexList.size()) {
        // if prev == 0. Add a segment's all indexes
        // eachSegmentIndexList.size() != prev, adding the last remaining indexes of this segment
        segmentIndexGroupList
            .add(new SegmentIndexGroup(segment, prev, eachSegmentIndexList.size() - 1));
      }
    }
    // adding the last segmentList data
    indexListForEachThread.add(segmentIndexGroupList);
    processedFileCount += filesCount;
    if (processedFileCount != totalFiles) {
      // this should not happen
      throw new RuntimeException(" not all the files processed ");
    }
    if (indexListForEachThread.size() < numOfThreadsForPruning) {
      // If the total indexes fitted in lesser number of threads than numOfThreadsForPruning.
      // Launch only that many threads where indexes are fitted while grouping.
      LOG.info("indexes is distributed in " + indexListForEachThread.size() + " threads");
      numOfThreadsForPruning = indexListForEachThread.size();
    }
    LOG.info(
        "Number of threads selected for multi-thread block pruning is " + numOfThreadsForPruning
            + ". total files: " + totalFiles + ". total segments: " + segments.size());
    List<Future<Void>> results = new ArrayList<>(numOfThreadsForPruning);
    final Map<Segment, List<ExtendedBlocklet>> prunedBlockletMap =
        new ConcurrentHashMap<>(segments.size());
    final ExecutorService executorService = Executors.newFixedThreadPool(numOfThreadsForPruning);
    final String threadName = Thread.currentThread().getName();
    for (int i = 0; i < numOfThreadsForPruning; i++) {
      final List<SegmentIndexGroup> segmentIndexGroups = indexListForEachThread.get(i);
      results.add(executorService.submit(new Callable<Void>() {
        @Override
        public Void call() throws IOException {
          Thread.currentThread().setName(threadName);
          for (SegmentIndexGroup segmentIndexGroup : segmentIndexGroups) {
            List<ExtendedBlocklet> pruneBlocklets = new ArrayList<>();
            List<Index> indexList = indexes.get(segmentIndexGroup.getSegment());
            SegmentProperties segmentProperties =
                segmentPropertiesFetcher.getSegmentPropertiesFromIndex(indexList.get(0));
            Segment segment = segmentIndexGroup.getSegment();
            boolean isExternalSegment = segment.getSegmentPath() != null;
            if (filter.isResolvedOnSegment(segmentProperties)) {
              FilterExecutor filterExecutor;
              if (!isExternalSegment) {
                filterExecutor = FilterUtil
                    .getFilterExecutorTree(filter.getResolver(), segmentProperties, null,
                        table.getMinMaxCacheColumns(segmentProperties), false);
              } else {
                filterExecutor = FilterUtil
                    .getFilterExecutorTree(filter.getExternalSegmentResolver(), segmentProperties,
                        null, table.getMinMaxCacheColumns(segmentProperties), false);
              }
              for (int i = segmentIndexGroup.getFromIndex();
                   i <= segmentIndexGroup.getToIndex(); i++) {
                List<Blocklet> dmPruneBlocklets;
                if (!isExternalSegment) {
                  dmPruneBlocklets = indexList.get(i)
                      .prune(filter.getResolver(), segmentProperties, filterExecutor, table);
                } else {
                  dmPruneBlocklets = indexList.get(i)
                      .prune(filter.getExternalSegmentResolver(), segmentProperties, filterExecutor,
                          table);
                }
                pruneBlocklets.addAll(addSegmentId(
                    blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),
                    segment));
              }
            } else {
              Expression filterExpression = filter.getNewCopyOfExpression();
              FilterExecutor filterExecutor;
              if (!isExternalSegment) {
                filterExecutor = FilterUtil.getFilterExecutorTree(
                    new IndexFilter(segmentProperties, table, filterExpression).getResolver(),
                    segmentProperties, null, table.getMinMaxCacheColumns(segmentProperties), false);
              } else {
                filterExecutor = FilterUtil.getFilterExecutorTree(
                    new IndexFilter(segmentProperties, table, filterExpression)
                        .getExternalSegmentResolver(), segmentProperties, null,
                    table.getMinMaxCacheColumns(segmentProperties), false);
              }
              for (int i = segmentIndexGroup.getFromIndex();
                   i <= segmentIndexGroup.getToIndex(); i++) {
                List<Blocklet> dmPruneBlocklets;
                if (!isExternalSegment) {
                  dmPruneBlocklets = indexList.get(i)
                      .prune(filterExpression, segmentProperties, table, filterExecutor);
                } else {
                  dmPruneBlocklets = indexList.get(i)
                      .prune(filter.getExternalSegmentFilter(), segmentProperties, table,
                          filterExecutor);
                }
                pruneBlocklets.addAll(addSegmentId(
                    blockletDetailsFetcher.getExtendedBlocklets(dmPruneBlocklets, segment),
                    segment));
              }
            }
            synchronized (prunedBlockletMap) {
              List<ExtendedBlocklet> pruneBlockletsExisting =
                  prunedBlockletMap.get(segmentIndexGroup.getSegment());
              if (pruneBlockletsExisting != null) {
                pruneBlockletsExisting.addAll(pruneBlocklets);
              } else {
                prunedBlockletMap.put(segmentIndexGroup.getSegment(), pruneBlocklets);
              }
            }
          }
          return null;
        }
      }));
    }
    executorService.shutdown();
    try {
      executorService.awaitTermination(2, TimeUnit.HOURS);
    } catch (InterruptedException e) {
      LOG.error("Error in pruning index in multi-thread: " + e.getMessage());
    }
    // check for error
    for (Future<Void> result : results) {
      try {
        result.get();
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    }
    for (Map.Entry<Segment, List<ExtendedBlocklet>> entry : prunedBlockletMap.entrySet()) {
      blocklets.addAll(entry.getValue());
    }
    return blocklets;
  }

  private List<ExtendedBlocklet> addSegmentId(List<ExtendedBlocklet> pruneBlocklets,
      Segment segment) {
    for (ExtendedBlocklet blocklet : pruneBlocklets) {
      blocklet.setSegment(segment);
    }
    return pruneBlocklets;
  }

  /**
   * This is used for making the index distributable.
   * It takes the valid segments and returns all the indexes as distributable objects so that
   * it can be distributed across machines.
   *
   * @return
   */
  public List<IndexInputSplit> toDistributable(List<Segment> allSegments) {
    List<IndexInputSplit> distributableList = new ArrayList<>();
    List<Segment> segments = getCarbonSegments(allSegments);
    for (Segment segment : segments) {
      distributableList.addAll(indexFactory.toDistributable(segment));
    }
    return distributableList;
  }

  public IndexInputSplitWrapper toDistributableSegment(Segment segment, String uniqueId)
      throws IOException {
    return indexFactory.toDistributableSegment(segment, indexSchema, identifier, uniqueId);
  }

  /**
   * This method returns all the indexes corresponding to the distributable object
   *
   * @param distributable
   * @return
   * @throws IOException
   */
  public List<Index> getTableIndexes(IndexInputSplit distributable) throws IOException {
    return indexFactory.getIndexes(distributable);
  }

  /**
   * This method is used from any machine after it is distributed. It takes the distributable object
   * to prune the filters.
   *
   * @param distributable
   * @param filterExp
   * @return
   */
  public List<ExtendedBlocklet> prune(List<Index> indices, IndexInputSplit distributable,
      FilterResolverIntf filterExp, List<PartitionSpec> partitions) throws IOException {
    List<ExtendedBlocklet> detailedBlocklets = new ArrayList<>();
    List<Blocklet> blocklets = new ArrayList<>();
    Set<Path> partitionsToPrune = getPartitionLocations(partitions);
    SegmentProperties segmentProperties = segmentPropertiesFetcher
        .getSegmentProperties(distributable.getSegment(), partitionsToPrune);
    FilterExecutor filterExecutor = FilterUtil
        .getFilterExecutorTree(filterExp, segmentProperties,
            null, table.getMinMaxCacheColumns(segmentProperties),
            false);
    for (Index index : indices) {
      blocklets.addAll(index.prune(filterExp, segmentProperties, filterExecutor, table));
    }
    BlockletSerializer serializer = new BlockletSerializer();
    String writePath =
        identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + indexSchema
            .getIndexName();
    if (indexFactory.getIndexLevel() == IndexLevel.FG) {
      FileFactory.mkdirs(writePath);
    }
    for (Blocklet blocklet : blocklets) {
      ExtendedBlocklet detailedBlocklet = blockletDetailsFetcher
          .getExtendedBlocklet(blocklet, distributable.getSegment());
      if (indexFactory.getIndexLevel() == IndexLevel.FG) {
        String blockletWritePath =
            writePath + CarbonCommonConstants.FILE_SEPARATOR + System.nanoTime();
        detailedBlocklet.setIndexWriterPath(blockletWritePath);
        serializer.serializeBlocklet((FineGrainBlocklet) blocklet, blockletWritePath);
      }
      detailedBlocklet.setSegment(distributable.getSegment());
      detailedBlocklets.add(detailedBlocklet);
    }
    return detailedBlocklets;
  }

  /**
   * Clear only the indexes of the segments
   * @param segmentIds list of segmentIds to be cleared from cache.
   */
  public void clear(List<String> segmentIds) {
    for (String segment: segmentIds) {
      indexFactory.clear(segment);
    }
  }

  /**
   * Clears all index
   */
  public void clear() {
    if (null != indexFactory) {
      indexFactory.clear();
    }
  }

  /**
   * delete only the index of the segments
   */
  public void deleteIndexData(List<Segment> allSegments) throws IOException {
    List<Segment> segments = getCarbonSegments(allSegments);
    for (Segment segment: segments) {
      indexFactory.deleteIndexData(segment);
    }
  }

  /**
   * delete index data if any
   */
  public void deleteIndexData() {
    indexFactory.deleteIndexData();
  }

  /**
   * delete index data for a segment if any
   */
  public void deleteSegmentIndexData(String segmentNo) throws IOException {
    indexFactory.deleteSegmentIndexData(segmentNo);
  }

  public IndexSchema getIndexSchema() {
    return indexSchema;
  }

  public IndexFactory getIndexFactory() {
    return indexFactory;
  }

  @Override
  public void onEvent(Event event, OperationContext opContext) {
    indexFactory.fireEvent(event);
  }

  /**
   * Prune the index of the given segments and return the Map of blocklet path and row count
   *
   * @param allSegments
   * @param partitions
   * @return
   * @throws IOException
   */
  public Map<String, Long> getBlockRowCount(List<Segment> allSegments,
      final List<PartitionSpec> partitions, TableIndex defaultIndex)
      throws IOException {
    List<Segment> segments = getCarbonSegments(allSegments);
    Map<String, Long> blockletToRowCountMap = new HashMap<>();
    for (Segment segment : segments) {
      List<CoarseGrainIndex> indexes = defaultIndex.getIndexFactory().getIndexes(segment);
      for (CoarseGrainIndex index : indexes) {
        if (null != partitions) {
          // if it has partitioned index but there is no partitioned information stored, it means
          // partitions are dropped so return empty list.
          if (index.validatePartitionInfo(partitions)) {
            return new HashMap<>();
          }
        }
        index.getRowCountForEachBlock(segment, partitions, blockletToRowCountMap);
      }
    }
    return blockletToRowCountMap;
  }

  /**
   * Get the mapping of blocklet path and row count for all blocks. This method skips the
   * validation of partition info for countStar job with index server enabled.
   */
  public Map<String, Long> getBlockRowCount(TableIndex defaultIndex, List<Segment> allSegments,
      final List<PartitionSpec> partitions) throws IOException {
    List<Segment> segments = getCarbonSegments(allSegments);
    Map<String, Long> blockletToRowCountMap = new HashMap<>();
    for (Segment segment : segments) {
      List<CoarseGrainIndex> indexes = defaultIndex.getIndexFactory().getIndexes(segment);
      for (CoarseGrainIndex index : indexes) {
        index.getRowCountForEachBlock(segment, partitions, blockletToRowCountMap);
      }
    }
    return blockletToRowCountMap;
  }

  /**
   * Prune the index of the given segments and return the Map of blocklet path and row count
   *
   * @param allSegments
   * @param partitions
   * @return
   * @throws IOException
   */
  public long getRowCount(List<Segment> allSegments, final List<PartitionSpec> partitions,
      TableIndex defaultIndex) throws IOException {
    List<Segment> segments = getCarbonSegments(allSegments);
    long totalRowCount = 0L;
    for (Segment segment : segments) {
      List<CoarseGrainIndex> indexes = defaultIndex.getIndexFactory().getIndexes(segment);
      for (CoarseGrainIndex index : indexes) {
        totalRowCount += index.getRowCount(segment, partitions);
      }
    }
    return totalRowCount;
  }

  /**
   * Method to prune the segments based on task min/max values
   *
   */
  public List<Segment> pruneSegments(List<Segment> segments, FilterResolverIntf filterExp)
      throws IOException {
    List<Segment> prunedSegments = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    for (Segment segment : segments) {
      List<Index> indices = indexFactory.getIndexes(segment);
      for (Index index : indices) {
        if (index.isScanRequired(filterExp)) {
          // If any one task in a given segment contains the data that means the segment need to
          // be scanned and we need to validate further indexes in the same segment
          prunedSegments.add(segment);
          break;
        }
      }
    }
    return prunedSegments;
  }

}