/**
 * 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.
 */

#include "ColumnDecoder.h"

#include "Column.h"

std::vector<bool> deserializeNullIndicators(MyStringBuffer &buffer,
                                            int32_t positionCount) {
  uint8_t mayHaveNullByte = buffer.getChar();

  bool mayHaveNull = mayHaveNullByte != 0;
  if (!mayHaveNull) {
    return {};
  }

  return deserializeBooleanArray(buffer, positionCount);
}

std::vector<bool> deserializeBooleanArray(MyStringBuffer &buffer,
                                          int32_t size) {
  const int32_t packedSize = (size + 7) / 8;
  std::vector<uint8_t> packedBytes(packedSize);
  for (int i = 0; i < packedSize; i++) {
    packedBytes[i] = buffer.getChar();
  }

  std::vector<bool> output(size);
  int currentByte = 0;
  const int fullGroups = size & ~0b111;

  for (int pos = 0; pos < fullGroups; pos += 8) {
    const uint8_t b = packedBytes[currentByte++];
    output[pos + 0] = (b & 0b10000000) != 0;
    output[pos + 1] = (b & 0b01000000) != 0;
    output[pos + 2] = (b & 0b00100000) != 0;
    output[pos + 3] = (b & 0b00010000) != 0;
    output[pos + 4] = (b & 0b00001000) != 0;
    output[pos + 5] = (b & 0b00000100) != 0;
    output[pos + 6] = (b & 0b00000010) != 0;
    output[pos + 7] = (b & 0b00000001) != 0;
  }

  if ((size & 0b111) > 0) {
    const uint8_t b = packedBytes.back();
    uint8_t mask = 0b10000000;

    for (int pos = fullGroups; pos < size; pos++) {
      output[pos] = (b & mask) != 0;
      mask >>= 1;
    }
  }

  return output;
}

std::unique_ptr<Column>
BaseColumnDecoder::readColumn(MyStringBuffer &buffer,
                              TSDataType::TSDataType dataType,
                              int32_t positionCount) {
  return nullptr;
}

std::unique_ptr<Column>
Int32ArrayColumnDecoder::readColumn(MyStringBuffer &buffer,
                                    TSDataType::TSDataType dataType,
                                    int32_t positionCount) {
  auto nullIndicators = deserializeNullIndicators(buffer, positionCount);

  switch (dataType) {
  case TSDataType::INT32:
  case TSDataType::DATE: {
    std::vector<int32_t> intValues(positionCount);
    for (int32_t i = 0; i < positionCount; i++) {
      if (!nullIndicators.empty() && nullIndicators[i])
        continue;
      intValues[i] = buffer.getInt();
    }
    return std::unique_ptr<IntColumn>(
        new IntColumn(0, positionCount, nullIndicators, intValues));
  }
  case TSDataType::FLOAT: {
    std::vector<float> floatValues(positionCount);
    for (int32_t i = 0; i < positionCount; i++) {
      if (!nullIndicators.empty() && nullIndicators[i])
        continue;
      floatValues[i] = buffer.getFloat();
    }
    return std::unique_ptr<FloatColumn>(
        new FloatColumn(0, positionCount, nullIndicators, floatValues));
  }
  default:
    throw IoTDBException("Invalid data type for Int32ArrayColumnDecoder");
  }
}

std::unique_ptr<Column>
Int64ArrayColumnDecoder::readColumn(MyStringBuffer &buffer,
                                    TSDataType::TSDataType dataType,
                                    int32_t positionCount) {
  auto nullIndicators = deserializeNullIndicators(buffer, positionCount);

  switch (dataType) {
  case TSDataType::INT64:
  case TSDataType::TIMESTAMP: {
    std::vector<int64_t> values(positionCount);
    for (int32_t i = 0; i < positionCount; i++) {
      if (!nullIndicators.empty() && nullIndicators[i])
        continue;
      values[i] = buffer.getInt64();
    }
    return std::unique_ptr<LongColumn>(
        new LongColumn(0, positionCount, nullIndicators, values));
  }
  case TSDataType::DOUBLE: {
    std::vector<double> values(positionCount);
    for (int32_t i = 0; i < positionCount; i++) {
      if (!nullIndicators.empty() && nullIndicators[i])
        continue;
      values[i] = buffer.getDouble();
    }
    return std::unique_ptr<DoubleColumn>(
        new DoubleColumn(0, positionCount, nullIndicators, values));
  }
  default:
    throw IoTDBException("Invalid data type for Int64ArrayColumnDecoder");
  }
}

std::unique_ptr<Column>
ByteArrayColumnDecoder::readColumn(MyStringBuffer &buffer,
                                   TSDataType::TSDataType dataType,
                                   int32_t positionCount) {
  if (dataType != TSDataType::BOOLEAN) {
    throw IoTDBException("Invalid data type for ByteArrayColumnDecoder");
  }

  auto nullIndicators = deserializeNullIndicators(buffer, positionCount);
  auto values = deserializeBooleanArray(buffer, positionCount);
  return std::unique_ptr<BooleanColumn>(
      new BooleanColumn(0, positionCount, nullIndicators, values));
}

std::unique_ptr<Column>
BinaryArrayColumnDecoder::readColumn(MyStringBuffer &buffer,
                                     TSDataType::TSDataType dataType,
                                     int32_t positionCount) {
  if (dataType != TSDataType::TEXT) {
    throw IoTDBException("Invalid data type for BinaryArrayColumnDecoder");
  }

  auto nullIndicators = deserializeNullIndicators(buffer, positionCount);
  std::vector<std::shared_ptr<Binary>> values(positionCount);

  for (int32_t i = 0; i < positionCount; i++) {
    if (!nullIndicators.empty() && nullIndicators[i])
      continue;

    int32_t length = buffer.getInt();
    if (length < 0) {
      throw IoTDBException("BinaryArrayColumnDecoder: negative TEXT length");
    }

    std::vector<uint8_t> value(length);
    for (int32_t j = 0; j < length; j++) {
      value[j] = buffer.getChar();
    }

    values[i] = std::make_shared<Binary>(value);
  }

  return std::unique_ptr<BinaryColumn>(
      new BinaryColumn(0, positionCount, nullIndicators, values));
}

std::unique_ptr<Column>
RunLengthColumnDecoder::readColumn(MyStringBuffer &buffer,
                                   TSDataType::TSDataType dataType,
                                   int32_t positionCount) {
  uint8_t encodingByte = buffer.getChar();

  auto columnEncoding = static_cast<ColumnEncoding>(encodingByte);
  auto decoder = getColumnDecoder(columnEncoding);

  auto column = decoder->readColumn(buffer, dataType, 1);
  if (!column) {
    throw IoTDBException("Failed to read inner column");
  }
  return std::unique_ptr<RunLengthEncodedColumn>(
      new RunLengthEncodedColumn(move(column), positionCount));
}