/*
 * 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/api.h>
#include <gtest/gtest.h>

#include <algorithm>
#include <chrono>
#include <cmath>
#include <limits>
#include <thread>
#include <tuple>

#include "test_utils.h"

class LogTableTest : public ::testing::Test {
   protected:
    fluss::Admin& admin() { return fluss_test::FlussTestEnvironment::Instance()->GetAdmin(); }

    fluss::Connection& connection() {
        return fluss_test::FlussTestEnvironment::Instance()->GetConnection();
    }
};

TEST_F(LogTableTest, AppendRecordBatchAndScan) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_append_record_batch_and_scan_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("c1", fluss::DataType::Int())
                      .AddColumn("c2", fluss::DataType::String())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(3)
                                .SetBucketKeys({"c1"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    // Create append writer
    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));

    // Append Arrow record batches
    {
        auto c1 = arrow::Int32Builder();
        c1.AppendValues({1, 2, 3}).ok();
        auto c2 = arrow::StringBuilder();
        c2.AppendValues({"a1", "a2", "a3"}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("c1", arrow::int32()), arrow::field("c2", arrow::utf8())}),
            3, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }

    {
        auto c1 = arrow::Int32Builder();
        c1.AppendValues({4, 5, 6}).ok();
        auto c2 = arrow::StringBuilder();
        c2.AppendValues({"a4", "a5", "a6"}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("c1", arrow::int32()), arrow::field("c2", arrow::utf8())}),
            3, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }

    ASSERT_OK(append_writer.Flush());

    // Create scanner and subscribe to all 3 buckets
    fluss::Table scan_table;
    ASSERT_OK(conn.GetTable(table_path, scan_table));
    int32_t num_buckets = scan_table.GetTableInfo().num_buckets;
    ASSERT_EQ(num_buckets, 3) << "Table should have 3 buckets";

    auto table_scan = scan_table.NewScan();
    fluss::LogScanner log_scanner;
    ASSERT_OK(table_scan.CreateLogScanner(log_scanner));

    for (int32_t bucket_id = 0; bucket_id < num_buckets; ++bucket_id) {
        ASSERT_OK(log_scanner.Subscribe(bucket_id, fluss::EARLIEST_OFFSET));
    }

    // Poll for records across all buckets
    std::vector<std::pair<int32_t, std::string>> records;
    fluss_test::PollRecords(log_scanner, 6, [](const fluss::ScanRecord& rec) {
        return std::make_pair(rec.row.GetInt32(0), std::string(rec.row.GetString(1)));
    }, records);
    ASSERT_EQ(records.size(), 6u) << "Expected 6 records";
    std::sort(records.begin(), records.end());

    std::vector<std::pair<int32_t, std::string>> expected = {
        {1, "a1"}, {2, "a2"}, {3, "a3"}, {4, "a4"}, {5, "a5"}, {6, "a6"}};
    EXPECT_EQ(records, expected);

    // Verify per-bucket iteration via BucketRecords
    {
        fluss::Table bucket_table;
        ASSERT_OK(conn.GetTable(table_path, bucket_table));
        auto bucket_scan = bucket_table.NewScan();
        fluss::LogScanner bucket_scanner;
        ASSERT_OK(bucket_scan.CreateLogScanner(bucket_scanner));

        for (int32_t bid = 0; bid < num_buckets; ++bid) {
            ASSERT_OK(bucket_scanner.Subscribe(bid, fluss::EARLIEST_OFFSET));
        }

        std::vector<std::pair<int32_t, std::string>> bucket_records;
        auto bucket_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
        size_t buckets_with_data = 0;
        while (bucket_records.size() < 6 && std::chrono::steady_clock::now() < bucket_deadline) {
            fluss::ScanRecords scan_records;
            ASSERT_OK(bucket_scanner.Poll(500, scan_records));

            // Iterate by bucket
            for (size_t b = 0; b < scan_records.BucketCount(); ++b) {
                auto bkt_records = scan_records.BucketAt(b);
                if (!bkt_records.Empty()) {
                    buckets_with_data++;
                }
                for (auto rec : bkt_records) {
                    bucket_records.emplace_back(rec.row.GetInt32(0),
                                                std::string(rec.row.GetString(1)));
                }
            }
        }

        ASSERT_EQ(bucket_records.size(), 6u) << "Expected 6 records via per-bucket iteration";
        EXPECT_GT(buckets_with_data, 1u) << "Records should be distributed across multiple buckets";

        std::sort(bucket_records.begin(), bucket_records.end());
        EXPECT_EQ(bucket_records, expected);
    }

    // Test unsubscribe
    ASSERT_OK(log_scanner.Unsubscribe(0));

    // Verify unsubscribe_partition fails on a non-partitioned table
    auto unsub_result = log_scanner.UnsubscribePartition(0, 0);
    ASSERT_FALSE(unsub_result.Ok())
        << "unsubscribe_partition should fail on a non-partitioned table";

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, LimitScan) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_limit_scan_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("c1", fluss::DataType::Int())
                      .AddColumn("c2", fluss::DataType::String())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"c1"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));
    {
        auto c1 = arrow::Int32Builder();
        c1.AppendValues({1, 2, 3, 4, 5}).ok();
        auto c2 = arrow::StringBuilder();
        c2.AppendValues({"a", "b", "c", "d", "e"}).ok();
        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("c1", arrow::int32()), arrow::field("c2", arrow::utf8())}),
            5, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie()});
        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }
    ASSERT_OK(append_writer.Flush());

    int64_t table_id = table.GetTableInfo().table_id;
    fluss::TableBucket bucket{table_id, 0};

    fluss::BatchScanner scanner;
    ASSERT_OK(table.NewScan().Limit(3).CreateBucketBatchScanner(bucket, scanner));
    EXPECT_TRUE(scanner.Bucket() == bucket);

    fluss::ArrowRecordBatches first;
    ASSERT_OK(scanner.NextBatch(first));
    ASSERT_FALSE(first.Empty()) << "first NextBatch should return a batch";
    int64_t rows = 0;
    for (const auto& b : first) {
        EXPECT_EQ(b->GetTableId(), table_id);
        EXPECT_EQ(b->GetBucketId(), 0);
        rows += b->NumRows();
    }
    // The server may return fewer rows than the limit, but never more.
    EXPECT_GT(rows, 0);
    EXPECT_LE(rows, 3);

    fluss::ArrowRecordBatches spent;
    ASSERT_OK(scanner.NextBatch(spent));
    EXPECT_TRUE(spent.Empty()) << "scanner must be spent after one batch";

    fluss::BatchScanner scanner2;
    ASSERT_OK(table.NewScan().Limit(10).CreateBucketBatchScanner(bucket, scanner2));
    fluss::ArrowRecordBatches all;
    ASSERT_OK(scanner2.CollectAllBatches(all));
    int64_t total = 0;
    for (const auto& b : all) {
        total += b->NumRows();
    }
    EXPECT_EQ(total, 5) << "limit 10 over a 5-row bucket returns all rows";

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, LimitScanProjection) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_limit_scan_projection_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("c1", fluss::DataType::Int())
                      .AddColumn("c2", fluss::DataType::String())
                      .AddColumn("c3", fluss::DataType::BigInt())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"c1"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));
    {
        auto c1 = arrow::Int32Builder();
        c1.AppendValues({1, 2}).ok();
        auto c2 = arrow::StringBuilder();
        c2.AppendValues({"a", "b"}).ok();
        auto c3 = arrow::Int64Builder();
        c3.AppendValues({100, 200}).ok();
        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("c1", arrow::int32()), arrow::field("c2", arrow::utf8()),
                           arrow::field("c3", arrow::int64())}),
            2, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie(), c3.Finish().ValueOrDie()});
        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }
    ASSERT_OK(append_writer.Flush());

    fluss::TableBucket bucket{table.GetTableInfo().table_id, 0};

    // Projecting [c1, c3] skips the middle c2 string column.
    fluss::BatchScanner scanner;
    ASSERT_OK(
        table.NewScan().ProjectByIndex({0, 2}).Limit(10).CreateBucketBatchScanner(bucket, scanner));
    fluss::ArrowRecordBatches all;
    ASSERT_OK(scanner.CollectAllBatches(all));
    ASSERT_EQ(all.Size(), 1u);

    auto rb = all[0]->GetArrowRecordBatch();
    ASSERT_EQ(rb->num_columns(), 2);
    EXPECT_EQ(rb->schema()->field(0)->name(), "c1");
    EXPECT_EQ(rb->schema()->field(1)->name(), "c3");
    ASSERT_EQ(rb->num_rows(), 2);
    auto c1a = std::static_pointer_cast<arrow::Int32Array>(rb->column(0));
    auto c3a = std::static_pointer_cast<arrow::Int64Array>(rb->column(1));
    EXPECT_EQ(c1a->Value(0), 1);
    EXPECT_EQ(c1a->Value(1), 2);
    EXPECT_EQ(c3a->Value(0), 100);
    EXPECT_EQ(c3a->Value(1), 200);

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, LimitScanErrors) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_limit_scan_errors_cpp");

    auto schema = fluss::Schema::NewBuilder().AddColumn("c1", fluss::DataType::Int()).Build();
    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"c1"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    int64_t table_id = table.GetTableInfo().table_id;

    {
        fluss::BatchScanner s;
        EXPECT_FALSE(
            table.NewScan().CreateBucketBatchScanner(fluss::TableBucket{table_id, 0}, s).Ok());
    }
    for (int32_t bad : {0, -5}) {
        fluss::BatchScanner s;
        EXPECT_FALSE(table.NewScan()
                         .Limit(bad)
                         .CreateBucketBatchScanner(fluss::TableBucket{table_id, 0}, s)
                         .Ok());
    }
    for (const auto& bad :
         {fluss::TableBucket{table_id + 9999, 0}, fluss::TableBucket{table_id, 99}}) {
        fluss::BatchScanner s;
        EXPECT_FALSE(table.NewScan().Limit(1).CreateBucketBatchScanner(bad, s).Ok());
    }
    {
        fluss::LogScanner s;
        EXPECT_FALSE(table.NewScan().Limit(5).CreateLogScanner(s).Ok());
        fluss::LogScanner s2;
        EXPECT_FALSE(table.NewScan().Limit(5).CreateRecordBatchLogScanner(s2).Ok());
    }
    ASSERT_OK(adm.DropTable(table_path, false));

    // A non-ARROW (INDEXED) log table rejects a limit scan.
    fluss::TablePath indexed_path("fluss", "test_limit_scan_indexed_cpp");
    auto indexed_descriptor = fluss::TableDescriptor::NewBuilder()
                                  .SetSchema(schema)
                                  .SetBucketCount(1)
                                  .SetBucketKeys({"c1"})
                                  .SetLogFormat("INDEXED")
                                  .SetProperty("table.replication.factor", "1")
                                  .Build();
    fluss_test::CreateTable(adm, indexed_path, indexed_descriptor);
    fluss::Table indexed_table;
    ASSERT_OK(conn.GetTable(indexed_path, indexed_table));
    {
        fluss::BatchScanner s;
        fluss::TableBucket b{indexed_table.GetTableInfo().table_id, 0};
        EXPECT_FALSE(indexed_table.NewScan().Limit(1).CreateBucketBatchScanner(b, s).Ok());
    }
    ASSERT_OK(adm.DropTable(indexed_path, false));
}

TEST_F(LogTableTest, ListOffsets) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_list_offsets_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    // Wait for table initialization
    std::this_thread::sleep_for(std::chrono::seconds(2));

    // Earliest offset should be 0 for empty table
    std::unordered_map<int32_t, int64_t> earliest_offsets;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Earliest(), earliest_offsets));
    EXPECT_EQ(earliest_offsets[0], 0) << "Earliest offset should be 0 for bucket 0";

    // Latest offset should be 0 for empty table
    std::unordered_map<int32_t, int64_t> latest_offsets;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Latest(), latest_offsets));
    EXPECT_EQ(latest_offsets[0], 0) << "Latest offset should be 0 for empty table";

    auto before_append_ms =
        std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch())
            .count();

    // Append records
    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));

    {
        auto id_builder = arrow::Int32Builder();
        id_builder.AppendValues({1, 2, 3}).ok();
        auto name_builder = arrow::StringBuilder();
        name_builder.AppendValues({"alice", "bob", "charlie"}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema(
                {arrow::field("id", arrow::int32()), arrow::field("name", arrow::utf8())}),
            3, {id_builder.Finish().ValueOrDie(), name_builder.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }
    ASSERT_OK(append_writer.Flush());

    std::this_thread::sleep_for(std::chrono::seconds(1));

    auto after_append_ms =
        std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::system_clock::now().time_since_epoch())
            .count();

    // Latest offset after appending should be 3
    std::unordered_map<int32_t, int64_t> latest_after;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Latest(), latest_after));
    EXPECT_EQ(latest_after[0], 3) << "Latest offset should be 3 after appending 3 records";

    // Earliest offset should still be 0
    std::unordered_map<int32_t, int64_t> earliest_after;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Earliest(), earliest_after));
    EXPECT_EQ(earliest_after[0], 0) << "Earliest offset should still be 0";

    // Timestamp before append should resolve to offset 0
    std::unordered_map<int32_t, int64_t> ts_offsets;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Timestamp(before_append_ms),
                              ts_offsets));
    EXPECT_EQ(ts_offsets[0], 0)
        << "Timestamp before append should resolve to offset 0";

    // Timestamp after append should resolve to offset 3
    std::unordered_map<int32_t, int64_t> ts_after_offsets;
    ASSERT_OK(adm.ListOffsets(table_path, {0}, fluss::OffsetSpec::Timestamp(after_append_ms),
                              ts_after_offsets));
    EXPECT_EQ(ts_after_offsets[0], 3)
        << "Timestamp after append should resolve to offset 3";

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, TestProject) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_project_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("col_a", fluss::DataType::Int())
                      .AddColumn("col_b", fluss::DataType::String())
                      .AddColumn("col_c", fluss::DataType::Int())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    // Append 3 records
    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));

    {
        auto col_a_builder = arrow::Int32Builder();
        col_a_builder.AppendValues({1, 2, 3}).ok();
        auto col_b_builder = arrow::StringBuilder();
        col_b_builder.AppendValues({"x", "y", "z"}).ok();
        auto col_c_builder = arrow::Int32Builder();
        col_c_builder.AppendValues({10, 20, 30}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("col_a", arrow::int32()),
                           arrow::field("col_b", arrow::utf8()),
                           arrow::field("col_c", arrow::int32())}),
            3,
            {col_a_builder.Finish().ValueOrDie(), col_b_builder.Finish().ValueOrDie(),
             col_c_builder.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }
    ASSERT_OK(append_writer.Flush());

    // Test project_by_name: select col_b and col_c only
    {
        fluss::Table proj_table;
        ASSERT_OK(conn.GetTable(table_path, proj_table));
        auto scan = proj_table.NewScan();
        scan.ProjectByName({"col_b", "col_c"});
        fluss::LogScanner scanner;
        ASSERT_OK(scan.CreateLogScanner(scanner));

        ASSERT_OK(scanner.Subscribe(0, 0));

        fluss::ScanRecords records;
        ASSERT_OK(scanner.Poll(10000, records));

        ASSERT_EQ(records.Count(), 3u) << "Should have 3 records with project_by_name";

        std::vector<std::string> expected_col_b = {"x", "y", "z"};
        std::vector<int32_t> expected_col_c = {10, 20, 30};

        // Collect and sort by col_c to get deterministic order
        std::vector<std::pair<std::string, int32_t>> collected;
        for (auto rec : records) {
            collected.emplace_back(std::string(rec.row.GetString(0)), rec.row.GetInt32(1));
        }
        std::sort(collected.begin(), collected.end(),
                  [](const auto& a, const auto& b) { return a.second < b.second; });

        for (size_t i = 0; i < 3; ++i) {
            EXPECT_EQ(collected[i].first, expected_col_b[i]) << "col_b mismatch at index " << i;
            EXPECT_EQ(collected[i].second, expected_col_c[i]) << "col_c mismatch at index " << i;
        }
    }

    // Test project by column indices: select col_b (1) and col_a (0) in that order
    {
        fluss::Table proj_table;
        ASSERT_OK(conn.GetTable(table_path, proj_table));
        auto scan = proj_table.NewScan();
        scan.ProjectByIndex({1, 0});
        fluss::LogScanner scanner;
        ASSERT_OK(scan.CreateLogScanner(scanner));

        ASSERT_OK(scanner.Subscribe(0, 0));

        fluss::ScanRecords records;
        ASSERT_OK(scanner.Poll(10000, records));

        ASSERT_EQ(records.Count(), 3u);

        std::vector<std::string> expected_col_b = {"x", "y", "z"};
        std::vector<int32_t> expected_col_a = {1, 2, 3};

        std::vector<std::pair<std::string, int32_t>> collected;
        for (auto rec : records) {
            collected.emplace_back(std::string(rec.row.GetString(0)), rec.row.GetInt32(1));
        }
        std::sort(collected.begin(), collected.end(),
                  [](const auto& a, const auto& b) { return a.second < b.second; });

        for (size_t i = 0; i < 3; ++i) {
            EXPECT_EQ(collected[i].first, expected_col_b[i]) << "col_b mismatch at index " << i;
            EXPECT_EQ(collected[i].second, expected_col_a[i]) << "col_a mismatch at index " << i;
        }
    }

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, TestPollBatches) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_poll_batches_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    std::this_thread::sleep_for(std::chrono::seconds(1));

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateRecordBatchLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    // Test 1: Empty table should return empty result
    {
        fluss::ArrowRecordBatches batches;
        ASSERT_OK(scanner.PollRecordBatch(500, batches));
        ASSERT_TRUE(batches.Empty());
    }

    // Append data
    auto table_append = table.NewAppend();
    fluss::AppendWriter writer;
    ASSERT_OK(table_append.CreateWriter(writer));

    auto make_batch = [](std::vector<int32_t> ids, std::vector<std::string> names) {
        auto id_builder = arrow::Int32Builder();
        id_builder.AppendValues(ids).ok();
        auto name_builder = arrow::StringBuilder();
        name_builder.AppendValues(names).ok();
        return arrow::RecordBatch::Make(
            arrow::schema(
                {arrow::field("id", arrow::int32()), arrow::field("name", arrow::utf8())}),
            static_cast<int64_t>(ids.size()),
            {id_builder.Finish().ValueOrDie(), name_builder.Finish().ValueOrDie()});
    };

    ASSERT_OK(writer.AppendArrowBatch(make_batch({1, 2}, {"a", "b"})));
    ASSERT_OK(writer.AppendArrowBatch(make_batch({3, 4}, {"c", "d"})));
    ASSERT_OK(writer.AppendArrowBatch(make_batch({5, 6}, {"e", "f"})));
    ASSERT_OK(writer.Flush());

    // Extract ids from Arrow batches
    auto extract_ids = [](const fluss::ArrowRecordBatches& batches) {
        std::vector<int32_t> ids;
        for (const auto& batch : batches) {
            auto arr =
                std::static_pointer_cast<arrow::Int32Array>(batch->GetArrowRecordBatch()->column(0));
            for (int64_t i = 0; i < arr->length(); ++i) {
                ids.push_back(arr->Value(i));
            }
        }
        return ids;
    };

    // Test 2: Poll until we get all 6 records
    std::vector<int32_t> all_ids;
    fluss_test::PollRecordBatches(scanner, 6, extract_ids, all_ids);
    ASSERT_EQ(all_ids, (std::vector<int32_t>{1, 2, 3, 4, 5, 6}));

    // Test 3: Append more and verify offset continuation (no duplicates)
    ASSERT_OK(writer.AppendArrowBatch(make_batch({7, 8}, {"g", "h"})));
    ASSERT_OK(writer.Flush());

    std::vector<int32_t> new_ids;
    fluss_test::PollRecordBatches(scanner, 2, extract_ids, new_ids);
    ASSERT_EQ(new_ids, (std::vector<int32_t>{7, 8}));

    // Test 4: Subscribing from mid-offset should truncate batch
    {
        fluss::Table trunc_table;
        ASSERT_OK(conn.GetTable(table_path, trunc_table));
        auto trunc_scan = trunc_table.NewScan();
        fluss::LogScanner trunc_scanner;
        ASSERT_OK(trunc_scan.CreateRecordBatchLogScanner(trunc_scanner));
        ASSERT_OK(trunc_scanner.Subscribe(0, 3));

        std::vector<int32_t> trunc_ids;
        fluss_test::PollRecordBatches(trunc_scanner, 5, extract_ids, trunc_ids);
        ASSERT_EQ(trunc_ids, (std::vector<int32_t>{4, 5, 6, 7, 8}));
    }

    // Test 5: Projection should only return requested columns
    {
        fluss::Table proj_table;
        ASSERT_OK(conn.GetTable(table_path, proj_table));
        auto proj_scan = proj_table.NewScan();
        proj_scan.ProjectByName({"id"});
        fluss::LogScanner proj_scanner;
        ASSERT_OK(proj_scan.CreateRecordBatchLogScanner(proj_scanner));
        ASSERT_OK(proj_scanner.Subscribe(0, 0));

        fluss::ArrowRecordBatches proj_batches;
        ASSERT_OK(proj_scanner.PollRecordBatch(10000, proj_batches));

        ASSERT_FALSE(proj_batches.Empty());
        EXPECT_EQ(proj_batches[0]->GetArrowRecordBatch()->num_columns(), 1)
            << "Projected batch should have 1 column (id), not 2";
    }

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, AllSupportedDatatypes) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_log_all_datatypes_cpp");

    // Create a log table with all supported datatypes
    auto schema =
        fluss::Schema::NewBuilder()
            .AddColumn("col_tinyint", fluss::DataType::TinyInt())
            .AddColumn("col_smallint", fluss::DataType::SmallInt())
            .AddColumn("col_int", fluss::DataType::Int())
            .AddColumn("col_bigint", fluss::DataType::BigInt())
            .AddColumn("col_float", fluss::DataType::Float())
            .AddColumn("col_double", fluss::DataType::Double())
            .AddColumn("col_boolean", fluss::DataType::Boolean())
            .AddColumn("col_char", fluss::DataType::Char(10))
            .AddColumn("col_string", fluss::DataType::String())
            .AddColumn("col_decimal", fluss::DataType::Decimal(10, 2))
            .AddColumn("col_date", fluss::DataType::Date())
            .AddColumn("col_time", fluss::DataType::Time())
            .AddColumn("col_timestamp", fluss::DataType::Timestamp())
            .AddColumn("col_timestamp_ltz", fluss::DataType::TimestampLtz())
            .AddColumn("col_bytes", fluss::DataType::Bytes())
            .AddColumn("col_binary", fluss::DataType::Binary(4))
            .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    size_t field_count = table.GetTableInfo().schema.columns.size();

    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));

    // Test data
    int32_t col_tinyint = 127;
    int32_t col_smallint = 32767;
    int32_t col_int = 2147483647;
    int64_t col_bigint = 9223372036854775807LL;
    float col_float = 3.14f;
    double col_double = 2.718281828459045;
    bool col_boolean = true;
    std::string col_char = "hello";
    std::string col_string = "world of fluss rust client";
    std::string col_decimal = "123.45";
    auto col_date = fluss::Date::FromDays(20476);           // 2026-01-23
    auto col_time = fluss::Time::FromMillis(36827000);       // 10:13:47
    auto col_timestamp = fluss::Timestamp::FromMillisNanos(1769163227123, 456000);
    auto col_timestamp_ltz = fluss::Timestamp::FromMillisNanos(1769163227123, 456000);
    std::vector<uint8_t> col_bytes = {'b', 'i', 'n', 'a', 'r', 'y', ' ', 'd', 'a', 't', 'a'};
    std::vector<uint8_t> col_binary = {0xDE, 0xAD, 0xBE, 0xEF};

    // Append a row with all datatypes
    {
        fluss::GenericRow row(field_count);
        row.SetInt32(0, col_tinyint);
        row.SetInt32(1, col_smallint);
        row.SetInt32(2, col_int);
        row.SetInt64(3, col_bigint);
        row.SetFloat32(4, col_float);
        row.SetFloat64(5, col_double);
        row.SetBool(6, col_boolean);
        row.SetString(7, col_char);
        row.SetString(8, col_string);
        row.SetDecimal(9, col_decimal);
        row.SetDate(10, col_date);
        row.SetTime(11, col_time);
        row.SetTimestampNtz(12, col_timestamp);
        row.SetTimestampLtz(13, col_timestamp_ltz);
        row.SetBytes(14, col_bytes);
        row.SetBytes(15, col_binary);
        ASSERT_OK(append_writer.Append(row));
    }

    // Append a row with null values
    {
        fluss::GenericRow row_with_nulls(field_count);
        for (size_t i = 0; i < field_count; ++i) {
            row_with_nulls.SetNull(i);
        }
        ASSERT_OK(append_writer.Append(row_with_nulls));
    }

    ASSERT_OK(append_writer.Flush());

    // Scan the records
    fluss::Table scan_table;
    ASSERT_OK(conn.GetTable(table_path, scan_table));
    auto table_scan = scan_table.NewScan();
    fluss::LogScanner log_scanner;
    ASSERT_OK(table_scan.CreateLogScanner(log_scanner));
    ASSERT_OK(log_scanner.Subscribe(0, 0));

    // Poll until we get 2 records
    std::vector<fluss::ScanRecord> all_records;
    fluss_test::PollRecords(log_scanner, 2,
        [](const fluss::ScanRecord& rec) { return rec; }, all_records);
    ASSERT_EQ(all_records.size(), 2u) << "Expected 2 records";

    // Verify first record (all values)
    auto& row = all_records[0].row;

    EXPECT_EQ(row.GetInt32(0), col_tinyint) << "col_tinyint mismatch";
    EXPECT_EQ(row.GetInt32(1), col_smallint) << "col_smallint mismatch";
    EXPECT_EQ(row.GetInt32(2), col_int) << "col_int mismatch";
    EXPECT_EQ(row.GetInt64(3), col_bigint) << "col_bigint mismatch";
    EXPECT_NEAR(row.GetFloat32(4), col_float, 1e-6f) << "col_float mismatch";
    EXPECT_NEAR(row.GetFloat64(5), col_double, 1e-15) << "col_double mismatch";
    EXPECT_EQ(row.GetBool(6), col_boolean) << "col_boolean mismatch";
    EXPECT_EQ(row.GetString(7), col_char) << "col_char mismatch";
    EXPECT_EQ(row.GetString(8), col_string) << "col_string mismatch";
    EXPECT_EQ(row.GetDecimalString(9), col_decimal) << "col_decimal mismatch";
    EXPECT_EQ(row.GetDate(10).days_since_epoch, col_date.days_since_epoch) << "col_date mismatch";
    EXPECT_EQ(row.GetTime(11).millis_since_midnight, col_time.millis_since_midnight)
        << "col_time mismatch";
    EXPECT_EQ(row.GetTimestamp(12).epoch_millis, col_timestamp.epoch_millis)
        << "col_timestamp millis mismatch";
    EXPECT_EQ(row.GetTimestamp(12).nano_of_millisecond, col_timestamp.nano_of_millisecond)
        << "col_timestamp nanos mismatch";
    EXPECT_EQ(row.GetTimestamp(13).epoch_millis, col_timestamp_ltz.epoch_millis)
        << "col_timestamp_ltz millis mismatch";
    EXPECT_EQ(row.GetTimestamp(13).nano_of_millisecond, col_timestamp_ltz.nano_of_millisecond)
        << "col_timestamp_ltz nanos mismatch";

    auto [bytes_ptr, bytes_len] = row.GetBytes(14);
    EXPECT_EQ(bytes_len, col_bytes.size()) << "col_bytes length mismatch";
    EXPECT_TRUE(std::memcmp(bytes_ptr, col_bytes.data(), bytes_len) == 0)
        << "col_bytes mismatch";

    auto [binary_ptr, binary_len] = row.GetBytes(15);
    EXPECT_EQ(binary_len, col_binary.size()) << "col_binary length mismatch";
    EXPECT_TRUE(std::memcmp(binary_ptr, col_binary.data(), binary_len) == 0)
        << "col_binary mismatch";

    // Verify second record (all nulls)
    auto& null_row = all_records[1].row;
    for (size_t i = 0; i < field_count; ++i) {
        EXPECT_TRUE(null_row.IsNull(i)) << "column " << i << " should be null";
    }

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, PartitionedTableAppendScan) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_partitioned_log_append_cpp");

    // Create a partitioned log table
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("region", fluss::DataType::String())
                      .AddColumn("value", fluss::DataType::BigInt())
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetPartitionKeys({"region"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    // Create partitions
    fluss_test::CreatePartitions(adm, table_path, "region", {"US", "EU"});

    // Wait for partitions
    std::this_thread::sleep_for(std::chrono::seconds(2));

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    auto table_append = table.NewAppend();
    fluss::AppendWriter append_writer;
    ASSERT_OK(table_append.CreateWriter(append_writer));

    // Append rows
    struct TestData {
        int32_t id;
        std::string region;
        int64_t value;
    };
    std::vector<TestData> test_data = {{1, "US", 100}, {2, "US", 200}, {3, "EU", 300}, {4, "EU", 400}};

    for (const auto& d : test_data) {
        fluss::GenericRow row(3);
        row.SetInt32(0, d.id);
        row.SetString(1, d.region);
        row.SetInt64(2, d.value);
        ASSERT_OK(append_writer.Append(row));
    }
    ASSERT_OK(append_writer.Flush());

    // Append arrow batches per partition
    {
        auto id_builder = arrow::Int32Builder();
        id_builder.AppendValues({5, 6}).ok();
        auto region_builder = arrow::StringBuilder();
        region_builder.AppendValues({"US", "US"}).ok();
        auto value_builder = arrow::Int64Builder();
        value_builder.AppendValues({500, 600}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("id", arrow::int32()),
                           arrow::field("region", arrow::utf8()),
                           arrow::field("value", arrow::int64())}),
            2,
            {id_builder.Finish().ValueOrDie(), region_builder.Finish().ValueOrDie(),
             value_builder.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }

    {
        auto id_builder = arrow::Int32Builder();
        id_builder.AppendValues({7, 8}).ok();
        auto region_builder = arrow::StringBuilder();
        region_builder.AppendValues({"EU", "EU"}).ok();
        auto value_builder = arrow::Int64Builder();
        value_builder.AppendValues({700, 800}).ok();

        auto batch = arrow::RecordBatch::Make(
            arrow::schema({arrow::field("id", arrow::int32()),
                           arrow::field("region", arrow::utf8()),
                           arrow::field("value", arrow::int64())}),
            2,
            {id_builder.Finish().ValueOrDie(), region_builder.Finish().ValueOrDie(),
             value_builder.Finish().ValueOrDie()});

        ASSERT_OK(append_writer.AppendArrowBatch(batch));
    }
    ASSERT_OK(append_writer.Flush());

    // Test list partition offsets
    std::unordered_map<int32_t, int64_t> us_offsets;
    ASSERT_OK(adm.ListPartitionOffsets(table_path, "US", {0}, fluss::OffsetSpec::Latest(),
                                       us_offsets));
    EXPECT_EQ(us_offsets[0], 4) << "US partition should have 4 records";

    std::unordered_map<int32_t, int64_t> eu_offsets;
    ASSERT_OK(adm.ListPartitionOffsets(table_path, "EU", {0}, fluss::OffsetSpec::Latest(),
                                       eu_offsets));
    EXPECT_EQ(eu_offsets[0], 4) << "EU partition should have 4 records";

    // Subscribe to all partitions and scan
    fluss::Table scan_table;
    ASSERT_OK(conn.GetTable(table_path, scan_table));
    auto table_scan = scan_table.NewScan();
    fluss::LogScanner log_scanner;
    ASSERT_OK(table_scan.CreateLogScanner(log_scanner));

    std::vector<fluss::PartitionInfo> partition_infos;
    ASSERT_OK(adm.ListPartitionInfos(table_path, partition_infos));

    for (const auto& pi : partition_infos) {
        ASSERT_OK(log_scanner.SubscribePartitionBuckets(pi.partition_id, 0, 0));
    }

    // Collect all records
    using Record = std::tuple<int32_t, std::string, int64_t>;
    auto extract_record = [](const fluss::ScanRecord& rec) -> Record {
        return {rec.row.GetInt32(0), std::string(rec.row.GetString(1)), rec.row.GetInt64(2)};
    };
    std::vector<Record> collected;
    fluss_test::PollRecords(log_scanner, 8, extract_record, collected);

    ASSERT_EQ(collected.size(), 8u) << "Expected 8 records total";
    std::sort(collected.begin(), collected.end());

    std::vector<Record> expected = {{1, "US", 100},  {2, "US", 200},  {3, "EU", 300},
                                    {4, "EU", 400},  {5, "US", 500},  {6, "US", 600},
                                    {7, "EU", 700},  {8, "EU", 800}};
    EXPECT_EQ(collected, expected);

    // Test unsubscribe_partition: unsubscribe EU, should only get US data
    {
        fluss::Table unsub_table;
        ASSERT_OK(conn.GetTable(table_path, unsub_table));
        auto unsub_scan = unsub_table.NewScan();
        fluss::LogScanner unsub_scanner;
        ASSERT_OK(unsub_scan.CreateLogScanner(unsub_scanner));

        int64_t eu_partition_id = -1;
        for (const auto& pi : partition_infos) {
            ASSERT_OK(unsub_scanner.SubscribePartitionBuckets(pi.partition_id, 0, 0));
            if (pi.partition_name == "EU") {
                eu_partition_id = pi.partition_id;
            }
        }
        ASSERT_GE(eu_partition_id, 0) << "EU partition should exist";

        ASSERT_OK(unsub_scanner.UnsubscribePartition(eu_partition_id, 0));

        std::vector<Record> us_only;
        fluss_test::PollRecords(unsub_scanner, 4, extract_record, us_only);

        ASSERT_EQ(us_only.size(), 4u) << "Should receive exactly 4 US records";
        for (const auto& [id, region, val] : us_only) {
            EXPECT_EQ(region, "US") << "After unsubscribe EU, only US data should be read";
        }
    }

    // Test subscribe_partition_buckets (batch subscribe)
    {
        fluss::Table batch_table;
        ASSERT_OK(conn.GetTable(table_path, batch_table));
        auto batch_scan = batch_table.NewScan();
        fluss::LogScanner batch_scanner;
        ASSERT_OK(batch_scan.CreateLogScanner(batch_scanner));

        std::vector<fluss::PartitionBucketSubscription> subs;
        for (const auto& pi : partition_infos) {
            subs.push_back({pi.partition_id, 0, 0});
        }
        ASSERT_OK(batch_scanner.SubscribePartitionBuckets(subs));

        std::vector<Record> batch_collected;
        fluss_test::PollRecords(batch_scanner, 8, extract_record, batch_collected);
        ASSERT_EQ(batch_collected.size(), 8u);
        std::sort(batch_collected.begin(), batch_collected.end());
        EXPECT_EQ(batch_collected, expected);
    }

    ASSERT_OK(adm.DropTable(table_path, false));
}

// ============================================================================
// Array data type tests
// ============================================================================

TEST_F(LogTableTest, AppendAndScanWithArray) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_append_scan_with_array_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("tags", fluss::DataType::Array(fluss::DataType::String()))
                      .AddColumn("scores", fluss::DataType::Array(fluss::DataType::Int()))
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"id"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    auto info = table.GetTableInfo();
    ASSERT_GE(info.schema.columns.size(), 3u);
    const auto& tags_type = info.schema.columns[1].data_type;
    ASSERT_EQ(tags_type.id(), fluss::TypeId::Array);
    ASSERT_NE(tags_type.element_type(), nullptr);
    ASSERT_EQ(tags_type.element_type()->id(), fluss::TypeId::String);
    const auto& scores_type = info.schema.columns[2].data_type;
    ASSERT_EQ(scores_type.id(), fluss::TypeId::Array);
    ASSERT_NE(scores_type.element_type(), nullptr);
    ASSERT_EQ(scores_type.element_type()->id(), fluss::TypeId::Int);

    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));

    {
        auto row = table.NewRow();
        row.Set("id", 1);

        fluss::ArrayWriter tags(2, fluss::DataType::String());
        tags.SetString(0, "hello");
        tags.SetString(1, "world");
        row.SetArray(1, std::move(tags));

        fluss::ArrayWriter scores(3, fluss::DataType::Int());
        scores.SetInt32(0, 10);
        scores.SetInt32(1, 20);
        scores.SetInt32(2, 30);
        row.SetArray(2, std::move(scores));

        ASSERT_OK(append_writer.Append(row));
    }
    {
        auto row = table.NewRow();
        row.Set("id", 2);

        fluss::ArrayWriter tags(1, fluss::DataType::String());
        tags.SetNull(0);
        row.SetArray(1, std::move(tags));

        fluss::ArrayWriter scores(0, fluss::DataType::Int());
        row.SetArray(2, std::move(scores));

        ASSERT_OK(append_writer.Append(row));
    }

    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    struct Record {
        int32_t id;
        size_t tag_count;
        std::vector<std::string> tags;
        size_t score_count;
        std::vector<int32_t> scores;
    };

    std::vector<Record> collected;
    auto extract = [](const fluss::ScanRecord& scan_rec) {
        const auto& rv = scan_rec.row;
        Record rec;
        rec.id = rv.GetInt32(0);

        auto tags = rv.GetValue(1);
        rec.tag_count = tags.Size();
        for (size_t i = 0; i < rec.tag_count; ++i) {
            auto el = tags.At(i);
            rec.tags.push_back(el.IsNull() ? "<null>" : el.GetString());
        }

        auto scores = rv.GetValue(2);
        rec.score_count = scores.Size();
        for (size_t i = 0; i < rec.score_count; ++i) {
            rec.scores.push_back(scores.At(i).GetInt32());
        }

        return rec;
    };

    fluss_test::PollRecords(scanner, 2, extract, collected);

    ASSERT_EQ(collected.size(), 2u);

    std::sort(collected.begin(), collected.end(),
              [](const Record& a, const Record& b) { return a.id < b.id; });

    EXPECT_EQ(collected[0].id, 1);
    ASSERT_EQ(collected[0].tag_count, 2u);
    EXPECT_EQ(collected[0].tags[0], "hello");
    EXPECT_EQ(collected[0].tags[1], "world");
    ASSERT_EQ(collected[0].score_count, 3u);
    EXPECT_EQ(collected[0].scores[0], 10);
    EXPECT_EQ(collected[0].scores[1], 20);
    EXPECT_EQ(collected[0].scores[2], 30);

    EXPECT_EQ(collected[1].id, 2);
    ASSERT_EQ(collected[1].tag_count, 1u);
    EXPECT_EQ(collected[1].tags[0], "<null>");
    ASSERT_EQ(collected[1].score_count, 0u);

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, AppendAndScanWithMapAndRow) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_append_scan_map_row_cpp");

    // MAP / ROW columns can't be built with the flat schema builder.
    auto arrow_schema = arrow::schema({
        arrow::field("id", arrow::int32()),
        arrow::field("attrs", arrow::map(arrow::utf8(), arrow::int32())),
        arrow::field("nested", arrow::struct_({arrow::field("seq", arrow::int32()),
                                               arrow::field("label", arrow::utf8())})),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema);

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));
    {
        auto row = table.NewRow();
        row.Set("id", 1);

        fluss::MapWriter attrs(2, fluss::DataType::String(), fluss::DataType::Int());
        attrs.SetKeyString("a");
        attrs.SetValueInt32(1);
        attrs.Commit();
        attrs.SetKeyString("b");
        attrs.SetValueInt32(2);
        attrs.Commit();
        row.Set("attrs", std::move(attrs));

        fluss::GenericRow nested(2);
        nested.SetInt32(0, 7);
        nested.SetString(1, "seven");
        row.Set("nested", std::move(nested));

        ASSERT_OK(append_writer.Append(row));
    }
    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    struct Record {
        int32_t id;
        size_t attr_count;
        std::string k0;
        int32_t v0;
        int32_t seq;
        std::string label;
    };
    std::vector<Record> collected;
    auto extract = [](const fluss::ScanRecord& scan_rec) {
        const auto& rv = scan_rec.row;
        Record rec;
        rec.id = rv.GetInt32(0);
        auto attrs = rv.GetValue(1);
        rec.attr_count = attrs.Size();
        rec.k0 = attrs.KeyAt(0).GetString();
        rec.v0 = attrs.ValueAt(0).GetInt32();
        auto nested = rv.GetValue("nested");
        rec.seq = nested.Field(0).GetInt32();
        rec.label = nested.Field(1).GetString();
        return rec;
    };
    fluss_test::PollRecords(scanner, 1, extract, collected);

    ASSERT_EQ(collected.size(), 1u);
    EXPECT_EQ(collected[0].id, 1);
    EXPECT_EQ(collected[0].attr_count, 2u);
    EXPECT_EQ(collected[0].k0, "a");
    EXPECT_EQ(collected[0].v0, 1);
    EXPECT_EQ(collected[0].seq, 7);
    EXPECT_EQ(collected[0].label, "seven");

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, ProjectionWithCompoundTypes) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_log_projection_compound_cpp");

    auto arrow_schema = arrow::schema({
        arrow::field("id", arrow::int32()),
        arrow::field("nested", arrow::struct_({arrow::field("seq", arrow::int32()),
                                               arrow::field("label", arrow::utf8())})),
        arrow::field("attrs", arrow::map(arrow::utf8(), arrow::int32())),
        arrow::field("tags", arrow::list(arrow::utf8())),
        arrow::field("extra", arrow::utf8()),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema);

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));
    {
        auto row = table.NewRow();
        row.SetInt32(0, 7);
        fluss::GenericRow nested(2);
        nested.SetInt32(0, 42);
        nested.SetString(1, "hello");
        row.SetRow(1, std::move(nested));
        fluss::MapWriter attrs(2, fluss::DataType::String(), fluss::DataType::Int());
        attrs.SetKeyString("x");
        attrs.SetValueInt32(1);
        attrs.Commit();
        attrs.SetKeyString("y");
        attrs.SetValueInt32(2);
        attrs.Commit();
        row.SetMap(2, std::move(attrs));
        fluss::ArrayWriter tags(2, fluss::DataType::String());
        tags.SetString(0, "alpha");
        tags.SetString(1, "beta");
        row.SetArray(3, std::move(tags));
        row.SetString(4, "ignore-me");
        ASSERT_OK(append_writer.Append(row));
    }
    ASSERT_OK(append_writer.Flush());

    // Project columns reordered, dropping `extra`: new layout is
    // [nested=0, attrs=1, tags=2, id=3].
    auto scan = table.NewScan();
    scan.ProjectByName({"nested", "attrs", "tags", "id"});
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    struct Rec {
        int32_t id;
        int32_t seq;
        std::string label;
        size_t attr_count;
        size_t tag_count;
        std::string tag0;
    };
    std::vector<Rec> collected;
    auto extract = [](const fluss::ScanRecord& sr) {
        const auto& rv = sr.row;
        Rec rec;
        auto nested = rv.GetValue(0);
        rec.seq = nested.Field(0).GetInt32();
        rec.label = nested.Field(1).GetString();
        auto m = rv.GetValue(1);
        rec.attr_count = m.Size();
        auto a = rv.GetValue(2);
        rec.tag_count = a.Size();
        rec.tag0 = a.At(0).GetString();
        rec.id = rv.GetInt32(3);
        return rec;
    };
    fluss_test::PollRecords(scanner, 1, extract, collected);

    ASSERT_EQ(collected.size(), 1u);
    EXPECT_EQ(collected[0].id, 7);
    EXPECT_EQ(collected[0].seq, 42);
    EXPECT_EQ(collected[0].label, "hello");
    EXPECT_EQ(collected[0].attr_count, 2u);
    EXPECT_EQ(collected[0].tag_count, 2u);
    EXPECT_EQ(collected[0].tag0, "alpha");

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, AppendAndScanWithNestedArray) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_append_scan_nested_array_cpp");

    auto schema =
        fluss::Schema::NewBuilder()
            .AddColumn("id", fluss::DataType::Int())
            .AddColumn("matrix",
                        fluss::DataType::Array(fluss::DataType::Array(fluss::DataType::Int())))
            .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"id"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));

    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));

    {
        auto row = table.NewRow();
        row.Set("id", 1);

        fluss::ArrayWriter inner1(2, fluss::DataType::Int());
        inner1.SetInt32(0, 1);
        inner1.SetInt32(1, 2);

        fluss::ArrayWriter inner2(2, fluss::DataType::Int());
        inner2.SetInt32(0, 3);
        inner2.SetInt32(1, 4);

        fluss::ArrayWriter outer(2, fluss::DataType::Array(fluss::DataType::Int()));
        outer.SetArray(0, std::move(inner1));
        outer.SetArray(1, std::move(inner2));

        row.SetArray(1, std::move(outer));
        ASSERT_OK(append_writer.Append(row));
    }

    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    struct Record {
        int32_t id;
        size_t outer_count;
        fluss::TypeId element_type;
        std::vector<std::vector<int32_t>> values;
    };

    std::vector<Record> collected;
    auto extract = [](const fluss::ScanRecord& scan_rec) {
        const auto& rv = scan_rec.row;
        Record rec;
        rec.id = rv.GetInt32(0);
        auto outer = rv.GetValue(1);
        rec.outer_count = outer.Size();
        rec.element_type = outer.Size() > 0 ? outer.At(0).Type() : fluss::TypeId::Array;
        rec.values.reserve(outer.Size());
        for (size_t i = 0; i < outer.Size(); ++i) {
            auto inner = outer.At(i);
            std::vector<int32_t> row;
            row.reserve(inner.Size());
            for (size_t j = 0; j < inner.Size(); ++j) {
                row.push_back(inner.At(j).GetInt32());
            }
            rec.values.push_back(std::move(row));
        }
        return rec;
    };

    fluss_test::PollRecords(scanner, 1, extract, collected);
    ASSERT_EQ(collected.size(), 1u);
    EXPECT_EQ(collected[0].id, 1);
    EXPECT_EQ(collected[0].outer_count, 2u);
    EXPECT_EQ(collected[0].element_type, fluss::TypeId::Array);
    ASSERT_EQ(collected[0].values.size(), 2u);
    EXPECT_EQ(collected[0].values[0], (std::vector<int32_t>{1, 2}));
    EXPECT_EQ(collected[0].values[1], (std::vector<int32_t>{3, 4}));

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, AppendAndScanWithArrayRichTypes) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_append_scan_array_rich_types_cpp");

    auto schema =
        fluss::Schema::NewBuilder()
            .AddColumn("id", fluss::DataType::Int())
            .AddColumn("arr_bytes", fluss::DataType::Array(fluss::DataType::Bytes()))
            .AddColumn("arr_date", fluss::DataType::Array(fluss::DataType::Date()))
            .AddColumn("arr_time", fluss::DataType::Array(fluss::DataType::Time()))
            .AddColumn("arr_ts", fluss::DataType::Array(fluss::DataType::Timestamp(6)))
            .AddColumn("arr_decimal", fluss::DataType::Array(fluss::DataType::Decimal(10, 2)))
            .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"id"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));

    {
        auto row = table.NewRow();
        row.Set("id", 1);

        fluss::ArrayWriter arr_bytes(2, fluss::DataType::Bytes());
        arr_bytes.SetBytes(0, std::vector<uint8_t>{0x10, 0x20, 0x30});
        arr_bytes.SetNull(1);
        row.SetArray(1, std::move(arr_bytes));

        fluss::ArrayWriter arr_date(2, fluss::DataType::Date());
        auto d0 = fluss::Date::FromDays(20000);
        arr_date.SetDate(0, d0);
        arr_date.SetNull(1);
        row.SetArray(2, std::move(arr_date));

        fluss::ArrayWriter arr_time(1, fluss::DataType::Time());
        auto t0 = fluss::Time::FromMillis(3600000);
        arr_time.SetTime(0, t0);
        row.SetArray(3, std::move(arr_time));

        fluss::ArrayWriter arr_ts(1, fluss::DataType::Timestamp(6));
        auto ts0 = fluss::Timestamp::FromMillisNanos(1769163227123, 456000);
        arr_ts.SetTimestampNtz(0, ts0);
        row.SetArray(4, std::move(arr_ts));

        fluss::ArrayWriter arr_decimal(2, fluss::DataType::Decimal(10, 2));
        arr_decimal.SetDecimal(0, "123.45");
        arr_decimal.SetNull(1);
        row.SetArray(5, std::move(arr_decimal));

        ASSERT_OK(append_writer.Append(row));
    }

    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    fluss::ScanRecords records;
    ASSERT_OK(scanner.Poll(10000, records));
    ASSERT_EQ(records.Count(), 1u);

    auto it = records.begin();
    ASSERT_TRUE(it != records.end());
    auto rec = *it;
    const auto& rv = rec.row;

    auto arr_bytes = rv.GetValue(1);
    EXPECT_EQ(arr_bytes.Size(), 2u);
    auto bytes0 = arr_bytes.At(0).GetBytes();
    ASSERT_EQ(bytes0.size(), 3u);
    EXPECT_EQ(bytes0[0], 0x10);
    EXPECT_EQ(bytes0[1], 0x20);
    EXPECT_EQ(bytes0[2], 0x30);
    EXPECT_TRUE(arr_bytes.At(1).IsNull());

    auto arr_date = rv.GetValue(2);
    EXPECT_EQ(arr_date.Size(), 2u);
    EXPECT_EQ(arr_date.At(0).GetDate().days_since_epoch, fluss::Date::FromDays(20000).days_since_epoch);
    EXPECT_TRUE(arr_date.At(1).IsNull());

    auto arr_time = rv.GetValue(3);
    EXPECT_EQ(arr_time.Size(), 1u);
    EXPECT_EQ(arr_time.At(0).GetTime().millis_since_midnight,
              fluss::Time::FromMillis(3600000).millis_since_midnight);

    auto arr_ts = rv.GetValue(4);
    EXPECT_EQ(arr_ts.Size(), 1u);
    auto ts = arr_ts.At(0).GetTimestamp();
    EXPECT_EQ(ts.epoch_millis, 1769163227123);
    EXPECT_EQ(ts.nano_of_millisecond, 456000);

    auto arr_dec = rv.GetValue(5);
    EXPECT_EQ(arr_dec.Size(), 2u);
    EXPECT_EQ(arr_dec.At(0).GetDecimalString(), "123.45");
    EXPECT_TRUE(arr_dec.At(1).IsNull());

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, ArrayApiValidationErrors) {
    // Type mismatch setter should fail through FFI Result propagation.
    {
        fluss::ArrayWriter bool_array(1, fluss::DataType::Boolean());
        bool threw = false;
        try {
            bool_array.SetInt32(0, 42);
        } catch (const std::exception&) {
            threw = true;
        }
        EXPECT_TRUE(threw);
    }

    auto& adm = admin();
    auto& conn = connection();
    fluss::TablePath table_path("fluss", "test_array_api_validation_errors_cpp");

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("vals", fluss::DataType::Array(fluss::DataType::Int()))
                      .Build();
    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"id"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));
    auto row = table.NewRow();
    row.Set("id", 1);
    fluss::ArrayWriter vals(2, fluss::DataType::Int());
    vals.SetInt32(0, 7);
    vals.SetNull(1);
    row.SetArray(1, std::move(vals));
    ASSERT_OK(append_writer.Append(row));
    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));
    fluss::ScanRecords records;
    ASSERT_OK(scanner.Poll(10000, records));
    ASSERT_EQ(records.Count(), 1u);
    auto it = records.begin();
    ASSERT_TRUE(it != records.end());
    auto rec = *it;

    auto view = rec.row.GetValue(1);
    EXPECT_EQ(view.Type(), fluss::TypeId::Array);
    EXPECT_EQ(view.Size(), 2u);
    EXPECT_TRUE(view.At(1).IsNull());

    // Out-of-bounds navigation throws.
    bool oob_threw = false;
    try {
        (void)view.At(5).GetInt32();
    } catch (const std::exception&) {
        oob_threw = true;
    }
    EXPECT_TRUE(oob_threw);

    // Wrong-type leaf read throws.
    bool wrong_type_threw = false;
    try {
        (void)view.At(0).GetInt64();
    } catch (const std::exception&) {
        wrong_type_threw = true;
    }
    EXPECT_TRUE(wrong_type_threw);

    // Typed read of a null element throws.
    bool null_typed_getter_threw = false;
    try {
        (void)view.At(1).GetInt32();
    } catch (const std::exception&) {
        null_typed_getter_threw = true;
    }
    EXPECT_TRUE(null_typed_getter_threw);

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, AppendAndScanWithArrayEncodingEdgeCases) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_array_encoding_edge_cases_cpp");

    auto schema =
        fluss::Schema::NewBuilder()
            .AddColumn("id", fluss::DataType::Int())
            .AddColumn("arr_long_str", fluss::DataType::Array(fluss::DataType::String()))
            .AddColumn("arr_big_decimal", fluss::DataType::Array(fluss::DataType::Decimal(22, 5)))
            .AddColumn("arr_ts_nano", fluss::DataType::Array(fluss::DataType::Timestamp(9)))
            .AddColumn("arr_float", fluss::DataType::Array(fluss::DataType::Float()))
            .AddColumn("arr_double", fluss::DataType::Array(fluss::DataType::Double()))
            .AddColumn("arr_binary", fluss::DataType::Array(fluss::DataType::Binary(4)))
            .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketCount(1)
                                .SetBucketKeys({"id"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    fluss::AppendWriter append_writer;
    ASSERT_OK(table.NewAppend().CreateWriter(append_writer));

    {
        auto row = table.NewRow();
        row.Set("id", 1);

        // >= 8 bytes forces the heap-pointer variable-length path (threshold: 7)
        fluss::ArrayWriter arr_long_str(2, fluss::DataType::String());
        arr_long_str.SetString(0, "abcdefgh");
        arr_long_str.SetString(1, "this is a much longer string that definitely exceeds inline");
        row.SetArray(1, std::move(arr_long_str));

        // precision > 18 forces non-compact decimal encoding
        fluss::ArrayWriter arr_big_decimal(2, fluss::DataType::Decimal(22, 5));
        arr_big_decimal.SetDecimal(0, "12345678901234567.12345");
        arr_big_decimal.SetDecimal(1, "-99999999999999999.99999");
        row.SetArray(2, std::move(arr_big_decimal));

        // precision > 3 forces non-compact timestamp (millis + nanos-of-millis)
        fluss::ArrayWriter arr_ts_nano(1, fluss::DataType::Timestamp(9));
        auto ts_nano = fluss::Timestamp::FromMillisNanos(1769163227123, 456789);
        arr_ts_nano.SetTimestampNtz(0, ts_nano);
        row.SetArray(3, std::move(arr_ts_nano));

        // IEEE 754 special values: NaN, +Infinity, -Infinity
        fluss::ArrayWriter arr_float(3, fluss::DataType::Float());
        arr_float.SetFloat32(0, std::numeric_limits<float>::quiet_NaN());
        arr_float.SetFloat32(1, std::numeric_limits<float>::infinity());
        arr_float.SetFloat32(2, -std::numeric_limits<float>::infinity());
        row.SetArray(4, std::move(arr_float));

        fluss::ArrayWriter arr_double(3, fluss::DataType::Double());
        arr_double.SetFloat64(0, std::numeric_limits<double>::quiet_NaN());
        arr_double.SetFloat64(1, std::numeric_limits<double>::infinity());
        arr_double.SetFloat64(2, -std::numeric_limits<double>::infinity());
        row.SetArray(5, std::move(arr_double));

        // Fixed-length binary
        fluss::ArrayWriter arr_binary(2, fluss::DataType::Binary(4));
        arr_binary.SetBytes(0, std::vector<uint8_t>{0xDE, 0xAD, 0xBE, 0xEF});
        arr_binary.SetNull(1);
        row.SetArray(6, std::move(arr_binary));

        ASSERT_OK(append_writer.Append(row));
    }

    ASSERT_OK(append_writer.Flush());

    auto scan = table.NewScan();
    fluss::LogScanner scanner;
    ASSERT_OK(scan.CreateLogScanner(scanner));
    ASSERT_OK(scanner.Subscribe(0, 0));

    fluss::ScanRecords records;
    ASSERT_OK(scanner.Poll(10000, records));
    ASSERT_EQ(records.Count(), 1u);

    auto it = records.begin();
    ASSERT_TRUE(it != records.end());
    auto rec = *it;
    const auto& rv = rec.row;

    // Long strings: heap-encoded variable-length round-trip
    auto strs = rv.GetValue(1);
    EXPECT_EQ(strs.Size(), 2u);
    EXPECT_EQ(strs.At(0).GetString(), "abcdefgh");
    EXPECT_EQ(strs.At(1).GetString(), "this is a much longer string that definitely exceeds inline");

    // Non-compact decimal (precision 22 > MAX_COMPACT_PRECISION 18)
    auto decs = rv.GetValue(2);
    EXPECT_EQ(decs.Size(), 2u);
    EXPECT_EQ(decs.At(0).GetDecimalString(), "12345678901234567.12345");
    EXPECT_EQ(decs.At(1).GetDecimalString(), "-99999999999999999.99999");

    // Non-compact timestamp (precision 9 > MAX_COMPACT_TIMESTAMP_PRECISION 3)
    auto tss = rv.GetValue(3);
    EXPECT_EQ(tss.Size(), 1u);
    auto ts = tss.At(0).GetTimestamp();
    EXPECT_EQ(ts.epoch_millis, 1769163227123);
    EXPECT_EQ(ts.nano_of_millisecond, 456789);

    // Float NaN / Infinity round-trip
    auto floats = rv.GetValue(4);
    EXPECT_EQ(floats.Size(), 3u);
    EXPECT_TRUE(std::isnan(floats.At(0).GetFloat32()));
    EXPECT_TRUE(std::isinf(floats.At(1).GetFloat32()));
    EXPECT_GT(floats.At(1).GetFloat32(), 0.0f);
    EXPECT_TRUE(std::isinf(floats.At(2).GetFloat32()));
    EXPECT_LT(floats.At(2).GetFloat32(), 0.0f);

    // Double NaN / Infinity round-trip
    auto doubles = rv.GetValue(5);
    EXPECT_EQ(doubles.Size(), 3u);
    EXPECT_TRUE(std::isnan(doubles.At(0).GetFloat64()));
    EXPECT_TRUE(std::isinf(doubles.At(1).GetFloat64()));
    EXPECT_GT(doubles.At(1).GetFloat64(), 0.0);
    EXPECT_TRUE(std::isinf(doubles.At(2).GetFloat64()));
    EXPECT_LT(doubles.At(2).GetFloat64(), 0.0);

    // Fixed-length binary round-trip
    auto bins = rv.GetValue(6);
    EXPECT_EQ(bins.Size(), 2u);
    auto bin = bins.At(0).GetBytes();
    ASSERT_EQ(bin.size(), 4u);
    EXPECT_EQ(bin[0], 0xDE);
    EXPECT_EQ(bin[1], 0xAD);
    EXPECT_EQ(bin[2], 0xBE);
    EXPECT_EQ(bin[3], 0xEF);
    EXPECT_TRUE(bins.At(1).IsNull());

    ASSERT_OK(adm.DropTable(table_path, false));
}

TEST_F(LogTableTest, ArrayWriterOverflowDetection) {
    // SetInt32 on TINYINT array must throw when value overflows i8 range (-128..127)
    {
        fluss::ArrayWriter tinyint_arr(1, fluss::DataType::TinyInt());
        EXPECT_EQ(tinyint_arr.Size(), 1u);
        bool threw = false;
        try {
            tinyint_arr.SetInt32(0, 1000);
        } catch (const std::exception& e) {
            threw = true;
            std::string msg(e.what());
            EXPECT_NE(msg.find("TINYINT"), std::string::npos);
        }
        EXPECT_TRUE(threw);
    }

    // SetInt32 on SMALLINT array must throw when value overflows i16 range (-32768..32767)
    {
        fluss::ArrayWriter smallint_arr(1, fluss::DataType::SmallInt());
        bool threw = false;
        try {
            smallint_arr.SetInt32(0, 40000);
        } catch (const std::exception& e) {
            threw = true;
            std::string msg(e.what());
            EXPECT_NE(msg.find("SMALLINT"), std::string::npos);
        }
        EXPECT_TRUE(threw);
    }

    // Negative overflow: -200 doesn't fit TINYINT
    {
        fluss::ArrayWriter tinyint_arr(1, fluss::DataType::TinyInt());
        bool threw = false;
        try {
            tinyint_arr.SetInt32(0, -200);
        } catch (const std::exception&) {
            threw = true;
        }
        EXPECT_TRUE(threw);
    }

    // Values within range must succeed
    {
        fluss::ArrayWriter tinyint_arr(1, fluss::DataType::TinyInt());
        EXPECT_NO_THROW(tinyint_arr.SetInt32(0, 127));
    }
    {
        fluss::ArrayWriter tinyint_arr(1, fluss::DataType::TinyInt());
        EXPECT_NO_THROW(tinyint_arr.SetInt32(0, -128));
    }
    {
        fluss::ArrayWriter smallint_arr(1, fluss::DataType::SmallInt());
        EXPECT_NO_THROW(smallint_arr.SetInt32(0, 32767));
    }
}

TEST_F(LogTableTest, MapWriterOverflowDetection) {
    // Keys/values go through SetKeyInt32/SetValueInt32 (always i32), so TINYINT /
    // SMALLINT must be range-checked like array elements.

    // TINYINT map value overflowing i8 (-128..127) must throw.
    {
        fluss::MapWriter m(1, fluss::DataType::String(), fluss::DataType::TinyInt());
        bool threw = false;
        try {
            m.SetValueInt32(1000);
        } catch (const std::exception& e) {
            threw = true;
            EXPECT_NE(std::string(e.what()).find("TINYINT"), std::string::npos);
        }
        EXPECT_TRUE(threw);
    }

    // SMALLINT map value overflowing i16 must throw.
    {
        fluss::MapWriter m(1, fluss::DataType::String(), fluss::DataType::SmallInt());
        bool threw = false;
        try {
            m.SetValueInt32(40000);
        } catch (const std::exception& e) {
            threw = true;
            EXPECT_NE(std::string(e.what()).find("SMALLINT"), std::string::npos);
        }
        EXPECT_TRUE(threw);
    }

    // Keys are checked the same way: a TINYINT key out of range throws.
    {
        fluss::MapWriter m(1, fluss::DataType::TinyInt(), fluss::DataType::Int());
        bool threw = false;
        try {
            m.SetKeyInt32(-200);
        } catch (const std::exception& e) {
            threw = true;
            EXPECT_NE(std::string(e.what()).find("TINYINT"), std::string::npos);
        }
        EXPECT_TRUE(threw);
    }

    // In-range key and value must succeed.
    {
        fluss::MapWriter m(1, fluss::DataType::TinyInt(), fluss::DataType::SmallInt());
        EXPECT_NO_THROW(m.SetKeyInt32(127));
        EXPECT_NO_THROW(m.SetValueInt32(32767));
    }
}

TEST_F(LogTableTest, NullabilityPreservedInTableInfo) {
    auto& adm = admin();
    auto& conn = connection();

    fluss::TablePath table_path("fluss", "test_nullability_table_info_cpp");

    auto schema =
        fluss::Schema::NewBuilder()
            .AddColumn("id", fluss::DataType::Int())
            .AddColumn("name", fluss::DataType::String())
            .AddColumn("tags", fluss::DataType::Array(fluss::DataType::String().NotNull()))
            .AddColumn("ids", fluss::DataType::Array(fluss::DataType::Int()).NotNull())
            .AddColumn("nested",
                       fluss::DataType::Array(
                           fluss::DataType::Array(fluss::DataType::Int()).NotNull()))
            .SetPrimaryKeys({"id"})
            .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetProperty("table.replication.factor", "1")
                                .Build();

    fluss_test::CreateTable(adm, table_path, table_descriptor);

    fluss::Table table;
    ASSERT_OK(conn.GetTable(table_path, table));
    auto info = table.GetTableInfo();

    ASSERT_EQ(info.schema.columns.size(), 5u);
    EXPECT_EQ(info.primary_keys, std::vector<std::string>{"id"});

    // Primary key columns are forced NOT NULL by schema normalization.
    EXPECT_EQ(info.schema.columns[0].data_type.id(), fluss::TypeId::Int);
    EXPECT_FALSE(info.schema.columns[0].data_type.nullable());

    // "name" STRING (nullable)
    EXPECT_EQ(info.schema.columns[1].data_type.id(), fluss::TypeId::String);
    EXPECT_TRUE(info.schema.columns[1].data_type.nullable());

    // "tags" ARRAY<STRING NOT NULL> (outer nullable)
    EXPECT_EQ(info.schema.columns[2].data_type.id(), fluss::TypeId::Array);
    EXPECT_TRUE(info.schema.columns[2].data_type.nullable());
    ASSERT_NE(info.schema.columns[2].data_type.element_type(), nullptr);
    EXPECT_FALSE(info.schema.columns[2].data_type.element_type()->nullable());

    // "ids" ARRAY<INT> NOT NULL (outer not null, element nullable)
    EXPECT_EQ(info.schema.columns[3].data_type.id(), fluss::TypeId::Array);
    EXPECT_FALSE(info.schema.columns[3].data_type.nullable());
    ASSERT_NE(info.schema.columns[3].data_type.element_type(), nullptr);
    EXPECT_TRUE(info.schema.columns[3].data_type.element_type()->nullable());

    // "nested" ARRAY<ARRAY<INT> NOT NULL> (outer nullable, inner array not null)
    EXPECT_EQ(info.schema.columns[4].data_type.id(), fluss::TypeId::Array);
    EXPECT_TRUE(info.schema.columns[4].data_type.nullable());
    ASSERT_NE(info.schema.columns[4].data_type.element_type(), nullptr);
    EXPECT_FALSE(info.schema.columns[4].data_type.element_type()->nullable());
    ASSERT_NE(info.schema.columns[4].data_type.element_type()->element_type(), nullptr);
    EXPECT_TRUE(info.schema.columns[4].data_type.element_type()->element_type()->nullable());

    ASSERT_OK(adm.DropTable(table_path, false));
}