/*
 * 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.metadata.datatype;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.BitSet;

import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
import org.apache.carbondata.core.scan.result.vector.impl.directread.ColumnarVectorWrapperDirectFactory;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.format.Encoding;

/**
 * Decimal converter to keep the data compact.
 */
public final class DecimalConverterFactory {

  public static final DecimalConverterFactory INSTANCE = new DecimalConverterFactory();

  private int[] minBytesForPrecision = minBytesForPrecision();

  private DecimalConverterFactory() {

  }

  private int computeMinBytesForPrecision(int precision) {
    int numBytes = 1;
    while (Math.pow(2.0, 8 * numBytes - 1) < Math.pow(10.0, precision)) {
      numBytes += 1;
    }
    return numBytes;
  }

  private int[] minBytesForPrecision() {
    int[] data = new int[39];
    for (int i = 0; i < data.length; i++) {
      data[i] = computeMinBytesForPrecision(i);
    }
    return data;
  }

  public enum DecimalConverterType {
    DECIMAL_LV(-1), DECIMAL_INT(4), DECIMAL_LONG(8), DECIMAL_UNSCALED(-1);

    private int sizeInBytes;

    DecimalConverterType(int sizeInBytes) {
      this.sizeInBytes = sizeInBytes;
    }

    public int getSizeInBytes() {
      return sizeInBytes;
    }

  }

  public interface DecimalConverter {

    Object convert(BigDecimal decimal);

    BigDecimal getDecimal(Object valueToBeConverted);

    void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, BitSet nullBitset,
        DataType pageType);

    int getSize();

    DecimalConverterType getDecimalConverterType();

  }

  public static class DecimalIntConverter implements DecimalConverter {

    protected int scale;

    DecimalIntConverter(int scale) {
      this.scale = scale;
    }

    @Override
    public Object convert(BigDecimal decimal) {
      long longValue = decimal.unscaledValue().longValue();
      return (int) longValue;
    }

    @Override
    public BigDecimal getDecimal(Object valueToBeConverted) {
      return BigDecimal.valueOf((Long) valueToBeConverted, scale);
    }

    @Override
    public void fillVector(Object valuesToBeConverted, int size,
        ColumnVectorInfo vectorInfo, BitSet nullBitSet, DataType pageType) {
      if (!(valuesToBeConverted instanceof byte[])) {
        throw new UnsupportedOperationException("This object type " + valuesToBeConverted.getClass()
            + " is not supported in this method");
      }
      // TODO we need to find way to directly set to vector with out conversion. This way is very
      // inefficient.
      CarbonColumnVector vector = getCarbonColumnVector(vectorInfo, nullBitSet);
      int precision;
      int newMeasureScale;
      if (vectorInfo.measure == null) {
        // complex primitive decimal flow comes as dimension
        precision = ((DecimalType) vector.getType()).getPrecision();
        newMeasureScale = ((DecimalType) vector.getType()).getScale();
        size = ColumnVectorInfo.getUpdatedPageSizeForChildVector(vectorInfo, size);
      } else {
        precision = vectorInfo.measure.getMeasure().getPrecision();
        newMeasureScale = vectorInfo.measure.getMeasure().getScale();
      }
      int shortSizeInBytes = DataTypes.SHORT.getSizeInBytes();
      int intSizeInBytes = DataTypes.INT.getSizeInBytes();
      int longSizeInBytes = DataTypes.LONG.getSizeInBytes();
      int shortIntSizeInBytes = DataTypes.SHORT_INT.getSizeInBytes();
      int lengthStoredInBytes;
      if (vectorInfo.encodings != null && vectorInfo.encodings.size() > 0 && CarbonUtil
          .hasEncoding(vectorInfo.encodings, Encoding.INT_LENGTH_COMPLEX_CHILD_BYTE_ARRAY)) {
        lengthStoredInBytes = intSizeInBytes;
      } else {
        // before to carbon 2.0, complex child length is stored as SHORT
        // for string, varchar, binary, date, decimal types
        lengthStoredInBytes = shortSizeInBytes;
      }
      byte[] data = (byte[]) valuesToBeConverted;
      if (pageType == DataTypes.BYTE || pageType == DataTypes.SHORT
          || pageType == DataTypes.SHORT_INT || pageType == DataTypes.INT
          || pageType == DataTypes.LONG) {
        for (int i = 0; i < size; i++) {
          if (nullBitSet.get(i)) {
            vector.putNull(i);
          } else {
            BigDecimal value;
            if (pageType == DataTypes.BYTE) {
              value = BigDecimal.valueOf(data[i], scale);
            } else if (pageType == DataTypes.SHORT) {
              value = BigDecimal
                  .valueOf(ByteUtil.toShortLittleEndian(data, i * shortSizeInBytes), scale);
            } else if (pageType == DataTypes.SHORT_INT) {
              value =
                  BigDecimal.valueOf(ByteUtil.valueOf3Bytes(data, i * shortIntSizeInBytes), scale);
            } else if (pageType == DataTypes.INT) {
              value =
                  BigDecimal.valueOf(ByteUtil.toIntLittleEndian(data, i * intSizeInBytes), scale);
            } else {
              value =
                  BigDecimal.valueOf(ByteUtil.toLongLittleEndian(data, i * longSizeInBytes), scale);
            }
            if (value.scale() < newMeasureScale) {
              value = value.setScale(newMeasureScale);
            }
            vector.putDecimal(i, value, precision);
          }
        }
      } else if (pageType == DataTypes.BYTE_ARRAY) {
        // complex primitive decimal dimension
        int offset = 0;
        int length;
        for (int j = 0; j < size; j++) {
          // here decimal data will be Length[4 byte], scale[1 byte], value[Length byte]
          if (lengthStoredInBytes == intSizeInBytes) {
            length = ByteBuffer.wrap(data, offset, lengthStoredInBytes).getInt();
          } else {
            length = ByteBuffer.wrap(data, offset, lengthStoredInBytes).getShort();
          }
          offset += lengthStoredInBytes;
          if (length == 0) {
            vector.putNull(j);
            continue;
          }
          BigDecimal value = DataTypeUtil.byteToBigDecimal(data, offset, length);
          if (value.scale() < newMeasureScale) {
            value = value.setScale(newMeasureScale);
          }
          vector.putDecimal(j, value, precision);
          offset += length;
        }
      }
    }

    @Override
    public int getSize() {
      return 4;
    }

    @Override
    public DecimalConverterType getDecimalConverterType() {
      return DecimalConverterType.DECIMAL_INT;
    }
  }

  public static class DecimalLongConverter extends DecimalIntConverter {

    DecimalLongConverter(int scale) {
      super(scale);
    }

    @Override
    public Object convert(BigDecimal decimal) {
      return decimal.unscaledValue().longValue();
    }

    @Override
    public BigDecimal getDecimal(Object valueToBeConverted) {
      return BigDecimal.valueOf((Long) valueToBeConverted, scale);
    }

    @Override
    public int getSize() {
      return 8;
    }

    @Override
    public DecimalConverterType getDecimalConverterType() {
      return DecimalConverterType.DECIMAL_LONG;
    }
  }

  public class DecimalUnscaledConverter extends DecimalIntConverter implements DecimalConverter {

    private int scale;

    private int numBytes;

    private byte[] decimalBuffer = new byte[minBytesForPrecision[38]];

    DecimalUnscaledConverter(int precision, int scale) {
      super(scale);
      this.scale = scale;
      this.numBytes = minBytesForPrecision[precision];
    }

    @Override
    public Object convert(BigDecimal decimal) {
      byte[] bytes = decimal.unscaledValue().toByteArray();
      byte[] fixedLengthBytes = null;
      if (bytes.length == numBytes) {
        // If the length of the underlying byte array of the unscaled `BigInteger` happens to be
        // `numBytes`, just reuse it, so that we don't bother copying it to `decimalBuffer`.
        fixedLengthBytes = bytes;
      } else {
        // Otherwise, the length must be less than `numBytes`.  In this case we copy contents of
        // the underlying bytes with padding sign bytes to `decimalBuffer` to form the result
        // fixed-length byte array.
        byte signByte = 0;
        if (bytes[0] < 0) {
          signByte = (byte) -1;
        } else {
          signByte = (byte) 0;
        }
        Arrays.fill(decimalBuffer, 0, numBytes - bytes.length, signByte);
        System.arraycopy(bytes, 0, decimalBuffer, numBytes - bytes.length, bytes.length);
        fixedLengthBytes = decimalBuffer;
      }
      byte[] value = new byte[numBytes];
      System.arraycopy(fixedLengthBytes, 0, value, 0, numBytes);
      return value;
    }

    @Override
    public BigDecimal getDecimal(Object valueToBeConverted) {
      BigInteger bigInteger = new BigInteger((byte[]) valueToBeConverted);
      return new BigDecimal(bigInteger, scale);
    }

    @Override
    public void fillVector(Object valuesToBeConverted, int size,
        ColumnVectorInfo vectorInfo, BitSet nullBitSet, DataType pageType) {
      if (valuesToBeConverted instanceof byte[]) {
        super.fillVector(valuesToBeConverted, size, vectorInfo, nullBitSet, pageType);
        return;
      }
      CarbonColumnVector vector = getCarbonColumnVector(vectorInfo, nullBitSet);
      int precision = vectorInfo.measure.getMeasure().getPrecision();
      int newMeasureScale = vectorInfo.measure.getMeasure().getScale();
      if (scale < newMeasureScale) {
        scale = newMeasureScale;
      }
      if (valuesToBeConverted instanceof byte[][]) {
        byte[][] data = (byte[][]) valuesToBeConverted;
        for (int i = 0; i < size; i++) {
          if (nullBitSet.get(i)) {
            vector.putNull(i);
          } else {
            BigInteger bigInteger = new BigInteger(data[i]);
            BigDecimal value = new BigDecimal(bigInteger, scale);
            if (value.scale() < newMeasureScale) {
              value = value.setScale(newMeasureScale);
            }
            vector.putDecimal(i, value, precision);
          }
        }
      }
    }

    @Override
    public int getSize() {
      return numBytes;
    }

    @Override
    public DecimalConverterType getDecimalConverterType() {
      return DecimalConverterType.DECIMAL_UNSCALED;
    }
  }

  public static class LVBytesDecimalConverter implements DecimalConverter {

    public static LVBytesDecimalConverter INSTANCE = new LVBytesDecimalConverter();

    @Override
    public Object convert(BigDecimal decimal) {
      return DataTypeUtil.bigDecimalToByte(decimal);
    }

    @Override
    public BigDecimal getDecimal(Object valueToBeConverted) {
      return DataTypeUtil.byteToBigDecimal((byte[]) valueToBeConverted);
    }

    @Override
    public void fillVector(Object valuesToBeConverted, int size,
        ColumnVectorInfo vectorInfo, BitSet nullBitSet, DataType pageType) {
      CarbonColumnVector vector = getCarbonColumnVector(vectorInfo, nullBitSet);
      //TODO handle complex child
      int precision = vectorInfo.measure.getMeasure().getPrecision();
      int newMeasureScale = vectorInfo.measure.getMeasure().getScale();
      if (valuesToBeConverted instanceof byte[][]) {
        byte[][] data = (byte[][]) valuesToBeConverted;
        for (int i = 0; i < size; i++) {
          if (nullBitSet.get(i)) {
            vector.putNull(i);
          } else {
            BigDecimal value = DataTypeUtil.byteToBigDecimal(data[i]);
            if (value.scale() < newMeasureScale) {
              value = value.setScale(newMeasureScale);
            }
            vector.putDecimal(i, value, precision);
          }
        }
      }
    }

    @Override
    public int getSize() {
      return -1;
    }

    @Override
    public DecimalConverterType getDecimalConverterType() {
      return DecimalConverterType.DECIMAL_LV;
    }
  }

  private static CarbonColumnVector getCarbonColumnVector(ColumnVectorInfo vectorInfo,
      BitSet nullBitSet) {
    CarbonColumnVector vector = vectorInfo.vector;
    BitSet deletedRows = vectorInfo.deletedRows;
    vector = ColumnarVectorWrapperDirectFactory
        .getDirectVectorWrapperFactory(vectorInfo, vector, vectorInfo.invertedIndex, nullBitSet,
            deletedRows, true, false);
    return vector;
  }

  public DecimalConverter getDecimalConverter(int precision, int scale) {
    if (precision < 0) {
      return new LVBytesDecimalConverter();
    } else if (precision <= 9) {
      return new DecimalIntConverter(scale);
    } else if (precision <= 18) {
      return new DecimalLongConverter(scale);
    } else {
      return new DecimalUnscaledConverter(precision, scale);
    }
  }

}