/*
 * 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.iotdb.tsfile.encoding.decoder;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;

import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.encoding.common.EndianType;
import org.apache.iotdb.tsfile.exception.encoding.TsFileDecodingException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;

/**
 * Abstract class for all rle decoder. Decoding values according to following
 * grammar: {@code
 * <length> <bitwidth> <encoded-data>}. For more information about rle format,
 * see RleEncoder
 */
public abstract class RleDecoder extends Decoder {

  private EndianType endianType;

  public EndianType getEndianType() {
    return endianType;
  }

  public void setEndianType(EndianType endianType) {
    this.endianType = endianType;
  }

  protected TSFileConfig config = TSFileDescriptor.getInstance().getConfig();
  /**
   * mode to indicate current encoding type 0 - RLE 1 - BIT_PACKED.
   */
  protected Mode mode;
  /**
   * bit width for bit-packing and rle to decode.
   */
  protected int bitWidth;
  /**
   * number of data left for reading in current buffer.
   */
  protected int currentCount;
  /**
   * how many bytes for all encoded data like [{@code <bitwidth> <encoded-data>}]
   * in inputstream.
   */
  protected int length;
  /**
   * a flag to indicate whether current pattern is end. false - need to start
   * reading a new page true - current page isn't over.
   */
  protected boolean isLengthAndBitWidthReaded;
  /**
   * buffer to save data format like [{@code <bitwidth> <encoded-data>}] for
   * decoder.
   */
  protected ByteBuffer byteCache;
  /**
   * number of bit-packing group in which is saved in header.
   */
  protected int bitPackingNum;

  /**
   * a constructor, init with endianType, default encoding is
   * <code>TSEncoding.RLE</code>.
   */
  public RleDecoder(EndianType endianType) {
    super(TSEncoding.RLE);
    this.endianType = endianType;
    reset();
  }

  @Override
  public void reset() {
    currentCount = 0;
    isLengthAndBitWidthReaded = false;
    bitPackingNum = 0;
    byteCache = ByteBuffer.allocate(0);
  }

  /**
   * get header for both rle and bit-packing current encode mode which is saved in
   * first bit of header.
   *
   * @return int value
   * @throws IOException cannot get header
   */
  public int getHeader() throws IOException {
    int header = ReadWriteForEncodingUtils.readUnsignedVarInt(byteCache);
    mode = (header & 1) == 0 ? Mode.RLE : Mode.BIT_PACKED;
    return header;
  }

  /**
   * get all encoded data according to mode.
   *
   * @throws IOException cannot read next value
   */
  protected void readNext() throws IOException {
    int header = getHeader();
    switch (mode) {
    case RLE:
      currentCount = header >> 1;
      readNumberInRle();
      break;
    case BIT_PACKED:
      callReadBitPackingBuffer(header);
      break;
    default:
      throw new TsFileDecodingException(String.format("tsfile-encoding IntRleDecoder: unknown encoding mode %s", mode));
    }
  }

  protected void callReadBitPackingBuffer(int header) throws IOException {
    int bitPackedGroupCount = header >> 1;
    // in last bit-packing group, there may be some useless value,
    // lastBitPackedNum indicates how many values is useful
    int lastBitPackedNum = ReadWriteIOUtils.read(byteCache);
    if (bitPackedGroupCount > 0) {

      currentCount = (bitPackedGroupCount - 1) * TSFileConfig.RLE_MIN_REPEATED_NUM + lastBitPackedNum;
      bitPackingNum = currentCount;
    } else {
      throw new TsFileDecodingException(
          String.format("tsfile-encoding IntRleDecoder: bitPackedGroupCount %d, smaller than 1", bitPackedGroupCount));
    }
    readBitPackingBuffer(bitPackedGroupCount, lastBitPackedNum);
  }

  /**
   * read length and bit width of current package before we decode number.
   *
   * @param buffer ByteBuffer
   */
  protected void readLengthAndBitWidth(ByteBuffer buffer) {
    length = ReadWriteForEncodingUtils.readUnsignedVarInt(buffer);
    byte[] tmp = new byte[length];
    buffer.get(tmp, 0, length);
    byteCache = ByteBuffer.wrap(tmp);
    isLengthAndBitWidthReaded = true;
    bitWidth = ReadWriteIOUtils.read(byteCache);
    initPacker();
  }

  /**
   * Check whether there is number left for reading.
   *
   * @param buffer decoded data saved in ByteBuffer
   * @return true or false to indicate whether there is number left
   * @throws IOException cannot check next value
   */
  @Override
  public boolean hasNext(ByteBuffer buffer) throws IOException {
    if (currentCount > 0 || buffer.remaining() > 0 || hasNextPackage()) {
      return true;
    }
    return false;
  }

  /**
   * Check whether there is another pattern left for reading.
   *
   * @return true or false to indicate whether there is another pattern left
   */
  protected boolean hasNextPackage() {
    return currentCount > 0 || byteCache.remaining() > 0;
  }

  protected abstract void initPacker();

  /**
   * Read rle package and save them in buffer.
   *
   * @throws IOException cannot read number
   */
  protected abstract void readNumberInRle() throws IOException;

  /**
   * Read bit-packing package and save them in buffer.
   *
   * @param bitPackedGroupCount number of group number
   * @param lastBitPackedNum    number of useful value in last group
   * @throws IOException cannot read bit pack
   */
  protected abstract void readBitPackingBuffer(int bitPackedGroupCount, int lastBitPackedNum) throws IOException;

  @Override
  public boolean readBoolean(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readBoolean is not supproted by RleDecoder");
  }

  @Override
  public short readShort(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readShort is not supproted by RleDecoder");
  }

  @Override
  public int readInt(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readInt is not supproted by RleDecoder");
  }

  @Override
  public long readLong(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readLong is not supproted by RleDecoder");
  }

  @Override
  public float readFloat(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readFloat is not supproted by RleDecoder");
  }

  @Override
  public double readDouble(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readDouble is not supproted by RleDecoder");
  }

  @Override
  public Binary readBinary(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readBinary is not supproted by RleDecoder");
  }

  @Override
  public BigDecimal readBigDecimal(ByteBuffer buffer) {
    throw new TsFileDecodingException("Method readBigDecimal is not supproted by RleDecoder");
  }

  protected enum Mode {
    RLE, BIT_PACKED
  }
}
