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.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));
      }
    }
  
  }
  