/**
 * @file   dd_compressor.cc
 *
 * @section LICENSE
 *
 * The MIT License
 *
 * @copyright Copyright (c) 2017-2020 TileDB, Inc.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 *
 * @section DESCRIPTION
 *
 * This file implements the double delta compressor class.
 */

#include "tiledb/sm/compressors/dd_compressor.h"
#include "tiledb/common/logger.h"
#include "tiledb/sm/buffer/buffer.h"
#include "tiledb/sm/buffer/const_buffer.h"
#include "tiledb/sm/buffer/preallocated_buffer.h"
#include "tiledb/sm/enums/datatype.h"

/* ****************************** */
/*             MACROS             */
/* ****************************** */

#define ABS(x) ((x) < 0) ? uint64_t(-(x)) : uint64_t(x)

using namespace tiledb::common;

namespace tiledb {
namespace sm {

const uint64_t DoubleDelta::OVERHEAD = 17;

/* ****************************** */
/*               API              */
/* ****************************** */

Status DoubleDelta::compress(
    Datatype type, ConstBuffer* input_buffer, Buffer* output_buffer) {
  switch (type) {
    case Datatype::INT8:
      return DoubleDelta::compress<int8_t>(input_buffer, output_buffer);
    case Datatype::UINT8:
      return DoubleDelta::compress<uint8_t>(input_buffer, output_buffer);
    case Datatype::INT16:
      return DoubleDelta::compress<int16_t>(input_buffer, output_buffer);
    case Datatype::UINT16:
      return DoubleDelta::compress<uint16_t>(input_buffer, output_buffer);
    case Datatype::INT32:
      return DoubleDelta::compress<int>(input_buffer, output_buffer);
    case Datatype::UINT32:
      return DoubleDelta::compress<uint32_t>(input_buffer, output_buffer);
    case Datatype::INT64:
      return DoubleDelta::compress<int64_t>(input_buffer, output_buffer);
    case Datatype::UINT64:
      return DoubleDelta::compress<uint64_t>(input_buffer, output_buffer);
    case Datatype::CHAR:
      return DoubleDelta::compress<char>(input_buffer, output_buffer);
    case Datatype::DATETIME_YEAR:
    case Datatype::DATETIME_MONTH:
    case Datatype::DATETIME_WEEK:
    case Datatype::DATETIME_DAY:
    case Datatype::DATETIME_HR:
    case Datatype::DATETIME_MIN:
    case Datatype::DATETIME_SEC:
    case Datatype::DATETIME_MS:
    case Datatype::DATETIME_US:
    case Datatype::DATETIME_NS:
    case Datatype::DATETIME_PS:
    case Datatype::DATETIME_FS:
    case Datatype::DATETIME_AS:
      return DoubleDelta::compress<int64_t>(input_buffer, output_buffer);
    case Datatype::STRING_ASCII:
    case Datatype::STRING_UTF8:
    case Datatype::STRING_UTF16:
    case Datatype::STRING_UTF32:
    case Datatype::STRING_UCS2:
    case Datatype::STRING_UCS4:
    case Datatype::ANY:
      return DoubleDelta::compress<uint8_t>(input_buffer, output_buffer);
    case Datatype::FLOAT32:
    case Datatype::FLOAT64:
      return LOG_STATUS(Status::CompressionError(
          "Cannot compress tile with DoubleDelta; Float "
          "datatypes are not supported"));
  }

  assert(false);
  return LOG_STATUS(Status::CompressionError(
      "Cannot compress tile with DoubleDelta; Not supported datatype"));
}

Status DoubleDelta::decompress(
    Datatype type,
    ConstBuffer* input_buffer,
    PreallocatedBuffer* output_buffer) {
  switch (type) {
    case Datatype::INT8:
      return DoubleDelta::decompress<int8_t>(input_buffer, output_buffer);
    case Datatype::UINT8:
      return DoubleDelta::decompress<uint8_t>(input_buffer, output_buffer);
    case Datatype::INT16:
      return DoubleDelta::decompress<int16_t>(input_buffer, output_buffer);
    case Datatype::UINT16:
      return DoubleDelta::decompress<uint16_t>(input_buffer, output_buffer);
    case Datatype::INT32:
      return DoubleDelta::decompress<int>(input_buffer, output_buffer);
    case Datatype::UINT32:
      return DoubleDelta::decompress<uint32_t>(input_buffer, output_buffer);
    case Datatype::INT64:
      return DoubleDelta::decompress<int64_t>(input_buffer, output_buffer);
    case Datatype::UINT64:
      return DoubleDelta::decompress<uint64_t>(input_buffer, output_buffer);
    case Datatype::CHAR:
      return DoubleDelta::decompress<char>(input_buffer, output_buffer);
    case Datatype::DATETIME_YEAR:
    case Datatype::DATETIME_MONTH:
    case Datatype::DATETIME_WEEK:
    case Datatype::DATETIME_DAY:
    case Datatype::DATETIME_HR:
    case Datatype::DATETIME_MIN:
    case Datatype::DATETIME_SEC:
    case Datatype::DATETIME_MS:
    case Datatype::DATETIME_US:
    case Datatype::DATETIME_NS:
    case Datatype::DATETIME_PS:
    case Datatype::DATETIME_FS:
    case Datatype::DATETIME_AS:
      return DoubleDelta::decompress<int64_t>(input_buffer, output_buffer);
    case Datatype::STRING_ASCII:
    case Datatype::STRING_UTF8:
    case Datatype::STRING_UTF16:
    case Datatype::STRING_UTF32:
    case Datatype::STRING_UCS2:
    case Datatype::STRING_UCS4:
    case Datatype::ANY:
      return DoubleDelta::decompress<uint8_t>(input_buffer, output_buffer);
    case Datatype::FLOAT32:
    case Datatype::FLOAT64:
      return LOG_STATUS(Status::CompressionError(
          "Cannot decompress tile with DoubleDelta; Float "
          "datatypes are not supported"));
  }

  assert(false);
  return LOG_STATUS(Status::CompressionError(
      "Cannot decompress tile with DoubleDelta; Not supported datatype"));
}

uint64_t DoubleDelta::overhead(uint64_t nbytes) {
  // DoubleDelta has a fixed size overhead
  (void)nbytes;
  return DoubleDelta::OVERHEAD;
}

/* ****************************** */
/*         PRIVATE METHODS        */
/* ****************************** */

template <class T>
Status DoubleDelta::compress(ConstBuffer* input_buffer, Buffer* output_buffer) {
  // Calculate number of values and handle trivial case
  uint64_t value_size = sizeof(T);
  uint64_t num = input_buffer->size() / value_size;
  assert(num > 0 && (input_buffer->size() % value_size == 0));

  // Calculate bitsize (ignoring the sign bit)
  auto in = (T*)input_buffer->data();
  unsigned int bitsize;
  RETURN_NOT_OK(compute_bitsize(in, num, &bitsize));
  assert(bitsize <= std::numeric_limits<uint8_t>::max());
  auto bitsize_c = static_cast<uint8_t>(bitsize);

  // Write bitsize and number of values
  RETURN_NOT_OK(output_buffer->write(&bitsize_c, sizeof(uint8_t)));
  RETURN_NOT_OK(output_buffer->write(&num, sizeof(uint64_t)));

  // Trivial case - no compression
  if (bitsize >= sizeof(T) * 8 - 1) {
    RETURN_NOT_OK(output_buffer->write(in, input_buffer->size()));
    return Status::Ok();
  }

  // Write first value
  RETURN_NOT_OK(output_buffer->write(&in[0], value_size));
  if (num == 1)
    return Status::Ok();

  // Write second value
  RETURN_NOT_OK(output_buffer->write(&in[1], value_size));
  if (num == 2)
    return Status::Ok();

  // Write double deltas
  int64_t prev_delta = int64_t(in[1]) - int64_t(in[0]);
  int bit_in_chunk = 63;  // Leftmost bit (MSB)
  uint64_t chunk = 0;
  for (uint64_t i = 2; i < num; ++i) {
    int64_t cur_delta = int64_t(in[i]) - int64_t(in[i - 1]);
    int64_t dd = cur_delta - prev_delta;
    RETURN_NOT_OK(
        write_double_delta(output_buffer, dd, bitsize, &chunk, &bit_in_chunk));
    prev_delta = cur_delta;
  }

  // Write whatever is left in the chunk
  if (bit_in_chunk < 63)
    RETURN_NOT_OK(output_buffer->write(&chunk, sizeof(uint64_t)));

  return Status::Ok();
}

template <class T>
Status DoubleDelta::compute_bitsize(
    T* in, uint64_t num, unsigned int* bitsize) {
  // Find maximum absolute double delta
  *bitsize = 0;
  int64_t max = 0;
  if (num <= 2) {
    *bitsize = 0;
    return Status::Ok();
  }
  int64_t prev_delta = int64_t(in[1]) - int64_t(in[0]);
  char delta_out_of_bounds = 0;
  for (uint64_t i = 2; i < num; ++i) {
    int64_t cur_delta = int64_t(in[i]) - int64_t(in[i - 1]);
    int64_t dd = cur_delta - prev_delta;
    delta_out_of_bounds |= (char)(cur_delta < 0 && prev_delta > 0 && dd > 0);
    delta_out_of_bounds |= (char)(cur_delta > 0 && prev_delta < 0 && dd < 0);
    max = std::max(std::abs(dd), max);
    prev_delta = cur_delta;
  }
  // Handle error
  if (delta_out_of_bounds) {
    return LOG_STATUS(
        Status::CompressionError("Cannot compress with DoubleDelta; Some "
                                 "negative double delta is out of bounds"));
  }
  // Calculate bitsize of the maximum absolute double delta
  do {
    ++(*bitsize);
    max >>= 1;
  } while (max);
  return Status::Ok();
}

template <class T>
Status DoubleDelta::decompress(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer) {
  // Read bitsize and number of values
  uint8_t bitsize_c = 0;
  uint64_t num = 0;
  uint64_t value_size = sizeof(T);
  RETURN_NOT_OK(input_buffer->read(&bitsize_c, sizeof(uint8_t)));
  RETURN_NOT_OK(input_buffer->read(&num, sizeof(uint64_t)));
  auto bitsize = static_cast<unsigned int>(bitsize_c);
  auto out = (T*)output_buffer->cur_data();

  // Trivial case - no compression
  if (bitsize >= sizeof(T) * 8 - 1) {
    RETURN_NOT_OK(output_buffer->write(
        input_buffer->cur_data(), input_buffer->nbytes_left_to_read()));
    return Status::Ok();
  }

  // Read first value
  T value;
  RETURN_NOT_OK(input_buffer->read(&value, value_size));
  RETURN_NOT_OK(output_buffer->write(&value, value_size));
  if (num == 1)
    return Status::Ok();

  // Read second value
  RETURN_NOT_OK(input_buffer->read(&value, value_size));
  RETURN_NOT_OK(output_buffer->write(&value, value_size));
  if (num == 2)
    return Status::Ok();

  // Read first chunk
  uint64_t chunk;
  RETURN_NOT_OK(input_buffer->read(&chunk, sizeof(uint64_t)));
  int bit_in_chunk = 63;  // Leftmost bit (MSB)

  // Decompress rest of the values
  int64_t dd;
  for (uint64_t i = 2; i < num; ++i) {
    RETURN_NOT_OK(
        read_double_delta(input_buffer, &dd, bitsize, &chunk, &bit_in_chunk));
    value = (T)(dd + 2 * (int64_t)out[i - 1] - (int64_t)out[i - 2]);
    RETURN_NOT_OK(output_buffer->write(&value, value_size));
  }

  return Status::Ok();
}

Status DoubleDelta::read_double_delta(
    ConstBuffer* buff,
    int64_t* double_delta,
    int bitsize,
    uint64_t* chunk,
    int* bit_in_chunk) {
  // Read sign
  uint64_t to_and = 1;
  to_and <<= (*bit_in_chunk);
  int sign = (((*chunk) & to_and) == 0) ? 1 : -1;
  --(*bit_in_chunk);

  // Read chunk and reset
  if (*bit_in_chunk < 0) {
    RETURN_NOT_OK(buff->read(chunk, sizeof(uint64_t)));
    *bit_in_chunk = 63;
  }

  // Read double delta
  int bits_left_to_read = bitsize;
  int bits_to_read_from_chunk = std::min(*bit_in_chunk + 1, bits_left_to_read);
  uint64_t tmp_chunk;
  int bit_in_dd = bitsize - 1;
  *double_delta = 0;
  while (bits_left_to_read > 0) {
    // Read from chunk
    if (bits_to_read_from_chunk > 0) {
      tmp_chunk = (((*chunk) << (63 - *bit_in_chunk)) >> (63 - bit_in_dd));
      (*double_delta) |= tmp_chunk;
      bit_in_dd -= bits_to_read_from_chunk;
      (*bit_in_chunk) -= bits_to_read_from_chunk;
      bits_left_to_read -= bits_to_read_from_chunk;
    }

    // Read chunk and reset
    if (*bit_in_chunk < 0 && buff->offset() != buff->size()) {
      RETURN_NOT_OK(buff->read(chunk, sizeof(uint64_t)));
      *bit_in_chunk = 63;
      bits_to_read_from_chunk = std::min(*bit_in_chunk + 1, bits_left_to_read);
    }
  }

  // Set sign
  (*double_delta) *= sign;

  return Status::Ok();
}

Status DoubleDelta::write_double_delta(
    Buffer* buff,
    int64_t double_delta,
    int bitsize,
    uint64_t* chunk,
    int* bit_in_chunk) {
  // Write sign
  uint64_t to_or = (double_delta < 0) ? 1 : 0;
  to_or <<= (*bit_in_chunk);
  (*chunk) |= to_or;
  --(*bit_in_chunk);

  // Write chunk and reset
  if (*bit_in_chunk < 0) {
    RETURN_NOT_OK(buff->write(chunk, sizeof(uint64_t)));
    *bit_in_chunk = 63;
    *chunk = 0;
  }

  // Write rest of bits
  int bits_left_to_write = bitsize;
  int bit_in_dd = bitsize - 1;
  int bits_to_fill_in_chunk = std::min((*bit_in_chunk) + 1, bits_left_to_write);
  uint64_t abs_dd = ABS(double_delta);
  uint64_t tmp_abs_dd;

  while (bits_left_to_write > 0) {
    if (bits_to_fill_in_chunk > 0) {
      tmp_abs_dd = (abs_dd << (63 - bit_in_dd));
      tmp_abs_dd >>= (63 - (*bit_in_chunk));
      (*chunk) |= tmp_abs_dd;
      (*bit_in_chunk) -= bits_to_fill_in_chunk;
      bit_in_dd -= bits_to_fill_in_chunk;
      bits_left_to_write -= bits_to_fill_in_chunk;
    }

    // Write chunk and reset
    if (*bit_in_chunk < 0) {
      RETURN_NOT_OK(buff->write(chunk, sizeof(uint64_t)));
      *bit_in_chunk = 63;
      *chunk = 0;
      bits_to_fill_in_chunk = std::min((*bit_in_chunk) + 1, bits_left_to_write);
    }
  }

  return Status::Ok();
}

// Explicit template instantiations

template Status DoubleDelta::compress<char>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<int8_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<uint8_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<int16_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<uint16_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<int>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<uint32_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<int64_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);
template Status DoubleDelta::compress<uint64_t>(
    ConstBuffer* input_buffer, Buffer* output_buffer);

template Status DoubleDelta::decompress<char>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<int8_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<uint8_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<int16_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<uint16_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<int>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<uint32_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<int64_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);
template Status DoubleDelta::decompress<uint64_t>(
    ConstBuffer* input_buffer, PreallocatedBuffer* output_buffer);

};  // namespace sm
}  // namespace tiledb