content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.carbondata.core.scan.result;
  
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.charset.Charset;
  import java.util.ArrayList;
  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.datastore.ReusableDataBuffer;
  import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
  import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
  import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
  import org.apache.carbondata.core.datastore.page.ColumnPage;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
  import org.apache.carbondata.core.mutate.DeleteDeltaVo;
  import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
  import org.apache.carbondata.core.scan.filter.GenericQueryType;
  import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
  import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
  import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
  import org.apache.carbondata.core.scan.scanner.LazyBlockletLoader;
  import org.apache.carbondata.core.scan.scanner.LazyPageLoader;
  import org.apache.carbondata.core.stats.QueryStatistic;
  import org.apache.carbondata.core.stats.QueryStatisticsConstants;
  import org.apache.carbondata.core.stats.QueryStatisticsModel;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.ReUsableByteArrayDataOutputStream;
  
  import org.apache.log4j.Logger;
  
  /**
   * Scanned result class which will store and provide the result on request
   */
  public abstract class BlockletScannedResult {
  
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(BlockletScannedResult.class.getName());
    /**
     * current row number
     */
    protected int currentRow = -1;
  
    protected int pageCounter;
    /**
     * matched rowId for each page
     */
    protected int[][] pageFilteredRowId;
    /**
     * key size of the fixed length column
     */
    protected int fixedLengthKeySize;
    /**
     * total number of filtered rows for each page
     */
    private int[] pageFilteredRowCount;
  
    /**
     * Filtered pages to be decoded and loaded to vector.
     */
    private int[] pageIdFiltered;
  
    /**
     * to keep track of number of rows process
     */
    protected int rowCounter;
    /**
     * dimension column data chunk
     */
    protected DimensionColumnPage[][] dimensionColumnPages;
  
    /**
     * Raw dimension chunks;
     */
    protected DimensionRawColumnChunk[] dimRawColumnChunks;
  
    /**
     * Raw dimension chunks;
     */
    protected MeasureRawColumnChunk[] msrRawColumnChunks;
    /**
     * measure column data chunk
     */
    protected ColumnPage[][] measureColumnPages;
    /**
     * dictionary column block index in file
     */
    protected int[] dictionaryColumnChunkIndexes;
  
    /**
     * no dictionary column chunk index in file
     */
    protected int[] noDictionaryColumnChunkIndexes;
  
    /**
     *
     */
    public Map<Integer, GenericQueryType> complexParentIndexToQueryMap;
  
    private int totalDimensionsSize;
  
    /**
     * blockedId which will be blockId + blocklet number in the block
     */
    private String blockletId;
  
    /**
     * parent block indexes
     */
    private int[] complexParentBlockIndexes;
  
    /**
     * blockletId+pageNumber to deleted record map
     */
    private Map<String, DeleteDeltaVo> deletedRecordMap;
  
    /**
     * current page delete delta vo
     */
    private DeleteDeltaVo currentDeleteDeltaVo;
  
    /**
     * actual blocklet number
     */
    private String blockletNumber;
  
    protected List<Integer> validRowIds;
  
    protected QueryStatisticsModel queryStatisticsModel;
  
    protected LazyBlockletLoader lazyBlockletLoader;
  
    private ReusableDataBuffer[] dimensionReusableBuffer;
  
    private ReusableDataBuffer[] measureReusableBuffer;
  
    // index used by dimensionReusableBuffer
    private int dimensionReusableBufferIndex;
  
    public BlockletScannedResult(BlockExecutionInfo blockExecutionInfo,
        QueryStatisticsModel queryStatisticsModel) {
      this.dimensionReusableBuffer = blockExecutionInfo.getDimensionReusableDataBuffer();
      this.measureReusableBuffer = blockExecutionInfo.getMeasureReusableDataBuffer();
      this.fixedLengthKeySize = blockExecutionInfo.getFixedLengthKeySize();
      this.noDictionaryColumnChunkIndexes = blockExecutionInfo.getNoDictionaryColumnChunkIndexes();
      this.dictionaryColumnChunkIndexes = blockExecutionInfo.getDictionaryColumnChunkIndex();
      this.complexParentIndexToQueryMap = blockExecutionInfo.getComplexDimensionInfoMap();
      this.complexParentBlockIndexes = blockExecutionInfo.getComplexColumnParentBlockIndexes();
      this.totalDimensionsSize = blockExecutionInfo.getProjectionDimensions().length;
      this.deletedRecordMap = blockExecutionInfo.getDeletedRecordsMap();
      this.queryStatisticsModel = queryStatisticsModel;
      validRowIds = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    }
  
    /**
     * Below method will be used to set the dimension chunks
     * which will be used to create a row
     *
     * @param columnPages dimension chunks used in query
     */
    public void setDimensionColumnPages(DimensionColumnPage[][] columnPages) {
      this.dimensionColumnPages = columnPages;
    }
  
    /**
     * Below method will be used to set the measure column chunks
     *
     * @param columnPages measure data chunks
     */
    public void setMeasureColumnPages(ColumnPage[][] columnPages) {
      this.measureColumnPages = columnPages;
    }
  
    public void setDimRawColumnChunks(DimensionRawColumnChunk[] dimRawColumnChunks) {
      this.dimRawColumnChunks = dimRawColumnChunks;
    }
  
    public void setMsrRawColumnChunks(MeasureRawColumnChunk[] msrRawColumnChunks) {
      this.msrRawColumnChunks = msrRawColumnChunks;
    }
  
    public LazyBlockletLoader getLazyBlockletLoader() {
      return lazyBlockletLoader;
    }
  
    public void setLazyBlockletLoader(LazyBlockletLoader lazyBlockletLoader) {
      this.lazyBlockletLoader = lazyBlockletLoader;
    }
  
    /**
     * Below method will be used to get the chunk based in measure ordinal
     *
     * @param ordinal measure ordinal
     * @return measure column chunk
     */
    public ColumnPage getMeasureChunk(int ordinal) {
      return measureColumnPages[ordinal][pageCounter];
    }
  
    /**
     * Below method will be used to get the key for all the dictionary dimensions
     * in integer array format which is present in the query
     *
     * @param rowId row id selected after scanning
     * @return return the dictionary key
     */
    protected int[] getDictionaryKeyIntegerArray(int rowId) {
      int[] completeKey = new int[totalDimensionsSize];
      int column = 0;
      for (int i = 0; i < this.dictionaryColumnChunkIndexes.length; i++) {
        column = dimensionColumnPages[dictionaryColumnChunkIndexes[i]][pageCounter]
            .fillSurrogateKey(rowId, column, completeKey);
      }
      return completeKey;
    }
  
    /**
     * Fill the column data of dictionary to vector
     */
    public void fillColumnarDictionaryBatch(ColumnVectorInfo[] vectorInfo) {
      int column = 0;
      for (int i = 0; i < this.dictionaryColumnChunkIndexes.length; i++) {
        column = dimensionColumnPages[dictionaryColumnChunkIndexes[i]][pageCounter]
            .fillVector(vectorInfo, column);
      }
    }
  
    /**
     * Fill the column data to vector
     */
    public void fillColumnarNoDictionaryBatch(ColumnVectorInfo[] vectorInfo) {
      for (int i = 0; i < this.noDictionaryColumnChunkIndexes.length; i++) {
        dimensionColumnPages[noDictionaryColumnChunkIndexes[i]][pageCounter]
            .fillVector(vectorInfo, i);
      }
    }
  
    /**
     * Fill the measure column data to vector
     */
    public void fillColumnarMeasureBatch(ColumnVectorInfo[] vectorInfo, int[] measuresOrdinal) {
      for (int i = 0; i < measuresOrdinal.length; i++) {
        vectorInfo[i].measureVectorFiller
            .fillMeasureVector(measureColumnPages[measuresOrdinal[i]][pageCounter], vectorInfo[i]);
      }
    }
  
    public void fillColumnarComplexBatch(ColumnVectorInfo[] vectorInfos) {
      ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
      ReUsableByteArrayDataOutputStream reusableDataOutput =
          new ReUsableByteArrayDataOutputStream(byteStream);
      boolean isExceptionThrown = false;
      for (ColumnVectorInfo vectorInfo : vectorInfos) {
        int offset = vectorInfo.offset;
        int len = offset + vectorInfo.size;
        int vectorOffset = vectorInfo.vectorOffset;
        CarbonColumnVector vector = vectorInfo.vector;
        for (int j = offset; j < len; j++) {
          try {
            vectorInfo.genericQueryType
                .parseBlocksAndReturnComplexColumnByteArray(dimRawColumnChunks, dimensionColumnPages,
                    pageFilteredRowId == null ? j : pageFilteredRowId[pageCounter][j], pageCounter,
                    reusableDataOutput);
            Object data = vectorInfo.genericQueryType
                .getDataBasedOnDataType(ByteBuffer.wrap(reusableDataOutput.getByteArray()));
            vector.putObject(vectorOffset++, data);
            reusableDataOutput.reset();
          } catch (IOException e) {
            isExceptionThrown = true;
            LOGGER.error(e.getMessage(), e);
          } finally {
            if (isExceptionThrown) {
              CarbonUtil.closeStreams(reusableDataOutput);
              CarbonUtil.closeStreams(byteStream);
            }
          }
        }
      }
      CarbonUtil.closeStreams(reusableDataOutput);
      CarbonUtil.closeStreams(byteStream);
    }
  
    /**
     * Fill the column data to vector
     */
    public void fillColumnarImplicitBatch(ColumnVectorInfo[] vectorInfo) {
      for (int i = 0; i < vectorInfo.length; i++) {
        ColumnVectorInfo columnVectorInfo = vectorInfo[i];
        CarbonColumnVector vector = columnVectorInfo.vector;
        int offset = columnVectorInfo.offset;
        int vectorOffset = columnVectorInfo.vectorOffset;
        int len = offset + columnVectorInfo.size;
        for (int j = offset; j < len; j++) {
          // Considering only String case now as we support only
          String data = getBlockletId();
          if (CarbonCommonConstants.CARBON_IMPLICIT_COLUMN_TUPLEID
              .equals(columnVectorInfo.dimension.getColumnName())) {
            data = data + CarbonCommonConstants.FILE_SEPARATOR + pageCounter
                + CarbonCommonConstants.FILE_SEPARATOR + (pageFilteredRowId == null ?
                j :
                pageFilteredRowId[pageCounter][j]);
          }
          vector.putByteArray(vectorOffset++,
              data.getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
        }
      }
    }
  
    /**
     * Just increment the counter in case of query only on measures.
     */
    public void incrementCounter() {
      rowCounter++;
      currentRow++;
    }
  
    /**
     * This method will add the delta to row counter
     *
     * @param delta
     */
    public void incrementCounter(int delta) {
      rowCounter += delta;
      currentRow += delta;
    }
  
    /**
     * Just increment the page counter and reset the remaining counters.
     */
    public void incrementPageCounter() {
      rowCounter = 0;
      currentRow = -1;
      pageCounter++;
      fillDataChunks();
      if (null != deletedRecordMap) {
        currentDeleteDeltaVo = deletedRecordMap.get(blockletNumber + "_" + pageCounter);
      }
    }
  
    /**
     * Just increment the page counter and reset the remaining counters.
     */
    public void incrementPageCounter(ColumnVectorInfo[] vectorInfos) {
      rowCounter = 0;
      currentRow = -1;
      pageCounter++;
      if (null != deletedRecordMap && pageCounter < pageIdFiltered.length) {
        currentDeleteDeltaVo =
            deletedRecordMap.get(blockletNumber + "_" + pageIdFiltered[pageCounter]);
      }
    }
  
    /**
     * This case is used only in case of compaction, since it does not use filter flow.
     */
    public void fillDataChunks() {
      freeDataChunkMemory();
      if (pageCounter >= pageFilteredRowCount.length) {
        return;
      }
      long startTime = System.currentTimeMillis();
      for (int i = 0; i < dimensionColumnPages.length; i++) {
        if (dimensionColumnPages[i][pageCounter] == null && dimRawColumnChunks[i] != null) {
          dimensionColumnPages[i][pageCounter] = dimRawColumnChunks[i]
              .convertToDimColDataChunkWithOutCache(pageCounter, null);
        }
      }
  
      for (int i = 0; i < measureColumnPages.length; i++) {
        if (measureColumnPages[i][pageCounter] == null && msrRawColumnChunks[i] != null) {
          measureColumnPages[i][pageCounter] = msrRawColumnChunks[i]
              .convertToColumnPageWithOutCache(pageCounter, null);
        }
      }
      QueryStatistic pageUncompressTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
          .get(QueryStatisticsConstants.PAGE_UNCOMPRESS_TIME);
      pageUncompressTime.addCountStatistic(QueryStatisticsConstants.PAGE_UNCOMPRESS_TIME,
          pageUncompressTime.getCount() + (System.currentTimeMillis() - startTime));
    }
  
    void fillChildDataChunks(CarbonColumnVector columnVector, CarbonDimension dimension,
        ColumnVectorInfo complexInfo) {
      columnVector.setLazyPage(new LazyPageLoader(lazyBlockletLoader, dimension.getOrdinal(), false,
          pageIdFiltered[pageCounter], complexInfo,
          dimensionReusableBuffer[dimensionReusableBufferIndex++]));
      if (columnVector.getChildrenVector() != null) {
        int i = 0;
        for (CarbonColumnVector childVector : columnVector.getChildrenVector()) {
          fillChildDataChunks(childVector, dimension.getListOfChildDimensions().get(i++),
              complexInfo);
        }
      }
    }
  
    /**
     * Fill all the vectors with data by decompressing/decoding the column page
     */
    public void fillDataChunks(ColumnVectorInfo[] dictionaryInfo, ColumnVectorInfo[] noDictionaryInfo,
        ColumnVectorInfo[] msrVectorInfo, int[] measuresOrdinal, ColumnVectorInfo[] complexInfo) {
      freeDataChunkMemory();
      if (pageCounter >= pageFilteredRowCount.length) {
        return;
      }
  
      for (int i = 0; i < this.dictionaryColumnChunkIndexes.length; i++) {
        dictionaryInfo[i].vector.setLazyPage(
            new LazyPageLoader(lazyBlockletLoader, dictionaryColumnChunkIndexes[i], false,
                pageIdFiltered[pageCounter], dictionaryInfo[i], dimensionReusableBuffer[i]));
      }
      dimensionReusableBufferIndex = dictionaryColumnChunkIndexes.length;
      for (int i = 0; i < this.noDictionaryColumnChunkIndexes.length; i++) {
        noDictionaryInfo[i].vector.setLazyPage(
            new LazyPageLoader(lazyBlockletLoader, noDictionaryColumnChunkIndexes[i], false,
                pageIdFiltered[pageCounter], noDictionaryInfo[i],
                dimensionReusableBuffer[dimensionReusableBufferIndex++]));
      }
      dimensionReusableBufferIndex =
          noDictionaryColumnChunkIndexes.length + dictionaryColumnChunkIndexes.length;
  
      for (int i = 0; i < this.complexParentBlockIndexes.length; i++) {
        complexInfo[i].vector.getColumnVector().setLazyPage(
            new LazyPageLoader(lazyBlockletLoader, complexParentBlockIndexes[i], false,
                pageIdFiltered[pageCounter], complexInfo[i],
                dimensionReusableBuffer[dimensionReusableBufferIndex++]));
        // set child pages
        int dimIndex = 0;
        for (CarbonColumnVector childVector : complexInfo[i].vector.getColumnVector()
            .getChildrenVector()) {
          fillChildDataChunks(childVector,
              complexInfo[i].dimension.getDimension().getListOfChildDimensions().get(dimIndex++),
              complexInfo[i]);
        }
      }
  
      for (int i = 0; i < measuresOrdinal.length; i++) {
        msrVectorInfo[i].vector.setLazyPage(
            new LazyPageLoader(lazyBlockletLoader, measuresOrdinal[i], true,
                pageIdFiltered[pageCounter], msrVectorInfo[i], measureReusableBuffer[i]));
      }
  
    }
  
    // free the memory for the last page chunk
    private void freeDataChunkMemory() {
      for (int i = 0; i < dimensionColumnPages.length; i++) {
        if (pageCounter > 0 && dimensionColumnPages[i][pageCounter - 1] != null) {
          dimensionColumnPages[i][pageCounter - 1].freeMemory();
          dimensionColumnPages[i][pageCounter - 1] = null;
        }
      }
      for (int i = 0; i < measureColumnPages.length; i++) {
        if (pageCounter > 0 && measureColumnPages[i][pageCounter - 1] != null) {
          measureColumnPages[i][pageCounter - 1].freeMemory();
          measureColumnPages[i][pageCounter - 1] = null;
        }
      }
      clearValidRowIdList();
    }
  
    public int numberOfPages() {
      return pageFilteredRowCount.length;
    }
  
    public int[] getPageIdFiltered() {
      return pageIdFiltered;
    }
  
    public void setPageIdFiltered(int[] pageIdFiltered) {
      this.pageIdFiltered = pageIdFiltered;
    }
  
    /**
     * Get total rows in the current page
     *
     * @return
     */
    public int getCurrentPageRowCount() {
      return pageFilteredRowCount[pageCounter];
    }
  
    public int getCurrentPageCounter() {
      return pageCounter;
    }
  
    /**
     * increment the counter.
     */
    public void setRowCounter(int rowCounter) {
      this.rowCounter = rowCounter;
    }
  
    /**
     * Below method will be used to get the dimension key array
     * for all the no dictionary dimension present in the query
     *
     * @param rowId row number
     * @return no dictionary keys for all no dictionary dimension
     */
    protected byte[][] getNoDictionaryKeyArray(int rowId) {
      byte[][] noDictionaryColumnsKeys = new byte[noDictionaryColumnChunkIndexes.length][];
      int position = 0;
      for (int i = 0; i < this.noDictionaryColumnChunkIndexes.length; i++) {
        noDictionaryColumnsKeys[position++] =
            dimensionColumnPages[noDictionaryColumnChunkIndexes[i]][pageCounter].getChunkData(rowId);
      }
      return noDictionaryColumnsKeys;
    }
  
    /**
     * This method will return the bitsets for valid row Id's to be scanned
     *
     * @param rowId
     * @param batchSize
     * @return
     */
  
    public abstract void fillValidRowIdsBatchFilling(int rowId, int batchSize);
  
    protected void clearValidRowIdList() {
      if (null != validRowIds && !validRowIds.isEmpty()) {
        validRowIds.clear();
      }
    }
  
    public List<Integer> getValidRowIds() {
      return validRowIds;
    }
  
    /**
     * Below method will be used to get the complex type keys array based
     * on row id for all the complex type dimension selected in query.
     * This method will be used to fill the data column wise
     *
     * @return complex type key array for all the complex dimension selected in query
     */
    protected List<byte[][]> getComplexTypeKeyArrayBatch() {
      List<byte[][]> complexTypeArrayList = new ArrayList<>(validRowIds.size());
      ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
      ReUsableByteArrayDataOutputStream reusableDataOutput =
          new ReUsableByteArrayDataOutputStream(byteStream);
      boolean isExceptionThrown = false;
      byte[][] complexTypeData = null;
      // everyTime it is initialized new as in case of prefetch it can modify the data
      for (int i = 0; i < validRowIds.size(); i++) {
        complexTypeData = new byte[complexParentBlockIndexes.length][];
        complexTypeArrayList.add(complexTypeData);
      }
      for (int i = 0; i < complexParentBlockIndexes.length; i++) {
        // get the genericQueryType for 1st column
        GenericQueryType genericQueryType =
            complexParentIndexToQueryMap.get(complexParentBlockIndexes[i]);
        for (int j = 0; j < validRowIds.size(); j++) {
          try {
            genericQueryType
                .parseBlocksAndReturnComplexColumnByteArray(dimRawColumnChunks, dimensionColumnPages,
                    validRowIds.get(j), pageCounter, reusableDataOutput);
            // get the key array in columnar way
            byte[][] complexKeyArray = complexTypeArrayList.get(j);
            complexKeyArray[i] = byteStream.toByteArray();
            reusableDataOutput.reset();
          } catch (IOException e) {
            isExceptionThrown = true;
            LOGGER.error(e.getMessage(), e);
          } finally {
            if (isExceptionThrown) {
              CarbonUtil.closeStreams(reusableDataOutput);
              CarbonUtil.closeStreams(byteStream);
            }
          }
        }
      }
      CarbonUtil.closeStreams(reusableDataOutput);
      CarbonUtil.closeStreams(byteStream);
      return complexTypeArrayList;
    }
  
    /**
     * @return blockletId
     */
    public String getBlockletId() {
      return blockletId;
    }
  
    /**
     * Set blocklet id, which looks like
     * "Part0/Segment_0/part-0-0_batchno0-0-1517155583332.carbondata/0"
     */
    public void setBlockletId(String blockletId, String blockletNumber) {
      this.blockletId = blockletId + CarbonCommonConstants.FILE_SEPARATOR + blockletNumber;
      this.blockletNumber = blockletNumber;
      // if deleted record map is present for this block
      // then get the first page deleted vo
      if (null != deletedRecordMap) {
        String key;
        if (pageIdFiltered != null) {
          key = this.blockletNumber + '_' + pageIdFiltered[pageCounter];
        } else {
          key = this.blockletNumber + '_' + pageCounter;
        }
        currentDeleteDeltaVo = deletedRecordMap.get(key);
      }
    }
  
    /**
     * Below method will be used to get the complex type keys array based
     * on row id for all the complex type dimension selected in query
     *
     * @param rowId row number
     * @return complex type key array for all the complex dimension selected in query
     */
    protected byte[][] getComplexTypeKeyArray(int rowId) {
      ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
      ReUsableByteArrayDataOutputStream reUsableDataOutput =
          new ReUsableByteArrayDataOutputStream(byteStream);
      boolean isExceptionThrown = false;
      byte[][] complexTypeData = new byte[complexParentBlockIndexes.length][];
      for (int i = 0; i < complexTypeData.length; i++) {
        GenericQueryType genericQueryType =
            complexParentIndexToQueryMap.get(complexParentBlockIndexes[i]);
        try {
          genericQueryType
              .parseBlocksAndReturnComplexColumnByteArray(dimRawColumnChunks, dimensionColumnPages,
                  rowId, pageCounter, reUsableDataOutput);
          complexTypeData[i] = byteStream.toByteArray();
          reUsableDataOutput.reset();
        } catch (IOException e) {
          isExceptionThrown = true;
          LOGGER.error(e.getMessage(), e);
        } finally {
          if (isExceptionThrown) {
            CarbonUtil.closeStreams(reUsableDataOutput);
            CarbonUtil.closeStreams(byteStream);
          }
        }
      }
      CarbonUtil.closeStreams(reUsableDataOutput);
      CarbonUtil.closeStreams(byteStream);
      return complexTypeData;
    }
  
    /**
     * to check whether any more row is present in the result
     *
     * @return
     */
    public boolean hasNext() {
      if (pageCounter
          < pageFilteredRowCount.length && rowCounter < this.pageFilteredRowCount[pageCounter]) {
        return true;
      } else if (pageCounter < pageFilteredRowCount.length) {
        pageCounter++;
        if (pageCounter >= pageFilteredRowCount.length) {
          return false;
        }
        if (this.pageFilteredRowCount[pageCounter] == 0) {
          return hasNext();
        }
        fillDataChunks();
        rowCounter = 0;
        currentRow = -1;
        if (null != deletedRecordMap) {
          currentDeleteDeltaVo = deletedRecordMap.get(blockletNumber + "_" + pageCounter);
        }
        return hasNext();
      }
      return false;
    }
  
    /**
     * Below method will be used to free the occupied memory
     */
    public void freeMemory() {
      // first free the dimension chunks
      if (null != dimensionColumnPages) {
        for (int i = 0; i < dimensionColumnPages.length; i++) {
          if (null != dimensionColumnPages[i]) {
            for (int j = 0; j < dimensionColumnPages[i].length; j++) {
              if (null != dimensionColumnPages[i][j]) {
                dimensionColumnPages[i][j].freeMemory();
                dimensionColumnPages[i][j] = null;
              }
            }
          }
        }
      }
      // free the measure data chunks
      if (null != measureColumnPages) {
        for (int i = 0; i < measureColumnPages.length; i++) {
          if (null != measureColumnPages[i]) {
            for (int j = 0; j < measureColumnPages[i].length; j++) {
              if (null != measureColumnPages[i][j]) {
                measureColumnPages[i][j].freeMemory();
                measureColumnPages[i][j] = null;
              }
            }
          }
        }
      }
      // free the raw chunks
      if (null != dimRawColumnChunks) {
        for (int i = 0; i < dimRawColumnChunks.length; i++) {
          if (null != dimRawColumnChunks[i]) {
            dimRawColumnChunks[i].freeMemory();
            dimRawColumnChunks[i] = null;
          }
        }
      }
      clearValidRowIdList();
      validRowIds = null;
    }
  
    /**
     * @param pageFilteredRowCount set total of number rows valid after scanning
     */
    public void setPageFilteredRowCount(int[] pageFilteredRowCount) {
      this.pageFilteredRowCount = pageFilteredRowCount;
      if (pageIdFiltered == null) {
        pageIdFiltered = new int[pageFilteredRowCount.length];
        for (int i = 0; i < pageIdFiltered.length; i++) {
          pageIdFiltered[i] = i;
        }
      }
    }
  
    /**
     * After applying filter it will return the  bit set with the valid row indexes
     * so below method will be used to set the row indexes
     */
    public void setPageFilteredRowId(int[][] pageFilteredRowId) {
      this.pageFilteredRowId = pageFilteredRowId;
    }
  
    public int getRowCounter() {
      return rowCounter;
    }
  
    /**
     * will return the current valid row id
     *
     * @return valid row id
     */
    public abstract int getCurrentRowId();
  
    /**
     * @return dictionary key array for all the dictionary dimension in integer array format
     * selected in query
     */
    public abstract int[] getDictionaryKeyIntegerArray();
  
    /**
     * Method to fill each dictionary column data column wise
     *
     * @param batchSize
     * @return
     */
    public List<byte[]> getDictionaryKeyArrayBatch(int batchSize) {
      // rowId from where computing need to start
      int startRowId = currentRow + 1;
      fillValidRowIdsBatchFilling(startRowId, batchSize);
      List<byte[]> dictionaryKeyArrayList = new ArrayList<>(validRowIds.size());
      int[] columnDataOffsets = null;
      byte[] completeKey = null;
      // everyTime it is initialized new as in case of prefetch it can modify the data
      for (int i = 0; i < validRowIds.size(); i++) {
        completeKey = new byte[fixedLengthKeySize];
        dictionaryKeyArrayList.add(completeKey);
      }
      // initialize offset array if data is present
      if (this.dictionaryColumnChunkIndexes.length > 0) {
        columnDataOffsets = new int[validRowIds.size()];
      }
      for (int i = 0; i < this.dictionaryColumnChunkIndexes.length; i++) {
        for (int j = 0; j < validRowIds.size(); j++) {
          columnDataOffsets[j] += dimensionColumnPages[dictionaryColumnChunkIndexes[i]][pageCounter]
              .fillRawData(validRowIds.get(j), columnDataOffsets[j], dictionaryKeyArrayList.get(j));
        }
      }
      return dictionaryKeyArrayList;
    }
  
    /**
     * Below method will be used to get the complex type key array
     *
     * @return complex type key array
     */
    public abstract byte[][] getComplexTypeKeyArray();
  
    /**
     * Below method will be used to get the complex type key array
     * This method will fill the data column wise for the given batch size
     *
     * @param batchSize
     * @return complex type key array
     */
    public abstract List<byte[][]> getComplexTypeKeyArrayBatch(int batchSize);
  
    /**
     * Below method will be used to get the no dictionary key
     * array for all the no dictionary dimension selected in query
     *
     * @return no dictionary key array for all the no dictionary dimension
     */
    public abstract byte[][] getNoDictionaryKeyArray();
  
    /**
     * Below method will be used to get the dimension key array
     * for all the no dictionary dimension present in the query
     * This method will fill the data column wise for the given batch size
     *
     * @return no dictionary keys for all no dictionary dimension
     */
    public List<byte[][]> getNoDictionaryKeyArrayBatch(int batchSize) {
      List<byte[][]> noDictionaryKeyArrayList = new ArrayList<>(validRowIds.size());
      byte[][] noDictionaryColumnsKeys = null;
      // everyTime it is initialized new as in case of prefetch it can modify the data
      for (int i = 0; i < validRowIds.size(); i++) {
        noDictionaryColumnsKeys = new byte[noDictionaryColumnChunkIndexes.length][];
        noDictionaryKeyArrayList.add(noDictionaryColumnsKeys);
      }
      int columnPosition = 0;
      for (int i = 0; i < this.noDictionaryColumnChunkIndexes.length; i++) {
        for (int j = 0; j < validRowIds.size(); j++) {
          byte[][] noDictionaryArray = noDictionaryKeyArrayList.get(j);
          noDictionaryArray[columnPosition] =
              dimensionColumnPages[noDictionaryColumnChunkIndexes[i]][pageCounter]
                  .getChunkData(validRowIds.get(j));
        }
        columnPosition++;
      }
      return noDictionaryKeyArrayList;
    }
  
    /**
     * Mark the filtered rows in columnar batch. These rows will not be added to vector batches later.
     * @param columnarBatch
     * @param startRow
     * @param size
     * @param vectorOffset
     */
    public int markFilteredRows(CarbonColumnarBatch columnarBatch, int startRow, int size,
        int vectorOffset) {
      int rowsFiltered = 0;
      if (currentDeleteDeltaVo != null) {
        int len = startRow + size;
        for (int i = startRow; i < len; i++) {
          int rowId = pageFilteredRowId != null ? pageFilteredRowId[pageCounter][i] : i;
          if (currentDeleteDeltaVo.containsRow(rowId)) {
            columnarBatch.markFiltered(vectorOffset);
            rowsFiltered++;
          }
          vectorOffset++;
        }
      }
      return rowsFiltered;
    }
  
    public DeleteDeltaVo getCurrentDeleteDeltaVo() {
      return currentDeleteDeltaVo;
    }
  
    /**
     * Below method will be used to check row got deleted
     *
     * @param rowId
     * @return is present in deleted row
     */
    public boolean containsDeletedRow(int rowId) {
      if (null != currentDeleteDeltaVo) {
        return currentDeleteDeltaVo.containsRow(rowId);
      }
      return false;
    }
  
    public int getBlockletNumber() {
      return Integer.parseInt(blockletNumber);
    }
  }
  