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 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.scan.result.vector.CarbonColumnVector;
  import org.apache.carbondata.core.scan.result.vector.CarbonDictionary;
  import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
  
  /**
   * This class is gives access to variable length dimension data chunk store
   */
  public class VariableLengthDimensionColumnPage extends AbstractDimensionColumnPage {
  
    /**
     * Constructor for this class
     * @param dataChunks           data chunk
     * @param invertedIndex        inverted index
     * @param invertedIndexReverse reverse inverted index
     * @param numberOfRows         number of rows
     * @param dictionary           carbon local dictionary for string column.
     * @param vectorInfo           vector to be filled with decoded column page.
     */
    public VariableLengthDimensionColumnPage(byte[] dataChunks, int[] invertedIndex,
        int[] invertedIndexReverse, int numberOfRows, DimensionStoreType dimStoreType,
        CarbonDictionary dictionary, ColumnVectorInfo vectorInfo, int dataLength) {
      boolean isExplicitSorted = isExplicitSorted(invertedIndex);
      long totalSize = 0;
      switch (dimStoreType) {
        case LOCAL_DICT:
          totalSize = null != invertedIndex ?
              (dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE)) :
              dataLength;
          break;
        case VARIABLE_INT_LENGTH:
        case VARIABLE_SHORT_LENGTH:
          totalSize = null != invertedIndex ?
              (dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) + (
                  numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE)) :
              (dataLength + (numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE));
          break;
        default:
          throw new UnsupportedOperationException("Invalidate dimension store type");
      }
      dataChunkStore = DimensionChunkStoreFactory.INSTANCE
          .getDimensionChunkStore(0, isExplicitSorted, numberOfRows, totalSize, dimStoreType,
              dictionary, vectorInfo != null, dataLength);
      if (vectorInfo != null) {
        dataChunkStore.fillVector(invertedIndex, invertedIndexReverse, dataChunks, vectorInfo);
      } else {
        dataChunkStore.putArray(invertedIndex, invertedIndexReverse, dataChunks);
      }
    }
  
    /**
     * 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) {
      // no required in this case because this column chunk is not the part of MDKey
      return 0;
    }
  
    /**
     * Converts to column dictionary integer value
     *
     * @param rowId
     * @param chunkIndex
     * @param outputSurrogateKey
     * @return
     */
    @Override
    public int fillSurrogateKey(int rowId, int chunkIndex, int[] outputSurrogateKey) {
      return chunkIndex + 1;
    }
  
    /**
     * @return whether column is dictionary column or not
     */
    @Override
    public boolean isNoDictionaryColumn() {
      return true;
    }
  
    /**
     * Fill the data to vector
     *
     * @param vectorInfo
     * @param chunkIndex
     * @return next column index
     */
    @Override
    public int fillVector(ColumnVectorInfo[] vectorInfo, int chunkIndex) {
      ColumnVectorInfo columnVectorInfo = vectorInfo[chunkIndex];
      CarbonColumnVector vector = columnVectorInfo.vector;
      int offset = columnVectorInfo.offset;
      int vectorOffset = columnVectorInfo.vectorOffset;
      int len = offset + columnVectorInfo.size;
      for (int i = offset; i < len; i++) {
        // Considering only String case now as we support only
        // string in no dictionary case at present.
        dataChunkStore.fillRow(i, vector, vectorOffset++);
      }
      return chunkIndex + 1;
    }
  
    /**
     * 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) {
      ColumnVectorInfo columnVectorInfo = vectorInfo[chunkIndex];
      CarbonColumnVector vector = columnVectorInfo.vector;
      int offset = columnVectorInfo.offset;
      int vectorOffset = columnVectorInfo.vectorOffset;
      int len = offset + columnVectorInfo.size;
      for (int i = offset; i < len; i++) {
        // Considering only String case now as we support only
        // string in no dictionary case at present.
        dataChunkStore.fillRow(filteredRowId[i], vector, vectorOffset++);
      }
      return chunkIndex + 1;
    }
  }
  