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

#ifndef ENCODING_PLAIN_ENCODER_H
#define ENCODING_PLAIN_ENCODER_H

#include <cstring>

#include "encoder.h"

#ifdef ENABLE_SIMD
#include "simde/x86/ssse3.h"
#endif

namespace storage {

class PlainEncoder : public Encoder {
   public:
    PlainEncoder() {}
    ~PlainEncoder() { destroy(); }
    void destroy() override { /* do nothing for PlainEncoder */
    }
    void reset() override { /* do thing for PlainEncoder */
    }

    FORCE_INLINE int encode(bool value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_i8(value ? 1 : 0, out_stream);
    }

    FORCE_INLINE int encode(int32_t value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_var_int(value, out_stream);
    }

    FORCE_INLINE int encode(int64_t value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_i64(value, out_stream);
    }

    FORCE_INLINE int encode(float value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_float(value, out_stream);
    }

    FORCE_INLINE int encode(double value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_double(value, out_stream);
    }

    FORCE_INLINE int encode(common::String value,
                            common::ByteStream& out_stream) override {
        return common::SerializationUtil::write_mystring(value, out_stream);
    }

    int flush(common::ByteStream& out_stream) override {
        // do nothing for PlainEncoder
        return common::E_OK;
    }

    int get_max_byte_size() override { return 0; }

    // Optimized batch encoding: directly byte-swap into ByteStream page buffer.
    // Avoids per-value write_buf overhead entirely — only calls acquire_buf()
    // once per page boundary crossing.
    int encode_batch(const int64_t* values, uint32_t count,
                     common::ByteStream& out_stream) override {
        if (count == 0) return common::E_OK;
        uint32_t offset = 0;
        while (offset < count) {
            common::ByteStream::Buffer buf = out_stream.acquire_buf();
            if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM;
            // How many int64 values fit in the remaining page space?
            uint32_t capacity = buf.len_ / 8;
            if (capacity == 0) {
                // Page has < 8 bytes left, fall back to write_buf for this one
                return Encoder::encode_batch(values + offset, count - offset,
                                             out_stream);
            }
            uint32_t batch = std::min(count - offset, capacity);
            uint8_t* dst = (uint8_t*)buf.buf_;
            const int64_t* src = values + offset;
            uint32_t i = 0;
#ifdef ENABLE_SIMD
            // SIMDe: byte-reverse 2 x int64 per iteration
            const simde__m128i bswap64_shuf = simde_mm_set_epi8(
                8, 9, 10, 11, 12, 13, 14, 15, 0, 1, 2, 3, 4, 5, 6, 7);
            for (; i + 2 <= batch; i += 2) {
                simde__m128i v =
                    simde_mm_loadu_si128((const simde__m128i*)&src[i]);
                v = simde_mm_shuffle_epi8(v, bswap64_shuf);
                simde_mm_storeu_si128((simde__m128i*)dst, v);
                dst += 16;
            }
#endif
            // Scalar tail
            for (; i < batch; i++) {
                uint64_t v = (uint64_t)src[i];
                dst[0] = (uint8_t)(v >> 56);
                dst[1] = (uint8_t)(v >> 48);
                dst[2] = (uint8_t)(v >> 40);
                dst[3] = (uint8_t)(v >> 32);
                dst[4] = (uint8_t)(v >> 24);
                dst[5] = (uint8_t)(v >> 16);
                dst[6] = (uint8_t)(v >> 8);
                dst[7] = (uint8_t)(v);
                dst += 8;
            }
            out_stream.buffer_used(batch * 8);
            offset += batch;
        }
        return common::E_OK;
    }

    int encode_batch(const double* values, uint32_t count,
                     common::ByteStream& out_stream) override {
        if (count == 0) return common::E_OK;
        uint32_t offset = 0;
        while (offset < count) {
            common::ByteStream::Buffer buf = out_stream.acquire_buf();
            if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM;
            uint32_t capacity = buf.len_ / 8;
            if (capacity == 0) {
                return Encoder::encode_batch(values + offset, count - offset,
                                             out_stream);
            }
            uint32_t batch = std::min(count - offset, capacity);
            uint8_t* dst = (uint8_t*)buf.buf_;
            const double* src = values + offset;
            uint32_t i = 0;
#ifdef ENABLE_SIMD
            // SIMDe: byte-reverse 2 x double (64-bit) per iteration
            const simde__m128i bswap64_shuf = simde_mm_set_epi8(
                8, 9, 10, 11, 12, 13, 14, 15, 0, 1, 2, 3, 4, 5, 6, 7);
            for (; i + 2 <= batch; i += 2) {
                simde__m128i v =
                    simde_mm_loadu_si128((const simde__m128i*)&src[i]);
                v = simde_mm_shuffle_epi8(v, bswap64_shuf);
                simde_mm_storeu_si128((simde__m128i*)dst, v);
                dst += 16;
            }
#endif
            // Scalar tail: round-trip the bits via memcpy to avoid the
            // strict-aliasing violation of reading a double through an
            // int64_t* (the old reinterpret_cast dispatch).
            for (; i < batch; i++) {
                uint64_t v;
                memcpy(&v, &src[i], sizeof(double));
                dst[0] = (uint8_t)(v >> 56);
                dst[1] = (uint8_t)(v >> 48);
                dst[2] = (uint8_t)(v >> 40);
                dst[3] = (uint8_t)(v >> 32);
                dst[4] = (uint8_t)(v >> 24);
                dst[5] = (uint8_t)(v >> 16);
                dst[6] = (uint8_t)(v >> 8);
                dst[7] = (uint8_t)(v);
                dst += 8;
            }
            out_stream.buffer_used(batch * 8);
            offset += batch;
        }
        return common::E_OK;
    }

    int encode_batch(const float* values, uint32_t count,
                     common::ByteStream& out_stream) override {
        if (count == 0) return common::E_OK;
        uint32_t offset = 0;
        while (offset < count) {
            common::ByteStream::Buffer buf = out_stream.acquire_buf();
            if (UNLIKELY(buf.buf_ == nullptr)) return common::E_OOM;
            uint32_t capacity = buf.len_ / 4;
            if (capacity == 0) {
                return Encoder::encode_batch(values + offset, count - offset,
                                             out_stream);
            }
            uint32_t batch = std::min(count - offset, capacity);
            uint8_t* dst = (uint8_t*)buf.buf_;
            const float* src = values + offset;
            uint32_t i = 0;
#ifdef ENABLE_SIMD
            // SIMDe: byte-reverse 4 x float (32-bit) per iteration
            const simde__m128i bswap32_shuf = simde_mm_set_epi8(
                12, 13, 14, 15, 8, 9, 10, 11, 4, 5, 6, 7, 0, 1, 2, 3);
            for (; i + 4 <= batch; i += 4) {
                simde__m128i v =
                    simde_mm_loadu_si128((const simde__m128i*)&src[i]);
                v = simde_mm_shuffle_epi8(v, bswap32_shuf);
                simde_mm_storeu_si128((simde__m128i*)dst, v);
                dst += 16;
            }
#endif
            for (; i < batch; i++) {
                uint32_t v;
                memcpy(&v, &src[i], sizeof(float));
                dst[0] = (uint8_t)(v >> 24);
                dst[1] = (uint8_t)(v >> 16);
                dst[2] = (uint8_t)(v >> 8);
                dst[3] = (uint8_t)(v);
                dst += 4;
            }
            out_stream.buffer_used(batch * 4);
            offset += batch;
        }
        return common::E_OK;
    }

    // Batch encode strings from Arrow-style offset+buffer layout.
    // Each string is serialized as: var_int(len) + raw bytes.
    int encode_string_batch(const char* buffer, const uint32_t* offsets,
                            uint32_t start_idx, uint32_t count,
                            common::ByteStream& out_stream) override {
        int ret = common::E_OK;
        for (uint32_t i = 0; i < count; i++) {
            uint32_t idx = start_idx + i;
            uint32_t len = offsets[idx + 1] - offsets[idx];
            if (RET_FAIL(common::SerializationUtil::write_var_int(
                    (int32_t)len, out_stream))) {
                return ret;
            }
            if (len > 0) {
                if (RET_FAIL(
                        out_stream.write_buf(buffer + offsets[idx], len))) {
                    return ret;
                }
            }
        }
        return ret;
    }
};

}  // end namespace storage
#endif  // ENCODING_PLAIN_ENCODER_H