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.datastore.chunk.reader.measure;
  
  import java.io.IOException;
  import java.util.List;
  
  import org.apache.carbondata.core.datastore.FileReader;
  import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
  import org.apache.carbondata.core.datastore.chunk.reader.MeasureColumnChunkReader;
  import org.apache.carbondata.core.datastore.compression.Compressor;
  import org.apache.carbondata.core.datastore.page.encoding.DefaultEncodingFactory;
  import org.apache.carbondata.core.datastore.page.encoding.EncodingFactory;
  import org.apache.carbondata.core.metadata.blocklet.BlockletInfo;
  
  /**
   * Measure block reader abstract class
   */
  public abstract class AbstractMeasureChunkReader implements MeasureColumnChunkReader {
    protected Compressor compressor;
  
    protected EncodingFactory encodingFactory = DefaultEncodingFactory.getInstance();
  
    /**
     * file path from which blocks will be read
     */
    protected String filePath;
    /**
     * measure column chunks offset
     */
    protected List<Long> measureColumnChunkOffsets;
    /**
     * measure column chunks length
     */
    protected List<Integer> measureColumnChunkLength;
  
    /**
     * Constructor to get minimum parameter to create instance of this class
     *
     * @param filePath file from which data will be read
     */
    public AbstractMeasureChunkReader(final BlockletInfo blockletInfo, String filePath) {
      this.filePath = filePath;
      this.measureColumnChunkOffsets = blockletInfo.getMeasureChunkOffsets();
      this.measureColumnChunkLength = blockletInfo.getMeasureChunksLength();
    }
  
    /**
     * Below method will be used to read the chunk based on block indexes
     * Reading logic of below method is: Except last column all the column chunk
     * can be read in group if not last column then read data of all the column
     * present in block index together then process it. For last column read is
     * separately and process
     *
     * @param fileReader   file reader to read the blocks from file
     * @param columnIndexRange blocks range to be read, columnIndexGroup[i] is one group, inside the
     *                         group, columnIndexGroup[i][0] is start column index,
     *                         and columnIndexGroup[i][1] is end column index
     * @return measure column chunks
     * @throws IOException
     */
    public MeasureRawColumnChunk[] readRawMeasureChunks(FileReader fileReader,
        int[][] columnIndexRange) throws IOException {
      // read the column chunk based on block index and add
      MeasureRawColumnChunk[] dataChunks =
          new MeasureRawColumnChunk[measureColumnChunkOffsets.size()];
      if (columnIndexRange.length == 0) {
        return dataChunks;
      }
      MeasureRawColumnChunk[] groupChunk = null;
      int index = 0;
      for (int i = 0; i < columnIndexRange.length - 1; i++) {
        index = 0;
        groupChunk = readRawMeasureChunksInGroup(
            fileReader, columnIndexRange[i][0], columnIndexRange[i][1]);
        for (int j = columnIndexRange[i][0]; j <= columnIndexRange[i][1]; j++) {
          dataChunks[j] = groupChunk[index++];
        }
      }
      if (columnIndexRange[columnIndexRange.length - 1][0] == measureColumnChunkOffsets.size() - 1) {
        dataChunks[columnIndexRange[columnIndexRange.length - 1][0]] =
            readRawMeasureChunk(fileReader, columnIndexRange[columnIndexRange.length - 1][0]);
      } else {
        groupChunk = readRawMeasureChunksInGroup(
            fileReader, columnIndexRange[columnIndexRange.length - 1][0],
            columnIndexRange[columnIndexRange.length - 1][1]);
        index = 0;
        for (int j = columnIndexRange[columnIndexRange.length - 1][0];
             j <= columnIndexRange[columnIndexRange.length - 1][1]; j++) {
          dataChunks[j] = groupChunk[index++];
        }
      }
      return dataChunks;
    }
  
    /**
     * Below method will be used to read measure chunk data in group.
     * This method will be useful to avoid multiple IO while reading the
     * data from
     *
     * @param fileReader               file reader to read the data
     * @param startColumnIndex first column index to be read
     * @param endColumnIndex   end column index to be read
     * @return measure raw chunkArray
     * @throws IOException
     */
    protected abstract MeasureRawColumnChunk[] readRawMeasureChunksInGroup(FileReader fileReader,
        int startColumnIndex, int endColumnIndex) throws IOException;
  }
  