// This file is licensed under the Elastic License 2.0. Copyright 2021 StarRocks Limited.

#pragma once

#include "column/bytes.h"
#include "column/column.h"
#include "util/slice.h"

namespace starrocks::vectorized {

class BinaryColumn final : public ColumnFactory<Column, BinaryColumn> {
    friend class ColumnFactory<Column, BinaryColumn>;

public:
    using ValueType = Slice;

    using Offset = uint32_t;
    using Offsets = Buffer<uint32_t>;

    using Bytes = starrocks::raw::RawVectorPad16<uint8_t>;

    using Container = Buffer<Slice>;

    // TODO(kks): when we create our own vector, we could let vector[-1] = 0,
    // and then we don't need explicitly emplace_back zero value
    BinaryColumn() { _offsets.emplace_back(0); }
    BinaryColumn(Bytes bytes, Offsets offsets) : _bytes(std::move(bytes)), _offsets(std::move(offsets)) {
        if (_offsets.empty()) {
            _offsets.emplace_back(0);
        }
    };

    // Copy constructor
    // NOTE: do *NOT* copy |_slices|
    BinaryColumn(const BinaryColumn& rhs) : _bytes(rhs._bytes), _offsets(rhs._offsets) {}

    // Move constructor
    // NOTE: do *NOT* copy |_slices|
    BinaryColumn(BinaryColumn&& rhs) : _bytes(std::move(rhs._bytes)), _offsets(std::move(rhs._offsets)) {}

    // Copy assignment
    BinaryColumn& operator=(const BinaryColumn& rhs) {
        BinaryColumn tmp(rhs);
        this->swap_column(tmp);
        return *this;
    }

    // Move assignment
    BinaryColumn& operator=(BinaryColumn&& rhs) {
        BinaryColumn tmp(std::move(rhs));
        this->swap_column(tmp);
        return *this;
    }

    ~BinaryColumn() override {
        if (!_offsets.empty()) {
            DCHECK_EQ(_bytes.size(), _offsets.back());
        } else {
            DCHECK_EQ(_bytes.size(), 0);
        }
    }

    bool low_cardinality() const override { return false; }
    bool is_binary() const override { return true; }

    const uint8_t* raw_data() const override {
        if (!_slices_cache) {
            _build_slices();
        }
        return reinterpret_cast<const uint8_t*>(_slices.data());
    }

    uint8_t* mutable_raw_data() override {
        if (!_slices_cache) {
            _build_slices();
        }
        return reinterpret_cast<uint8_t*>(_slices.data());
    }

    size_t size() const override { return _offsets.size() - 1; }

    size_t type_size() const override { return sizeof(Slice); }

    size_t byte_size() const override { return _bytes.size() * sizeof(uint8_t) + _offsets.size() * sizeof(Offset); }

    size_t byte_size(size_t from, size_t size) const override {
        DCHECK_LE(from + size, this->size()) << "Range error";
        return (_offsets[from + size] - _offsets[from]) + size * sizeof(Offset);
    }

    size_t byte_size(size_t idx) const override { return _offsets[idx + 1] - _offsets[idx] + sizeof(uint32_t); }

    Slice get_slice(size_t idx) const {
        return Slice(_bytes.data() + _offsets[idx], _offsets[idx + 1] - _offsets[idx]);
    }

    // For n value, the offsets size is n + 1
    // For example, for string "I","love","you"
    // the _bytes array is "Iloveyou"
    // the _offsets array is [0,1,5,8]
    void reserve(size_t n) override {
        // hard to know how the best reserve size of |_bytes|, inaccurate reserve may
        // affect the performance.
        // _bytes.reserve(n * 4);
        _offsets.reserve(n + 1);
        _slices_cache = false;
    }

    // If you know the size of the Byte array in advance, you can call this method,
    // n means the number of strings, byte_size is the total length of the string
    void reserve(size_t n, size_t byte_size) {
        _offsets.reserve(n + 1);
        _bytes.reserve(byte_size);
        _slices_cache = false;
    }

    void resize(size_t n) override {
        _offsets.resize(n + 1, _offsets.back());
        _bytes.resize(_offsets.back());
        _slices_cache = false;
    }

    void assign(size_t n, size_t idx) override;

    void remove_first_n_values(size_t count) override;

    void append(const Slice& str) {
        _bytes.insert(_bytes.end(), str.data, str.data + str.size);
        _offsets.emplace_back(_bytes.size());
        _slices_cache = false;
    }

    void append_datum(const Datum& datum) override {
        append(datum.get_slice());
        _slices_cache = false;
    }

    void append(const Column& src, size_t offset, size_t count) override;

    void append_selective(const Column& src, const uint32_t* indexes, uint32_t from, uint32_t size) override;

    void append_value_multiple_times(const Column& src, uint32_t index, uint32_t size) override;

    bool append_nulls(size_t count) override { return false; }

    void append_string(const std::string& str) {
        _bytes.insert(_bytes.end(), str.data(), str.data() + str.size());
        _offsets.emplace_back(_bytes.size());
        _slices_cache = false;
    }

    bool append_strings(const std::vector<Slice>& strs) override;

    bool append_strings_overflow(const std::vector<Slice>& strs, size_t max_length) override;

    bool append_continuous_strings(const std::vector<Slice>& strs) override;

    size_t append_numbers(const void* buff, size_t length) override { return -1; }

    void append_value_multiple_times(const void* value, size_t count) override;

    void append_default() override {
        _offsets.emplace_back(_bytes.size());
        _slices_cache = false;
    }

    void append_default(size_t count) override {
        _offsets.insert(_offsets.end(), count, _bytes.size());
        _slices_cache = false;
    }

    uint32_t max_one_element_serialize_size() const override;

    uint32_t serialize(size_t idx, uint8_t* pos) override;

    uint32_t serialize_default(uint8_t* pos) override;

    void serialize_batch(uint8_t* dst, Buffer<uint32_t>& slice_sizes, size_t chunk_size,
                         uint32_t max_one_row_size) override;

    const uint8_t* deserialize_and_append(const uint8_t* pos) override;

    void deserialize_and_append_batch(std::vector<Slice>& srcs, size_t batch_size) override;

    uint32_t serialize_size(size_t idx) const override { return sizeof(uint32_t) + _offsets[idx + 1] - _offsets[idx]; }

    size_t serialize_size() const override {
        DCHECK_EQ(_bytes.size(), _offsets.back());
        return byte_size() + sizeof(uint32_t) * 2; // _offsets size + _bytes size;
    }

    uint8_t* serialize_column(uint8_t* dst) override;

    const uint8_t* deserialize_column(const uint8_t* src) override;

    MutableColumnPtr clone_empty() const override { return create_mutable(); }

    ColumnPtr cut(size_t start, size_t length) const;
    size_t filter_range(const Column::Filter& filter, size_t start, size_t to) override;

    int compare_at(size_t left, size_t right, const Column& rhs, int nan_direction_hint) const override;

    void fnv_hash(uint32_t* hashes, uint32_t from, uint32_t to) const override;

    void crc32_hash(uint32_t* hash, uint32_t from, uint32_t to) const override;

    void put_mysql_row_buffer(MysqlRowBuffer* buf, size_t idx) const override;

    std::string get_name() const override { return "binary"; }

    Container& get_data() {
        if (!_slices_cache) {
            _build_slices();
        }
        return _slices;
    }
    const Container& get_data() const {
        if (!_slices_cache) {
            _build_slices();
        }
        return _slices;
    }

    Bytes& get_bytes() { return _bytes; }

    const Bytes& get_bytes() const { return _bytes; }

    Offsets& get_offset() { return _offsets; }
    const Offsets& get_offset() const { return _offsets; }

    Datum get(size_t n) const override { return Datum(get_slice(n)); }

    size_t container_memory_usage() const override {
        return _bytes.capacity() + _offsets.capacity() * sizeof(_offsets[0]) + _slices.capacity() * sizeof(_slices[0]);
    }

    size_t shrink_memory_usage() const override {
        return _bytes.size() * sizeof(uint8_t) + _offsets.size() * sizeof(_offsets[0]) +
               _slices.size() * sizeof(_slices[0]);
    }

    void swap_column(Column& rhs) override {
        auto& r = down_cast<BinaryColumn&>(rhs);
        using std::swap;
        swap(_delete_state, r._delete_state);
        swap(_bytes, r._bytes);
        swap(_offsets, r._offsets);
        swap(_slices, r._slices);
        swap(_slices_cache, r._slices_cache);
    }

    void reset_column() override {
        Column::reset_column();
        // TODO(zhuming): shrink size if needed.
        _bytes.clear();
        _offsets.resize(1, 0);
        _slices.clear();
        _slices_cache = false;
    }

    void invalidate_slice_cache() { _slices_cache = false; }

    std::string debug_item(uint32_t idx) const override;

    std::string debug_string() const override {
        std::stringstream ss;
        ss << "[";
        for (int i = 0; i < size() - 1; ++i) {
            ss << debug_item(i) << ", ";
        }
        ss << debug_item(size() - 1) << "]";
        return ss.str();
    }

    bool reach_capacity_limit() const override {
        return _bytes.size() >= Column::MAX_CAPACITY_LIMIT || _offsets.size() >= Column::MAX_CAPACITY_LIMIT ||
               _slices.size() >= Column::MAX_CAPACITY_LIMIT;
    }

private:
    void _build_slices() const;

    Bytes _bytes;
    Offsets _offsets;

    mutable Container _slices;
    mutable bool _slices_cache = false;
};

using Offsets = BinaryColumn::Offsets;
} // namespace starrocks::vectorized
