/*
 * 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.localdictionary.dictionaryholder;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.carbondata.core.cache.dictionary.DictionaryByteArrayWrapper;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.localdictionary.exception.DictionaryThresholdReachedException;
import org.apache.carbondata.core.util.CarbonProperties;

/**
 * Map based dictionary holder class, it will use map to hold
 * the dictionary key and its value
 */
public class MapBasedDictionaryStore implements DictionaryStore {

  /**
   * use to assign dictionary value to new key
   */
  private int lastAssignValue;

  /**
   * to maintain dictionary key value
   */
  private final Map<DictionaryByteArrayWrapper, Integer> dictionary;

  /**
   * maintaining array for reverse lookup
   * otherwise iterating every time in map for reverse lookup will be slowdown the performance
   * It will only maintain the reference
   */
  private DictionaryByteArrayWrapper[] referenceDictionaryArray;

  /**
   * dictionary threshold to check if threshold is reached
   */
  private int dictionaryThreshold;

  /**
   * dictionary threshold size in bytes
   */
  private long dictionarySizeThresholdInBytes;

  /**
   * for checking threshold is reached or not
   */
  private boolean isThresholdReached;

  /**
   * current data size
   */
  private long currentSize;

  public MapBasedDictionaryStore(int dictionaryThreshold) {
    this.dictionaryThreshold = dictionaryThreshold;
    this.dictionarySizeThresholdInBytes = Integer.parseInt(CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.CARBON_LOCAL_DICTIONARY_SIZE_THRESHOLD_IN_MB)) << 20;
    this.dictionary = new ConcurrentHashMap<>();
    this.referenceDictionaryArray = new DictionaryByteArrayWrapper[dictionaryThreshold];
  }

  /**
   * Below method will be used to add dictionary value to dictionary holder
   * if it is already present in the holder then it will return exiting dictionary value.
   *
   * @param data dictionary key
   * @return dictionary value
   */
  @Override
  public int putIfAbsent(byte[] data) throws DictionaryThresholdReachedException {
    // check if threshold has already reached
    checkIfThresholdReached();
    DictionaryByteArrayWrapper key = new DictionaryByteArrayWrapper(data);
    // get the dictionary value
    Integer value = dictionary.get(key);
    // if value is null then dictionary is not present in store
    if (null == value) {
      // acquire the lock
      synchronized (dictionary) {
        // check threshold
        checkIfThresholdReached();
        // get the value again as other thread might have added
        value = dictionary.get(key);
        // double checking
        if (null == value) {
          // increment the value
          value = ++lastAssignValue;
          currentSize += data.length;
          // if new value is greater than threshold
          if (value > dictionaryThreshold || currentSize > dictionarySizeThresholdInBytes) {
            // set the threshold boolean to true
            isThresholdReached = true;
            // throw exception
            checkIfThresholdReached();
          }
          // add to reference array
          // position is -1 as dictionary value starts from 1
          this.referenceDictionaryArray[value - 1] = key;
          dictionary.put(key, value);
        }
      }
    }
    return value;
  }

  private void checkIfThresholdReached() throws DictionaryThresholdReachedException {
    if (isThresholdReached) {
      if (currentSize > dictionarySizeThresholdInBytes) {
        throw new DictionaryThresholdReachedException(
            "Unable to generate dictionary. Dictionary Size crossed bytes: "
                + dictionarySizeThresholdInBytes);
      } else {
        throw new DictionaryThresholdReachedException(
            "Unable to generate dictionary value. Dictionary threshold reached");
      }
    }
  }

  /**
   * Below method to get the current size of dictionary
   *
   * @return
   */
  @Override
  public boolean isThresholdReached() {
    return isThresholdReached;
  }

  /**
   * Below method will be used to get the dictionary key based on value
   *
   * @param value dictionary value
   *              Caller will take of passing proper value
   * @return dictionary key based on value
   */
  @Override
  public byte[] getDictionaryKeyBasedOnValue(int value) {
    assert referenceDictionaryArray != null;
    // reference array index will be -1 of the value as dictionary value starts from 1
    return referenceDictionaryArray[value - 1].getData();
  }
}