/*
 * 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 <arrow/c/bridge.h>

#include <cassert>
#include <ctime>

#include "ffi_converter.hpp"
#include "fluss.hpp"
#include "lib.rs.h"
#include "rust/cxx.h"
#include "type_lowering.hpp"
// todo:  bindings/cpp/BUILD.bazel still doesn't declare Arrow include/link dependencies.
// In environments where Bazel does not already have Arrow available, this will fail at compile/link
// time.
#include <arrow/record_batch.h>
#include <arrow/type.h>

namespace fluss {

static constexpr int kSecondsPerDay = 24 * 60 * 60;

static std::time_t timegm_utc(std::tm* tm) {
#if defined(_WIN32)
    return _mkgmtime(tm);
#else
    return ::timegm(tm);
#endif
}

static std::tm gmtime_utc(std::time_t epoch_seconds) {
    std::tm tm{};
#if defined(_WIN32)
    gmtime_s(&tm, &epoch_seconds);
#else
    ::gmtime_r(&epoch_seconds, &tm);
#endif
    return tm;
}

Date Date::FromYMD(int year, int month, int day) {
    std::tm tm{};
    tm.tm_year = year - 1900;
    tm.tm_mon = month - 1;
    tm.tm_mday = day;
    std::time_t epoch_seconds = timegm_utc(&tm);
    return {static_cast<int32_t>(epoch_seconds / kSecondsPerDay)};
}

int Date::Year() const {
    std::time_t epoch_seconds = static_cast<std::time_t>(days_since_epoch) * kSecondsPerDay;
    std::tm tm = gmtime_utc(epoch_seconds);
    return tm.tm_year + 1900;
}

int Date::Month() const {
    std::time_t epoch_seconds = static_cast<std::time_t>(days_since_epoch) * kSecondsPerDay;
    std::tm tm = gmtime_utc(epoch_seconds);
    return tm.tm_mon + 1;
}

int Date::Day() const {
    std::time_t epoch_seconds = static_cast<std::time_t>(days_since_epoch) * kSecondsPerDay;
    std::tm tm = gmtime_utc(epoch_seconds);
    return tm.tm_mday;
}

// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define CHECK_INNER(name)                                                                 \
    do {                                                                                  \
        if (!inner_) throw std::logic_error(name ": not available (moved-from or null)"); \
    } while (0)

namespace {

ffi::ArrayWriterInner* make_array_writer_arrow(size_t size,
                                               std::shared_ptr<arrow::DataType> element,
                                               bool nullable) {
    auto schema = arrow::schema({arrow::field("element", std::move(element), nullable)});
    return ffi::new_array_writer_arrow(size, detail::export_arrow_schema(*schema)).into_raw();
}

ffi::MapWriterInner* make_map_writer_arrow(size_t capacity, std::shared_ptr<arrow::DataType> key,
                                           std::shared_ptr<arrow::DataType> value,
                                           bool value_nullable) {
    auto schema = arrow::schema({
        arrow::field("key", std::move(key), /*nullable=*/false),
        arrow::field("value", std::move(value), value_nullable),
    });
    return ffi::new_map_writer_arrow(capacity, detail::export_arrow_schema(*schema)).into_raw();
}

}  // namespace

// ============================================================================
// ArrayWriter — builder for array values backed by Rust ArrayWriterInner
// ============================================================================

ArrayWriter::ArrayWriter(size_t size, DataType element_type) : element_type_(std::move(element_type)) {
    if (detail::is_compound(element_type_)) {
        inner_ = make_array_writer_arrow(size, detail::to_arrow_type(element_type_),
                                         element_type_.nullable());
        return;
    }

    auto flat = utils::flatten_array_type(element_type_);
    int32_t leaf_type_id = flat.nesting > 0 ? flat.leaf_type : static_cast<int32_t>(element_type_.id());
    uint32_t leaf_precision = static_cast<uint32_t>(flat.nesting > 0 ? flat.leaf_precision
                                                                      : element_type_.precision());
    uint32_t leaf_scale = static_cast<uint32_t>(flat.nesting > 0 ? flat.leaf_scale : element_type_.scale());
    uint32_t array_nesting = static_cast<uint32_t>(flat.nesting);

    auto box = ffi::new_array_writer(size, leaf_type_id, leaf_precision, leaf_scale, array_nesting);
    inner_ = box.into_raw();
}

ArrayWriter::ArrayWriter(size_t size, std::shared_ptr<arrow::DataType> element_type)
    : element_type_(DataType::Int()) {  // placeholder; Rust holds the real element type
    inner_ = make_array_writer_arrow(size, std::move(element_type), /*nullable=*/true);
}

ArrayWriter::~ArrayWriter() noexcept { Destroy(); }

void ArrayWriter::Destroy() noexcept {
    if (inner_) {
        rust::Box<ffi::ArrayWriterInner>::from_raw(inner_);
        inner_ = nullptr;
    }
}

ArrayWriter::ArrayWriter(ArrayWriter&& other) noexcept
    : inner_(other.inner_), element_type_(std::move(other.element_type_)) {
    other.inner_ = nullptr;
}

ArrayWriter& ArrayWriter::operator=(ArrayWriter&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        element_type_ = std::move(other.element_type_);
        other.inner_ = nullptr;
    }
    return *this;
}

bool ArrayWriter::Available() const { return inner_ != nullptr; }

size_t ArrayWriter::Size() const noexcept {
    assert(inner_ && "ArrayWriter::Size called on moved-from instance");
    return inner_->aw_size();
}

// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define CHECK_AW(name)                                                                    \
    do {                                                                                  \
        if (!inner_) throw std::logic_error(name ": not available (moved-from or null)"); \
    } while (0)

void ArrayWriter::SetNull(size_t idx) { CHECK_AW("ArrayWriter"); inner_->aw_set_null(idx); }
void ArrayWriter::SetBool(size_t idx, bool v) { CHECK_AW("ArrayWriter"); inner_->aw_set_bool(idx, v); }
void ArrayWriter::SetInt32(size_t idx, int32_t v) { CHECK_AW("ArrayWriter"); inner_->aw_set_i32(idx, v); }
void ArrayWriter::SetInt64(size_t idx, int64_t v) { CHECK_AW("ArrayWriter"); inner_->aw_set_i64(idx, v); }
void ArrayWriter::SetFloat32(size_t idx, float v) { CHECK_AW("ArrayWriter"); inner_->aw_set_f32(idx, v); }
void ArrayWriter::SetFloat64(size_t idx, double v) { CHECK_AW("ArrayWriter"); inner_->aw_set_f64(idx, v); }

void ArrayWriter::SetString(size_t idx, const std::string& v) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_str(idx, v);
}

void ArrayWriter::SetBytes(size_t idx, const std::vector<uint8_t>& v) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_bytes(idx, rust::Slice<const uint8_t>(v.data(), v.size()));
}

void ArrayWriter::SetDate(size_t idx, fluss::Date d) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_date(idx, d.days_since_epoch);
}

void ArrayWriter::SetTime(size_t idx, fluss::Time t) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_time(idx, t.millis_since_midnight);
}

void ArrayWriter::SetTimestampNtz(size_t idx, fluss::Timestamp ts) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_ts_ntz(idx, ts.epoch_millis, ts.nano_of_millisecond);
}

void ArrayWriter::SetTimestampLtz(size_t idx, fluss::Timestamp ts) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_ts_ltz(idx, ts.epoch_millis, ts.nano_of_millisecond);
}

void ArrayWriter::SetDecimal(size_t idx, const std::string& value) {
    CHECK_AW("ArrayWriter");
    inner_->aw_set_decimal_str(idx, value);
}

void ArrayWriter::SetArray(size_t idx, ArrayWriter&& nested) {
    CHECK_AW("ArrayWriter");
    if (!nested.inner_) {
        throw std::logic_error("ArrayWriter::SetArray: nested writer not available");
    }
    inner_->aw_set_array(idx, *nested.inner_);
    nested.Destroy();
}

void ArrayWriter::SetRow(size_t idx, GenericRow&& row) {
    CHECK_AW("ArrayWriter");
    if (!row.inner_) {
        throw std::logic_error("ArrayWriter::SetRow: nested row not available");
    }
    inner_->aw_set_row(idx, *row.inner_);
}

void ArrayWriter::SetMap(size_t idx, MapWriter&& map) {
    CHECK_AW("ArrayWriter");
    if (!map.inner_) {
        throw std::logic_error("ArrayWriter::SetMap: nested map not available");
    }
    inner_->aw_set_map(idx, *map.inner_);
    map.Destroy();
}

// ============================================================================
// MapWriter — builder for map values backed by Rust MapWriterInner
// ============================================================================

MapWriter::MapWriter(size_t capacity, DataType key_type, DataType value_type)
    : key_type_(std::move(key_type)), value_type_(std::move(value_type)) {
    if (detail::is_compound(key_type_) || detail::is_compound(value_type_)) {
        inner_ = make_map_writer_arrow(capacity, detail::to_arrow_type(key_type_),
                                       detail::to_arrow_type(value_type_), value_type_.nullable());
        return;
    }

    // Keys are scalar; values may be a (possibly nested) ARRAY, so flatten them.
    auto value_flat = utils::flatten_array_type(value_type_);
    int32_t value_leaf = value_flat.nesting > 0 ? value_flat.leaf_type
                                                : static_cast<int32_t>(value_type_.id());
    uint32_t value_precision = static_cast<uint32_t>(
        value_flat.nesting > 0 ? value_flat.leaf_precision : value_type_.precision());
    uint32_t value_scale = static_cast<uint32_t>(
        value_flat.nesting > 0 ? value_flat.leaf_scale : value_type_.scale());
    uint32_t value_nesting = static_cast<uint32_t>(value_flat.nesting);

    auto box = ffi::new_map_writer(capacity,
                                   static_cast<int32_t>(key_type_.id()),
                                   static_cast<uint32_t>(key_type_.precision()),
                                   static_cast<uint32_t>(key_type_.scale()),
                                   value_leaf, value_precision, value_scale, value_nesting);
    inner_ = box.into_raw();
}

MapWriter::MapWriter(size_t capacity, std::shared_ptr<arrow::DataType> key_type,
                     std::shared_ptr<arrow::DataType> value_type)
    : key_type_(DataType::Int()), value_type_(DataType::Int()) {
    // Placeholders above; Rust holds the real key/value types.
    inner_ = make_map_writer_arrow(capacity, std::move(key_type), std::move(value_type),
                                   /*nullable=*/true);
}

MapWriter::~MapWriter() noexcept { Destroy(); }

void MapWriter::Destroy() noexcept {
    if (inner_) {
        rust::Box<ffi::MapWriterInner>::from_raw(inner_);
        inner_ = nullptr;
    }
}

MapWriter::MapWriter(MapWriter&& other) noexcept
    : inner_(other.inner_),
      key_type_(std::move(other.key_type_)),
      value_type_(std::move(other.value_type_)) {
    other.inner_ = nullptr;
}

MapWriter& MapWriter::operator=(MapWriter&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        key_type_ = std::move(other.key_type_);
        value_type_ = std::move(other.value_type_);
        other.inner_ = nullptr;
    }
    return *this;
}

bool MapWriter::Available() const { return inner_ != nullptr; }

void MapWriter::SetKeyBool(bool k) { CHECK_AW("MapWriter"); inner_->mw_key_bool(k); }
void MapWriter::SetKeyInt32(int32_t k) { CHECK_AW("MapWriter"); inner_->mw_key_i32(k); }
void MapWriter::SetKeyInt64(int64_t k) { CHECK_AW("MapWriter"); inner_->mw_key_i64(k); }
void MapWriter::SetKeyFloat32(float k) { CHECK_AW("MapWriter"); inner_->mw_key_f32(k); }
void MapWriter::SetKeyFloat64(double k) { CHECK_AW("MapWriter"); inner_->mw_key_f64(k); }
void MapWriter::SetKeyString(const std::string& k) { CHECK_AW("MapWriter"); inner_->mw_key_str(k); }
void MapWriter::SetKeyBytes(const std::vector<uint8_t>& k) {
    CHECK_AW("MapWriter");
    inner_->mw_key_bytes(rust::Slice<const uint8_t>(k.data(), k.size()));
}
void MapWriter::SetKeyDate(fluss::Date k) { CHECK_AW("MapWriter"); inner_->mw_key_date(k.days_since_epoch); }
void MapWriter::SetKeyTime(fluss::Time k) {
    CHECK_AW("MapWriter");
    inner_->mw_key_time(k.millis_since_midnight);
}
void MapWriter::SetKeyTimestamp(fluss::Timestamp k) {
    CHECK_AW("MapWriter");
    inner_->mw_key_timestamp(k.epoch_millis, k.nano_of_millisecond);
}
void MapWriter::SetKeyDecimal(const std::string& k) { CHECK_AW("MapWriter"); inner_->mw_key_decimal_str(k); }

void MapWriter::SetValueNull() { CHECK_AW("MapWriter"); inner_->mw_value_null(); }
void MapWriter::SetValueBool(bool v) { CHECK_AW("MapWriter"); inner_->mw_value_bool(v); }
void MapWriter::SetValueInt32(int32_t v) { CHECK_AW("MapWriter"); inner_->mw_value_i32(v); }
void MapWriter::SetValueInt64(int64_t v) { CHECK_AW("MapWriter"); inner_->mw_value_i64(v); }
void MapWriter::SetValueFloat32(float v) { CHECK_AW("MapWriter"); inner_->mw_value_f32(v); }
void MapWriter::SetValueFloat64(double v) { CHECK_AW("MapWriter"); inner_->mw_value_f64(v); }
void MapWriter::SetValueString(const std::string& v) { CHECK_AW("MapWriter"); inner_->mw_value_str(v); }
void MapWriter::SetValueBytes(const std::vector<uint8_t>& v) {
    CHECK_AW("MapWriter");
    inner_->mw_value_bytes(rust::Slice<const uint8_t>(v.data(), v.size()));
}
void MapWriter::SetValueDate(fluss::Date v) { CHECK_AW("MapWriter"); inner_->mw_value_date(v.days_since_epoch); }
void MapWriter::SetValueTime(fluss::Time v) {
    CHECK_AW("MapWriter");
    inner_->mw_value_time(v.millis_since_midnight);
}
void MapWriter::SetValueTimestamp(fluss::Timestamp v) {
    CHECK_AW("MapWriter");
    inner_->mw_value_timestamp(v.epoch_millis, v.nano_of_millisecond);
}
void MapWriter::SetValueDecimal(const std::string& v) {
    CHECK_AW("MapWriter");
    inner_->mw_value_decimal_str(v);
}
void MapWriter::SetValueRow(GenericRow&& v) {
    CHECK_AW("MapWriter");
    if (!v.inner_) {
        throw std::logic_error("MapWriter::SetValueRow: nested row not available");
    }
    inner_->mw_value_row(*v.inner_);
}
void MapWriter::SetValueMap(MapWriter&& v) {
    CHECK_AW("MapWriter");
    if (!v.inner_) {
        throw std::logic_error("MapWriter::SetValueMap: nested map not available");
    }
    inner_->mw_value_map(*v.inner_);
    v.Destroy();
}
void MapWriter::SetValueArray(ArrayWriter&& v) {
    CHECK_AW("MapWriter");
    if (!v.inner_) {
        throw std::logic_error("MapWriter::SetValueArray: nested array not available");
    }
    inner_->mw_value_array(*v.inner_);
    v.Destroy();
}

void MapWriter::Commit() { CHECK_AW("MapWriter"); inner_->mw_commit(); }

// ============================================================================
// Value — one recursive handle for complex (ARRAY/MAP/ROW) reads
// ============================================================================

Value::~Value() noexcept { Destroy(); }

void Value::Destroy() noexcept {
    if (inner_) {
        rust::Box<ffi::ValueInner>::from_raw(inner_);
        inner_ = nullptr;
    }
}

Value::Value(Value&& other) noexcept : inner_(other.inner_) { other.inner_ = nullptr; }

Value& Value::operator=(Value&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        other.inner_ = nullptr;
    }
    return *this;
}

// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define CHECK_VALUE()                                                                  \
    do {                                                                               \
        if (!inner_) throw std::logic_error("Value: not available (moved-from)");      \
    } while (0)

TypeId Value::Type() const noexcept {
    assert(inner_ && "Value::Type called on moved-from instance");
    return static_cast<TypeId>(inner_->v_type());
}
bool Value::IsNull() const noexcept {
    assert(inner_ && "Value::IsNull called on moved-from instance");
    return inner_->v_is_null();
}
bool Value::GetBool() const { CHECK_VALUE(); return inner_->v_get_bool(); }
int32_t Value::GetInt32() const { CHECK_VALUE(); return inner_->v_get_i32(); }
int64_t Value::GetInt64() const { CHECK_VALUE(); return inner_->v_get_i64(); }
float Value::GetFloat32() const { CHECK_VALUE(); return inner_->v_get_f32(); }
double Value::GetFloat64() const { CHECK_VALUE(); return inner_->v_get_f64(); }
std::string Value::GetString() const { CHECK_VALUE(); return std::string(inner_->v_get_str()); }
std::vector<uint8_t> Value::GetBytes() const {
    CHECK_VALUE();
    auto rv = inner_->v_get_bytes();
    return {rv.data(), rv.data() + rv.size()};
}
fluss::Date Value::GetDate() const { CHECK_VALUE(); return fluss::Date{inner_->v_get_date_days()}; }
fluss::Time Value::GetTime() const {
    CHECK_VALUE();
    return fluss::Time{inner_->v_get_time_millis()};
}
fluss::Timestamp Value::GetTimestamp() const {
    CHECK_VALUE();
    return fluss::Timestamp{inner_->v_get_ts_millis(), inner_->v_get_ts_nanos()};
}
std::string Value::GetDecimalString() const {
    CHECK_VALUE();
    return std::string(inner_->v_get_decimal_str());
}
size_t Value::Size() const { CHECK_VALUE(); return inner_->v_size(); }
size_t Value::FieldCount() const { CHECK_VALUE(); return inner_->v_field_count(); }
Value Value::At(size_t i) const {
    CHECK_VALUE();
    auto box = inner_->v_at(i);
    return Value(box.into_raw());
}
Value Value::KeyAt(size_t i) const {
    CHECK_VALUE();
    auto box = inner_->v_key_at(i);
    return Value(box.into_raw());
}
Value Value::ValueAt(size_t i) const {
    CHECK_VALUE();
    auto box = inner_->v_value_at(i);
    return Value(box.into_raw());
}
Value Value::Field(size_t i) const {
    CHECK_VALUE();
    auto box = inner_->v_field(i);
    return Value(box.into_raw());
}
Value Value::Field(const std::string& name) const {
    CHECK_VALUE();
    auto box = inner_->v_field_by_name(name);
    return Value(box.into_raw());
}

#undef CHECK_VALUE

// ============================================================================
// GenericRow — write-only row backed by opaque Rust GenericRowInner
// ============================================================================

GenericRow::GenericRow() {
    auto box = ffi::new_generic_row(0);
    inner_ = box.into_raw();
}

GenericRow::GenericRow(size_t field_count) {
    auto box = ffi::new_generic_row(field_count);
    inner_ = box.into_raw();
}

GenericRow::~GenericRow() noexcept { Destroy(); }

void GenericRow::Destroy() noexcept {
    if (inner_) {
        rust::Box<ffi::GenericRowInner>::from_raw(inner_);
        inner_ = nullptr;
    }
    column_map_.reset();
}

GenericRow::GenericRow(GenericRow&& other) noexcept
    : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
    other.inner_ = nullptr;
}

GenericRow& GenericRow::operator=(GenericRow&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        column_map_ = std::move(other.column_map_);
        other.inner_ = nullptr;
    }
    return *this;
}

bool GenericRow::Available() const { return inner_ != nullptr; }

void GenericRow::Reset() {
    CHECK_INNER("GenericRow");
    inner_->gr_reset();
}

void GenericRow::SetNull(size_t idx) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_null(idx);
}
void GenericRow::SetBool(size_t idx, bool v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_bool(idx, v);
}
void GenericRow::SetInt32(size_t idx, int32_t v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_i32(idx, v);
}
void GenericRow::SetInt64(size_t idx, int64_t v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_i64(idx, v);
}
void GenericRow::SetFloat32(size_t idx, float v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_f32(idx, v);
}
void GenericRow::SetFloat64(size_t idx, double v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_f64(idx, v);
}

void GenericRow::SetString(size_t idx, std::string v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_str(idx, v);
}

void GenericRow::SetBytes(size_t idx, std::vector<uint8_t> v) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_bytes(idx, rust::Slice<const uint8_t>(v.data(), v.size()));
}

void GenericRow::SetDate(size_t idx, fluss::Date d) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_date(idx, d.days_since_epoch);
}

void GenericRow::SetTime(size_t idx, fluss::Time t) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_time(idx, t.millis_since_midnight);
}

void GenericRow::SetTimestampNtz(size_t idx, fluss::Timestamp ts) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_ts_ntz(idx, ts.epoch_millis, ts.nano_of_millisecond);
}

void GenericRow::SetTimestampLtz(size_t idx, fluss::Timestamp ts) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_ts_ltz(idx, ts.epoch_millis, ts.nano_of_millisecond);
}

void GenericRow::SetDecimal(size_t idx, const std::string& value) {
    CHECK_INNER("GenericRow");
    inner_->gr_set_decimal_str(idx, value);
}

void GenericRow::SetArray(size_t idx, ArrayWriter&& writer) {
    CHECK_INNER("GenericRow");
    if (!writer.inner_) {
        throw std::logic_error("GenericRow::SetArray: ArrayWriter not available");
    }
    inner_->gr_set_array(idx, *writer.inner_);
    writer.Destroy();
}

void GenericRow::SetMap(size_t idx, MapWriter&& writer) {
    CHECK_INNER("GenericRow");
    if (!writer.inner_) {
        throw std::logic_error("GenericRow::SetMap: MapWriter not available");
    }
    inner_->gr_set_map(idx, *writer.inner_);
    writer.Destroy();
}

void GenericRow::SetRow(size_t idx, GenericRow&& nested) {
    CHECK_INNER("GenericRow");
    if (!nested.inner_) {
        throw std::logic_error("GenericRow::SetRow: nested row not available");
    }
    // gr_set_row moves the nested row's contents out; the emptied nested
    // handle is freed by `nested`'s own destructor at end of scope.
    inner_->gr_set_row(idx, *nested.inner_);
}

// ============================================================================
// ScanData — destructor must live in .cpp where rust::Box is visible
// ============================================================================

detail::ScanData::~ScanData() {
    if (raw) {
        rust::Box<ffi::ScanResultInner>::from_raw(raw);
    }
}

detail::PrefixData::~PrefixData() {
    if (raw) {
        rust::Box<ffi::PrefixLookupResultInner>::from_raw(raw);
    }
}

// ============================================================================
// RowView — zero-copy read-only row view for scan results
// ============================================================================

// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define CHECK_DATA(name)                                                                 \
    do {                                                                                 \
        if (!data_) throw std::logic_error(name ": not available (moved-from or null)"); \
    } while (0)

size_t RowView::FieldCount() const { return data_ ? data_->raw->sv_field_count() : 0; }

TypeId RowView::GetType(size_t idx) const {
    CHECK_DATA("RowView");
    return static_cast<TypeId>(data_->raw->sv_column_type(idx));
}

bool RowView::IsNull(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_is_null(bucket_idx_, rec_idx_, idx);
}
bool RowView::GetBool(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_get_bool(bucket_idx_, rec_idx_, idx);
}
int32_t RowView::GetInt32(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_get_i32(bucket_idx_, rec_idx_, idx);
}
int64_t RowView::GetInt64(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_get_i64(bucket_idx_, rec_idx_, idx);
}
float RowView::GetFloat32(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_get_f32(bucket_idx_, rec_idx_, idx);
}
double RowView::GetFloat64(size_t idx) const {
    CHECK_DATA("RowView");
    return data_->raw->sv_get_f64(bucket_idx_, rec_idx_, idx);
}

std::string_view RowView::GetString(size_t idx) const {
    CHECK_DATA("RowView");
    auto s = data_->raw->sv_get_str(bucket_idx_, rec_idx_, idx);
    return std::string_view(s.data(), s.size());
}

std::pair<const uint8_t*, size_t> RowView::GetBytes(size_t idx) const {
    CHECK_DATA("RowView");
    auto bytes = data_->raw->sv_get_bytes(bucket_idx_, rec_idx_, idx);
    return {bytes.data(), bytes.size()};
}

Date RowView::GetDate(size_t idx) const {
    CHECK_DATA("RowView");
    return Date{data_->raw->sv_get_date_days(bucket_idx_, rec_idx_, idx)};
}

Time RowView::GetTime(size_t idx) const {
    CHECK_DATA("RowView");
    return Time{data_->raw->sv_get_time_millis(bucket_idx_, rec_idx_, idx)};
}

Timestamp RowView::GetTimestamp(size_t idx) const {
    CHECK_DATA("RowView");
    return Timestamp{data_->raw->sv_get_ts_millis(bucket_idx_, rec_idx_, idx),
                     data_->raw->sv_get_ts_nanos(bucket_idx_, rec_idx_, idx)};
}

bool RowView::IsDecimal(size_t idx) const { return GetType(idx) == TypeId::Decimal; }

std::string RowView::GetDecimalString(size_t idx) const {
    CHECK_DATA("RowView");
    return std::string(data_->raw->sv_get_decimal_str(bucket_idx_, rec_idx_, idx));
}

Value RowView::GetValue(size_t idx) const {
    CHECK_DATA("RowView");
    auto box = data_->raw->sv_get_value(bucket_idx_, rec_idx_, idx);
    return Value(box.into_raw());
}

Value RowView::GetValue(const std::string& name) const { return GetValue(Resolve(name)); }

// ============================================================================
// PrefixLookupResult / PrefixRowView — read path for prefix lookups
// ============================================================================

size_t PrefixLookupResult::Size() const { return data_ ? data_->raw->plv_row_count() : 0; }

size_t PrefixRowView::FieldCount() const { return data_ ? data_->raw->plv_field_count() : 0; }

TypeId PrefixRowView::GetType(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return static_cast<TypeId>(data_->raw->plv_column_type(idx));
}

bool PrefixRowView::IsNull(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_is_null(rec_idx_, idx);
}
bool PrefixRowView::GetBool(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_get_bool(rec_idx_, idx);
}
int32_t PrefixRowView::GetInt32(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_get_i32(rec_idx_, idx);
}
int64_t PrefixRowView::GetInt64(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_get_i64(rec_idx_, idx);
}
float PrefixRowView::GetFloat32(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_get_f32(rec_idx_, idx);
}
double PrefixRowView::GetFloat64(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return data_->raw->plv_get_f64(rec_idx_, idx);
}
std::string_view PrefixRowView::GetString(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    auto s = data_->raw->plv_get_str(rec_idx_, idx);
    return std::string_view(s.data(), s.size());
}
std::pair<const uint8_t*, size_t> PrefixRowView::GetBytes(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    auto bytes = data_->raw->plv_get_bytes(rec_idx_, idx);
    return {bytes.data(), bytes.size()};
}
Date PrefixRowView::GetDate(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return Date{data_->raw->plv_get_date_days(rec_idx_, idx)};
}
Time PrefixRowView::GetTime(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return Time{data_->raw->plv_get_time_millis(rec_idx_, idx)};
}
Timestamp PrefixRowView::GetTimestamp(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return Timestamp{data_->raw->plv_get_ts_millis(rec_idx_, idx),
                     data_->raw->plv_get_ts_nanos(rec_idx_, idx)};
}
bool PrefixRowView::IsDecimal(size_t idx) const { return GetType(idx) == TypeId::Decimal; }
std::string PrefixRowView::GetDecimalString(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    return std::string(data_->raw->plv_get_decimal_str(rec_idx_, idx));
}

Value PrefixRowView::GetValue(size_t idx) const {
    CHECK_DATA("PrefixRowView");
    auto box = data_->raw->plv_get_value(rec_idx_, idx);
    return Value(box.into_raw());
}
Value PrefixRowView::GetValue(const std::string& name) const { return GetValue(Resolve(name)); }

// ============================================================================
// ScanRecords — backed by opaque Rust ScanResultInner
// ============================================================================

// ScanRecords constructor, destructor, move operations are all defaulted in the header.

size_t ScanRecords::Count() const { return data_ ? data_->raw->sv_record_count() : 0; }

bool ScanRecords::IsEmpty() const { return Count() == 0; }

ScanRecord ScanRecords::RecordAt(size_t bucket, size_t rec_idx) const {
    if (!data_) {
        throw std::logic_error("ScanRecords: not available (moved-from or null)");
    }
    return ScanRecord{data_->raw->sv_offset(bucket, rec_idx),
                      data_->raw->sv_timestamp(bucket, rec_idx),
                      static_cast<ChangeType>(data_->raw->sv_change_type(bucket, rec_idx)),
                      RowView(data_, bucket, rec_idx)};
}

static TableBucket to_table_bucket(const ffi::FfiBucketInfo& g) {
    return TableBucket{g.table_id, g.bucket_id,
                       g.has_partition_id ? std::optional<int64_t>(g.partition_id) : std::nullopt};
}

size_t ScanRecords::BucketCount() const { return data_ ? data_->raw->sv_bucket_infos().size() : 0; }

ScanRecord ScanRecords::Iterator::operator*() const {
    return owner_->RecordAt(bucket_idx_, rec_idx_);
}

ScanRecords::Iterator ScanRecords::begin() const { return Iterator(this, 0, 0); }

ScanRecords::Iterator& ScanRecords::Iterator::operator++() {
    ++rec_idx_;
    if (owner_->data_) {
        const auto& infos = owner_->data_->raw->sv_bucket_infos();
        while (bucket_idx_ < infos.size() && rec_idx_ >= infos[bucket_idx_].record_count) {
            rec_idx_ = 0;
            ++bucket_idx_;
        }
    }
    return *this;
}

std::vector<TableBucket> ScanRecords::Buckets() const {
    std::vector<TableBucket> result;
    if (!data_) return result;
    const auto& infos = data_->raw->sv_bucket_infos();
    result.reserve(infos.size());
    for (const auto& g : infos) {
        result.push_back(to_table_bucket(g));
    }
    return result;
}

BucketRecords ScanRecords::Records(const TableBucket& bucket) const {
    if (!data_) {
        return BucketRecords({}, bucket, 0, 0);
    }
    const auto& infos = data_->raw->sv_bucket_infos();
    for (size_t i = 0; i < infos.size(); ++i) {
        TableBucket tb = to_table_bucket(infos[i]);
        if (tb == bucket) {
            return BucketRecords(data_, std::move(tb), i, infos[i].record_count);
        }
    }
    return BucketRecords({}, bucket, 0, 0);
}

BucketRecords ScanRecords::BucketAt(size_t idx) const {
    if (!data_) {
        throw std::logic_error("ScanRecords: not available (moved-from or null)");
    }
    const auto& infos = data_->raw->sv_bucket_infos();
    if (idx >= infos.size()) {
        throw std::out_of_range("ScanRecords::BucketAt: index " + std::to_string(idx) +
                                " out of range (" + std::to_string(infos.size()) + " buckets)");
    }
    return BucketRecords(data_, to_table_bucket(infos[idx]), idx, infos[idx].record_count);
}

ScanRecord BucketRecords::operator[](size_t idx) const {
    if (idx >= count_) {
        throw std::out_of_range("BucketRecords: index " + std::to_string(idx) + " out of range (" +
                                std::to_string(count_) + " records)");
    }
    return ScanRecord{data_->raw->sv_offset(bucket_idx_, idx),
                      data_->raw->sv_timestamp(bucket_idx_, idx),
                      static_cast<ChangeType>(data_->raw->sv_change_type(bucket_idx_, idx)),
                      RowView(data_, bucket_idx_, idx)};
}

ScanRecord BucketRecords::Iterator::operator*() const { return owner_->operator[](idx_); }

// ============================================================================
// LookupResult — backed by opaque Rust LookupResultInner
// ============================================================================

LookupResult::LookupResult() noexcept = default;

LookupResult::~LookupResult() noexcept { Destroy(); }

void LookupResult::Destroy() noexcept {
    if (inner_) {
        rust::Box<ffi::LookupResultInner>::from_raw(inner_);
        inner_ = nullptr;
        column_map_.reset();
    }
}

LookupResult::LookupResult(LookupResult&& other) noexcept
    : inner_(other.inner_), column_map_(std::move(other.column_map_)) {
    other.inner_ = nullptr;
}

LookupResult& LookupResult::operator=(LookupResult&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        column_map_ = std::move(other.column_map_);
        other.inner_ = nullptr;
    }
    return *this;
}

void LookupResult::BuildColumnMap() const {
    if (!inner_) return;
    auto map = std::make_shared<detail::ColumnMap>();
    auto count = inner_->lv_field_count();
    for (size_t i = 0; i < count; ++i) {
        auto name = inner_->lv_column_name(i);
        (*map)[std::string(name.data(), name.size())] = {
            i, static_cast<TypeId>(inner_->lv_column_type(i))};
    }
    column_map_ = std::move(map);
}

bool LookupResult::Found() const { return inner_ && inner_->lv_found(); }

size_t LookupResult::FieldCount() const { return inner_ ? inner_->lv_field_count() : 0; }

TypeId LookupResult::GetType(size_t idx) const {
    CHECK_INNER("LookupResult");
    return static_cast<TypeId>(inner_->lv_column_type(idx));
}

bool LookupResult::IsNull(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_is_null(idx);
}
bool LookupResult::GetBool(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_get_bool(idx);
}
int32_t LookupResult::GetInt32(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_get_i32(idx);
}
int64_t LookupResult::GetInt64(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_get_i64(idx);
}
float LookupResult::GetFloat32(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_get_f32(idx);
}
double LookupResult::GetFloat64(size_t idx) const {
    CHECK_INNER("LookupResult");
    return inner_->lv_get_f64(idx);
}

std::string_view LookupResult::GetString(size_t idx) const {
    CHECK_INNER("LookupResult");
    auto s = inner_->lv_get_str(idx);
    return std::string_view(s.data(), s.size());
}

std::pair<const uint8_t*, size_t> LookupResult::GetBytes(size_t idx) const {
    CHECK_INNER("LookupResult");
    auto bytes = inner_->lv_get_bytes(idx);
    return {bytes.data(), bytes.size()};
}

Date LookupResult::GetDate(size_t idx) const {
    CHECK_INNER("LookupResult");
    return Date{inner_->lv_get_date_days(idx)};
}

Time LookupResult::GetTime(size_t idx) const {
    CHECK_INNER("LookupResult");
    return Time{inner_->lv_get_time_millis(idx)};
}

Timestamp LookupResult::GetTimestamp(size_t idx) const {
    CHECK_INNER("LookupResult");
    return Timestamp{inner_->lv_get_ts_millis(idx), inner_->lv_get_ts_nanos(idx)};
}

bool LookupResult::IsDecimal(size_t idx) const { return GetType(idx) == TypeId::Decimal; }

std::string LookupResult::GetDecimalString(size_t idx) const {
    CHECK_INNER("LookupResult");
    return std::string(inner_->lv_get_decimal_str(idx));
}

Value LookupResult::GetValue(size_t idx) const {
    CHECK_INNER("LookupResult");
    auto box = inner_->lv_get_value(idx);
    return Value(box.into_raw());
}

Value LookupResult::GetValue(const std::string& name) const {
    return GetValue(Resolve(name));
}

// ============================================================================
// Table
// ============================================================================

Table::Table() noexcept = default;

Table::Table(ffi::Table* table) noexcept : table_(table) {}

Table::~Table() noexcept { Destroy(); }

void Table::Destroy() noexcept {
    if (table_) {
        ffi::delete_table(table_);
        table_ = nullptr;
    }
}

Table::Table(Table&& other) noexcept
    : table_(other.table_), column_map_(std::move(other.column_map_)) {
    other.table_ = nullptr;
}

Table& Table::operator=(Table&& other) noexcept {
    if (this != &other) {
        Destroy();
        table_ = other.table_;
        column_map_ = std::move(other.column_map_);
        other.table_ = nullptr;
    }
    return *this;
}

bool Table::Available() const { return table_ != nullptr; }

TableAppend Table::NewAppend() { return TableAppend(table_); }

TableUpsert Table::NewUpsert() { return TableUpsert(table_); }

TableLookup Table::NewLookup() { return TableLookup(table_); }

TableScan Table::NewScan() { return TableScan(table_); }

Result Table::NewPrefixLookup(std::vector<std::string> lookup_columns, PrefixLookuper& out) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }

    rust::Vec<rust::String> cols;
    cols.reserve(lookup_columns.size());
    for (const auto& c : lookup_columns) {
        cols.push_back(rust::String(c));
    }

    auto ffi_result = table_->new_prefix_lookuper(std::move(cols));
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = PrefixLookuper(utils::ptr_from_ffi<ffi::PrefixLookuper>(ffi_result));
    }
    return result;
}

const std::shared_ptr<GenericRow::ColumnMap>& Table::GetColumnMap() const {
    if (!column_map_ && Available()) {
        auto info = GetTableInfo();
        column_map_ = std::make_shared<GenericRow::ColumnMap>();
        for (size_t i = 0; i < info.schema.columns.size(); ++i) {
            (*column_map_)[info.schema.columns[i].name] = {i,
                                                           info.schema.columns[i].data_type.id()};
        }
    }
    return column_map_;
}

GenericRow Table::NewRow() const {
    GenericRow row;
    row.column_map_ = GetColumnMap();
    return row;
}

TableInfo Table::GetTableInfo() const {
    if (!Available()) {
        return TableInfo{};
    }
    auto ffi_info = table_->get_table_info_from_table();
    return utils::from_ffi_table_info(ffi_info);
}

TablePath Table::GetTablePath() const {
    if (!Available()) {
        return TablePath{};
    }
    auto ffi_path = table_->get_table_path();
    return TablePath{std::string(ffi_path.database_name), std::string(ffi_path.table_name)};
}

bool Table::HasPrimaryKey() const {
    if (!Available()) {
        return false;
    }
    return table_->has_primary_key();
}

// ============================================================================
// TableAppend
// ============================================================================

TableAppend::TableAppend(ffi::Table* table) noexcept : table_(table) {}

Result TableAppend::CreateWriter(AppendWriter& out) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }

    auto ffi_result = table_->new_append_writer();
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = AppendWriter(utils::ptr_from_ffi<ffi::AppendWriter>(ffi_result));
    }
    return result;
}

// ============================================================================
// TableUpsert
// ============================================================================

TableUpsert::TableUpsert(ffi::Table* table) noexcept : table_(table) {}

TableUpsert& TableUpsert::PartialUpdateByIndex(std::vector<size_t> column_indices) {
    if (column_indices.empty()) {
        throw std::invalid_argument("PartialUpdateByIndex requires at least one column");
    }
    column_indices_ = std::move(column_indices);
    column_names_.clear();
    return *this;
}

TableUpsert& TableUpsert::PartialUpdateByName(std::vector<std::string> column_names) {
    if (column_names.empty()) {
        throw std::invalid_argument("PartialUpdateByName requires at least one column");
    }
    column_names_ = std::move(column_names);
    column_indices_.clear();
    return *this;
}

std::vector<size_t> TableUpsert::ResolveNameProjection() const {
    auto ffi_info = table_->get_table_info_from_table();
    const auto& columns = ffi_info.schema.columns;

    std::vector<size_t> indices;
    for (const auto& name : column_names_) {
        bool found = false;
        for (size_t i = 0; i < columns.size(); ++i) {
            if (std::string(columns[i].name) == name) {
                indices.push_back(i);
                found = true;
                break;
            }
        }
        if (!found) {
            throw std::runtime_error("Column '" + name + "' not found");
        }
    }
    return indices;
}

Result TableUpsert::CreateWriter(UpsertWriter& out) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }

    try {
        auto resolved_indices = !column_names_.empty() ? ResolveNameProjection() : column_indices_;

        rust::Vec<size_t> rust_indices;
        for (size_t idx : resolved_indices) {
            rust_indices.push_back(idx);
        }
        auto ffi_result = table_->create_upsert_writer(std::move(rust_indices));
        auto result = utils::from_ffi_result(ffi_result.result);
        if (result.Ok()) {
            out = UpsertWriter(utils::ptr_from_ffi<ffi::UpsertWriter>(ffi_result));
        }
        return result;
    } catch (const std::exception& e) {
        // ResolveNameProjection() may throw
        return utils::make_client_error(e.what());
    }
}

// ============================================================================
// TableLookup
// ============================================================================

TableLookup::TableLookup(ffi::Table* table) noexcept : table_(table) {}

Result TableLookup::CreateLookuper(Lookuper& out) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }

    auto ffi_result = table_->new_lookuper();
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = Lookuper(utils::ptr_from_ffi<ffi::Lookuper>(ffi_result));
    }
    return result;
}

// ============================================================================
// TableScan
// ============================================================================

TableScan::TableScan(ffi::Table* table) noexcept : table_(table) {}

TableScan& TableScan::ProjectByIndex(std::vector<size_t> column_indices) {
    projection_ = std::move(column_indices);
    name_projection_.clear();
    return *this;
}

TableScan& TableScan::ProjectByName(std::vector<std::string> column_names) {
    name_projection_ = std::move(column_names);
    projection_.clear();
    return *this;
}

std::vector<size_t> TableScan::ResolveNameProjection() const {
    auto ffi_info = table_->get_table_info_from_table();
    const auto& columns = ffi_info.schema.columns;

    std::vector<size_t> indices;
    for (const auto& name : name_projection_) {
        bool found = false;
        for (size_t i = 0; i < columns.size(); ++i) {
            if (std::string(columns[i].name) == name) {
                indices.push_back(i);
                found = true;
                break;
            }
        }
        if (!found) {
            throw std::runtime_error("Column '" + name + "' not found");
        }
    }
    return indices;
}

Result TableScan::CreateLogScanner(LogScanner& out) { return DoCreateScanner(out, false); }

Result TableScan::CreateRecordBatchLogScanner(LogScanner& out) {
    return DoCreateScanner(out, true);
}

Result TableScan::DoCreateScanner(LogScanner& out, bool is_record_batch) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }
    if (limit_.has_value()) {
        return utils::make_client_error(
            "Log scanners don't support limit pushdown; use CreateBucketBatchScanner()");
    }

    try {
        auto resolved_indices = !name_projection_.empty() ? ResolveNameProjection() : projection_;
        rust::Vec<size_t> rust_indices;
        for (size_t idx : resolved_indices) {
            rust_indices.push_back(idx);
        }
        auto ffi_result = table_->create_scanner(std::move(rust_indices), is_record_batch);
        auto result = utils::from_ffi_result(ffi_result.result);
        if (result.Ok()) {
            out.scanner_ = utils::ptr_from_ffi<ffi::LogScanner>(ffi_result);
        }
        return result;
    } catch (const std::exception& e) {
        // ResolveNameProjection() may throw
        return utils::make_client_error(e.what());
    }
}

TableScan& TableScan::Limit(int32_t row_number) {
    limit_ = row_number;
    return *this;
}

Result TableScan::CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out) {
    if (table_ == nullptr) {
        return utils::make_client_error("Table not available");
    }
    if (!limit_.has_value()) {
        return utils::make_client_error(
            "CreateBucketBatchScanner requires a limit set via Limit()");
    }

    try {
        auto resolved_indices = !name_projection_.empty() ? ResolveNameProjection() : projection_;
        rust::Vec<size_t> rust_indices;
        for (size_t idx : resolved_indices) {
            rust_indices.push_back(idx);
        }
        int64_t partition_id = bucket.partition_id.value_or(-1);
        auto ffi_result = table_->create_bucket_batch_scanner(
            std::move(rust_indices), *limit_, bucket.table_id, partition_id, bucket.bucket_id);
        auto result = utils::from_ffi_result(ffi_result.result);
        if (result.Ok()) {
            out.scanner_ = utils::ptr_from_ffi<ffi::BatchScanner>(ffi_result);
            out.bucket_ = bucket;
        }
        return result;
    } catch (const std::exception& e) {
        // ResolveNameProjection() may throw
        return utils::make_client_error(e.what());
    }
}

// ============================================================================
// WriteResult
// ============================================================================

WriteResult::WriteResult() noexcept = default;

WriteResult::WriteResult(ffi::WriteResult* inner) noexcept : inner_(inner) {}

WriteResult::~WriteResult() noexcept { Destroy(); }

void WriteResult::Destroy() noexcept {
    if (inner_) {
        ffi::delete_write_result(inner_);
        inner_ = nullptr;
    }
}

WriteResult::WriteResult(WriteResult&& other) noexcept : inner_(other.inner_) {
    other.inner_ = nullptr;
}

WriteResult& WriteResult::operator=(WriteResult&& other) noexcept {
    if (this != &other) {
        Destroy();
        inner_ = other.inner_;
        other.inner_ = nullptr;
    }
    return *this;
}

bool WriteResult::Available() const { return inner_ != nullptr; }

Result WriteResult::Wait() {
    if (!Available()) {
        return utils::make_ok();
    }

    auto ffi_result = inner_->wait();
    return utils::from_ffi_result(ffi_result);
}

// ============================================================================
// AppendWriter
// ============================================================================

AppendWriter::AppendWriter() noexcept = default;

AppendWriter::AppendWriter(ffi::AppendWriter* writer) noexcept : writer_(writer) {}

AppendWriter::~AppendWriter() noexcept { Destroy(); }

void AppendWriter::Destroy() noexcept {
    if (writer_) {
        ffi::delete_append_writer(writer_);
        writer_ = nullptr;
    }
}

AppendWriter::AppendWriter(AppendWriter&& other) noexcept : writer_(other.writer_) {
    other.writer_ = nullptr;
}

AppendWriter& AppendWriter::operator=(AppendWriter&& other) noexcept {
    if (this != &other) {
        Destroy();
        writer_ = other.writer_;
        other.writer_ = nullptr;
    }
    return *this;
}

bool AppendWriter::Available() const { return writer_ != nullptr; }

Result AppendWriter::Append(const GenericRow& row) {
    WriteResult wr;
    return Append(row, wr);
}

Result AppendWriter::Append(const GenericRow& row, WriteResult& out) {
    if (!Available()) {
        return utils::make_client_error("AppendWriter not available");
    }
    if (!row.Available()) {
        return utils::make_client_error("GenericRow not available");
    }

    auto ffi_result = writer_->append(*row.inner_);
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = WriteResult(utils::ptr_from_ffi<ffi::WriteResult>(ffi_result));
    }
    return result;
}

Result AppendWriter::AppendArrowBatch(const std::shared_ptr<arrow::RecordBatch>& batch) {
    WriteResult wr;
    return AppendArrowBatch(batch, wr);
}

Result AppendWriter::AppendArrowBatch(const std::shared_ptr<arrow::RecordBatch>& batch,
                                      WriteResult& out) {
    if (!Available()) {
        return utils::make_client_error("AppendWriter not available");
    }
    if (!batch) {
        return utils::make_client_error("Arrow RecordBatch is null");
    }

    // Export via Arrow C Data Interface
    struct ArrowArray c_array;
    struct ArrowSchema c_schema;
    auto status = arrow::ExportRecordBatch(*batch, &c_array, &c_schema);
    if (!status.ok()) {
        return utils::make_client_error("Failed to export Arrow batch: " + status.ToString());
    }

    // Heap-allocate for Rust ownership transfer
    auto* array_heap = new ArrowArray(std::move(c_array));
    auto* schema_heap = new ArrowSchema(std::move(c_schema));

    // Rust takes ownership of both pointers immediately via Box::from_raw(),
    // so after this call C++ must NOT free them.
    auto ffi_result = writer_->append_arrow_batch(reinterpret_cast<size_t>(array_heap),
                                                  reinterpret_cast<size_t>(schema_heap));
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out.Destroy();
        out.inner_ = utils::ptr_from_ffi<ffi::WriteResult>(ffi_result);
    }
    return result;
}

Result AppendWriter::Flush() {
    if (!Available()) {
        return utils::make_client_error("AppendWriter not available");
    }

    auto ffi_result = writer_->flush();
    return utils::from_ffi_result(ffi_result);
}

// ============================================================================
// UpsertWriter
// ============================================================================

UpsertWriter::UpsertWriter() noexcept = default;

UpsertWriter::UpsertWriter(ffi::UpsertWriter* writer) noexcept : writer_(writer) {}

UpsertWriter::~UpsertWriter() noexcept { Destroy(); }

void UpsertWriter::Destroy() noexcept {
    if (writer_) {
        ffi::delete_upsert_writer(writer_);
        writer_ = nullptr;
    }
}

UpsertWriter::UpsertWriter(UpsertWriter&& other) noexcept : writer_(other.writer_) {
    other.writer_ = nullptr;
}

UpsertWriter& UpsertWriter::operator=(UpsertWriter&& other) noexcept {
    if (this != &other) {
        Destroy();
        writer_ = other.writer_;
        other.writer_ = nullptr;
    }
    return *this;
}

bool UpsertWriter::Available() const { return writer_ != nullptr; }

Result UpsertWriter::Upsert(const GenericRow& row) {
    WriteResult wr;
    return Upsert(row, wr);
}

Result UpsertWriter::Upsert(const GenericRow& row, WriteResult& out) {
    if (!Available()) {
        return utils::make_client_error("UpsertWriter not available");
    }
    if (!row.Available()) {
        return utils::make_client_error("GenericRow not available");
    }

    auto ffi_result = writer_->upsert(*row.inner_);
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = WriteResult(utils::ptr_from_ffi<ffi::WriteResult>(ffi_result));
    }
    return result;
}

Result UpsertWriter::Delete(const GenericRow& row) {
    WriteResult wr;
    return Delete(row, wr);
}

Result UpsertWriter::Delete(const GenericRow& row, WriteResult& out) {
    if (!Available()) {
        return utils::make_client_error("UpsertWriter not available");
    }
    if (!row.Available()) {
        return utils::make_client_error("GenericRow not available");
    }

    auto ffi_result = writer_->delete_row(*row.inner_);
    auto result = utils::from_ffi_result(ffi_result.result);
    if (result.Ok()) {
        out = WriteResult(utils::ptr_from_ffi<ffi::WriteResult>(ffi_result));
    }
    return result;
}

Result UpsertWriter::Flush() {
    if (!Available()) {
        return utils::make_client_error("UpsertWriter not available");
    }

    auto ffi_result = writer_->upsert_flush();
    return utils::from_ffi_result(ffi_result);
}

// ============================================================================
// Lookuper
// ============================================================================

Lookuper::Lookuper() noexcept = default;

Lookuper::Lookuper(ffi::Lookuper* lookuper) noexcept : lookuper_(lookuper) {}

Lookuper::~Lookuper() noexcept { Destroy(); }

void Lookuper::Destroy() noexcept {
    if (lookuper_) {
        ffi::delete_lookuper(lookuper_);
        lookuper_ = nullptr;
    }
}

Lookuper::Lookuper(Lookuper&& other) noexcept : lookuper_(other.lookuper_) {
    other.lookuper_ = nullptr;
}

Lookuper& Lookuper::operator=(Lookuper&& other) noexcept {
    if (this != &other) {
        Destroy();
        lookuper_ = other.lookuper_;
        other.lookuper_ = nullptr;
    }
    return *this;
}

bool Lookuper::Available() const { return lookuper_ != nullptr; }

Result Lookuper::Lookup(const GenericRow& pk_row, LookupResult& out) {
    if (!Available()) {
        return utils::make_client_error("Lookuper not available");
    }
    if (!pk_row.Available()) {
        return utils::make_client_error("GenericRow not available");
    }

    auto result_box = lookuper_->lookup(*pk_row.inner_);
    if (result_box->lv_has_error()) {
        return utils::make_error(result_box->lv_error_code(),
                                 std::string(result_box->lv_error_message()));
    }

    out.Destroy();
    out.inner_ = result_box.into_raw();
    return utils::make_ok();
}

// ============================================================================
// PrefixLookuper
// ============================================================================

PrefixLookuper::PrefixLookuper() noexcept = default;

PrefixLookuper::PrefixLookuper(ffi::PrefixLookuper* lookuper) noexcept : lookuper_(lookuper) {}

PrefixLookuper::~PrefixLookuper() noexcept { Destroy(); }

void PrefixLookuper::Destroy() noexcept {
    if (lookuper_) {
        ffi::delete_prefix_lookuper(lookuper_);
        lookuper_ = nullptr;
    }
}

PrefixLookuper::PrefixLookuper(PrefixLookuper&& other) noexcept : lookuper_(other.lookuper_) {
    other.lookuper_ = nullptr;
}

PrefixLookuper& PrefixLookuper::operator=(PrefixLookuper&& other) noexcept {
    if (this != &other) {
        Destroy();
        lookuper_ = other.lookuper_;
        other.lookuper_ = nullptr;
    }
    return *this;
}

bool PrefixLookuper::Available() const { return lookuper_ != nullptr; }

Result PrefixLookuper::PrefixLookup(const GenericRow& prefix_row, PrefixLookupResult& out) {
    if (!Available()) {
        return utils::make_client_error("PrefixLookuper not available");
    }
    if (!prefix_row.Available()) {
        return utils::make_client_error("GenericRow not available");
    }

    auto result_box = lookuper_->prefix_lookup(*prefix_row.inner_);
    if (result_box->plv_has_error()) {
        return utils::make_error(result_box->plv_error_code(),
                                 std::string(result_box->plv_error_message()));
    }

    // Take ownership of the FFI box first (~PrefixData calls from_raw), so the
    // column-map loop below can't leak it if a string/map allocation throws.
    // The map is built eagerly and shared by all PrefixRowViews.
    auto data = std::make_shared<detail::PrefixData>(result_box.into_raw(), detail::ColumnMap{});
    auto col_count = data->raw->plv_field_count();
    for (size_t i = 0; i < col_count; ++i) {
        auto name = data->raw->plv_column_name(i);
        data->columns[std::string(name.data(), name.size())] = {
            i, static_cast<TypeId>(data->raw->plv_column_type(i))};
    }
    out.data_ = std::move(data);
    return utils::make_ok();
}

// ============================================================================
// LogScanner
// ============================================================================

LogScanner::LogScanner() noexcept = default;

LogScanner::LogScanner(ffi::LogScanner* scanner) noexcept : scanner_(scanner) {}

LogScanner::~LogScanner() noexcept { Destroy(); }

void LogScanner::Destroy() noexcept {
    if (scanner_) {
        ffi::delete_log_scanner(scanner_);
        scanner_ = nullptr;
    }
}

LogScanner::LogScanner(LogScanner&& other) noexcept : scanner_(other.scanner_) {
    other.scanner_ = nullptr;
}

LogScanner& LogScanner::operator=(LogScanner&& other) noexcept {
    if (this != &other) {
        Destroy();
        scanner_ = other.scanner_;
        other.scanner_ = nullptr;
    }
    return *this;
}

bool LogScanner::Available() const { return scanner_ != nullptr; }

Result LogScanner::Subscribe(int32_t bucket_id, int64_t start_offset) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto ffi_result = scanner_->subscribe(bucket_id, start_offset);
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::Subscribe(const std::vector<BucketSubscription>& bucket_offsets) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    rust::Vec<ffi::FfiBucketSubscription> rust_subs;
    for (const auto& sub : bucket_offsets) {
        ffi::FfiBucketSubscription ffi_sub;
        ffi_sub.bucket_id = sub.bucket_id;
        ffi_sub.offset = sub.offset;
        rust_subs.push_back(ffi_sub);
    }

    auto ffi_result = scanner_->subscribe_buckets(std::move(rust_subs));
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::SubscribePartitionBuckets(int64_t partition_id, int32_t bucket_id,
                                             int64_t start_offset) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto ffi_result = scanner_->subscribe_partition(partition_id, bucket_id, start_offset);
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::SubscribePartitionBuckets(
    const std::vector<PartitionBucketSubscription>& subscriptions) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    rust::Vec<ffi::FfiPartitionBucketSubscription> rust_subs;
    for (const auto& sub : subscriptions) {
        ffi::FfiPartitionBucketSubscription ffi_sub;
        ffi_sub.partition_id = sub.partition_id;
        ffi_sub.bucket_id = sub.bucket_id;
        ffi_sub.offset = sub.offset;
        rust_subs.push_back(ffi_sub);
    }

    auto ffi_result = scanner_->subscribe_partition_buckets(std::move(rust_subs));
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::Unsubscribe(int32_t bucket_id) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto ffi_result = scanner_->unsubscribe(bucket_id);
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::UnsubscribePartition(int64_t partition_id, int32_t bucket_id) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto ffi_result = scanner_->unsubscribe_partition(partition_id, bucket_id);
    return utils::from_ffi_result(ffi_result);
}

Result LogScanner::Poll(int64_t timeout_ms, ScanRecords& out) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto result_box = scanner_->poll(timeout_ms);
    if (result_box->sv_has_error()) {
        return utils::make_error(result_box->sv_error_code(),
                                 std::string(result_box->sv_error_message()));
    }

    // Wrap raw pointer in ScanData immediately so it's never leaked on exception.
    auto data = std::make_shared<detail::ScanData>(result_box.into_raw(), detail::ColumnMap{});
    // Build column map eagerly — shared by all RowViews/BucketRecords.
    auto col_count = data->raw->sv_column_count();
    for (size_t i = 0; i < col_count; ++i) {
        auto name = data->raw->sv_column_name(i);
        data->columns[std::string(name.data(), name.size())] = {
            i, static_cast<TypeId>(data->raw->sv_column_type(i))};
    }
    out.data_ = std::move(data);
    return utils::make_ok();
}

ArrowRecordBatch::ArrowRecordBatch(std::shared_ptr<arrow::RecordBatch> batch, int64_t table_id,
                                   int64_t partition_id, int32_t bucket_id,
                                   int64_t base_offset) noexcept
    : batch_(std::move(batch)),
      table_id_(table_id),
      partition_id_(partition_id),
      bucket_id_(bucket_id),
      base_offset_(base_offset) {}

bool ArrowRecordBatch::Available() const { return batch_ != nullptr; }

int64_t ArrowRecordBatch::NumRows() const {
    if (!Available()) return 0;
    return batch_->num_rows();
}

int64_t ArrowRecordBatch::GetTableId() const {
    if (!Available()) return 0;
    return this->table_id_;
}

int64_t ArrowRecordBatch::GetPartitionId() const {
    if (!Available()) return -1;
    return this->partition_id_;
}

int32_t ArrowRecordBatch::GetBucketId() const {
    if (!Available()) return -1;
    return this->bucket_id_;
}

int64_t ArrowRecordBatch::GetBaseOffset() const {
    if (!Available()) return -1;
    return this->base_offset_;
}

int64_t ArrowRecordBatch::GetLastOffset() const {
    if (!Available()) return -1;
    return this->base_offset_ + this->NumRows() - 1;
}

namespace detail {
// Imports FFI Arrow batches (C Data Interface) into C++ ArrowRecordBatch
// wrappers. A friend of ArrowRecordBatch so it can build the wrapper's private
// ctor; shared by LogScanner::PollRecordBatch and BatchScanner.
struct ArrowBatchImporter {
    static Result Import(ffi::FfiArrowRecordBatches& src, ArrowRecordBatches& out) {
        out.batches.clear();
        for (const auto& ffi_batch : src.batches) {
            auto* c_array = reinterpret_cast<struct ArrowArray*>(ffi_batch.array_ptr);
            auto* c_schema = reinterpret_cast<struct ArrowSchema*>(ffi_batch.schema_ptr);

            auto import_result = arrow::ImportRecordBatch(c_array, c_schema);
            // Free the Rust-allocated container structures whether or not the
            // import succeeded, to avoid leaks.
            ffi::free_arrow_ffi_structures(ffi_batch.array_ptr, ffi_batch.schema_ptr);
            if (!import_result.ok()) {
                return utils::make_client_error("Failed to import Arrow record batch: " +
                                                import_result.status().ToString());
            }
            out.batches.push_back(std::unique_ptr<ArrowRecordBatch>(new ArrowRecordBatch(
                import_result.ValueOrDie(), ffi_batch.table_id, ffi_batch.partition_id,
                ffi_batch.bucket_id, ffi_batch.base_offset)));
        }
        return utils::make_ok();
    }
};
}  // namespace detail

Result LogScanner::PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) {
    if (!Available()) {
        return utils::make_client_error("LogScanner not available");
    }

    auto ffi_result = scanner_->poll_record_batch(timeout_ms);
    auto result = utils::from_ffi_result(ffi_result.result);
    if (!result.Ok()) {
        return result;
    }
    return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out);
}

// ============================================================================
// BatchScanner
// ============================================================================

BatchScanner::BatchScanner() noexcept = default;

BatchScanner::BatchScanner(ffi::BatchScanner* scanner) noexcept : scanner_(scanner) {}

BatchScanner::~BatchScanner() noexcept { Destroy(); }

void BatchScanner::Destroy() noexcept {
    if (scanner_) {
        ffi::delete_batch_scanner(scanner_);
        scanner_ = nullptr;
    }
}

BatchScanner::BatchScanner(BatchScanner&& other) noexcept
    : scanner_(other.scanner_), bucket_(other.bucket_) {
    other.scanner_ = nullptr;
}

BatchScanner& BatchScanner::operator=(BatchScanner&& other) noexcept {
    if (this != &other) {
        Destroy();
        scanner_ = other.scanner_;
        bucket_ = other.bucket_;
        other.scanner_ = nullptr;
    }
    return *this;
}

bool BatchScanner::Available() const { return scanner_ != nullptr; }

Result BatchScanner::NextBatch(ArrowRecordBatches& out) {
    if (!Available()) {
        return utils::make_client_error("BatchScanner not available");
    }
    auto ffi_result = scanner_->next_batch();
    auto result = utils::from_ffi_result(ffi_result.result);
    if (!result.Ok()) {
        return result;
    }
    return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out);
}

Result BatchScanner::CollectAllBatches(ArrowRecordBatches& out) {
    if (!Available()) {
        return utils::make_client_error("BatchScanner not available");
    }
    auto ffi_result = scanner_->collect_all_batches();
    auto result = utils::from_ffi_result(ffi_result.result);
    if (!result.Ok()) {
        return result;
    }
    return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out);
}

}  // namespace fluss