/*
 * 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.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;

import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.page.statistics.SimpleStatsResult;
import org.apache.carbondata.core.metadata.ValueEncoderMeta;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.datatype.DecimalType;
import org.apache.carbondata.core.metadata.schema.table.Writable;
import org.apache.carbondata.core.util.DataTypeUtil;

/**
 * It holds metadata for one column page
 */
public class ColumnPageEncoderMeta extends ValueEncoderMeta implements Writable {

  private static final long serialVersionUID = 1905162071950251407L;

  // column spec of this column
  private transient TableSpec.ColumnSpec columnSpec;

  // storage data type of this column, it could be different from data type in the column spec
  private DataType storeDataType;

  // compressor name for compressing and decompressing this column.
  // Make it protected for RLEEncoderMeta
  protected String compressorName;

  // Whether the flow should go to fill complete vector while decoding the page.
  private transient boolean fillCompleteVector;

  public ColumnPageEncoderMeta() {
  }

  public ColumnPageEncoderMeta(TableSpec.ColumnSpec columnSpec, DataType storeDataType,
      String compressorName) {
    if (columnSpec == null) {
      throw new IllegalArgumentException("column spec must not be null");
    }
    if (storeDataType == null) {
      throw new IllegalArgumentException("store data type must not be null");
    }
    if (compressorName == null) {
      throw new IllegalArgumentException("compressor must not be null");
    }
    this.columnSpec = columnSpec;
    this.storeDataType = storeDataType;
    this.compressorName = compressorName;
    setType(DataType.convertType(storeDataType));
  }

  public ColumnPageEncoderMeta(TableSpec.ColumnSpec columnSpec, DataType storeDataType,
      SimpleStatsResult stats, String compressorName) {
    this(columnSpec, storeDataType, compressorName);
    if (stats != null) {
      setDecimal(stats.getDecimalCount());
      setMaxValue(stats.getMax());
      setMinValue(stats.getMin());
    }
  }

  public DataType getStoreDataType() {
    return storeDataType;
  }

  @Override
  public void write(DataOutput out) throws IOException {
    columnSpec.write(out);
    out.writeByte(storeDataType.getId());
    out.writeInt(getDecimal());
    out.writeByte(getDataTypeSelected());
    writeMinMax(out);
    out.writeUTF(compressorName);
  }

  @Override
  public void readFields(DataInput in) throws IOException {
    columnSpec = new TableSpec.ColumnSpec();
    columnSpec.readFields(in);
    storeDataType = DataTypes.valueOf(in.readByte());
    if (DataTypes.isDecimal(storeDataType)) {
      DecimalType decimalType = (DecimalType) storeDataType;
      decimalType.setPrecision(columnSpec.getPrecision());
      decimalType.setScale(columnSpec.getScale());
    }

    setDecimal(in.readInt());
    setDataTypeSelected(in.readByte());
    readMinMax(in);
    compressorName = in.readUTF();
  }

  private void writeMinMax(DataOutput out) throws IOException {
    DataType dataType = columnSpec.getSchemaDataType();
    if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
      out.writeByte((byte) getMaxValue());
      out.writeByte((byte) getMinValue());
      out.writeLong(0L); // unique value is obsoleted, maintain for compatibility
    } else if (dataType == DataTypes.SHORT) {
      out.writeShort((short) getMaxValue());
      out.writeShort((short) getMinValue());
      out.writeLong(0L); // unique value is obsoleted, maintain for compatibility
    } else if (dataType == DataTypes.INT) {
      out.writeInt((int) getMaxValue());
      out.writeInt((int) getMinValue());
      out.writeLong(0L); // unique value is obsoleted, maintain for compatibility
    } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
      out.writeLong((Long) getMaxValue());
      out.writeLong((Long) getMinValue());
      out.writeLong(0L); // unique value is obsoleted, maintain for compatibility
    } else if (dataType == DataTypes.DOUBLE) {
      out.writeDouble((Double) getMaxValue());
      out.writeDouble((Double) getMinValue());
      out.writeDouble(0d); // unique value is obsoleted, maintain for compatibility
    } else if (dataType == DataTypes.FLOAT) {
      out.writeFloat((Float) getMaxValue());
      out.writeFloat((Float) getMinValue());
      out.writeFloat(0f); // unique value is obsoleted, maintain for compatibility
    } else if (DataTypes.isDecimal(dataType)) {
      byte[] maxAsBytes = getMaxAsBytes(columnSpec.getSchemaDataType());
      byte[] minAsBytes = getMinAsBytes(columnSpec.getSchemaDataType());
      byte[] unique = DataTypeUtil.bigDecimalToByte(BigDecimal.ZERO);
      out.writeShort((short) maxAsBytes.length);
      out.write(maxAsBytes);
      out.writeShort((short) minAsBytes.length);
      out.write(minAsBytes);
      // unique value is obsoleted, maintain for compatibility
      out.writeShort((short) unique.length);
      out.write(unique);
      if (DataTypes.isDecimal(dataType)) {
        DecimalType decimalType = (DecimalType) dataType;
        out.writeInt(decimalType.getScale());
        out.writeInt(decimalType.getPrecision());
      } else {
        out.writeInt(-1);
        out.writeInt(-1);
      }
    } else if (dataType == DataTypes.BYTE_ARRAY || dataType == DataTypes.BINARY) {
      // for complex type, it will come here, ignoring stats for complex type
      // TODO: support stats for complex type
    } else {
      throw new IllegalArgumentException("invalid data type: " + storeDataType);
    }
  }

  private void readMinMax(DataInput in) throws IOException {
    DataType dataType = columnSpec.getSchemaDataType();
    if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
      this.setMaxValue(in.readByte());
      this.setMinValue(in.readByte());
      in.readLong();  // for non exist value which is obsoleted, it is backward compatibility;
    } else if (dataType == DataTypes.SHORT) {
      this.setMaxValue(in.readShort());
      this.setMinValue(in.readShort());
      in.readLong();  // for non exist value which is obsoleted, it is backward compatibility;
    } else if (dataType == DataTypes.INT) {
      this.setMaxValue(in.readInt());
      this.setMinValue(in.readInt());
      in.readLong();  // for non exist value which is obsoleted, it is backward compatibility;
    } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
      this.setMaxValue(in.readLong());
      this.setMinValue(in.readLong());
      in.readLong();  // for non exist value which is obsoleted, it is backward compatibility;
    } else if (dataType == DataTypes.DOUBLE) {
      this.setMaxValue(in.readDouble());
      this.setMinValue(in.readDouble());
      in.readDouble(); // for non exist value which is obsoleted, it is backward compatibility;
    } else if (dataType == DataTypes.FLOAT) {
      this.setMaxValue(in.readFloat());
      this.setMinValue(in.readFloat());
      in.readFloat(); // for non exist value which is obsoleted, it is backward compatibility;
    } else if (DataTypes.isDecimal(dataType)) {
      byte[] max = new byte[in.readShort()];
      in.readFully(max);
      this.setMaxValue(DataTypeUtil.byteToBigDecimal(max));
      byte[] min = new byte[in.readShort()];
      in.readFully(min);
      this.setMinValue(DataTypeUtil.byteToBigDecimal(min));
      // unique value is obsoleted, maintain for compatibility
      short uniqueLength = in.readShort();
      in.readFully(new byte[uniqueLength]);
      // scale field is obsoleted. It is stored in the schema data type in columnSpec
      in.readInt();
      // precision field is obsoleted. It is stored in the schema data type in columnSpec
      in.readInt();
    } else if (dataType == DataTypes.BYTE_ARRAY || dataType == DataTypes.BINARY) {
      // for complex type, it will come here, ignoring stats for complex type
      // TODO: support stats for complex type
    } else {
      throw new IllegalArgumentException("invalid data type: " + storeDataType);
    }
  }

  private byte[] getMaxAsBytes(DataType dataType) {
    return getValueAsBytes(getMaxValue(), dataType);
  }

  private byte[] getMinAsBytes(DataType dataType) {
    return getValueAsBytes(getMinValue(), dataType);
  }

  /**
   * convert value to byte array
   */
  private byte[] getValueAsBytes(Object value, DataType dataType) {
    ByteBuffer b;
    if (dataType == DataTypes.BYTE_ARRAY) {
      b = ByteBuffer.allocate(8);
      b.putLong((byte) value);
      b.flip();
      return b.array();
    } else if (dataType == DataTypes.SHORT) {
      b = ByteBuffer.allocate(8);
      b.putLong((short) value);
      b.flip();
      return b.array();
    } else if (dataType == DataTypes.INT) {
      b = ByteBuffer.allocate(8);
      b.putLong((int) value);
      b.flip();
      return b.array();
    } else if (dataType == DataTypes.LONG) {
      b = ByteBuffer.allocate(8);
      b.putLong((long) value);
      b.flip();
      return b.array();
    } else if (dataType == DataTypes.DOUBLE) {
      b = ByteBuffer.allocate(8);
      b.putDouble((double) value);
      b.flip();
      return b.array();
    } else if (DataTypes.isDecimal(dataType)) {
      return DataTypeUtil.bigDecimalToByte((BigDecimal) value);
    } else if (dataType == DataTypes.STRING || dataType == DataTypes.TIMESTAMP
        || dataType == DataTypes.DATE) {
      return (byte[]) value;
    } else {
      throw new IllegalArgumentException("Invalid data type: " + storeDataType);
    }
  }

  public int getScale() {
    if (DataTypes.isDecimal(columnSpec.getSchemaDataType())) {
      return columnSpec.getScale();
    }
    throw new UnsupportedOperationException();
  }

  public int getPrecision() {
    if (DataTypes.isDecimal(columnSpec.getSchemaDataType())) {
      return columnSpec.getPrecision();
    }
    throw new UnsupportedOperationException();
  }

  public TableSpec.ColumnSpec getColumnSpec() {
    return columnSpec;
  }

  public String getCompressorName() {
    return compressorName;
  }

  public DataType getSchemaDataType() {
    return columnSpec.getSchemaDataType();
  }

  public boolean isFillCompleteVector() {
    return fillCompleteVector;
  }

  public void setFillCompleteVector(boolean fillCompleteVector) {
    this.fillCompleteVector = fillCompleteVector;
  }
}