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.impl;
  
  import java.util.function.IntUnaryOperator;
  
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.datastore.chunk.store.DimensionChunkStoreFactory;
  import org.apache.carbondata.core.datastore.chunk.store.DimensionChunkStoreFactory.DimensionStoreType;
  import org.apache.carbondata.core.metadata.datatype.DataType;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
  import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
  
  /**
   * This class is gives access to fixed length dimension data chunk store
   */
  public class FixedLengthDimensionColumnPage extends AbstractDimensionColumnPage {
  
    /**
     * Constructor
     *
     * @param dataChunk            data chunk
     * @param invertedIndex        inverted index
     * @param invertedIndexReverse reverse inverted index
     * @param numberOfRows         number of rows
     * @param columnValueSize      size of each column value
     */
    public FixedLengthDimensionColumnPage(byte[] dataChunk, int[] invertedIndex,
        int[] invertedIndexReverse, int numberOfRows, int columnValueSize, int dataLength) {
      boolean isExplicitSorted = isExplicitSorted(invertedIndex);
      long totalSize = isExplicitSorted ?
          dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
          dataLength;
      dataChunkStore = DimensionChunkStoreFactory.INSTANCE
          .getDimensionChunkStore(columnValueSize, isExplicitSorted, numberOfRows, totalSize,
              DimensionStoreType.FIXED_LENGTH, null, false, dataLength);
      dataChunkStore.putArray(invertedIndex, invertedIndexReverse, dataChunk);
    }
  
    /**
     * Constructor
     *
     * @param dataChunk            data chunk
     * @param invertedIndex        inverted index
     * @param invertedIndexReverse reverse inverted index
     * @param numberOfRows         number of rows
     * @param columnValueSize      size of each column value
     * @param vectorInfo           vector to be filled with decoded column page.
     */
    public FixedLengthDimensionColumnPage(byte[] dataChunk, int[] invertedIndex,
        int[] invertedIndexReverse, int numberOfRows, int columnValueSize,
        ColumnVectorInfo vectorInfo, int dataLength) {
      boolean isExplicitSorted = isExplicitSorted(invertedIndex);
      long totalSize = isExplicitSorted ?
          dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
          dataLength;
      dataChunkStore = DimensionChunkStoreFactory.INSTANCE
          .getDimensionChunkStore(columnValueSize, isExplicitSorted, numberOfRows, totalSize,
              DimensionStoreType.FIXED_LENGTH, null, vectorInfo != null, dataLength);
      if (vectorInfo == null) {
        dataChunkStore.putArray(invertedIndex, invertedIndexReverse, dataChunk);
      } else {
        dataChunkStore.fillVector(invertedIndex, invertedIndexReverse, dataChunk, vectorInfo);
      }
    }
  
    /**
     * Below method will be used to fill the data based on offset and row id
     *
     * @param rowId            row id of the chunk
     * @param offset           offset from which data need to be filed
     * @param data             data to filed
     * @return how many bytes was copied
     */
    @Override
    public int fillRawData(int rowId, int offset, byte[] data) {
      dataChunkStore.fillRow(rowId, data, offset);
      return dataChunkStore.getColumnValueSize();
    }
  
    /**
     * Converts to column dictionary integer value
     *
     * @param rowId
     * @param chunkIndex
     * @param outputSurrogateKey
     * @return
     */
    @Override
    public int fillSurrogateKey(int rowId, int chunkIndex, int[] outputSurrogateKey) {
      outputSurrogateKey[chunkIndex] = dataChunkStore.getSurrogate(rowId);
      return chunkIndex + 1;
    }
  
    private int fillVector(ColumnVectorInfo[] vectorInfo, int chunkIndex,
        IntUnaryOperator function) {
      ColumnVectorInfo columnVectorInfo = vectorInfo[chunkIndex];
      int offset = columnVectorInfo.offset;
      int vectorOffset = columnVectorInfo.vectorOffset;
      int len = columnVectorInfo.size + offset;
      CarbonColumnVector vector = columnVectorInfo.vector;
      for (int j = offset; j < len; j++) {
        int dict = dataChunkStore.getSurrogate(function.applyAsInt(j));
        if (columnVectorInfo.directDictionaryGenerator == null) {
          vector.putInt(vectorOffset++, dict);
        } else {
          Object valueFromSurrogate =
              columnVectorInfo.directDictionaryGenerator.getValueFromSurrogate(dict);
          if (valueFromSurrogate == null) {
            vector.putNull(vectorOffset++);
          } else {
            DataType dataType = columnVectorInfo.directDictionaryGenerator.getReturnType();
            if (dataType == DataTypes.INT) {
              vector.putInt(vectorOffset++, (int) valueFromSurrogate);
            } else if (dataType == DataTypes.LONG) {
              vector.putLong(vectorOffset++, (long) valueFromSurrogate);
            } else {
              throw new IllegalArgumentException("unsupported data type: " +
                  columnVectorInfo.directDictionaryGenerator.getReturnType());
            }
          }
        }
      }
      return chunkIndex + 1;
    }
  
  
    /**
     * Fill the data to vector
     *
     * @param vectorInfo
     * @param chunkIndex
     * @return next column index
     */
    @Override
    public int fillVector(ColumnVectorInfo[] vectorInfo, int chunkIndex) {
      return fillVector(vectorInfo, chunkIndex, j -> j);
    }
  
    /**
     * Fill the data to vector
     *
     * @param filteredRowId
     * @param vectorInfo
     * @param chunkIndex
     * @return next column index
     */
    @Override
    public int fillVector(int[] filteredRowId, ColumnVectorInfo[] vectorInfo,
        int chunkIndex) {
      return fillVector(vectorInfo, chunkIndex, j -> filteredRowId[j]);
    }
  }
  