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

import java.io.IOException;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datamap.DataMapFilter;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datamap.dev.DataMapModel;
import org.apache.carbondata.core.datamap.dev.cgdatamap.CoarseGrainDataMap;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.indexstore.AbstractMemoryDMStore;
import org.apache.carbondata.core.indexstore.BlockMetaInfo;
import org.apache.carbondata.core.indexstore.Blocklet;
import org.apache.carbondata.core.indexstore.ExtendedBlocklet;
import org.apache.carbondata.core.indexstore.PartitionSpec;
import org.apache.carbondata.core.indexstore.SafeMemoryDMStore;
import org.apache.carbondata.core.indexstore.UnsafeMemoryDMStore;
import org.apache.carbondata.core.indexstore.row.DataMapRow;
import org.apache.carbondata.core.indexstore.row.DataMapRowImpl;
import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.metadata.ColumnarFormatVersion;
import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
import org.apache.carbondata.core.metadata.blocklet.index.BlockletIndex;
import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex;
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.ColumnSchema;
import org.apache.carbondata.core.profiler.ExplainCollector;
import org.apache.carbondata.core.scan.expression.Expression;
import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
import org.apache.carbondata.core.scan.filter.FilterUtil;
import org.apache.carbondata.core.scan.filter.executer.FilterExecuter;
import org.apache.carbondata.core.scan.filter.executer.ImplicitColumnFilterExecutor;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.BlockletDataMapUtil;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.DataFileFooterConverter;
import org.apache.carbondata.core.util.path.CarbonTablePath;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * Datamap implementation for block.
 */
public class BlockDataMap extends CoarseGrainDataMap
    implements BlockletDataMapRowIndexes, Serializable {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(BlockDataMap.class.getName());

  protected static final long serialVersionUID = -2170289352240810993L;
  /**
   * for CACHE_LEVEL=BLOCK and legacy store default blocklet id will be -1
   */
  private static final short BLOCK_DEFAULT_BLOCKLET_ID = -1;
  /**
   * store which will hold all the block or blocklet entries in one task
   */
  protected AbstractMemoryDMStore memoryDMStore;
  /**
   * task summary holder store
   */
  protected AbstractMemoryDMStore taskSummaryDMStore;
  /**
   * index of segmentProperties in the segmentProperties holder
   */
  protected transient SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper
      segmentPropertiesWrapper;
  /**
   * flag to check for store from 1.1 or any prior version
   */
  protected boolean isLegacyStore;
  /**
   * flag to be used for forming the complete file path from file name. It will be true in case of
   * partition table and non transactional table
   */
  protected boolean isFilePathStored;

  @Override
  public void init(DataMapModel dataMapModel)
      throws IOException, MemoryException {
    long startTime = System.currentTimeMillis();
    assert (dataMapModel instanceof BlockletDataMapModel);
    BlockletDataMapModel blockletDataMapInfo = (BlockletDataMapModel) dataMapModel;
    DataFileFooterConverter fileFooterConverter =
        new DataFileFooterConverter(dataMapModel.getConfiguration());
    List<DataFileFooter> indexInfo = null;
    if (blockletDataMapInfo.getIndexInfos() == null || blockletDataMapInfo.getIndexInfos()
        .isEmpty()) {
      indexInfo = fileFooterConverter
          .getIndexInfo(blockletDataMapInfo.getFilePath(), blockletDataMapInfo.getFileData(),
              blockletDataMapInfo.getCarbonTable().isTransactionalTable());
    } else {
      // when index info is already read and converted to data file footer object
      indexInfo = blockletDataMapInfo.getIndexInfos();
    }
    Path path = new Path(blockletDataMapInfo.getFilePath());
    // store file path only in case of partition table, non transactional table and flat folder
    // structure
    byte[] filePath;
    boolean isPartitionTable = blockletDataMapInfo.getCarbonTable().isHivePartitionTable();
    if (isPartitionTable || !blockletDataMapInfo.getCarbonTable().isTransactionalTable() ||
        blockletDataMapInfo.getCarbonTable().isSupportFlatFolder() ||
        // if the segment data is written in tablepath then no need to store whole path of file.
        !blockletDataMapInfo.getFilePath().startsWith(
            blockletDataMapInfo.getCarbonTable().getTablePath())) {
      filePath = path.getParent().toString().getBytes(CarbonCommonConstants.DEFAULT_CHARSET);
      isFilePathStored = true;
    } else {
      filePath = new byte[0];
    }
    byte[] fileName = path.getName().getBytes(CarbonCommonConstants.DEFAULT_CHARSET);
    byte[] segmentId =
        blockletDataMapInfo.getSegmentId().getBytes(CarbonCommonConstants.DEFAULT_CHARSET);
    if (!indexInfo.isEmpty()) {
      DataFileFooter fileFooter = indexInfo.get(0);
      // store for 1.1 or any prior version will not have any blocklet information in file footer
      isLegacyStore = fileFooter.getBlockletList() == null;
      // init segment properties and create schema
      SegmentProperties segmentProperties = initSegmentProperties(blockletDataMapInfo, fileFooter);
      createMemorySchema(blockletDataMapInfo);
      createSummaryDMStore(blockletDataMapInfo);
      CarbonRowSchema[] taskSummarySchema = getTaskSummarySchema();
      // check for legacy store and load the metadata
      DataMapRowImpl summaryRow =
          loadMetadata(taskSummarySchema, segmentProperties, blockletDataMapInfo, indexInfo);
      finishWriting(taskSummarySchema, filePath, fileName, segmentId, summaryRow);
    }
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          "Time taken to load blocklet datamap from file : " + dataMapModel.getFilePath() + " is "
              + (System.currentTimeMillis() - startTime));
    }
  }

  private void finishWriting(CarbonRowSchema[] taskSummarySchema, byte[] filePath, byte[] fileName,
      byte[] segmentId, DataMapRowImpl summaryRow) throws MemoryException {
    if (memoryDMStore != null) {
      memoryDMStore.finishWriting();
    }
    if (null != taskSummaryDMStore) {
      addTaskSummaryRowToUnsafeMemoryStore(taskSummarySchema, summaryRow, filePath, fileName,
          segmentId);
      taskSummaryDMStore.finishWriting();
    }
  }

  /**
   * Method to check the cache level and load metadata based on that information
   *
   * @param segmentProperties
   * @param blockletDataMapInfo
   * @param indexInfo
   * @throws IOException
   * @throws MemoryException
   */
  protected DataMapRowImpl loadMetadata(CarbonRowSchema[] taskSummarySchema,
      SegmentProperties segmentProperties, BlockletDataMapModel blockletDataMapInfo,
      List<DataFileFooter> indexInfo) throws IOException, MemoryException {
    if (isLegacyStore) {
      return loadBlockInfoForOldStore(taskSummarySchema, segmentProperties, blockletDataMapInfo,
          indexInfo);
    } else {
      return loadBlockMetaInfo(taskSummarySchema, segmentProperties, blockletDataMapInfo,
          indexInfo);
    }
  }

  /**
   * initialise segment properties
   *
   * @param fileFooter
   * @throws IOException
   */
  private SegmentProperties initSegmentProperties(BlockletDataMapModel blockletDataMapInfo,
      DataFileFooter fileFooter) {
    List<ColumnSchema> columnInTable = fileFooter.getColumnInTable();
    int[] columnCardinality = fileFooter.getSegmentInfo().getColumnCardinality();
    segmentPropertiesWrapper = SegmentPropertiesAndSchemaHolder.getInstance()
        .addSegmentProperties(blockletDataMapInfo.getCarbonTable(),
            columnInTable, columnCardinality, blockletDataMapInfo.getSegmentId());
    return segmentPropertiesWrapper.getSegmentProperties();
  }

  /**
   * This is old store scenario, here blocklet information is not available in index
   * file so load only block info. Old store refers to store in 1.1 or prior to 1.1 version
   *
   * @param blockletDataMapInfo
   * @param indexInfo
   * @throws IOException
   * @throws MemoryException
   */
  protected DataMapRowImpl loadBlockInfoForOldStore(CarbonRowSchema[] taskSummarySchema,
      SegmentProperties segmentProperties, BlockletDataMapModel blockletDataMapInfo,
      List<DataFileFooter> indexInfo) throws IOException, MemoryException {
    DataMapRowImpl summaryRow = null;
    CarbonRowSchema[] schema = getFileFooterEntrySchema();
    boolean[] minMaxFlag = new boolean[segmentProperties.getColumnsValueSize().length];
    FilterUtil.setMinMaxFlagForLegacyStore(minMaxFlag, segmentProperties);
    long totalRowCount = 0;
    for (DataFileFooter fileFooter : indexInfo) {
      TableBlockInfo blockInfo = fileFooter.getBlockInfo().getTableBlockInfo();
      BlockMetaInfo blockMetaInfo =
          blockletDataMapInfo.getBlockMetaInfoMap().get(blockInfo.getFilePath());
      // Here it loads info about all blocklets of index
      // Only add if the file exists physically. There are scenarios which index file exists inside
      // merge index but related carbondata files are deleted. In that case we first check whether
      // the file exists physically or not
      if (null != blockMetaInfo) {
        BlockletIndex blockletIndex = fileFooter.getBlockletIndex();
        BlockletMinMaxIndex minMaxIndex = blockletIndex.getMinMaxIndex();
        summaryRow = loadToUnsafeBlock(schema, taskSummarySchema, fileFooter, segmentProperties,
            getMinMaxCacheColumns(), blockInfo.getFilePath(), summaryRow,
            blockMetaInfo, minMaxIndex.getMinValues(), minMaxIndex.getMaxValues(), minMaxFlag);
        totalRowCount += fileFooter.getNumberOfRows();
      }
    }
    List<Short> blockletCountList = new ArrayList<>();
    blockletCountList.add((short) 0);
    byte[] blockletCount = convertRowCountFromShortToByteArray(blockletCountList);
    // set the total row count
    summaryRow.setLong(totalRowCount, TASK_ROW_COUNT);
    summaryRow.setByteArray(blockletCount, taskSummarySchema.length - 1);
    setMinMaxFlagForTaskSummary(summaryRow, taskSummarySchema, segmentProperties, minMaxFlag);
    return summaryRow;
  }

  protected void setMinMaxFlagForTaskSummary(DataMapRow summaryRow,
      CarbonRowSchema[] taskSummarySchema, SegmentProperties segmentProperties,
      boolean[] minMaxFlag) {
    // add min max flag for all the dimension columns
    boolean[] minMaxFlagValuesForColumnsToBeCached = BlockletDataMapUtil
        .getMinMaxFlagValuesForColumnsToBeCached(segmentProperties, getMinMaxCacheColumns(),
            minMaxFlag);
    addMinMaxFlagValues(summaryRow, taskSummarySchema[TASK_MIN_MAX_FLAG],
        minMaxFlagValuesForColumnsToBeCached, TASK_MIN_MAX_FLAG);
  }

  /**
   * Method to load block metadata information
   *
   * @param blockletDataMapInfo
   * @param indexInfo
   * @throws IOException
   * @throws MemoryException
   */
  private DataMapRowImpl loadBlockMetaInfo(CarbonRowSchema[] taskSummarySchema,
      SegmentProperties segmentProperties, BlockletDataMapModel blockletDataMapInfo,
      List<DataFileFooter> indexInfo) throws IOException, MemoryException {
    String tempFilePath = null;
    DataFileFooter previousDataFileFooter = null;
    int footerCounter = 0;
    byte[][] blockMinValues = null;
    byte[][] blockMaxValues = null;
    DataMapRowImpl summaryRow = null;
    List<Short> blockletCountInEachBlock = new ArrayList<>(indexInfo.size());
    short totalBlockletsInOneBlock = 0;
    boolean isLastFileFooterEntryNeedToBeAdded = false;
    CarbonRowSchema[] schema = getFileFooterEntrySchema();
    // flag for each block entry
    boolean[] minMaxFlag = new boolean[segmentProperties.getColumnsValueSize().length];
    Arrays.fill(minMaxFlag, true);
    // min max flag for task summary
    boolean[] taskSummaryMinMaxFlag = new boolean[segmentProperties.getColumnsValueSize().length];
    Arrays.fill(taskSummaryMinMaxFlag, true);
    long totalRowCount = 0;
    for (DataFileFooter fileFooter : indexInfo) {
      TableBlockInfo blockInfo = fileFooter.getBlockInfo().getTableBlockInfo();
      BlockMetaInfo blockMetaInfo =
          blockletDataMapInfo.getBlockMetaInfoMap().get(blockInfo.getFilePath());
      footerCounter++;
      if (blockMetaInfo != null) {
        // this variable will be used for adding the DataMapRow entry every time a unique block
        // path is encountered
        if (null == tempFilePath) {
          tempFilePath = blockInfo.getFilePath();
          // 1st time assign the min and max values from the current file footer
          blockMinValues = fileFooter.getBlockletIndex().getMinMaxIndex().getMinValues();
          blockMaxValues = fileFooter.getBlockletIndex().getMinMaxIndex().getMaxValues();
          updateMinMaxFlag(fileFooter, minMaxFlag);
          updateMinMaxFlag(fileFooter, taskSummaryMinMaxFlag);
          previousDataFileFooter = fileFooter;
          totalBlockletsInOneBlock++;
        } else if (blockInfo.getFilePath().equals(tempFilePath)) {
          // After iterating over all the blocklets that belong to one block we need to compute the
          // min and max at block level. So compare min and max values and update if required
          BlockletMinMaxIndex currentFooterMinMaxIndex =
              fileFooter.getBlockletIndex().getMinMaxIndex();
          blockMinValues =
              compareAndUpdateMinMax(currentFooterMinMaxIndex.getMinValues(), blockMinValues, true);
          blockMaxValues =
              compareAndUpdateMinMax(currentFooterMinMaxIndex.getMaxValues(), blockMaxValues,
                  false);
          updateMinMaxFlag(fileFooter, minMaxFlag);
          updateMinMaxFlag(fileFooter, taskSummaryMinMaxFlag);
          totalBlockletsInOneBlock++;
        }
        // as one task contains entries for all the blocklets we need iterate and load only the
        // with unique file path because each unique path will correspond to one
        // block in the task. OR condition is to handle the loading of last file footer
        if (!blockInfo.getFilePath().equals(tempFilePath) || footerCounter == indexInfo.size()) {
          TableBlockInfo previousBlockInfo =
              previousDataFileFooter.getBlockInfo().getTableBlockInfo();
          summaryRow = loadToUnsafeBlock(schema, taskSummarySchema, previousDataFileFooter,
              segmentProperties, getMinMaxCacheColumns(), previousBlockInfo.getFilePath(),
              summaryRow,
              blockletDataMapInfo.getBlockMetaInfoMap().get(previousBlockInfo.getFilePath()),
              blockMinValues, blockMaxValues, minMaxFlag);
          totalRowCount += previousDataFileFooter.getNumberOfRows();
          minMaxFlag = new boolean[segmentProperties.getColumnsValueSize().length];
          Arrays.fill(minMaxFlag, true);
          // flag to check whether last file footer entry is different from previous entry.
          // If yes then it need to be added at last
          isLastFileFooterEntryNeedToBeAdded =
              (footerCounter == indexInfo.size()) && (!blockInfo.getFilePath()
                  .equals(tempFilePath));
          // assign local variables values using the current file footer
          tempFilePath = blockInfo.getFilePath();
          blockMinValues = fileFooter.getBlockletIndex().getMinMaxIndex().getMinValues();
          blockMaxValues = fileFooter.getBlockletIndex().getMinMaxIndex().getMaxValues();
          updateMinMaxFlag(fileFooter, minMaxFlag);
          updateMinMaxFlag(fileFooter, taskSummaryMinMaxFlag);
          previousDataFileFooter = fileFooter;
          blockletCountInEachBlock.add(totalBlockletsInOneBlock);
          // for next block count will start from 1 because a row is created whenever a new file
          // path comes. Here already a new file path has come so the count should start from 1
          totalBlockletsInOneBlock = 1;
        }
      }
    }
    // add the last file footer entry
    if (isLastFileFooterEntryNeedToBeAdded) {
      summaryRow =
          loadToUnsafeBlock(schema, taskSummarySchema, previousDataFileFooter, segmentProperties,
              getMinMaxCacheColumns(),
              previousDataFileFooter.getBlockInfo().getTableBlockInfo().getFilePath(), summaryRow,
              blockletDataMapInfo.getBlockMetaInfoMap()
                  .get(previousDataFileFooter.getBlockInfo().getTableBlockInfo().getFilePath()),
              blockMinValues, blockMaxValues, minMaxFlag);
      totalRowCount += previousDataFileFooter.getNumberOfRows();
      blockletCountInEachBlock.add(totalBlockletsInOneBlock);
    }
    byte[] blockletCount = convertRowCountFromShortToByteArray(blockletCountInEachBlock);
    // set the total row count
    summaryRow.setLong(totalRowCount, TASK_ROW_COUNT);
    // blocklet count index is the last index
    summaryRow.setByteArray(blockletCount, taskSummarySchema.length - 1);
    setMinMaxFlagForTaskSummary(summaryRow, taskSummarySchema, segmentProperties,
        taskSummaryMinMaxFlag);
    return summaryRow;
  }

  protected void updateMinMaxFlag(DataFileFooter fileFooter, boolean[] minMaxFlag) {
    BlockletDataMapUtil
        .updateMinMaxFlag(fileFooter.getBlockletIndex().getMinMaxIndex(), minMaxFlag);
  }

  private byte[] convertRowCountFromShortToByteArray(List<Short> blockletCountInEachBlock) {
    int bufferSize = blockletCountInEachBlock.size() * 2;
    ByteBuffer byteBuffer = ByteBuffer.allocate(bufferSize);
    for (Short blockletCount : blockletCountInEachBlock) {
      byteBuffer.putShort(blockletCount);
    }
    byteBuffer.rewind();
    return byteBuffer.array();
  }

  protected void setLocations(String[] locations, DataMapRow row, int ordinal)
      throws UnsupportedEncodingException {
    // Add location info
    String locationStr = StringUtils.join(locations, ',');
    row.setByteArray(locationStr.getBytes(CarbonCommonConstants.DEFAULT_CHARSET), ordinal);
  }

  /**
   * Load information for the block.It is the case can happen only for old stores
   * where blocklet information is not available in index file. So load only block information
   * and read blocklet information in executor.
   */
  protected DataMapRowImpl loadToUnsafeBlock(CarbonRowSchema[] schema,
      CarbonRowSchema[] taskSummarySchema, DataFileFooter fileFooter,
      SegmentProperties segmentProperties, List<CarbonColumn> minMaxCacheColumns, String filePath,
      DataMapRowImpl summaryRow, BlockMetaInfo blockMetaInfo, byte[][] minValues,
      byte[][] maxValues, boolean[] minMaxFlag) {
    // Add one row to maintain task level min max for segment pruning
    if (summaryRow == null) {
      summaryRow = new DataMapRowImpl(taskSummarySchema);
    }
    DataMapRow row = new DataMapRowImpl(schema);
    int ordinal = 0;
    int taskMinMaxOrdinal = 1;
    // get min max values for columns to be cached
    byte[][] minValuesForColumnsToBeCached = BlockletDataMapUtil
        .getMinMaxForColumnsToBeCached(segmentProperties, minMaxCacheColumns, minValues);
    byte[][] maxValuesForColumnsToBeCached = BlockletDataMapUtil
        .getMinMaxForColumnsToBeCached(segmentProperties, minMaxCacheColumns, maxValues);
    boolean[] minMaxFlagValuesForColumnsToBeCached = BlockletDataMapUtil
        .getMinMaxFlagValuesForColumnsToBeCached(segmentProperties, minMaxCacheColumns, minMaxFlag);
    row.setRow(addMinMax(schema[ordinal], minValuesForColumnsToBeCached), ordinal);
    // compute and set task level min values
    addTaskMinMaxValues(summaryRow, taskSummarySchema, taskMinMaxOrdinal,
        minValuesForColumnsToBeCached, TASK_MIN_VALUES_INDEX, true);
    ordinal++;
    taskMinMaxOrdinal++;
    row.setRow(addMinMax(schema[ordinal], maxValuesForColumnsToBeCached), ordinal);
    // compute and set task level max values
    addTaskMinMaxValues(summaryRow, taskSummarySchema, taskMinMaxOrdinal,
        maxValuesForColumnsToBeCached, TASK_MAX_VALUES_INDEX, false);
    ordinal++;
    // add total rows in one carbondata file
    row.setInt((int) fileFooter.getNumberOfRows(), ordinal++);
    // add file name
    byte[] filePathBytes =
        getFileNameFromPath(filePath).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
    row.setByteArray(filePathBytes, ordinal++);
    // add version number
    row.setShort(fileFooter.getVersionId().number(), ordinal++);
    // add schema updated time
    row.setLong(fileFooter.getSchemaUpdatedTimeStamp(), ordinal++);
    // add block offset
    row.setLong(fileFooter.getBlockInfo().getTableBlockInfo().getBlockOffset(), ordinal++);
    try {
      setLocations(blockMetaInfo.getLocationInfo(), row, ordinal++);
      // store block size
      row.setLong(blockMetaInfo.getSize(), ordinal++);
      // add min max flag for all the dimension columns
      addMinMaxFlagValues(row, schema[ordinal], minMaxFlagValuesForColumnsToBeCached, ordinal);
      memoryDMStore.addIndexRow(schema, row);
    } catch (Exception e) {
      String message = "Load to unsafe failed for block: " + filePath;
      LOGGER.error(message, e);
      throw new RuntimeException(message, e);
    }
    return summaryRow;
  }

  protected void addMinMaxFlagValues(DataMapRow row, CarbonRowSchema carbonRowSchema,
      boolean[] minMaxFlag, int ordinal) {
    CarbonRowSchema[] minMaxFlagSchema =
        ((CarbonRowSchema.StructCarbonRowSchema) carbonRowSchema).getChildSchemas();
    DataMapRow minMaxFlagRow = new DataMapRowImpl(minMaxFlagSchema);
    int flagOrdinal = 0;
    // min value adding
    for (int i = 0; i < minMaxFlag.length; i++) {
      minMaxFlagRow.setBoolean(minMaxFlag[i], flagOrdinal++);
    }
    row.setRow(minMaxFlagRow, ordinal);
  }

  protected String getFileNameFromPath(String filePath) {
    return CarbonTablePath.getCarbonDataFileName(filePath);
  }

  protected String getFilePath() {
    if (isFilePathStored) {
      return getTableTaskInfo(SUMMARY_INDEX_PATH);
    }
    // create the segment directory path
    String tablePath = segmentPropertiesWrapper.getTableIdentifier().getTablePath();
    String segmentId = getTableTaskInfo(SUMMARY_SEGMENTID);
    return CarbonTablePath.getSegmentPath(tablePath, segmentId);
  }

  protected String getFileNameWithFilePath(DataMapRow dataMapRow, String filePath) {
    String fileName = filePath + CarbonCommonConstants.FILE_SEPARATOR + new String(
        dataMapRow.getByteArray(FILE_PATH_INDEX), CarbonCommonConstants.DEFAULT_CHARSET_CLASS)
        + CarbonTablePath.getCarbonDataExtension();
    return FileFactory.getUpdatedFilePath(fileName);
  }

  private void addTaskSummaryRowToUnsafeMemoryStore(CarbonRowSchema[] taskSummarySchema,
      DataMapRow summaryRow, byte[] filePath, byte[] fileName, byte[] segmentId) {
    // write the task summary info to unsafe memory store
    if (null != summaryRow) {
      summaryRow.setByteArray(fileName, SUMMARY_INDEX_FILE_NAME);
      summaryRow.setByteArray(segmentId, SUMMARY_SEGMENTID);
      if (filePath.length > 0) {
        summaryRow.setByteArray(filePath, SUMMARY_INDEX_PATH);
      }
      try {
        taskSummaryDMStore.addIndexRow(taskSummarySchema, summaryRow);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  }

  protected DataMapRow addMinMax(CarbonRowSchema carbonRowSchema,
      byte[][] minValues) {
    CarbonRowSchema[] minSchemas =
        ((CarbonRowSchema.StructCarbonRowSchema) carbonRowSchema).getChildSchemas();
    DataMapRow minRow = new DataMapRowImpl(minSchemas);
    int minOrdinal = 0;
    // min value adding
    for (int i = 0; i < minValues.length; i++) {
      minRow.setByteArray(minValues[i], minOrdinal++);
    }
    return minRow;
  }

  /**
   * This method will compute min/max values at task level
   *
   * @param taskMinMaxRow
   * @param carbonRowSchema
   * @param taskMinMaxOrdinal
   * @param minMaxValue
   * @param ordinal
   * @param isMinValueComparison
   */
  protected void addTaskMinMaxValues(DataMapRow taskMinMaxRow, CarbonRowSchema[] carbonRowSchema,
      int taskMinMaxOrdinal, byte[][] minMaxValue, int ordinal, boolean isMinValueComparison) {
    DataMapRow row = taskMinMaxRow.getRow(ordinal);
    byte[][] updatedMinMaxValues = null;
    if (null == row) {
      CarbonRowSchema[] minSchemas =
          ((CarbonRowSchema.StructCarbonRowSchema) carbonRowSchema[taskMinMaxOrdinal])
              .getChildSchemas();
      row = new DataMapRowImpl(minSchemas);
      updatedMinMaxValues = minMaxValue;
    } else {
      byte[][] existingMinMaxValues = getMinMaxValue(taskMinMaxRow, ordinal);
      updatedMinMaxValues =
          compareAndUpdateMinMax(minMaxValue, existingMinMaxValues, isMinValueComparison);
    }
    int minMaxOrdinal = 0;
    // min/max value adding
    for (int i = 0; i < updatedMinMaxValues.length; i++) {
      row.setByteArray(updatedMinMaxValues[i], minMaxOrdinal++);
    }
    taskMinMaxRow.setRow(row, ordinal);
  }

  /**
   * This method will do min/max comparison of values and update if required
   *
   * @param minMaxValueCompare1
   * @param minMaxValueCompare2
   * @param isMinValueComparison
   */
  private byte[][] compareAndUpdateMinMax(byte[][] minMaxValueCompare1,
      byte[][] minMaxValueCompare2, boolean isMinValueComparison) {
    // Compare and update min max values
    byte[][] updatedMinMaxValues = new byte[minMaxValueCompare1.length][];
    System.arraycopy(minMaxValueCompare1, 0, updatedMinMaxValues, 0, minMaxValueCompare1.length);
    for (int i = 0; i < minMaxValueCompare1.length; i++) {
      int compare = ByteUtil.UnsafeComparer.INSTANCE
          .compareTo(minMaxValueCompare2[i], minMaxValueCompare1[i]);
      if (isMinValueComparison) {
        if (compare < 0) {
          updatedMinMaxValues[i] = minMaxValueCompare2[i];
        }
      } else if (compare > 0) {
        updatedMinMaxValues[i] = minMaxValueCompare2[i];
      }
    }
    return updatedMinMaxValues;
  }

  protected void createMemorySchema(BlockletDataMapModel blockletDataMapModel)
      throws MemoryException {
    memoryDMStore = getMemoryDMStore(blockletDataMapModel.isAddToUnsafe());
  }

  /**
   * Creates the schema to store summary information or the information which can be stored only
   * once per datamap. It stores datamap level max/min of each column and partition information of
   * datamap
   *
   * @throws MemoryException
   */
  protected void createSummaryDMStore(BlockletDataMapModel blockletDataMapModel)
      throws MemoryException {
    taskSummaryDMStore = getMemoryDMStore(blockletDataMapModel.isAddToUnsafe());
  }

  @Override
  public boolean isScanRequired(FilterResolverIntf filterExp) {
    FilterExecuter filterExecuter = FilterUtil.getFilterExecuterTree(
        filterExp, getSegmentProperties(), null, getMinMaxCacheColumns(), false);
    DataMapRow unsafeRow = taskSummaryDMStore
        .getDataMapRow(getTaskSummarySchema(), taskSummaryDMStore.getRowCount() - 1);
    boolean isScanRequired = FilterExpressionProcessor
        .isScanRequired(filterExecuter, getMinMaxValue(unsafeRow, TASK_MAX_VALUES_INDEX),
            getMinMaxValue(unsafeRow, TASK_MIN_VALUES_INDEX),
            getMinMaxFlag(unsafeRow, TASK_MIN_MAX_FLAG));
    if (isScanRequired) {
      return true;
    }
    return false;
  }

  protected List<CarbonColumn> getMinMaxCacheColumns() {
    return segmentPropertiesWrapper.getMinMaxCacheColumns();
  }

  /**
   * for CACHE_LEVEL=BLOCK, each entry in memoryDMStore is for a block
   * if data is not legacy store, we can get blocklet count from taskSummaryDMStore
   */
  protected short getBlockletNumOfEntry(int index) {
    if (isLegacyStore) {
      // dummy value
      return 0;
    } else {
      final byte[] bytes = getBlockletRowCountForEachBlock();
      // if the segment data is written in tablepath
      // then the reuslt of getBlockletRowCountForEachBlock will be empty.
      if (bytes.length == 0) {
        return 0;
      } else {
        return ByteBuffer.wrap(bytes).getShort(index * CarbonCommonConstants.SHORT_SIZE_IN_BYTE);
      }
    }
  }

  // get total block number in this datamap
  public int getTotalBlocks() {
    if (isLegacyStore) {
      // dummy value
      return 0;
    } else {
      return memoryDMStore.getRowCount();
    }
  }

  // get total blocklet number in this datamap
  protected int getTotalBlocklets() {
    ByteBuffer byteBuffer = ByteBuffer.wrap(getBlockletRowCountForEachBlock());
    int sum = 0;
    while (byteBuffer.hasRemaining()) {
      sum += byteBuffer.getShort();
    }
    return sum;
  }

  @Override
  public long getRowCount(Segment segment, List<PartitionSpec> partitions) {
    long totalRowCount =
        taskSummaryDMStore.getDataMapRow(getTaskSummarySchema(), 0).getLong(TASK_ROW_COUNT);
    if (totalRowCount == 0) {
      Map<String, Long> blockletToRowCountMap = new HashMap<>();
      getRowCountForEachBlock(segment, partitions, blockletToRowCountMap);
      for (long blockletRowCount : blockletToRowCountMap.values()) {
        totalRowCount += blockletRowCount;
      }
    } else {
      if (taskSummaryDMStore.getRowCount() == 0) {
        return 0L;
      }
    }
    return totalRowCount;
  }

  public Map<String, Long> getRowCountForEachBlock(Segment segment, List<PartitionSpec> partitions,
      Map<String, Long> blockletToRowCountMap) {
    if (memoryDMStore.getRowCount() == 0) {
      return new HashMap<>();
    }
    // if it has partitioned datamap but there is no partitioned information stored, it means
    // partitions are dropped so return empty list.
    if (partitions != null) {
      if (!validatePartitionInfo(partitions)) {
        return new HashMap<>();
      }
    }
    CarbonRowSchema[] schema = getFileFooterEntrySchema();
    int numEntries = memoryDMStore.getRowCount();
    for (int i = 0; i < numEntries; i++) {
      DataMapRow dataMapRow = memoryDMStore.getDataMapRow(schema, i);
      String fileName = new String(dataMapRow.getByteArray(FILE_PATH_INDEX),
          CarbonCommonConstants.DEFAULT_CHARSET_CLASS) + CarbonTablePath.getCarbonDataExtension();
      int rowCount = dataMapRow.getInt(ROW_COUNT_INDEX);
      // prepend segment number with the blocklet file path
      String blockletMapKey = segment.getSegmentNo() + "," + fileName;
      Long existingCount = blockletToRowCountMap.get(blockletMapKey);
      if (null != existingCount) {
        blockletToRowCountMap.put(blockletMapKey, (long) rowCount + existingCount);
      } else {
        blockletToRowCountMap.put(blockletMapKey, (long) rowCount);
      }
    }
    return blockletToRowCountMap;
  }

  private List<Blocklet> prune(FilterResolverIntf filterExp) {
    if (memoryDMStore.getRowCount() == 0) {
      return new ArrayList<>();
    }
    List<Blocklet> blocklets = new ArrayList<>();
    CarbonRowSchema[] schema = getFileFooterEntrySchema();
    String filePath = getFilePath();
    int numEntries = memoryDMStore.getRowCount();
    int totalBlocklets = 0;
    if (ExplainCollector.enabled()) {
      totalBlocklets = getTotalBlocklets();
    }
    int hitBlocklets = 0;
    if (filterExp == null) {
      for (int i = 0; i < numEntries; i++) {
        DataMapRow dataMapRow = memoryDMStore.getDataMapRow(schema, i);
        blocklets.add(createBlocklet(dataMapRow, getFileNameWithFilePath(dataMapRow, filePath),
            getBlockletId(dataMapRow), false));
      }
      hitBlocklets = totalBlocklets;
    } else {
      // Remove B-tree jump logic as start and end key prepared is not
      // correct for old store scenarios
      int entryIndex = 0;
      FilterExecuter filterExecuter = FilterUtil.getFilterExecuterTree(
          filterExp, getSegmentProperties(), null, getMinMaxCacheColumns(), false);
      // flag to be used for deciding whether use min/max in executor pruning for BlockletDataMap
      boolean useMinMaxForPruning = useMinMaxForExecutorPruning(filterExp);
      // min and max for executor pruning
      while (entryIndex < numEntries) {
        DataMapRow row = memoryDMStore.getDataMapRow(schema, entryIndex);
        boolean[] minMaxFlag = getMinMaxFlag(row, BLOCK_MIN_MAX_FLAG);
        String fileName = getFileNameWithFilePath(row, filePath);
        short blockletId = getBlockletId(row);
        boolean isValid =
            addBlockBasedOnMinMaxValue(filterExecuter, getMinMaxValue(row, MAX_VALUES_INDEX),
                getMinMaxValue(row, MIN_VALUES_INDEX), minMaxFlag, fileName, blockletId);
        if (isValid) {
          blocklets.add(createBlocklet(row, fileName, blockletId, useMinMaxForPruning));
          if (ExplainCollector.enabled()) {
            hitBlocklets += getBlockletNumOfEntry(entryIndex);
          }
        }
        entryIndex++;
      }
    }
    if (ExplainCollector.enabled()) {
      if (isLegacyStore) {
        ExplainCollector.setShowPruningInfo(false);
      } else {
        ExplainCollector.setShowPruningInfo(true);
        ExplainCollector.addTotalBlocklets(totalBlocklets);
        ExplainCollector.addTotalBlocks(getTotalBlocks());
        ExplainCollector.addDefaultDataMapPruningHit(hitBlocklets);
      }
    }
    return blocklets;
  }

  protected boolean useMinMaxForExecutorPruning(FilterResolverIntf filterResolverIntf) {
    return false;
  }

  @Override
  public List<Blocklet> prune(Expression expression, SegmentProperties properties,
      List<PartitionSpec> partitions, CarbonTable carbonTable) throws IOException {
    return prune(new DataMapFilter(properties, carbonTable, expression).getResolver(), properties,
        partitions);
  }

  @Override
  public List<Blocklet> prune(FilterResolverIntf filterExp, SegmentProperties segmentProperties,
      List<PartitionSpec> partitions) {
    if (memoryDMStore.getRowCount() == 0) {
      return new ArrayList<>();
    }
    // if it has partitioned datamap but there is no partitioned information stored, it means
    // partitions are dropped so return empty list.
    if (partitions != null) {
      if (!validatePartitionInfo(partitions)) {
        return new ArrayList<>();
      }
    }
    // Prune with filters if the partitions are existed in this datamap
    // changed segmentProperties to this.segmentProperties to make sure the pruning with its own
    // segmentProperties.
    // Its a temporary fix. The Interface DataMap.prune(FilterResolverIntf filterExp,
    // SegmentProperties segmentProperties, List<PartitionSpec> partitions) should be corrected
    return prune(filterExp);
  }

  private boolean validatePartitionInfo(List<PartitionSpec> partitions) {
    // First get the partitions which are stored inside datamap.
    String[] fileDetails = getFileDetails();
    // Check the exact match of partition information inside the stored partitions.
    boolean found = false;
    Path folderPath = new Path(fileDetails[0]);
    for (PartitionSpec spec : partitions) {
      if (folderPath.equals(spec.getLocation()) && isCorrectUUID(fileDetails, spec)) {
        found = true;
        break;
      }
    }
    return found;
  }

  @Override
  public void finish() {

  }

  private boolean isCorrectUUID(String[] fileDetails, PartitionSpec spec) {
    boolean needToScan = false;
    if (spec.getUuid() != null) {
      String[] split = spec.getUuid().split("_");
      if (split[0].equals(fileDetails[2]) && CarbonTablePath.DataFileUtil
          .getTimeStampFromFileName(fileDetails[1]).equals(split[1])) {
        needToScan = true;
      }
    } else {
      needToScan = true;
    }
    return needToScan;
  }

  /**
   * select the blocks based on column min and max value
   *
   * @param filterExecuter
   * @param maxValue
   * @param minValue
   * @param minMaxFlag
   * @param filePath
   * @param blockletId
   * @return
   */
  private boolean addBlockBasedOnMinMaxValue(FilterExecuter filterExecuter, byte[][] maxValue,
      byte[][] minValue, boolean[] minMaxFlag, String filePath, int blockletId) {
    BitSet bitSet = null;
    if (filterExecuter instanceof ImplicitColumnFilterExecutor) {
      String uniqueBlockPath = filePath.substring(filePath.lastIndexOf("/Part") + 1);
      // this case will come in case of old store where index file does not contain the
      // blocklet information
      if (blockletId != -1) {
        uniqueBlockPath = uniqueBlockPath + CarbonCommonConstants.FILE_SEPARATOR + blockletId;
      }
      bitSet = ((ImplicitColumnFilterExecutor) filterExecuter)
          .isFilterValuesPresentInBlockOrBlocklet(maxValue, minValue, uniqueBlockPath, minMaxFlag);
    } else {
      bitSet = filterExecuter.isScanRequired(maxValue, minValue, minMaxFlag);
    }
    if (!bitSet.isEmpty()) {
      return true;
    } else {
      return false;
    }
  }

  public ExtendedBlocklet getDetailedBlocklet(String blockletId) {
    if (isLegacyStore) {
      throw new UnsupportedOperationException("With legacy store only BlockletDataMap is allowed."
          + " In order to use other dataMaps upgrade to new store.");
    }
    int absoluteBlockletId = Integer.parseInt(blockletId);
    return createBlockletFromRelativeBlockletId(absoluteBlockletId);
  }

  /**
   * Method to get the relative blocklet ID. Absolute blocklet ID is the blocklet Id as per
   * task level but relative blocklet ID is id as per carbondata file/block level
   *
   * @param absoluteBlockletId
   * @return
   */
  private ExtendedBlocklet createBlockletFromRelativeBlockletId(int absoluteBlockletId) {
    short relativeBlockletId = -1;
    int rowIndex = 0;
    // return 0 if absoluteBlockletId is 0
    if (absoluteBlockletId == 0) {
      relativeBlockletId = (short) absoluteBlockletId;
    } else {
      int diff = absoluteBlockletId;
      ByteBuffer byteBuffer = ByteBuffer.wrap(getBlockletRowCountForEachBlock());
      // Example: absoluteBlockletID = 17, blockletRowCountForEachBlock = {4,3,2,5,7}
      // step1: diff = 17-4, diff = 13
      // step2: diff = 13-3, diff = 10
      // step3: diff = 10-2, diff = 8
      // step4: diff = 8-5, diff = 3
      // step5: diff = 3-7, diff = -4 (satisfies <= 0)
      // step6: relativeBlockletId = -4+7, relativeBlockletId = 3 (4th index starting from 0)
      while (byteBuffer.hasRemaining()) {
        short blockletCount = byteBuffer.getShort();
        diff = diff - blockletCount;
        if (diff < 0) {
          relativeBlockletId = (short) (diff + blockletCount);
          break;
        }
        rowIndex++;
      }
    }
    DataMapRow row =
        memoryDMStore.getDataMapRow(getFileFooterEntrySchema(), rowIndex);
    String filePath = getFilePath();
    return createBlocklet(row, getFileNameWithFilePath(row, filePath), relativeBlockletId,
        false);
  }

  private byte[] getBlockletRowCountForEachBlock() {
    // taskSummary DM store will  have only one row
    CarbonRowSchema[] taskSummarySchema = getTaskSummarySchema();
    return taskSummaryDMStore
        .getDataMapRow(taskSummarySchema, taskSummaryDMStore.getRowCount() - 1)
        .getByteArray(taskSummarySchema.length - 1);
  }

  /**
   * Get the index file name of the blocklet data map
   *
   * @return
   */
  public String getTableTaskInfo(int index) {
    DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(getTaskSummarySchema(), 0);
    try {
      return new String(unsafeRow.getByteArray(index), CarbonCommonConstants.DEFAULT_CHARSET);
    } catch (UnsupportedEncodingException e) {
      // should never happen!
      throw new IllegalArgumentException("UTF8 encoding is not supported", e);
    }
  }

  private byte[][] getMinMaxValue(DataMapRow row, int index) {
    DataMapRow minMaxRow = row.getRow(index);
    byte[][] minMax = new byte[minMaxRow.getColumnCount()][];
    for (int i = 0; i < minMax.length; i++) {
      minMax[i] = minMaxRow.getByteArray(i);
    }
    return minMax;
  }

  private boolean[] getMinMaxFlag(DataMapRow row, int index) {
    DataMapRow minMaxFlagRow = row.getRow(index);
    boolean[] minMaxFlag = new boolean[minMaxFlagRow.getColumnCount()];
    for (int i = 0; i < minMaxFlag.length; i++) {
      minMaxFlag[i] = minMaxFlagRow.getBoolean(i);
    }
    return minMaxFlag;
  }

  protected short getBlockletId(DataMapRow dataMapRow) {
    return BLOCK_DEFAULT_BLOCKLET_ID;
  }

  protected ExtendedBlocklet createBlocklet(DataMapRow row, String fileName, short blockletId,
      boolean useMinMaxForPruning) {
    short versionNumber = row.getShort(VERSION_INDEX);
    ExtendedBlocklet blocklet = new ExtendedBlocklet(fileName, blockletId + "", false,
        ColumnarFormatVersion.valueOf(versionNumber));
    blocklet.setDataMapRow(row);
    blocklet.setColumnCardinality(getColumnCardinality());
    blocklet.setLegacyStore(isLegacyStore);
    blocklet.setUseMinMaxForPruning(useMinMaxForPruning);
    return blocklet;
  }

  private String[] getFileDetails() {
    try {
      String[] fileDetails = new String[3];
      DataMapRow unsafeRow = taskSummaryDMStore.getDataMapRow(getTaskSummarySchema(), 0);
      fileDetails[0] = new String(unsafeRow.getByteArray(SUMMARY_INDEX_PATH),
          CarbonCommonConstants.DEFAULT_CHARSET);
      fileDetails[1] = new String(unsafeRow.getByteArray(SUMMARY_INDEX_FILE_NAME),
          CarbonCommonConstants.DEFAULT_CHARSET);
      fileDetails[2] = new String(unsafeRow.getByteArray(SUMMARY_SEGMENTID),
          CarbonCommonConstants.DEFAULT_CHARSET);
      return fileDetails;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  @Override
  public void clear() {
    if (memoryDMStore != null) {
      memoryDMStore.freeMemory();
    }
    // clear task min/max unsafe memory
    if (null != taskSummaryDMStore) {
      taskSummaryDMStore.freeMemory();
    }
  }

  public long getMemorySize() {
    long memoryUsed = 0L;
    if (memoryDMStore != null) {
      memoryUsed += memoryDMStore.getMemoryUsed();
    }
    if (null != taskSummaryDMStore) {
      memoryUsed += taskSummaryDMStore.getMemoryUsed();
    }
    return memoryUsed;
  }

  protected SegmentProperties getSegmentProperties() {
    return segmentPropertiesWrapper.getSegmentProperties();
  }

  public int[] getColumnCardinality() {
    return segmentPropertiesWrapper.getColumnCardinality();
  }

  public List<ColumnSchema> getColumnSchema() {
    return segmentPropertiesWrapper.getColumnsInTable();
  }

  protected AbstractMemoryDMStore getMemoryDMStore(boolean addToUnsafe)
      throws MemoryException {
    AbstractMemoryDMStore memoryDMStore;
    if (addToUnsafe) {
      memoryDMStore = new UnsafeMemoryDMStore();
    } else {
      memoryDMStore = new SafeMemoryDMStore();
    }
    return memoryDMStore;
  }

  protected CarbonRowSchema[] getFileFooterEntrySchema() {
    return segmentPropertiesWrapper.getBlockFileFooterEntrySchema();
  }

  protected CarbonRowSchema[] getTaskSummarySchema() {
    try {
      return segmentPropertiesWrapper.getTaskSummarySchemaForBlock(true, isFilePathStored);
    } catch (MemoryException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * This method will ocnvert safe to unsafe memory DM store
   *
   * @throws MemoryException
   */
  public void convertToUnsafeDMStore() throws MemoryException {
    if (memoryDMStore instanceof SafeMemoryDMStore) {
      UnsafeMemoryDMStore unsafeMemoryDMStore = memoryDMStore.convertToUnsafeDMStore(
          getFileFooterEntrySchema());
      memoryDMStore.freeMemory();
      memoryDMStore = unsafeMemoryDMStore;
    }
    if (taskSummaryDMStore instanceof SafeMemoryDMStore) {
      UnsafeMemoryDMStore unsafeSummaryMemoryDMStore =
          taskSummaryDMStore.convertToUnsafeDMStore(getTaskSummarySchema());
      taskSummaryDMStore.freeMemory();
      taskSummaryDMStore = unsafeSummaryMemoryDMStore;
    }
  }

  public SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper getSegmentPropertiesWrapper() {
    return segmentPropertiesWrapper;
  }

  @Override
  public int getNumberOfEntries() {
    if (memoryDMStore != null) {
      if (memoryDMStore.getRowCount() == 0) {
        // so that one datamap considered as one record
        return 1;
      } else {
        return memoryDMStore.getRowCount();
      }
    } else {
      // legacy store
      return 1;
    }
  }
}