/*
 * 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.store.impl;

import java.util.Arrays;
import java.util.BitSet;

import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.chunk.store.DimensionDataChunkStore;
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.CarbonDictionary;
import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
import org.apache.carbondata.core.scan.result.vector.impl.directread.ColumnarVectorWrapperDirectFactory;
import org.apache.carbondata.core.scan.result.vector.impl.directread.ConvertibleVector;
import org.apache.carbondata.core.util.CarbonUtil;

/**
 * Dimension chunk store for local dictionary encoded data.
 * It's a decorator over dimension chunk store
 */
public class LocalDictDimensionDataChunkStore implements DimensionDataChunkStore {

  private DimensionDataChunkStore dimensionDataChunkStore;

  private CarbonDictionary dictionary;

  private int dataLength;

  public LocalDictDimensionDataChunkStore(DimensionDataChunkStore dimensionDataChunkStore,
      CarbonDictionary dictionary, int dataLength) {
    this.dimensionDataChunkStore = dimensionDataChunkStore;
    this.dictionary = dictionary;
    this.dataLength = dataLength;
  }

  /**
   * Below method will be used to put the rows and its metadata in off-heap
   *
   * @param invertedIndex        inverted index to be stored
   * @param invertedIndexReverse inverted index reverse to be stored
   * @param data                 data to be stored
   */
  public void putArray(int[] invertedIndex, int[] invertedIndexReverse, byte[] data) {
    this.dimensionDataChunkStore.putArray(invertedIndex, invertedIndexReverse, data);
  }

  @Override
  public void fillVector(int[] invertedIndex, int[] invertedIndexReverse, byte[] data,
      ColumnVectorInfo vectorInfo) {
    int columnValueSize = dimensionDataChunkStore.getColumnValueSize();
    int rowsNum = dataLength / columnValueSize;
    CarbonColumnVector vector = vectorInfo.vector;
    if (vector.getType().isComplexType()) {
      if (DataTypes.isStructType(vector.getType())) {
        int deletedRow = vectorInfo.deletedRows != null ? vectorInfo.deletedRows.cardinality() : 0;
        rowsNum = dataLength - deletedRow;
      } else {
        // this is not required to be changed in the case of the array because
        // datalength of the array already taking care of deleted rows in
        // ColumnVectorInfo.getUpdatedPageSizeForChildVector
        rowsNum = dataLength;
      }
      vector = vectorInfo.vectorStack.peek();
      CarbonColumnVector sliceVector = vector.getColumnVector();
      // use rowsNum as positionCount in order to create dictionary block
      sliceVector.setPositionCount(rowsNum);
      sliceVector.setIsLocalDictEnabledForComplextype(true);
    }
    if (!dictionary.isDictionaryUsed()) {
      vector.setDictionary(dictionary);
      dictionary.setDictionaryUsed();
    }
    BitSet nullBitset = new BitSet();
    CarbonColumnVector dictionaryVector = ColumnarVectorWrapperDirectFactory
        .getDirectVectorWrapperFactory(vectorInfo, vector.getDictionaryVector(), invertedIndex,
            nullBitset, vectorInfo.deletedRows, false, true);
    vector = ColumnarVectorWrapperDirectFactory
        .getDirectVectorWrapperFactory(vectorInfo, vector, invertedIndex, nullBitset,
            vectorInfo.deletedRows, false, false);
    // this check is in case of array of string type
    if (vectorInfo.vector.getType().isComplexType()
        && dictionaryVector instanceof CarbonColumnVectorImpl
        && ((CarbonColumnVectorImpl) dictionaryVector).getIntArraySize() < rowsNum) {
      ((CarbonColumnVectorImpl) dictionaryVector).increaseIntArraySize(rowsNum);
    }
    for (int i = 0; i < rowsNum; i++) {
      int surrogate = CarbonUtil.getSurrogateInternal(data, i * columnValueSize, columnValueSize);
      // If complex string primitive value is null then surrogate will be unequal to
      // MEMBER_DEFAULT_VAL_SURROGATE_KEY. Therefore check should be using MEMBER_DEFAULT_VAL_ARRAY
      if (surrogate == CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY || Arrays
          .equals(CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY,
              dictionary.getDictionaryValue(surrogate))) {
        vector.putNull(i);
        dictionaryVector.putNull(i);
      } else {
        // if vector is 'ColumnarVectorWrapperDirectWithDeleteDelta', it needs to call 'putNotNull'
        // to increase 'counter', otherwise it will set the null value to the wrong index.
        vector.putNotNull(i);
        dictionaryVector.putInt(i, surrogate);
      }
    }
    if (dictionaryVector instanceof ConvertibleVector) {
      ((ConvertibleVector) dictionaryVector).convert();
    }
  }

  @Override
  public byte[] getRow(int rowId) {
    return dictionary.getDictionaryValue(dimensionDataChunkStore.getSurrogate(rowId));
  }

  @Override
  public void fillRow(int rowId, CarbonColumnVector vector, int vectorRow) {
    if (!dictionary.isDictionaryUsed()) {
      vector.setDictionary(dictionary);
      dictionary.setDictionaryUsed();
    }
    int surrogate = dimensionDataChunkStore.getSurrogate(rowId);
    if (surrogate == CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY) {
      vector.putNull(vectorRow);
      vector.getDictionaryVector().putNull(vectorRow);
      return;
    }
    vector.putNotNull(vectorRow);
    vector.getDictionaryVector().putInt(vectorRow, dimensionDataChunkStore.getSurrogate(rowId));
  }

  @Override
  public void fillRow(int rowId, byte[] buffer, int offset) {
    throw new UnsupportedOperationException("Operation not supported");
  }

  @Override
  public int getInvertedIndex(int rowId) {
    return this.dimensionDataChunkStore.getInvertedIndex(rowId);
  }

  @Override
  public int getInvertedReverseIndex(int rowId) {
    return this.dimensionDataChunkStore.getInvertedReverseIndex(rowId);
  }

  @Override
  public int getSurrogate(int rowId) {
    throw new UnsupportedOperationException("Operation not supported");
  }

  @Override
  public int getColumnValueSize() {
    throw new UnsupportedOperationException("Operation not supported");
  }

  @Override
  public boolean isExplicitSorted() {
    return this.dimensionDataChunkStore.isExplicitSorted();
  }

  @Override
  public int compareTo(int rowId, byte[] compareValue) {
    return dimensionDataChunkStore.compareTo(rowId, compareValue);
  }

  /**
   * Below method will be used to free the memory occupied by the column chunk
   */
  @Override
  public void freeMemory() {
    if (null != dimensionDataChunkStore) {
      this.dimensionDataChunkStore.freeMemory();
      this.dictionary = null;
      this.dimensionDataChunkStore = null;
    }
  }
}