/*
 * 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.page.encoding;

import java.math.BigDecimal;

import org.apache.carbondata.core.datastore.ColumnType;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.datastore.page.DecimalColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.adaptive.AdaptiveDeltaFloatingCodec;
import org.apache.carbondata.core.datastore.page.encoding.adaptive.AdaptiveDeltaIntegralCodec;
import org.apache.carbondata.core.datastore.page.encoding.adaptive.AdaptiveFloatingCodec;
import org.apache.carbondata.core.datastore.page.encoding.adaptive.AdaptiveIntegralCodec;
import org.apache.carbondata.core.datastore.page.encoding.compress.DirectCompressCodec;
import org.apache.carbondata.core.datastore.page.encoding.dimension.legacy.ComplexDimensionIndexCodec;
import org.apache.carbondata.core.datastore.page.encoding.dimension.legacy.DirectDictDimensionIndexCodec;
import org.apache.carbondata.core.datastore.page.encoding.dimension.legacy.PlainDimensionIndexCodec;
import org.apache.carbondata.core.datastore.page.statistics.PrimitivePageStatsCollector;
import org.apache.carbondata.core.datastore.page.statistics.SimpleStatsResult;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
import org.apache.carbondata.core.util.DataTypeUtil;

/**
 * Default factory will select encoding base on column page data type and statistics
 */
public class DefaultEncodingFactory extends EncodingFactory {

  private static final int THREE_BYTES_MAX = (int) Math.pow(2, 23) - 1;
  private static final int THREE_BYTES_MIN = -THREE_BYTES_MAX - 1;

  private static final boolean newWay = false;

  private static EncodingFactory encodingFactory = new DefaultEncodingFactory();

  public static EncodingFactory getInstance() {
    // TODO: make it configurable after added new encodingFactory
    return encodingFactory;
  }

  @Override
  public ColumnPageEncoder createEncoder(TableSpec.ColumnSpec columnSpec, ColumnPage inputPage) {
    // TODO: add log
    // choose the encoding type for measure type and no dictionary primitive type columns
    if (columnSpec instanceof TableSpec.MeasureSpec || (
        DataTypeUtil.isPrimitiveColumn(columnSpec.getSchemaDataType())
            && columnSpec.getColumnType() == ColumnType.PLAIN_VALUE)) {
      return createEncoderForMeasureOrNoDictionaryPrimitive(inputPage, columnSpec);
    } else {
      if (newWay) {
        return createEncoderForDimension((TableSpec.DimensionSpec) columnSpec, inputPage);
      } else {
        assert columnSpec instanceof TableSpec.DimensionSpec;
        return createEncoderForDimensionLegacy((TableSpec.DimensionSpec) columnSpec);
      }
    }
  }

  private ColumnPageEncoder createEncoderForDimension(TableSpec.DimensionSpec columnSpec,
      ColumnPage inputPage) {
    switch (columnSpec.getColumnType()) {
      case DIRECT_DICTIONARY:
      case PLAIN_VALUE:
        return new DirectCompressCodec(inputPage.getDataType()).createEncoder(null);
      case COMPLEX:
        return new ComplexDimensionIndexCodec(false, false).createEncoder(null);
      default:
        throw new RuntimeException("unsupported dimension type: " + columnSpec.getColumnType());
    }
  }

  private ColumnPageEncoder createEncoderForDimensionLegacy(TableSpec.DimensionSpec dimensionSpec) {
    switch (dimensionSpec.getColumnType()) {
      case DIRECT_DICTIONARY:
        return new DirectDictDimensionIndexCodec(
            dimensionSpec.isInSortColumns(),
            dimensionSpec.isInSortColumns() && dimensionSpec.isDoInvertedIndex())
            .createEncoder(null);
      case PLAIN_VALUE:
        return new PlainDimensionIndexCodec(dimensionSpec.isInSortColumns(),
            dimensionSpec.isInSortColumns() && dimensionSpec.isDoInvertedIndex(),
            dimensionSpec.getSchemaDataType() == DataTypes.VARCHAR
                || dimensionSpec.getSchemaDataType() == DataTypes.BINARY).createEncoder(null);
      default:
        throw new RuntimeException("unsupported dimension type: " +
            dimensionSpec.getColumnType());
    }
  }

  private ColumnPageEncoder createEncoderForMeasureOrNoDictionaryPrimitive(ColumnPage columnPage,
      TableSpec.ColumnSpec columnSpec) {

    SimpleStatsResult stats = columnPage.getStatistics();
    DataType dataType = stats.getDataType();
    if (dataType == DataTypes.BOOLEAN
        || dataType == DataTypes.BYTE_ARRAY
        || columnPage.getDataType() == DataTypes.BINARY) {
      return new DirectCompressCodec(columnPage.getDataType()).createEncoder(null);
    } else if (dataType == DataTypes.BYTE ||
        dataType == DataTypes.SHORT ||
        dataType == DataTypes.INT ||
        dataType == DataTypes.LONG ||
        dataType == DataTypes.TIMESTAMP) {
      return selectCodecByAlgorithmForIntegral(stats, false, columnSpec).createEncoder(null);
    } else if (DataTypes.isDecimal(dataType)) {
      return createEncoderForDecimalDataTypeMeasure(columnPage, columnSpec);
    } else if (dataType == DataTypes.FLOAT || dataType == DataTypes.DOUBLE) {
      return selectCodecByAlgorithmForFloating(stats, false, columnSpec).createEncoder(null);
    } else {
      throw new RuntimeException("unsupported data type: " + stats.getDataType());
    }
  }

  private ColumnPageEncoder createEncoderForDecimalDataTypeMeasure(ColumnPage columnPage,
      TableSpec.ColumnSpec columnSpec) {
    DecimalConverterFactory.DecimalConverterType decimalConverterType =
        ((DecimalColumnPage) columnPage).getDecimalConverter().getDecimalConverterType();
    switch (decimalConverterType) {
      case DECIMAL_INT:
      case DECIMAL_LONG:
        return selectCodecByAlgorithmForDecimal(columnPage.getStatistics(), decimalConverterType,
            columnSpec)
            .createEncoder(null);
      default:
        return new DirectCompressCodec(columnPage.getDataType()).createEncoder(null);
    }
  }

  private static DataType fitLongMinMax(long max, long min) {
    if (max <= Byte.MAX_VALUE && min >= Byte.MIN_VALUE) {
      return DataTypes.BYTE;
    } else if (max <= Short.MAX_VALUE && min >= Short.MIN_VALUE) {
      return DataTypes.SHORT;
    } else if (max <= THREE_BYTES_MAX && min >= THREE_BYTES_MIN) {
      return DataTypes.SHORT_INT;
    } else if (max <= Integer.MAX_VALUE && min >= Integer.MIN_VALUE) {
      return DataTypes.INT;
    } else {
      return DataTypes.LONG;
    }
  }

  private static DataType fitMinMax(DataType dataType, Object max, Object min) {
    if (dataType == DataTypes.BYTE || dataType == DataTypes.BOOLEAN) {
      return fitLongMinMax((byte) max, (byte) min);
    } else if (dataType == DataTypes.SHORT) {
      return fitLongMinMax((short) max, (short) min);
    } else if (dataType == DataTypes.INT) {
      return fitLongMinMax((int) max, (int) min);
    } else if ((dataType == DataTypes.LONG) || (dataType == DataTypes.TIMESTAMP)) {
      return fitLongMinMax((long) max, (long) min);
    } else if (dataType == DataTypes.DOUBLE) {
      return fitLongMinMax((long) (double) max, (long) (double) min);
    } else {
      throw new RuntimeException("internal error: " + dataType);
    }
  }

  private static DataType fitMinMaxForDecimalType(DataType dataType, Object max, Object min,
      DecimalConverterFactory.DecimalConverterType decimalConverterType) {
    long maxValue = ((BigDecimal) max).unscaledValue().longValue();
    long minValue = ((BigDecimal) min).unscaledValue().longValue();
    switch (decimalConverterType) {
      case DECIMAL_INT:
        return fitLongMinMax((int) maxValue, (int) minValue);
      case DECIMAL_LONG:
        return fitLongMinMax(maxValue, minValue);
      default:
        throw new RuntimeException("internal error: " + dataType);
    }
  }

  private static DataType fitDeltaForDecimalType(DataType dataType, Object max, Object min,
      DecimalConverterFactory.DecimalConverterType decimalConverterType) {
    long maxValue = ((BigDecimal) max).unscaledValue().longValue();
    long minValue = ((BigDecimal) min).unscaledValue().longValue();
    switch (decimalConverterType) {
      case DECIMAL_INT:
        long value = maxValue - minValue;
        return compareMinMaxAndSelectDataType(value);
      case DECIMAL_LONG:
        return DataTypes.LONG;
      default:
        throw new RuntimeException("internal error: " + dataType);
    }
  }

  // fit the long input value into minimum data type
  private static DataType fitDelta(DataType dataType, Object max, Object min) {
    // use long data type to calculate delta to avoid overflow
    long value;
    if (dataType == DataTypes.BYTE || dataType == DataTypes.BOOLEAN) {
      value = (long) (byte) max - (long) (byte) min;
    } else if (dataType == DataTypes.SHORT) {
      value = (long) (short) max - (long) (short) min;
    } else if (dataType == DataTypes.INT) {
      value = (long) (int) max - (long) (int) min;
    } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
      value = (long) max - (long) min;
      // The subtraction overflowed iff the operands have opposing signs
      // and the result's sign differs from the minuend.
      boolean overflow = (((long) max ^ (long) min) & ((long) max ^ value)) < 0;
      if (overflow) {
        return DataTypes.LONG;
      }
    } else if (dataType == DataTypes.DOUBLE) {
      return DataTypes.LONG;
    } else {
      throw new RuntimeException("internal error: " + dataType);
    }
    return compareMinMaxAndSelectDataType(value);
  }

  private static DataType compareMinMaxAndSelectDataType(long value) {
    return fitLongMinMax(value, value);
  }

  /**
   * choose between adaptive encoder or delta adaptive encoder, based on whose target data type
   * size is smaller
   */
  static ColumnPageCodec selectCodecByAlgorithmForIntegral(SimpleStatsResult stats,
      boolean isComplexPrimitive, TableSpec.ColumnSpec columnSpec) {
    DataType srcDataType = stats.getDataType();
    DataType adaptiveDataType = fitMinMax(stats.getDataType(), stats.getMax(), stats.getMin());

    DataType deltaDataType = fitDelta(stats.getDataType(), stats.getMax(), stats.getMin());
    // for complex primitive, if source and destination data type is same, use adaptive encoding.
    if (!isComplexPrimitive) {
      // in case of decimal datatype, check if the decimal converter type is Int or Long and based
      // on that get size in bytes
      if (Math.min(adaptiveDataType.getSizeInBytes(), deltaDataType.getSizeInBytes()) == srcDataType
          .getSizeInBytes()) {
        // no effect to use adaptive or delta, use compression only
        return new DirectCompressCodec(stats.getDataType());
      }
    }
    boolean isInvertedIndex = isInvertedIndex(isComplexPrimitive, columnSpec);
    if (adaptiveDataType.getSizeInBytes() <= deltaDataType.getSizeInBytes()) {
      // choose adaptive encoding
      return new AdaptiveIntegralCodec(stats.getDataType(), adaptiveDataType, stats,
          isInvertedIndex);
    } else {
      // choose delta adaptive encoding
      return new AdaptiveDeltaIntegralCodec(stats.getDataType(), deltaDataType, stats,
          isInvertedIndex);
    }
  }

  /**
   * Check whether the column is sort column and inverted index column
   *
   * @param isComplexPrimitive
   * @param columnSpec
   * @return
   */
  private static boolean isInvertedIndex(boolean isComplexPrimitive,
      TableSpec.ColumnSpec columnSpec) {
    boolean isSort;
    boolean isInvertedIndex = false;
    if (columnSpec instanceof TableSpec.DimensionSpec && !isComplexPrimitive) {
      isSort = ((TableSpec.DimensionSpec) columnSpec).isInSortColumns();
      isInvertedIndex = isSort && ((TableSpec.DimensionSpec) columnSpec).isDoInvertedIndex();
    }
    return isInvertedIndex;
  }

  // choose between upscale adaptive encoder or upscale delta adaptive encoder,
  // based on whose target data type size is smaller
  static ColumnPageCodec selectCodecByAlgorithmForFloating(SimpleStatsResult stats,
      boolean isComplexPrimitive, TableSpec.ColumnSpec columnSpec) {
    DataType srcDataType = stats.getDataType();
    double maxValue;
    double minValue;
    if (srcDataType == DataTypes.FLOAT) {
      maxValue = (float) stats.getMax();
      minValue = (float) stats.getMin();
    } else {
      maxValue = (double) stats.getMax();
      minValue = (double) stats.getMin();
    }
    int decimalCount = stats.getDecimalCount();

    // For Complex Type primitive we should always choose adaptive path
    // as LV format will be reduced to only V format. Therefore inorder
    // to do that decimal count should be actual count instead of -1.
    if (isComplexPrimitive && decimalCount == -1 && stats instanceof PrimitivePageStatsCollector) {
      decimalCount = ((PrimitivePageStatsCollector)stats).getDecimalForComplexPrimitive();
    }

    //Here we should use the Max abs as max to getDatatype, let's say -1 and -10000000, -1 is max,
    //but we can't use -1 to getDatatype, we should use -10000000.
    double absMaxValue = Math.max(Math.abs(maxValue), Math.abs(minValue));
    if (srcDataType == DataTypes.FLOAT && decimalCount == 0) {
      return getColumnPageCodec(stats, isComplexPrimitive, columnSpec, srcDataType, maxValue,
          minValue, decimalCount, absMaxValue);
    } else if (decimalCount == 0) {
      // short, int, long
      return selectCodecByAlgorithmForIntegral(stats, false, columnSpec);
    } else if (decimalCount < 0 && !isComplexPrimitive) {
      return new DirectCompressCodec(DataTypes.DOUBLE);
    } else {
      return getColumnPageCodec(stats, isComplexPrimitive, columnSpec, srcDataType, maxValue,
          minValue, decimalCount, absMaxValue);
    }
  }

  private static ColumnPageCodec getColumnPageCodec(SimpleStatsResult stats,
      boolean isComplexPrimitive, TableSpec.ColumnSpec columnSpec, DataType srcDataType,
      double maxValue, double minValue, int decimalCount, double absMaxValue) {
    // double
    // If absMaxValue exceeds LONG.MAX_VALUE, then go for direct compression
    if ((Math.pow(10, decimalCount) * absMaxValue) > Long.MAX_VALUE) {
      return new DirectCompressCodec(DataTypes.DOUBLE);
    } else {
      long max = (long) (Math.pow(10, decimalCount) * absMaxValue);
      DataType adaptiveDataType = fitLongMinMax(max, 0);
      DataType deltaDataType = compareMinMaxAndSelectDataType(
          (long) (Math.pow(10, decimalCount) * (maxValue - minValue)));
      if (adaptiveDataType.getSizeInBytes() > deltaDataType.getSizeInBytes()) {
        return new AdaptiveDeltaFloatingCodec(srcDataType, deltaDataType, stats,
            isInvertedIndex(isComplexPrimitive, columnSpec));
      } else if (adaptiveDataType.getSizeInBytes() < DataTypes.DOUBLE.getSizeInBytes() || (
          (isComplexPrimitive) && (adaptiveDataType.getSizeInBytes() == DataTypes.DOUBLE
              .getSizeInBytes()))) {
        return new AdaptiveFloatingCodec(srcDataType, adaptiveDataType, stats,
            isInvertedIndex(isComplexPrimitive, columnSpec));
      } else {
        return new DirectCompressCodec(DataTypes.DOUBLE);
      }
    }
  }

  /**
   * choose between adaptive encoder or delta adaptive encoder, based on whose target data type
   * size is smaller for decimal data type
   */
  static ColumnPageCodec selectCodecByAlgorithmForDecimal(SimpleStatsResult stats,
      DecimalConverterFactory.DecimalConverterType decimalConverterType,
      TableSpec.ColumnSpec columnSpec) {
    DataType srcDataType = stats.getDataType();
    DataType adaptiveDataType =
        fitMinMaxForDecimalType(stats.getDataType(), stats.getMax(), stats.getMin(),
            decimalConverterType);
    DataType deltaDataType;

    if (adaptiveDataType == DataTypes.LONG) {
      deltaDataType = DataTypes.LONG;
    } else {
      deltaDataType = fitDeltaForDecimalType(stats.getDataType(), stats.getMax(), stats.getMin(),
          decimalConverterType);
    }
    // in case of decimal data type check if the decimal converter type is Int or Long and based on
    // that get size in bytes
    if (Math.min(adaptiveDataType.getSizeInBytes(), deltaDataType.getSizeInBytes()) == srcDataType
        .getSizeInBytes()) {
      // no effect to use adaptive or delta, use compression only
      return new DirectCompressCodec(stats.getDataType());
    }
    if (adaptiveDataType.getSizeInBytes() <= deltaDataType.getSizeInBytes()) {
      // choose adaptive encoding
      return new AdaptiveIntegralCodec(stats.getDataType(), adaptiveDataType, stats,
          isInvertedIndex(columnSpec.getColumnType() == ColumnType.COMPLEX_PRIMITIVE, columnSpec));
    } else {
      // choose delta adaptive encoding
      return new AdaptiveDeltaIntegralCodec(stats.getDataType(), deltaDataType, stats,
          isInvertedIndex(columnSpec.getColumnType() == ColumnType.COMPLEX_PRIMITIVE, columnSpec));
    }
  }

}