/*
 * 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.scan.executor.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.block.AbstractIndex;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.index.IndexFilter;
import org.apache.carbondata.core.index.Segment;
import org.apache.carbondata.core.indexstore.BlockletDetailInfo;
import org.apache.carbondata.core.indexstore.blockletindex.BlockletDataRefNode;
import org.apache.carbondata.core.indexstore.blockletindex.IndexWrapper;
import org.apache.carbondata.core.memory.UnsafeMemoryManager;
import org.apache.carbondata.core.metadata.blocklet.BlockletInfo;
import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
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.QueryExecutor;
import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.core.scan.executor.util.QueryUtil;
import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
import org.apache.carbondata.core.scan.filter.FilterUtil;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.ProjectionDimension;
import org.apache.carbondata.core.scan.model.ProjectionMeasure;
import org.apache.carbondata.core.scan.model.QueryModel;
import org.apache.carbondata.core.stats.QueryStatistic;
import org.apache.carbondata.core.stats.QueryStatisticsConstants;
import org.apache.carbondata.core.util.BlockletIndexUtil;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonTimeStatisticsFactory;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
import org.apache.carbondata.core.util.ThreadLocalTaskInfo;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

/**
 * This class provides a skeletal implementation of the {@link QueryExecutor}
 * interface to minimize the effort required to implement this interface. This
 * will be used to prepare all the properties required for query execution
 */
public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(AbstractQueryExecutor.class.getName());
  /**
   * holder for query properties which will be used to execute the query
   */
  protected QueryExecutorProperties queryProperties;

  // whether to clear/free unsafe memory or not
  private boolean freeUnsafeMemory;

  /**
   * query result iterator which will execute the query
   * and give the result
   */
  protected CarbonIterator queryIterator;

  public AbstractQueryExecutor(Configuration configuration) {
    ThreadLocalSessionInfo.setConfigurationToCurrentThread(configuration);
    queryProperties = new QueryExecutorProperties();
  }

  public void setExecutorService(ExecutorService executorService) {
    // add executor service for query execution
    queryProperties.executorService = executorService;
  }

  /**
   * Below method will be used to fill the executor properties based on query
   * model it will parse the query model and get the detail and fill it in
   * query properties
   *
   * @param queryModel
   */
  protected void initQuery(QueryModel queryModel) throws IOException {
    LOGGER.info("Query will be executed on table: " + queryModel.getAbsoluteTableIdentifier()
        .getCarbonTableIdentifier().getTableName());
    this.freeUnsafeMemory = queryModel.isFreeUnsafeMemory();
    // Initializing statistics list to record the query statistics
    // creating copy on write to handle concurrent scenario
    queryProperties.queryStatisticsRecorder = queryModel.getStatisticsRecorder();
    if (null == queryProperties.queryStatisticsRecorder) {
      queryProperties.queryStatisticsRecorder =
          CarbonTimeStatisticsFactory.createExecutorRecorder(queryModel.getQueryId());
      queryModel.setStatisticsRecorder(queryProperties.queryStatisticsRecorder);
    }
    QueryStatistic queryStatistic = new QueryStatistic();
    // sort the block info
    // so block will be loaded in sorted order this will be required for
    // query execution
    Collections.sort(queryModel.getTableBlockInfos());
    queryProperties.dataBlocks = getDataBlocks(queryModel);
    queryStatistic
        .addStatistics(QueryStatisticsConstants.LOAD_BLOCKS_EXECUTOR, System.currentTimeMillis());
    queryProperties.queryStatisticsRecorder.recordStatistics(queryStatistic);

    // as aggregation will be executed in following order
    // 1.aggregate dimension expression
    // 2. expression
    // 3. query measure
    // so calculating the index of the expression start index
    // and measure column start index
    queryProperties.filterMeasures = new HashSet<>();
    queryProperties.complexFilterDimension = new HashSet<>();
    if (queryModel.getIndexFilter() != null) {
      QueryUtil.getAllFilterDimensionsAndMeasures(queryModel.getIndexFilter().getResolver(),
          queryProperties.complexFilterDimension, queryProperties.filterMeasures);
    }

    queryStatistic = new QueryStatistic();
    queryStatistic
        .addStatistics(QueryStatisticsConstants.LOAD_DICTIONARY, System.currentTimeMillis());
    queryProperties.queryStatisticsRecorder.recordStatistics(queryStatistic);
  }

  /**
   * Method returns the block(s) on which query will get executed
   *
   * @param queryModel
   * @return
   * @throws IOException
   */
  private List<AbstractIndex> getDataBlocks(QueryModel queryModel) throws IOException {
    Map<String, List<TableBlockInfo>> listMap = new LinkedHashMap<>();
    // this is introduced to handle the case when CACHE_LEVEL=BLOCK and there are few other indexes
    // like lucene, Bloom created on the table. In that case all the indexes will do blocklet
    // level pruning and blockInfo entries will be repeated with different blockletIds
    Map<String, DataFileFooter> filePathToFileFooterMapping = new HashMap<>();
    Map<String, SegmentProperties> filePathToSegmentPropertiesMap = new HashMap<>();
    for (TableBlockInfo blockInfo : queryModel.getTableBlockInfos()) {
      List<TableBlockInfo> tableBlockInfos = listMap.get(blockInfo.getFilePath());
      if (tableBlockInfos == null) {
        tableBlockInfos = new ArrayList<>();
        listMap.put(blockInfo.getFilePath(), tableBlockInfos);
      }
      SegmentProperties segmentProperties =
          filePathToSegmentPropertiesMap.get(blockInfo.getFilePath());
      BlockletDetailInfo blockletDetailInfo = blockInfo.getDetailInfo();
      // This case can come in 2 scenarios:
      // 1. old stores (1.1 or any prior version to 1.1) where blocklet information is not
      // available so read the blocklet information from block file
      // 2. CACHE_LEVEL is set to block
      // 3. CACHE_LEVEL is BLOCKLET but filter column min/max is not cached in driver
      if (null == blockletDetailInfo || blockletDetailInfo.getBlockletInfo() == null
          || blockletDetailInfo.isUseMinMaxForPruning()) {
        if (null != blockletDetailInfo) {
          blockInfo.setBlockOffset(blockletDetailInfo.getBlockFooterOffset());
        }
        DataFileFooter fileFooter = filePathToFileFooterMapping.get(blockInfo.getFilePath());
        if (null != blockInfo.getDataFileFooter()) {
          fileFooter = blockInfo.getDataFileFooter();
        }
        if (null == fileFooter) {
          blockInfo.setDetailInfo(null);
          fileFooter = CarbonUtil.readMetadataFile(blockInfo);
          // In case of non transactional table just set columnUniqueId as columnName to support
          // backward compatibility. non transactional tables column uniqueId is always equal to
          // columnName
          if (!queryModel.getTable().isTransactionalTable()) {
            QueryUtil.updateColumnUniqueIdForNonTransactionTable(fileFooter.getColumnInTable());
          }
          filePathToFileFooterMapping.put(blockInfo.getFilePath(), fileFooter);
          if (null == blockletDetailInfo) {
            blockletDetailInfo = QueryUtil.getBlockletDetailInfo(fileFooter, blockInfo);
          }
          blockInfo.setDetailInfo(blockletDetailInfo);
        }
        if (null == segmentProperties) {
          segmentProperties = new SegmentProperties(fileFooter.getColumnInTable());
          createFilterExpression(queryModel, segmentProperties);
          updateColumns(queryModel, fileFooter.getColumnInTable(), blockInfo.getFilePath());
          filePathToSegmentPropertiesMap.put(blockInfo.getFilePath(), segmentProperties);
        }
        readAndFillBlockletInfo(tableBlockInfos, blockInfo, blockletDetailInfo, fileFooter);
      } else {
        if (null == segmentProperties) {
          segmentProperties = new SegmentProperties(blockInfo.getDetailInfo().getColumnSchemas());
          createFilterExpression(queryModel, segmentProperties);
          updateColumns(queryModel, blockInfo.getDetailInfo().getColumnSchemas(),
              blockInfo.getFilePath());
          filePathToSegmentPropertiesMap.put(blockInfo.getFilePath(), segmentProperties);
        }
        tableBlockInfos.add(blockInfo);
      }
    }
    List<AbstractIndex> indexList = new ArrayList<>();
    for (List<TableBlockInfo> tableBlockInfos : listMap.values()) {
      indexList.add(new IndexWrapper(tableBlockInfos,
          filePathToSegmentPropertiesMap.get(tableBlockInfos.get(0).getFilePath())));
    }
    return indexList;
  }

  /**
   * It updates dimensions and measures of query model. In few scenarios like SDK user can configure
   * sort options per load, so if first load has c1 as integer column and configure as sort column
   * then carbon treat that as dimension.But in second load if user change the sort option then the
   * c1 become measure as by default integers are measures. So this method updates the measures to
   * dimensions and vice versa as per the index file schema.
   */
  private void updateColumns(QueryModel queryModel, List<ColumnSchema> columnsInTable,
      String filePath) throws IOException {
    if (queryModel.getTable().isTransactionalTable()) {
      return;
    }
    // First validate the schema of the carbondata file if the same column name have different
    // datatype
    boolean sameColumnSchemaList = BlockletIndexUtil
        .isSameColumnAndDifferentDatatypeInSchema(columnsInTable,
            queryModel.getTable().getTableInfo().getFactTable().getListOfColumns());
    if (!sameColumnSchemaList) {
      LOGGER.error("Datatype of the common columns present in " + filePath + " doesn't match with"
          + "the column's datatype in table schema");
      throw new IOException("All common columns present in the files doesn't have same datatype. "
          + "Unsupported operation on nonTransactional table. Check logs.");
    }
    List<ProjectionDimension> dimensions = queryModel.getProjectionDimensions();
    List<ProjectionMeasure> measures = queryModel.getProjectionMeasures();
    List<ProjectionDimension> updatedDims = new ArrayList<>();
    List<ProjectionMeasure> updatedMsrs = new ArrayList<>();

    // Check and update dimensions to measures if it is measure in index file schema
    for (ProjectionDimension dimension : dimensions) {
      int index = columnsInTable.indexOf(dimension.getDimension().getColumnSchema());
      if (index > -1) {
        if (!columnsInTable.get(index).isDimensionColumn()) {
          ProjectionMeasure measure = new ProjectionMeasure(
              new CarbonMeasure(columnsInTable.get(index), dimension.getDimension().getOrdinal(),
                  dimension.getDimension().getSchemaOrdinal()));
          measure.setOrdinal(dimension.getOrdinal());
          updatedMsrs.add(measure);
        } else {
          updatedDims.add(dimension);
        }
      } else {
        updatedDims.add(dimension);
      }
    }

    // Check and update measure to dimension if it is dimension in index file schema.
    for (ProjectionMeasure measure : measures) {
      int index = columnsInTable.indexOf(measure.getMeasure().getColumnSchema());
      if (index > -1) {
        if (columnsInTable.get(index).isDimensionColumn()) {
          ProjectionDimension dimension = new ProjectionDimension(
              new CarbonDimension(columnsInTable.get(index), measure.getMeasure().getOrdinal(), -1,
                  measure.getMeasure().getSchemaOrdinal()));
          dimension.setOrdinal(measure.getOrdinal());
          updatedDims.add(dimension);
        } else {
          updatedMsrs.add(measure);
        }
      } else {
        updatedMsrs.add(measure);
      }
    }
    // Clear and update the query model projections.
    dimensions.clear();
    dimensions.addAll(updatedDims);
    measures.clear();
    measures.addAll(updatedMsrs);
  }

  private void createFilterExpression(QueryModel queryModel, SegmentProperties properties) {
    if (queryModel.getIndexFilter() != null) {
      if (!queryModel.getIndexFilter().isResolvedOnSegment(properties)) {
        IndexFilter expression = new IndexFilter(properties, queryModel.getTable(),
            queryModel.getIndexFilter().getExpression());
        queryModel.setIndexFilter(expression);
      }
    }
  }

  /**
   * Read the file footer of block file and get the blocklets to query
   */
  private void readAndFillBlockletInfo(List<TableBlockInfo> tableBlockInfos,
      TableBlockInfo blockInfo, BlockletDetailInfo blockletDetailInfo, DataFileFooter fileFooter) {
    List<BlockletInfo> blockletList = fileFooter.getBlockletList();
    // cases when blockletID will be -1
    // 1. In case of legacy store
    // 2. In case CACHE_LEVEL is block and no other index apart from blocklet index is
    // created for a table
    // In all above cases entries will be according to the number of blocks and not according to
    // number of blocklets
    if (blockletDetailInfo.getBlockletId() != -1) {
      // fill the info only for given blockletId in detailInfo
      BlockletInfo blockletInfo = blockletList.get(blockletDetailInfo.getBlockletId());
      fillBlockletInfoToTableBlock(tableBlockInfos, blockInfo, fileFooter,
          blockletInfo, blockletDetailInfo.getBlockletId());
    } else {
      short count = 0;
      for (BlockletInfo blockletInfo : blockletList) {
        fillBlockletInfoToTableBlock(tableBlockInfos, blockInfo, fileFooter,
            blockletInfo, count);
        count++;
      }
    }
  }

  private void fillBlockletInfoToTableBlock(List<TableBlockInfo> tableBlockInfos,
      TableBlockInfo blockInfo, DataFileFooter fileFooter, BlockletInfo blockletInfo,
      short blockletId) {
    TableBlockInfo info = blockInfo.copy();
    BlockletDetailInfo detailInfo = info.getDetailInfo();
    // set column schema details
    detailInfo.setColumnSchemas(fileFooter.getColumnInTable());
    detailInfo.setRowCount(blockletInfo.getNumberOfRows());
    byte[][] maxValues = blockletInfo.getBlockletIndex().getMinMaxIndex().getMaxValues();
    byte[][] minValues = blockletInfo.getBlockletIndex().getMinMaxIndex().getMinValues();
    blockletInfo.getBlockletIndex().getMinMaxIndex().setMaxValues(maxValues);
    blockletInfo.getBlockletIndex().getMinMaxIndex().setMinValues(minValues);
    detailInfo.setBlockletInfo(blockletInfo);
    detailInfo.setBlockletId(blockletId);
    detailInfo.setPagesCount((short) blockletInfo.getNumberOfPages());
    tableBlockInfos.add(info);
  }

  protected List<BlockExecutionInfo> getBlockExecutionInfos(QueryModel queryModel)
      throws IOException {
    initQuery(queryModel);
    List<BlockExecutionInfo> blockExecutionInfoList = new ArrayList<BlockExecutionInfo>();
    // fill all the block execution infos for all the blocks selected in
    // query
    // and query will be executed based on that infos
    ReusableDataBuffer[] dimensionReusableDataBuffers = null;
    ReusableDataBuffer[] measureReusableDataBuffers = null;

    for (int i = 0; i < queryProperties.dataBlocks.size(); i++) {
      AbstractIndex abstractIndex = queryProperties.dataBlocks.get(i);
      BlockletDataRefNode dataRefNode =
          (BlockletDataRefNode) abstractIndex.getDataRefNode();
      final BlockExecutionInfo blockExecutionInfoForBlock =
          getBlockExecutionInfoForBlock(
              queryModel,
              abstractIndex,
              dataRefNode.numberOfNodes(),
              dataRefNode.getBlockInfos().get(0).getFilePath(),
              dataRefNode.getBlockInfos().get(0).getDeletedDeltaFilePath(),
              dataRefNode.getBlockInfos().get(0).getSegment());
      if (null == dimensionReusableDataBuffers || null == measureReusableDataBuffers) {
        dimensionReusableDataBuffers = blockExecutionInfoForBlock.getDimensionReusableDataBuffer();
        measureReusableDataBuffers = blockExecutionInfoForBlock.getMeasureReusableDataBuffer();
      } else {
        if (dimensionReusableDataBuffers.length == blockExecutionInfoForBlock
            .getDimensionReusableDataBuffer().length) {
          blockExecutionInfoForBlock.setDimensionReusableDataBuffer(dimensionReusableDataBuffers);
        }
        if (measureReusableDataBuffers.length == blockExecutionInfoForBlock
            .getMeasureReusableDataBuffer().length) {
          blockExecutionInfoForBlock.setMeasureReusableDataBuffer(measureReusableDataBuffers);
        }
      }
      blockExecutionInfoList.add(blockExecutionInfoForBlock);
    }
    if (null != queryModel.getStatisticsRecorder()) {
      QueryStatistic queryStatistic = new QueryStatistic();
      queryStatistic.addCountStatistic(QueryStatisticsConstants.SCAN_BLOCKS_NUM,
          blockExecutionInfoList.size());
      queryModel.getStatisticsRecorder().recordStatistics(queryStatistic);
    }
    return blockExecutionInfoList;
  }

  /**
   * Below method will be used to get the block execution info which is
   * required to execute any block  based on query model
   *
   * @param queryModel query model from user query
   * @param blockIndex block index
   * @return block execution info
   */
  private BlockExecutionInfo getBlockExecutionInfoForBlock(QueryModel queryModel,
      AbstractIndex blockIndex, int numberOfBlockletToScan, String filePath,
      String[] deleteDeltaFiles, Segment segment) {
    BlockExecutionInfo blockExecutionInfo = new BlockExecutionInfo();
    SegmentProperties segmentProperties = blockIndex.getSegmentProperties();
    // set actual query dimensions and measures. It may differ in case of restructure scenarios
    RestructureUtil.actualProjectionOfSegment(blockExecutionInfo, queryModel, segmentProperties);
    // below is to get only those dimension in query which is present in the
    // table block
    List<ProjectionDimension> projectDimensions = RestructureUtil
        .createDimensionInfoAndGetCurrentBlockQueryDimension(blockExecutionInfo,
            blockExecutionInfo.getActualQueryDimensions(), segmentProperties.getDimensions(),
            segmentProperties.getComplexDimensions(),
            blockExecutionInfo.getActualQueryMeasures().length,
            queryModel.getTable().getTableInfo().isTransactionalTable());
    boolean isStandardTable = CarbonUtil.isStandardCarbonTable(queryModel.getTable());
    String blockId = CarbonUtil
        .getBlockId(queryModel.getAbsoluteTableIdentifier(), filePath, segment.getSegmentNo(),
            queryModel.getTable().getTableInfo().isTransactionalTable(),
            isStandardTable, queryModel.getTable().isHivePartitionTable());
    if (!isStandardTable) {
      blockExecutionInfo.setBlockId(CarbonTablePath.getShortBlockIdForPartitionTable(blockId));
    } else {
      blockExecutionInfo.setBlockId(CarbonTablePath.getShortBlockId(blockId));
    }
    blockExecutionInfo.setDeleteDeltaFilePath(deleteDeltaFiles);
    blockExecutionInfo.setStartBlockletIndex(0);
    blockExecutionInfo.setNumberOfBlockletToScan(numberOfBlockletToScan);
    blockExecutionInfo.setProjectionDimensions(projectDimensions
        .toArray(new ProjectionDimension[0]));
    // get measures present in the current block
    List<ProjectionMeasure> projectionMeasures = RestructureUtil
        .createMeasureInfoAndGetCurrentBlockQueryMeasures(blockExecutionInfo,
            blockExecutionInfo.getActualQueryMeasures(), segmentProperties.getMeasures(),
            queryModel.getTable().getTableInfo().isTransactionalTable());
    blockExecutionInfo.setProjectionMeasures(
        projectionMeasures.toArray(new ProjectionMeasure[projectionMeasures.size()]));
    blockExecutionInfo.setDataBlock(blockIndex);
    // setting whether raw record query or not
    blockExecutionInfo.setRawRecordDetailQuery(queryModel.isForcedDetailRawQuery());
    // total number dimension
    blockExecutionInfo
        .setTotalNumberDimensionToRead(
            segmentProperties.getDimensionOrdinalToChunkMapping().size());
    blockExecutionInfo.setReadOnlyDelta(queryModel.isReadOnlyDelta());
    if (queryModel.isReadPageByPage()) {
      blockExecutionInfo.setPrefetchBlocklet(false);
      LOGGER.info("Query prefetch is: false, read page by page");
    } else {
      LOGGER.info("Query prefetch is: " + queryModel.isPreFetchData());
      blockExecutionInfo.setPrefetchBlocklet(queryModel.isPreFetchData());
    }
    // In case of fg index it should not go to direct fill.
    boolean fgIndexPathPresent = false;
    for (TableBlockInfo blockInfo : queryModel.getTableBlockInfos()) {
      fgIndexPathPresent = blockInfo.getIndexWriterPath() != null;
      if (fgIndexPathPresent) {
        queryModel.setDirectVectorFill(false);
        break;
      }
    }

    blockExecutionInfo.setDirectVectorFill(queryModel.isDirectVectorFill());
    blockExecutionInfo.setTotalNumberOfMeasureToRead(
        segmentProperties.getMeasuresOrdinalToChunkMapping().size());
    blockExecutionInfo.setComplexDimensionInfoMap(
        QueryUtil.getComplexDimensionsMap(
            projectDimensions,
            segmentProperties.getDimensionOrdinalToChunkMapping(),
            queryProperties.complexFilterDimension));
    if (null != queryModel.getIndexFilter()) {
      FilterResolverIntf filterResolverIntf;
      if (!filePath.startsWith(queryModel.getTable().getTablePath())) {
        filterResolverIntf = queryModel.getIndexFilter().getExternalSegmentResolver();
      } else {
        // loading the filter executor tree for filter evaluation
        filterResolverIntf = queryModel.getIndexFilter().getResolver();
      }
      blockExecutionInfo.setFilterExecutorTree(
          FilterUtil.getFilterExecutorTree(filterResolverIntf, segmentProperties,
              blockExecutionInfo.getComplexDimensionInfoMap(), false));
    }
    // expression measure
    List<CarbonMeasure> expressionMeasures =
        new ArrayList<CarbonMeasure>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    // setting all the dimension chunk indexes to be read from file
    int numberOfElementToConsider = 0;
    // list of dimensions to be projected
    Set<Integer> allProjectionListDimensionIndexes = new LinkedHashSet<>();
    // create a list of filter dimensions present in the current block
    Set<CarbonDimension> currentBlockFilterDimensions =
        getCurrentBlockFilterDimensions(queryProperties.complexFilterDimension, segmentProperties);
    int[] dimensionChunkIndexes = QueryUtil.getDimensionChunkIndexes(projectDimensions,
        segmentProperties.getDimensionOrdinalToChunkMapping(),
        currentBlockFilterDimensions, allProjectionListDimensionIndexes);
    ReusableDataBuffer[] dimensionBuffer = new ReusableDataBuffer[projectDimensions.size()];
    for (int i = 0; i < dimensionBuffer.length; i++) {
      dimensionBuffer[i] = new ReusableDataBuffer();
    }
    blockExecutionInfo.setDimensionReusableDataBuffer(dimensionBuffer);
    int numberOfColumnToBeReadInOneIO = Integer.parseInt(CarbonProperties.getInstance()
        .getProperty(CarbonV3DataFormatConstants.NUMBER_OF_COLUMN_TO_READ_IN_IO,
            CarbonV3DataFormatConstants.NUMBER_OF_COLUMN_TO_READ_IN_IO_DEFAULT_VALUE));

    if (dimensionChunkIndexes.length > 0) {
      numberOfElementToConsider = dimensionChunkIndexes[dimensionChunkIndexes.length - 1]
          == segmentProperties.getBlockToDimensionOrdinalMapping().size() - 1 ?
          dimensionChunkIndexes.length - 1 :
          dimensionChunkIndexes.length;
      blockExecutionInfo.setAllSelectedDimensionColumnIndexRange(
          CarbonUtil.getRangeIndex(dimensionChunkIndexes, numberOfElementToConsider,
              numberOfColumnToBeReadInOneIO));
    } else {
      blockExecutionInfo.setAllSelectedDimensionColumnIndexRange(new int[0][0]);
    }
    // get the list of updated filter measures present in the current block
    Set<CarbonMeasure> filterMeasures =
        getCurrentBlockFilterMeasures(queryProperties.filterMeasures, segmentProperties);
    // list of measures to be projected
    List<Integer> allProjectionListMeasureIndexes = new ArrayList<>();
    int[] measureChunkIndexes = QueryUtil.getMeasureChunkIndexes(
        projectionMeasures, expressionMeasures,
        segmentProperties.getMeasuresOrdinalToChunkMapping(), filterMeasures,
        allProjectionListMeasureIndexes);
    ReusableDataBuffer[] measureBuffer =
        new ReusableDataBuffer[allProjectionListMeasureIndexes.size()];
    for (int i = 0; i < measureBuffer.length; i++) {
      measureBuffer[i] = new ReusableDataBuffer();
    }
    blockExecutionInfo.setMeasureReusableDataBuffer(measureBuffer);
    if (measureChunkIndexes.length > 0) {

      numberOfElementToConsider = measureChunkIndexes[measureChunkIndexes.length - 1]
          == segmentProperties.getMeasures().size() - 1 ?
          measureChunkIndexes.length - 1 :
          measureChunkIndexes.length;
      // setting all the measure chunk indexes to be read from file
      blockExecutionInfo.setAllSelectedMeasureIndexRange(
          CarbonUtil.getRangeIndex(
              measureChunkIndexes, numberOfElementToConsider,
              numberOfColumnToBeReadInOneIO));
    } else {
      blockExecutionInfo.setAllSelectedMeasureIndexRange(new int[0][0]);
    }
    // setting the indexes of list of dimension in projection list
    blockExecutionInfo.setProjectionListDimensionIndexes(ArrayUtils.toPrimitive(
        allProjectionListDimensionIndexes
            .toArray(new Integer[allProjectionListDimensionIndexes.size()])));
    // setting the indexes of list of measures in projection list
    blockExecutionInfo.setProjectionListMeasureIndexes(ArrayUtils.toPrimitive(
        allProjectionListMeasureIndexes
            .toArray(new Integer[allProjectionListMeasureIndexes.size()])));
    // setting the size of fixed key column (dictionary column)
    blockExecutionInfo
        .setFixedLengthKeySize(getKeySize(projectDimensions, segmentProperties));
    List<Integer> dictionaryColumnChunkIndex = new ArrayList<Integer>();
    List<Integer> noDictionaryColumnChunkIndex = new ArrayList<Integer>();
    // get the block index to be read from file for query dimension
    // for both dictionary columns and no dictionary columns
    QueryUtil.fillQueryDimensionChunkIndexes(projectDimensions,
        segmentProperties.getDimensionOrdinalToChunkMapping(), dictionaryColumnChunkIndex,
        noDictionaryColumnChunkIndex);
    int[] queryDictionaryColumnChunkIndexes = ArrayUtils.toPrimitive(
        dictionaryColumnChunkIndex.toArray(new Integer[dictionaryColumnChunkIndex.size()]));
    // need to sort the dictionary column as for all dimension
    // column key will be filled based on key order
    if (!queryModel.isForcedDetailRawQuery()) {
      Arrays.sort(queryDictionaryColumnChunkIndexes);
    }
    blockExecutionInfo.setDictionaryColumnChunkIndex(queryDictionaryColumnChunkIndexes);
    // setting the no dictionary column block indexes
    blockExecutionInfo.setNoDictionaryColumnChunkIndexes(ArrayUtils.toPrimitive(
        noDictionaryColumnChunkIndex.toArray(new Integer[noDictionaryColumnChunkIndex.size()])));
    blockExecutionInfo.setComplexColumnParentBlockIndexes(
        getComplexDimensionParentBlockIndexes(projectDimensions));
    blockExecutionInfo.setVectorBatchCollector(queryModel.isVectorReader());
    DataTypeUtil.setDataTypeConverter(queryModel.getConverter());
    blockExecutionInfo.setRequiredRowId(queryModel.isRequiredRowId());
    return blockExecutionInfo;
  }

  /**
   * This method will be used to get fixed key length size this will be used
   * to create a row from column chunk
   *
   * @param queryDimension    query dimension
   * @param blockMetadataInfo block metadata info
   * @return key size
   */
  private int getKeySize(List<ProjectionDimension> queryDimension,
      SegmentProperties blockMetadataInfo) {
    // add the dimension block ordinal for each dictionary column
    // existing in the current block dimensions. Set is used because in case of column groups
    // ordinal of columns in a column group will be same
    Set<Integer> fixedLengthDimensionOrdinal =
        new HashSet<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    int counter = 0;
    while (counter < queryDimension.size()) {
      if (queryDimension.get(counter).getDimension().getNumberOfChild() > 0) {
        counter += queryDimension.get(counter).getDimension().getNumberOfChild();
      } else if (queryDimension.get(counter).getDimension().getDataType() != DataTypes.DATE) {
        counter++;
      } else {
        fixedLengthDimensionOrdinal.add(blockMetadataInfo.getDimensionOrdinalToChunkMapping()
            .get(queryDimension.get(counter).getDimension().getOrdinal()));
        counter++;
      }
    }
    int[] dictionaryColumnOrdinal = ArrayUtils.toPrimitive(
        fixedLengthDimensionOrdinal.toArray(new Integer[fixedLengthDimensionOrdinal.size()]));
    // calculate the size of existing query dictionary columns in this block
    if (blockMetadataInfo.getNumberOfDictDimensions() > 0) {
      int[] eachColumnValueSize = blockMetadataInfo.createDimColumnValueLength();
      int keySize = 0;
      for (int i = 0; i < dictionaryColumnOrdinal.length; i++) {
        keySize += eachColumnValueSize[dictionaryColumnOrdinal[i]];
      }
      return keySize;
    }
    return 0;
  }

  private int[] getComplexDimensionParentBlockIndexes(List<ProjectionDimension> queryDimensions) {
    List<Integer> parentBlockIndexList = new ArrayList<Integer>();
    for (ProjectionDimension queryDimension : queryDimensions) {
      if (queryDimension.getDimension().getDataType().isComplexType()) {
        if (null != queryDimension.getDimension().getComplexParentDimension()) {
          if (queryDimension.getDimension().isComplex()) {
            parentBlockIndexList.add(queryDimension.getDimension().getOrdinal());
          } else {
            parentBlockIndexList.add(queryDimension.getParentDimension().getOrdinal());
          }
        } else {
          parentBlockIndexList.add(queryDimension.getDimension().getOrdinal());
        }
      }
    }
    return ArrayUtils
        .toPrimitive(parentBlockIndexList.toArray(new Integer[parentBlockIndexList.size()]));
  }

  /**
   * This method will create the updated list of filter measures present in the current block
   *
   * @param queryFilterMeasures
   * @param segmentProperties
   * @return
   */
  private Set<CarbonMeasure> getCurrentBlockFilterMeasures(Set<CarbonMeasure> queryFilterMeasures,
      SegmentProperties segmentProperties) {
    if (!queryFilterMeasures.isEmpty()) {
      Set<CarbonMeasure> updatedFilterMeasures = new HashSet<>(queryFilterMeasures.size());
      for (CarbonMeasure queryMeasure : queryFilterMeasures) {
        CarbonMeasure measureFromCurrentBlock =
            segmentProperties.getMeasureFromCurrentBlock(queryMeasure);
        if (null != measureFromCurrentBlock) {
          updatedFilterMeasures.add(measureFromCurrentBlock);
        }
      }
      return updatedFilterMeasures;
    } else {
      return queryFilterMeasures;
    }
  }

  /**
   * This method will create the updated list of filter dimensions present in the current block
   *
   * @param queryFilterDimensions
   * @param segmentProperties
   * @return
   */
  private Set<CarbonDimension> getCurrentBlockFilterDimensions(
      Set<CarbonDimension> queryFilterDimensions, SegmentProperties segmentProperties) {
    if (!queryFilterDimensions.isEmpty()) {
      Set<CarbonDimension> updatedFilterDimensions = new HashSet<>(queryFilterDimensions.size());
      for (CarbonDimension queryDimension : queryFilterDimensions) {
        CarbonDimension dimensionFromCurrentBlock =
            segmentProperties.getDimensionFromCurrentBlock(queryDimension);
        if (null != dimensionFromCurrentBlock) {
          updatedFilterDimensions.add(dimensionFromCurrentBlock);
        }
      }
      return updatedFilterDimensions;
    } else {
      return queryFilterDimensions;
    }
  }

  /**
   * Below method will be used to finish the execution
   *
   * @throws QueryExecutionException
   */
  @Override
  public void finish() throws QueryExecutionException {
    CarbonUtil.clearBlockCache(queryProperties.dataBlocks);
    Throwable exceptionOccurred = null;
    if (null != queryIterator) {
      // catch if there is any exception so that it can be rethrown after clearing all the resources
      // else if any exception is thrown from this point executor service will not be terminated
      try {
        queryIterator.close();
      } catch (Throwable e) {
        exceptionOccurred = e;
      }
    }
    // clear all the unsafe memory used for the given task ID only if it is necessary to be cleared
    if (freeUnsafeMemory) {
      UnsafeMemoryManager.INSTANCE
          .freeMemoryAll(ThreadLocalTaskInfo.getCarbonTaskInfo().getTaskId());
      ThreadLocalTaskInfo.clearCarbonTaskInfo();
    }
    if (null != queryProperties.executorService) {
      // In case of limit query when number of limit records is already found so executors
      // must stop all the running execution otherwise it will keep running and will hit
      // the query performance.
      queryProperties.executorService.shutdownNow();
    }
    // if there is any exception re throw the exception
    if (null != exceptionOccurred) {
      throw new QueryExecutionException(exceptionOccurred);
    }
    DataTypeUtil.clearFormatter();
  }

}