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

#include <arrow/api.h>

#include "test_utils.h"

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

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

TEST_F(KvTableTest, UpsertDeleteAndLookup) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .AddColumn("age", fluss::DataType::BigInt())
                      .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));

    // Create upsert writer
    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    // Upsert 3 rows (fire-and-forget, then flush)
    struct TestData {
        int32_t id;
        std::string name;
        int64_t age;
    };
    std::vector<TestData> test_data = {{1, "Verso", 32}, {2, "Noco", 25}, {3, "Esquie", 35}};

    for (const auto& d : test_data) {
        fluss::GenericRow row(3);
        row.SetInt32(0, d.id);
        row.SetString(1, d.name);
        row.SetInt64(2, d.age);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_writer.Flush());

    // Create lookuper
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    // Verify lookup results
    for (const auto& d : test_data) {
        fluss::GenericRow key(3);
        key.SetInt32(0, d.id);

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found()) << "Row with id=" << d.id << " should exist";

        EXPECT_EQ(result.GetInt32(0), d.id) << "id mismatch";
        EXPECT_EQ(result.GetString(1), d.name) << "name mismatch";
        EXPECT_EQ(result.GetInt64(2), d.age) << "age mismatch";
    }

    // Update record with id=1 (await acknowledgment)
    {
        fluss::GenericRow updated_row(3);
        updated_row.SetInt32(0, 1);
        updated_row.SetString(1, "Verso");
        updated_row.SetInt64(2, 33);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(updated_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify the update
    {
        fluss::GenericRow key(3);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt64(2), 33) << "Age should be updated";
        EXPECT_EQ(result.GetString(1), "Verso") << "Name should remain unchanged";
    }

    // Delete record with id=1 (await acknowledgment)
    {
        fluss::GenericRow delete_row(3);
        delete_row.SetInt32(0, 1);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Delete(delete_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify deletion
    {
        fluss::GenericRow key(3);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_FALSE(result.Found()) << "Record 1 should not exist after delete";
    }

    // Verify other records still exist
    for (int id : {2, 3}) {
        fluss::GenericRow key(3);
        key.SetInt32(0, id);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found()) << "Record " << id
                                    << " should still exist after deleting record 1";
    }

    // Lookup non-existent key
    {
        fluss::GenericRow key(3);
        key.SetInt32(0, 999);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_FALSE(result.Found()) << "Non-existent key should return not found";
    }

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

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

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

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

    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));

    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));
    std::vector<std::pair<int32_t, std::string>> rows = {
        {1, "Verso"}, {2, "Noco"}, {3, "Esquie"}, {4, "Aline"}, {5, "Gustave"}};
    for (const auto& [id, name] : rows) {
        fluss::GenericRow row(2);
        row.SetInt32(0, id);
        row.SetString(1, name);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_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 scanned = 0;
    for (const auto& b : first) {
        EXPECT_EQ(b->GetBucketId(), 0);
        scanned += b->NumRows();
    }
    EXPECT_GT(scanned, 0);
    EXPECT_LE(scanned, 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(100).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);

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

TEST_F(KvTableTest, LookupWithNestedArray) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("matrix",
                                 fluss::DataType::Array(fluss::DataType::Array(fluss::DataType::Int())))
                      .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 upsert = table.NewUpsert();
    fluss::UpsertWriter writer;
    ASSERT_OK(upsert.CreateWriter(writer));

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

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

        fluss::ArrayWriter inner2(2, fluss::DataType::Int());
        inner2.SetInt32(0, 21);
        inner2.SetInt32(1, 22);

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

        ASSERT_OK(writer.Upsert(row));
        ASSERT_OK(writer.Flush());
    }

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

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

    fluss::LookupResult result;
    ASSERT_OK(lookuper.Lookup(key, result));
    ASSERT_TRUE(result.Found());
    auto outer = result.GetValue("matrix");
    EXPECT_EQ(outer.Type(), fluss::TypeId::Array);
    ASSERT_EQ(outer.Size(), 2u);

    auto first = outer.At(0);
    EXPECT_EQ(first.Type(), fluss::TypeId::Array);
    ASSERT_EQ(first.Size(), 2u);
    EXPECT_EQ(first.At(0).GetInt32(), 11);
    EXPECT_EQ(first.At(1).GetInt32(), 12);

    auto second = outer.At(1);
    ASSERT_EQ(second.Size(), 2u);
    EXPECT_EQ(second.At(0).GetInt32(), 21);
    EXPECT_EQ(second.At(1).GetInt32(), 22);

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

TEST_F(KvTableTest, LookupComplexTypesMatrix) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto row_seq_label = arrow::struct_(
        {arrow::field("seq", arrow::int32()), arrow::field("label", arrow::utf8())});

    auto arrow_schema = arrow::schema({
        arrow::field("id", arrow::int32()),
        arrow::field("m_str_int", arrow::map(arrow::utf8(), arrow::int32())),
        arrow::field("m_str_row", arrow::map(arrow::utf8(), row_seq_label)),
        arrow::field("m_str_map",
                     arrow::map(arrow::utf8(), arrow::map(arrow::utf8(), arrow::int32()))),
        arrow::field("m_str_arr", arrow::map(arrow::utf8(), arrow::list(arrow::int32()))),
        arrow::field("arr_map", arrow::list(arrow::map(arrow::utf8(), arrow::int32()))),
        arrow::field("arr_row", arrow::list(row_seq_label)),
        arrow::field("r_deep", arrow::struct_({arrow::field(
                                   "inner", arrow::struct_({arrow::field("n", arrow::int32())}))})),
        arrow::field("r_with_arr",
                     arrow::struct_({arrow::field("f_int", arrow::int32()),
                                     arrow::field("f_arr", arrow::list(arrow::int32()))})),
        // row_rich: every scalar type + an array field in one ROW.
        arrow::field("r_rich",
                     arrow::struct_({
                         arrow::field("f_bool", arrow::boolean()),
                         arrow::field("f_int", arrow::int32()),
                         arrow::field("f_long", arrow::int64()),
                         arrow::field("f_float", arrow::float32()),
                         arrow::field("f_double", arrow::float64()),
                         arrow::field("f_str", arrow::utf8()),
                         arrow::field("f_bytes", arrow::binary()),
                         arrow::field("f_decimal", arrow::decimal128(10, 2)),
                         arrow::field("f_date", arrow::date32()),
                         arrow::field("f_time", arrow::time32(arrow::TimeUnit::MILLI)),
                         arrow::field("f_ts_ntz", arrow::timestamp(arrow::TimeUnit::MICRO)),
                         arrow::field("f_ts_ltz", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")),
                         arrow::field("f_binary", arrow::fixed_size_binary(4)),
                         arrow::field("f_arr", arrow::list(arrow::int32())),
                     })),
        arrow::field("m_str_tiny", arrow::map(arrow::utf8(), arrow::int8())),
        arrow::field("arr_small", arrow::list(arrow::int16())),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema, {"id"});

    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 upsert = table.NewUpsert();
    fluss::UpsertWriter writer;
    ASSERT_OK(upsert.CreateWriter(writer));

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

        // map<string,int> — second entry has a NULL value.
        {
            fluss::MapWriter m(2, fluss::DataType::String(), fluss::DataType::Int());
            m.SetKeyString("a");
            m.SetValueInt32(1);
            m.Commit();
            m.SetKeyString("b");
            m.SetValueNull();
            m.Commit();
            row.Set("m_str_int", std::move(m));
        }
        // map<string, row<seq,label>> — value is a ROW, so the Arrow ctor.
        {
            fluss::MapWriter m(1, arrow::utf8(), row_seq_label);
            m.SetKeyString("k");
            fluss::GenericRow v(2);
            v.SetInt32(0, 7);
            v.SetString(1, "seven");
            m.SetValueRow(std::move(v));
            m.Commit();
            row.Set("m_str_row", std::move(m));
        }
        // map<string, map<string,int>>
        {
            fluss::MapWriter m(1, arrow::utf8(), arrow::map(arrow::utf8(), arrow::int32()));
            m.SetKeyString("k");
            fluss::MapWriter inner(1, fluss::DataType::String(), fluss::DataType::Int());
            inner.SetKeyString("x");
            inner.SetValueInt32(9);
            inner.Commit();
            m.SetValueMap(std::move(inner));
            m.Commit();
            row.Set("m_str_map", std::move(m));
        }
        // map<string, array<int>> — value array fits the flat ctor.
        {
            fluss::MapWriter m(1, fluss::DataType::String(),
                               fluss::DataType::Array(fluss::DataType::Int()));
            m.SetKeyString("k");
            fluss::ArrayWriter v(2, fluss::DataType::Int());
            v.SetInt32(0, 10);
            v.SetInt32(1, 20);
            m.SetValueArray(std::move(v));
            m.Commit();
            row.Set("m_str_arr", std::move(m));
        }
        // array<map<string,int>> — element is a MAP, so the Arrow ctor.
        {
            fluss::ArrayWriter a(1, arrow::map(arrow::utf8(), arrow::int32()));
            fluss::MapWriter e(1, fluss::DataType::String(), fluss::DataType::Int());
            e.SetKeyString("p");
            e.SetValueInt32(5);
            e.Commit();
            a.SetMap(0, std::move(e));
            row.Set("arr_map", std::move(a));
        }
        // array<row<seq,label>> — element is a ROW, so the Arrow ctor.
        {
            fluss::ArrayWriter a(2, row_seq_label);
            fluss::GenericRow e0(2);
            e0.SetInt32(0, 1);
            e0.SetString(1, "one");
            fluss::GenericRow e1(2);
            e1.SetInt32(0, 2);
            e1.SetString(1, "two");
            a.SetRow(0, std::move(e0));
            a.SetRow(1, std::move(e1));
            row.Set("arr_row", std::move(a));
        }
        // row<inner: row<n>>
        {
            fluss::GenericRow inner(1);
            inner.SetInt32(0, 42);
            fluss::GenericRow outer(1);
            outer.SetRow(0, std::move(inner));
            row.Set("r_deep", std::move(outer));
        }
        // row<f_int, f_arr: array<int>>
        {
            fluss::GenericRow r(2);
            r.SetInt32(0, 100);
            fluss::ArrayWriter arr(3, fluss::DataType::Int());
            arr.SetInt32(0, 1);
            arr.SetInt32(1, 2);
            arr.SetInt32(2, 3);
            r.SetArray(1, std::move(arr));
            row.Set("r_with_arr", std::move(r));
        }
        // row_rich: exercise every Value leaf getter + an array field.
        {
            fluss::GenericRow rr(14);
            rr.SetBool(0, true);
            rr.SetInt32(1, 100000);
            rr.SetInt64(2, 9876543210LL);
            rr.SetFloat32(3, 3.5F);
            rr.SetFloat64(4, 2.5);
            rr.SetString(5, "hello world");
            rr.SetBytes(6, {'b', 'i', 'n'});
            rr.SetDecimal(7, "123.45");
            rr.SetDate(8, fluss::Date{20476});
            rr.SetTime(9, fluss::Time{36827123});
            rr.SetTimestampNtz(10, fluss::Timestamp{1769163227123LL, 456000});
            rr.SetTimestampLtz(11, fluss::Timestamp{1769163227456LL, 0});
            rr.SetBytes(12, {1, 2, 3, 4});
            fluss::ArrayWriter farr(3, fluss::DataType::Int());
            farr.SetInt32(0, 7);
            farr.SetNull(1);
            farr.SetInt32(2, 11);
            rr.SetArray(13, std::move(farr));
            row.Set("r_rich", std::move(rr));
        }
        // map<string,tinyint>
        {
            fluss::MapWriter m(2, fluss::DataType::String(), fluss::DataType::TinyInt());
            m.SetKeyString("lo");
            m.SetValueInt32(-128);
            m.Commit();
            m.SetKeyString("hi");
            m.SetValueInt32(127);
            m.Commit();
            row.Set("m_str_tiny", std::move(m));
        }
        // array<smallint>
        {
            fluss::ArrayWriter a(2, fluss::DataType::SmallInt());
            a.SetInt32(0, 1000);
            a.SetInt32(1, -2000);
            row.Set("arr_small", std::move(a));
        }

        ASSERT_OK(writer.Upsert(row));
        ASSERT_OK(writer.Flush());
    }

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));
    auto key = table.NewRow();
    key.Set("id", 1);
    fluss::LookupResult result;
    ASSERT_OK(lookuper.Lookup(key, result));
    ASSERT_TRUE(result.Found());

    // map<string,int> — entry 1 has a NULL value.
    {
        auto m = result.GetValue("m_str_int");
        EXPECT_EQ(m.Type(), fluss::TypeId::Map);
        ASSERT_EQ(m.Size(), 2u);
        EXPECT_EQ(m.KeyAt(0).GetString(), "a");
        EXPECT_FALSE(m.ValueAt(0).IsNull());
        EXPECT_EQ(m.ValueAt(0).GetInt32(), 1);
        EXPECT_EQ(m.KeyAt(1).GetString(), "b");
        EXPECT_TRUE(m.ValueAt(1).IsNull());
    }
    // map<string, row<seq,label>>
    {
        auto m = result.GetValue("m_str_row");
        ASSERT_EQ(m.Size(), 1u);
        EXPECT_EQ(m.KeyAt(0).GetString(), "k");
        auto v = m.ValueAt(0);
        EXPECT_EQ(v.Field(0).GetInt32(), 7);
        EXPECT_EQ(v.Field(1).GetString(), "seven");
    }
    // map<string, map<string,int>>
    {
        auto inner = result.GetValue("m_str_map").ValueAt(0);
        ASSERT_EQ(inner.Size(), 1u);
        EXPECT_EQ(inner.KeyAt(0).GetString(), "x");
        EXPECT_EQ(inner.ValueAt(0).GetInt32(), 9);
    }
    // map<string, array<int>>
    {
        auto av = result.GetValue("m_str_arr").ValueAt(0);
        ASSERT_EQ(av.Size(), 2u);
        EXPECT_EQ(av.At(0).GetInt32(), 10);
        EXPECT_EQ(av.At(1).GetInt32(), 20);
    }
    // array<map<string,int>>
    {
        auto a = result.GetValue("arr_map");
        ASSERT_EQ(a.Size(), 1u);
        auto e = a.At(0);
        EXPECT_EQ(e.KeyAt(0).GetString(), "p");
        EXPECT_EQ(e.ValueAt(0).GetInt32(), 5);
    }
    // array<row<seq,label>>
    {
        auto a = result.GetValue("arr_row");
        EXPECT_EQ(a.Type(), fluss::TypeId::Array);
        ASSERT_EQ(a.Size(), 2u);
        EXPECT_EQ(a.At(0).Field(0).GetInt32(), 1);
        EXPECT_EQ(a.At(0).Field(1).GetString(), "one");
        EXPECT_EQ(a.At(1).Field(0).GetInt32(), 2);
        EXPECT_EQ(a.At(1).Field(1).GetString(), "two");
    }
    // row<inner: row<n>>
    {
        auto inner = result.GetValue("r_deep").Field(0);
        EXPECT_EQ(inner.Field(0).GetInt32(), 42);
    }
    // row<f_int, f_arr>
    {
        auto r = result.GetValue("r_with_arr");
        EXPECT_EQ(r.Type(), fluss::TypeId::Row);
        EXPECT_EQ(r.Field(0).GetInt32(), 100);
        EXPECT_EQ(r.Field("f_int").GetInt32(), 100);  // ROW field by name
        auto arr = r.Field(1);
        ASSERT_EQ(arr.Size(), 3u);
        EXPECT_EQ(arr.At(2).GetInt32(), 3);
    }
    // row_rich — every leaf getter on one Value handle.
    {
        auto rr = result.GetValue("r_rich");
        ASSERT_EQ(rr.FieldCount(), 14u);
        EXPECT_TRUE(rr.Field(0).GetBool());
        EXPECT_EQ(rr.Field(1).GetInt32(), 100000);
        EXPECT_EQ(rr.Field(2).GetInt64(), 9876543210LL);
        EXPECT_FLOAT_EQ(rr.Field(3).GetFloat32(), 3.5F);
        EXPECT_DOUBLE_EQ(rr.Field(4).GetFloat64(), 2.5);
        EXPECT_EQ(rr.Field(5).GetString(), "hello world");
        auto by = rr.Field(6).GetBytes();
        ASSERT_EQ(by.size(), 3u);
        EXPECT_EQ(by[0], 'b');
        EXPECT_EQ(rr.Field(7).GetDecimalString(), "123.45");
        EXPECT_EQ(rr.Field(8).GetDate().days_since_epoch, 20476);
        EXPECT_EQ(rr.Field(9).GetTime().millis_since_midnight, 36827123);
        EXPECT_EQ(rr.Field(10).GetTimestamp().epoch_millis, 1769163227123LL);
        EXPECT_EQ(rr.Field(11).GetTimestamp().epoch_millis, 1769163227456LL);
        auto bin = rr.Field(12).GetBytes();
        ASSERT_EQ(bin.size(), 4u);
        EXPECT_EQ(bin[3], 4);
        auto fa = rr.Field(13);
        ASSERT_EQ(fa.Size(), 3u);
        EXPECT_TRUE(fa.At(1).IsNull());
        EXPECT_EQ(fa.At(2).GetInt32(), 11);
    }
    // map<string,tinyint>
    {
        auto m = result.GetValue("m_str_tiny");
        ASSERT_EQ(m.Size(), 2u);
        EXPECT_EQ(m.ValueAt(0).GetInt32(), -128);
        EXPECT_EQ(m.ValueAt(1).GetInt32(), 127);
    }
    // array<smallint>
    {
        auto a = result.GetValue("arr_small");
        ASSERT_EQ(a.Size(), 2u);
        EXPECT_EQ(a.At(0).GetInt32(), 1000);
        EXPECT_EQ(a.At(1).GetInt32(), -2000);
    }

    // Row 2 (id=2) — every compound column NULL.
    {
        const int column_count = arrow_schema->num_fields();
        auto row = table.NewRow();
        row.SetInt32(0, 2);
        for (int i = 1; i < column_count; ++i) {
            row.SetNull(i);
        }
        ASSERT_OK(writer.Upsert(row));
        ASSERT_OK(writer.Flush());

        auto key2 = table.NewRow();
        key2.SetInt32(0, 2);
        fluss::LookupResult result2;
        ASSERT_OK(lookuper.Lookup(key2, result2));
        ASSERT_TRUE(result2.Found());
        EXPECT_EQ(result2.GetInt32(0), 2);
        for (int i = 1; i < column_count; ++i) {
            EXPECT_TRUE(result2.IsNull(i)) << "column " << i << " should be null";
        }
    }

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

// Regression: timestamp map values built via the Arrow MapWriter ctor must pick
// NTZ vs LTZ from the declared value type (previously always wrote NTZ).
TEST_F(KvTableTest, MapWithTimestampValuesNtzAndLtz) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto arrow_schema = arrow::schema({
        arrow::field("id", arrow::int32()),
        arrow::field("mn", arrow::map(arrow::utf8(), arrow::timestamp(arrow::TimeUnit::MICRO))),
        arrow::field("ml",
                     arrow::map(arrow::utf8(), arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"))),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema, {"id"});

    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));
    fluss::UpsertWriter writer;
    ASSERT_OK(table.NewUpsert().CreateWriter(writer));

    const fluss::Timestamp ntz_val{1769163227123LL, 456000};
    const fluss::Timestamp ltz_val{1700000000789LL, 123000};
    {
        auto row = table.NewRow();
        row.SetInt32(0, 1);
        // mn: map<string, timestamp> (NTZ), via the Arrow ctor.
        fluss::MapWriter mn(1, arrow::utf8(), arrow::timestamp(arrow::TimeUnit::MICRO));
        mn.SetKeyString("k");
        mn.SetValueTimestamp(ntz_val);
        mn.Commit();
        row.SetMap(1, std::move(mn));
        // ml: map<string, timestamp_ltz> (LTZ), via the Arrow ctor.
        fluss::MapWriter ml(1, arrow::utf8(), arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"));
        ml.SetKeyString("k");
        ml.SetValueTimestamp(ltz_val);
        ml.Commit();
        row.SetMap(2, std::move(ml));
        ASSERT_OK(writer.Upsert(row));
        ASSERT_OK(writer.Flush());
    }

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));
    auto key = table.NewRow();
    key.SetInt32(0, 1);
    fluss::LookupResult result;
    ASSERT_OK(lookuper.Lookup(key, result));
    ASSERT_TRUE(result.Found());

    auto mn = result.GetValue("mn");
    ASSERT_EQ(mn.Size(), 1u);
    EXPECT_EQ(mn.ValueAt(0).GetTimestamp().epoch_millis, ntz_val.epoch_millis);
    EXPECT_EQ(mn.ValueAt(0).GetTimestamp().nano_of_millisecond, ntz_val.nano_of_millisecond);

    auto ml = result.GetValue("ml");
    ASSERT_EQ(ml.Size(), 1u);
    EXPECT_EQ(ml.ValueAt(0).GetTimestamp().epoch_millis, ltz_val.epoch_millis);
    EXPECT_EQ(ml.ValueAt(0).GetTimestamp().nano_of_millisecond, ltz_val.nano_of_millisecond);

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

// Deeply nested MAP/ROW via native DataType::Map / DataType::Row only — no Arrow.
TEST_F(KvTableTest, NativeNestedBuilderNoArrow) {
    auto& adm = admin();
    auto& conn = connection();

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

    // array<row<seq: int, attrs: map<string, int>>>
    auto event = fluss::DataType::Row({
        {"seq", fluss::DataType::Int()},
        {"attrs", fluss::DataType::Map(fluss::DataType::String(), fluss::DataType::Int())},
    });
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("events", fluss::DataType::Array(event))
                      .AddColumn("profile", fluss::DataType::Row({
                                                {"name", fluss::DataType::String()},
                                                {"score", fluss::DataType::Double()},
                                            }))
                      .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));
    fluss::UpsertWriter writer;
    ASSERT_OK(table.NewUpsert().CreateWriter(writer));

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

        // events = [ {0, {"a":0}}, {1, {"a":10,"b":11}} ]
        fluss::ArrayWriter events(2, event);
        for (int i = 0; i < 2; i++) {
            fluss::GenericRow ev(2);
            ev.SetInt32(0, i);
            fluss::MapWriter attrs(static_cast<size_t>(i + 1), fluss::DataType::String(),
                                   fluss::DataType::Int());
            attrs.SetKeyString("a");
            attrs.SetValueInt32(i * 10);
            attrs.Commit();
            if (i == 1) {
                attrs.SetKeyString("b");
                attrs.SetValueInt32(11);
                attrs.Commit();
            }
            ev.SetMap(1, std::move(attrs));
            events.SetRow(i, std::move(ev));
        }
        row.Set("events", std::move(events));

        fluss::GenericRow profile(2);
        profile.SetString(0, "alice");
        profile.SetFloat64(1, 9.5);
        row.Set("profile", std::move(profile));

        ASSERT_OK(writer.Upsert(row));
        ASSERT_OK(writer.Flush());
    }

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));
    auto key = table.NewRow();
    key.SetInt32(0, 1);
    fluss::LookupResult result;
    ASSERT_OK(lookuper.Lookup(key, result));
    ASSERT_TRUE(result.Found());

    auto events = result.GetValue("events");  // ARRAY<ROW<int, MAP<string,int>>>
    ASSERT_EQ(events.Type(), fluss::TypeId::Array);
    ASSERT_EQ(events.Size(), 2u);
    EXPECT_EQ(events.At(0).Field("seq").GetInt32(), 0);
    EXPECT_EQ(events.At(0).Field("attrs").ValueAt(0).GetInt32(), 0);
    auto e1_attrs = events.At(1).Field("attrs");
    ASSERT_EQ(e1_attrs.Size(), 2u);
    EXPECT_EQ(e1_attrs.KeyAt(1).GetString(), "b");
    EXPECT_EQ(e1_attrs.ValueAt(1).GetInt32(), 11);

    auto profile = result.GetValue("profile");  // ROW<name, score>
    EXPECT_EQ(profile.Field("name").GetString(), "alice");
    EXPECT_DOUBLE_EQ(profile.Field("score").GetFloat64(), 9.5);

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

TEST_F(KvTableTest, ComplexTypesPartialUpdate) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto arrow_schema = arrow::schema({
        arrow::field("id", arrow::int32()),
        arrow::field("name", arrow::utf8()),
        arrow::field("score", arrow::int64()),
        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())),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema, {"id"});

    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 upsert = table.NewUpsert();
        fluss::UpsertWriter w;
        ASSERT_OK(upsert.CreateWriter(w));
        auto row = table.NewRow();
        row.SetInt32(0, 1);
        row.SetString(1, "Verso");
        row.SetInt64(2, 100);
        fluss::GenericRow nested(2);
        nested.SetInt32(0, 10);
        nested.SetString(1, "alpha");
        row.SetRow(3, std::move(nested));
        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.SetMap(4, std::move(attrs));
        fluss::ArrayWriter tags(2, fluss::DataType::String());
        tags.SetString(0, "x");
        tags.SetString(1, "y");
        row.SetArray(5, std::move(tags));
        ASSERT_OK(w.Upsert(row));
        ASSERT_OK(w.Flush());
    }

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));
    auto key = table.NewRow();
    key.SetInt32(0, 1);

    // Partial update of a scalar column — compound columns must be preserved.
    {
        auto pu = table.NewUpsert();
        pu.PartialUpdateByName({"id", "score"});
        fluss::UpsertWriter pw;
        ASSERT_OK(pu.CreateWriter(pw));
        auto row = table.NewRow();
        row.SetInt32(0, 1);
        row.SetNull(1);
        row.SetInt64(2, 420);
        row.SetNull(3);
        row.SetNull(4);
        row.SetNull(5);
        ASSERT_OK(pw.Upsert(row));
        ASSERT_OK(pw.Flush());

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetString(1), "Verso");
        EXPECT_EQ(result.GetInt64(2), 420);
        auto n = result.GetValue(3);
        EXPECT_EQ(n.Field(0).GetInt32(), 10);
        EXPECT_EQ(n.Field(1).GetString(), "alpha");
        EXPECT_EQ(result.GetValue(4).Size(), 2u);
        EXPECT_EQ(result.GetValue(5).Size(), 2u);
    }

    // Partial update of the ROW column — other compound columns preserved.
    {
        auto pu = table.NewUpsert();
        pu.PartialUpdateByName({"id", "nested"});
        fluss::UpsertWriter pw;
        ASSERT_OK(pu.CreateWriter(pw));
        auto row = table.NewRow();
        row.SetInt32(0, 1);
        row.SetNull(1);
        row.SetNull(2);
        fluss::GenericRow nn(2);
        nn.SetInt32(0, 99);
        nn.SetString(1, "omega");
        row.SetRow(3, std::move(nn));
        row.SetNull(4);
        row.SetNull(5);
        ASSERT_OK(pw.Upsert(row));
        ASSERT_OK(pw.Flush());

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt64(2), 420);
        auto n = result.GetValue(3);
        EXPECT_EQ(n.Field(0).GetInt32(), 99);
        EXPECT_EQ(n.Field(1).GetString(), "omega");
        EXPECT_EQ(result.GetValue(4).Size(), 2u);
        EXPECT_EQ(result.GetValue(5).Size(), 2u);
    }

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

TEST_F(KvTableTest, PartitionedComplexTypes) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto arrow_schema = arrow::schema({
        arrow::field("region", arrow::utf8()),
        arrow::field("user_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())),
    });
    auto schema = fluss::Schema::FromArrow(arrow_schema, {"region", "user_id"});

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetPartitionKeys({"region"})
                                .SetProperty("table.replication.factor", "1")
                                .Build();
    fluss_test::CreateTable(adm, table_path, table_descriptor);
    fluss_test::CreatePartitions(adm, table_path, "region", {"US", "EU"});

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

    fluss::UpsertWriter writer;
    ASSERT_OK(table.NewUpsert().CreateWriter(writer));

    struct Rec {
        const char* region;
        int32_t user_id;
        int32_t seq;
        const char* label;
    };
    const Rec data[] = {{"US", 1, 11, "alpha"}, {"EU", 2, 22, "beta"}};

    for (const auto& d : data) {
        auto row = table.NewRow();
        row.SetString(0, d.region);
        row.SetInt32(1, d.user_id);
        fluss::GenericRow nested(2);
        nested.SetInt32(0, d.seq);
        nested.SetString(1, d.label);
        row.SetRow(2, std::move(nested));
        fluss::MapWriter attrs(1, fluss::DataType::String(), fluss::DataType::Int());
        attrs.SetKeyString(d.label);
        attrs.SetValueInt32(d.seq);
        attrs.Commit();
        row.SetMap(3, std::move(attrs));
        ASSERT_OK(writer.Upsert(row));
    }
    ASSERT_OK(writer.Flush());

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    for (const auto& d : data) {
        auto key = table.NewRow();
        key.SetString(0, d.region);
        key.SetInt32(1, d.user_id);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        auto nested = result.GetValue(2);
        EXPECT_EQ(nested.Field(0).GetInt32(), d.seq);
        EXPECT_EQ(nested.Field(1).GetString(), d.label);
        auto attrs = result.GetValue(3);
        ASSERT_EQ(attrs.Size(), 1u);
        EXPECT_EQ(attrs.KeyAt(0).GetString(), d.label);
        EXPECT_EQ(attrs.ValueAt(0).GetInt32(), d.seq);
    }

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

TEST_F(KvTableTest, LookupArrayValidationErrors) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("vals", fluss::DataType::Array(fluss::DataType::Int()))
                      .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 upsert = table.NewUpsert();
    fluss::UpsertWriter writer;
    ASSERT_OK(upsert.CreateWriter(writer));

    auto row = table.NewRow();
    row.Set("id", 1);
    fluss::ArrayWriter vals(2, fluss::DataType::Int());
    vals.SetInt32(0, 99);
    vals.SetNull(1);
    row.Set("vals", std::move(vals));
    ASSERT_OK(writer.Upsert(row));
    ASSERT_OK(writer.Flush());

    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    auto key = table.NewRow();
    key.Set("id", 1);
    fluss::LookupResult result;
    ASSERT_OK(lookuper.Lookup(key, result));
    ASSERT_TRUE(result.Found());

    auto view = result.GetValue("vals");
    EXPECT_EQ(view.Type(), fluss::TypeId::Array);
    ASSERT_EQ(view.Size(), 2u);
    EXPECT_EQ(view.At(0).GetInt32(), 99);
    EXPECT_TRUE(view.At(1).IsNull());

    // A 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);

    // A 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(KvTableTest, CompositePrimaryKeys) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("region", fluss::DataType::String())
                      .AddColumn("score", fluss::DataType::BigInt())
                      .AddColumn("user_id", fluss::DataType::Int())
                      .SetPrimaryKeys({"region", "user_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 table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    // Insert records with composite keys
    struct TestData {
        std::string region;
        int32_t user_id;
        int64_t score;
    };
    std::vector<TestData> test_data = {
        {"US", 1, 100}, {"US", 2, 200}, {"EU", 1, 150}, {"EU", 2, 250}};

    for (const auto& d : test_data) {
        auto row = table.NewRow();
        row.Set("region", d.region);
        row.Set("score", d.score);
        row.Set("user_id", d.user_id);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_writer.Flush());

    // Create lookuper
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    // Lookup (US, 1) - should return score 100
    {
        auto key = table.NewRow();
        key.Set("region", "US");
        key.Set("user_id", 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt64("score"), 100) << "Score for (US, 1) should be 100";
    }

    // Lookup (EU, 2) - should return score 250
    {
        auto key = table.NewRow();
        key.Set("region", "EU");
        key.Set("user_id", 2);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt64("score"), 250) << "Score for (EU, 2) should be 250";
    }

    // Update (US, 1) score (await acknowledgment)
    {
        auto update_row = table.NewRow();
        update_row.Set("region", "US");
        update_row.Set("user_id", 1);
        update_row.Set("score", static_cast<int64_t>(500));
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(update_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify update
    {
        auto key = table.NewRow();
        key.Set("region", "US");
        key.Set("user_id", 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt64("score"), 500) << "Row score should be updated";
    }

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

TEST_F(KvTableTest, PrefixLookupByBucketKey) {
    auto& adm = admin();
    auto& conn = connection();

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

    // Bucket key (a, b) is a strict prefix of the PK (a, b, c), enabling prefix lookup on (a, b).
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("a", fluss::DataType::Int())
                      .AddColumn("b", fluss::DataType::String())
                      .AddColumn("c", fluss::DataType::BigInt())
                      .AddColumn("d", fluss::DataType::String())
                      .SetPrimaryKeys({"a", "b", "c"})
                      .Build();

    auto table_descriptor = fluss::TableDescriptor::NewBuilder()
                                .SetSchema(schema)
                                .SetBucketKeys({"a", "b"})
                                .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_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    struct TestData {
        int32_t a;
        std::string b;
        int64_t c;
        std::string d;
    };
    std::vector<TestData> test_data = {
        {1, "aaaaaaaaa", 1, "value1"},
        {1, "aaaaaaaaa", 2, "value2"},
        {1, "aaaaaaaaa", 3, "value3"},
        {2, "aaaaaaaaa", 4, "value4"},
    };

    for (const auto& row_data : test_data) {
        auto row = table.NewRow();
        row.Set("a", row_data.a);
        row.Set("b", row_data.b);
        row.Set("c", row_data.c);
        row.Set("d", row_data.d);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_writer.Flush());

    fluss::PrefixLookuper prefix_lookuper;
    ASSERT_OK(table.NewPrefixLookup({"a", "b"}, prefix_lookuper));

    // Prefix (1, "aaaaaaaaa") matches 3 rows, returned in primary-key order.
    {
        auto key = table.NewRow();
        key.Set("a", 1);
        key.Set("b", "aaaaaaaaa");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        ASSERT_EQ(result.Size(), 3u);
        for (size_t i = 0; i < result.Size(); ++i) {
            auto row = result.GetRow(i);
            EXPECT_EQ(row.GetInt32("a"), 1);
            EXPECT_EQ(row.GetString("b"), "aaaaaaaaa");
            EXPECT_EQ(row.GetInt64("c"), static_cast<int64_t>(i) + 1);
            EXPECT_EQ(row.GetString("d"), "value" + std::to_string(i + 1));
        }
    }

    // Prefix (2, "aaaaaaaaa") matches a single row.
    {
        auto key = table.NewRow();
        key.Set("a", 2);
        key.Set("b", "aaaaaaaaa");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        ASSERT_EQ(result.Size(), 1u);
        auto row = result.GetRow(0);
        EXPECT_EQ(row.GetInt64("c"), 4);
        EXPECT_EQ(row.GetString("d"), "value4");
    }

    // A prefix with no matching rows yields an empty result.
    {
        auto key = table.NewRow();
        key.Set("a", 3);
        key.Set("b", "a");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        EXPECT_TRUE(result.IsEmpty());
    }

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

TEST_F(KvTableTest, PrefixLookupValidationErrors) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("a", fluss::DataType::Int())
                      .AddColumn("b", fluss::DataType::String())
                      .AddColumn("c", fluss::DataType::BigInt())
                      .SetPrimaryKeys({"a", "b", "c"})
                      .Build();

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

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

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

    // Lookup columns must list the bucket keys (a, b) in order.
    {
        fluss::PrefixLookuper lookuper;
        auto result = table.NewPrefixLookup({"b", "a"}, lookuper);
        ASSERT_FALSE(result.Ok());
        EXPECT_EQ(result.error_code, fluss::ErrorCode::CLIENT_ERROR);
        EXPECT_NE(result.error_message.find("must contain all bucket keys"), std::string::npos);
    }

    // Lookup columns must equal the bucket keys, not a longer prefix of the PK.
    {
        fluss::PrefixLookuper lookuper;
        auto result = table.NewPrefixLookup({"a", "b", "c"}, lookuper);
        ASSERT_FALSE(result.Ok());
        EXPECT_EQ(result.error_code, fluss::ErrorCode::CLIENT_ERROR);
        EXPECT_NE(result.error_message.find("must contain all bucket keys"), std::string::npos);
    }

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

TEST_F(KvTableTest, PrefixLookupPartitioned) {
    auto& adm = admin();
    auto& conn = connection();

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

    // Partitioned by region; bucket key (a, b) is a prefix of the PK minus the partition column.
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("region", fluss::DataType::String())
                      .AddColumn("a", fluss::DataType::Int())
                      .AddColumn("b", fluss::DataType::String())
                      .AddColumn("c", fluss::DataType::BigInt())
                      .AddColumn("d", fluss::DataType::String())
                      .SetPrimaryKeys({"region", "a", "b", "c"})
                      .Build();

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

    fluss_test::CreateTable(adm, table_path, table_descriptor);
    fluss_test::CreatePartitions(adm, table_path, "region", {"US", "EU"});

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

    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    struct TestData {
        std::string region;
        int32_t a;
        std::string b;
        int64_t c;
        std::string d;
    };
    std::vector<TestData> test_data = {
        {"US", 1, "aaaaaaaaa", 1, "us-1"},
        {"US", 1, "aaaaaaaaa", 2, "us-2"},
        {"US", 2, "aaaaaaaaa", 3, "us-3"},
        {"EU", 1, "aaaaaaaaa", 4, "eu-1"},
        {"EU", 1, "bbbbbbbbb", 5, "eu-2"},
    };

    for (const auto& row_data : test_data) {
        auto row = table.NewRow();
        row.Set("region", row_data.region);
        row.Set("a", row_data.a);
        row.Set("b", row_data.b);
        row.Set("c", row_data.c);
        row.Set("d", row_data.d);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_writer.Flush());

    fluss::PrefixLookuper prefix_lookuper;
    ASSERT_OK(table.NewPrefixLookup({"region", "a", "b"}, prefix_lookuper));

    // (US, 1, "aaaaaaaaa") matches 2 rows.
    {
        auto key = table.NewRow();
        key.Set("region", "US");
        key.Set("a", 1);
        key.Set("b", "aaaaaaaaa");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        ASSERT_EQ(result.Size(), 2u);
        for (size_t i = 0; i < result.Size(); ++i) {
            auto row = result.GetRow(i);
            EXPECT_EQ(row.GetString("region"), "US");
            EXPECT_EQ(row.GetInt32("a"), 1);
            EXPECT_EQ(row.GetString("b"), "aaaaaaaaa");
        }
    }

    // (EU, 1, "bbbbbbbbb") matches a single row.
    {
        auto key = table.NewRow();
        key.Set("region", "EU");
        key.Set("a", 1);
        key.Set("b", "bbbbbbbbb");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        ASSERT_EQ(result.Size(), 1u);
        EXPECT_EQ(result.GetRow(0).GetString("d"), "eu-2");
    }

    // A non-existent partition yields an empty result.
    {
        auto key = table.NewRow();
        key.Set("region", "APAC");
        key.Set("a", 1);
        key.Set("b", "aaaaaaaaa");
        fluss::PrefixLookupResult result;
        ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
        EXPECT_TRUE(result.IsEmpty());
    }

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

TEST_F(KvTableTest, PartialUpdate) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .AddColumn("age", fluss::DataType::BigInt())
                      .AddColumn("score", fluss::DataType::BigInt())
                      .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));

    // Insert initial record with all columns
    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    {
        fluss::GenericRow row(4);
        row.SetInt32(0, 1);
        row.SetString(1, "Verso");
        row.SetInt64(2, 32);
        row.SetInt64(3, 6942);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify initial record
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    {
        fluss::GenericRow key(4);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt32(0), 1);
        EXPECT_EQ(result.GetString(1), "Verso");
        EXPECT_EQ(result.GetInt64(2), 32);
        EXPECT_EQ(result.GetInt64(3), 6942);
    }

    // Create partial update writer to update only score column
    auto partial_upsert = table.NewUpsert();
    partial_upsert.PartialUpdateByName({"id", "score"});
    fluss::UpsertWriter partial_writer;
    ASSERT_OK(partial_upsert.CreateWriter(partial_writer));

    // Update only the score column (await acknowledgment)
    {
        fluss::GenericRow partial_row(4);
        partial_row.SetInt32(0, 1);
        partial_row.SetNull(1);  // not in partial update
        partial_row.SetNull(2);  // not in partial update
        partial_row.SetInt64(3, 420);
        fluss::WriteResult wr;
        ASSERT_OK(partial_writer.Upsert(partial_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify partial update - name and age should remain unchanged
    {
        fluss::GenericRow key(4);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt32(0), 1) << "id should remain 1";
        EXPECT_EQ(result.GetString(1), "Verso") << "name should remain unchanged";
        EXPECT_EQ(result.GetInt64(2), 32) << "age should remain unchanged";
        EXPECT_EQ(result.GetInt64(3), 420) << "score should be updated to 420";
    }

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

TEST_F(KvTableTest, PartialUpdateByIndex) {
    auto& adm = admin();
    auto& conn = connection();

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

    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .AddColumn("age", fluss::DataType::BigInt())
                      .AddColumn("score", fluss::DataType::BigInt())
                      .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));

    // Insert initial record with all columns
    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    {
        fluss::GenericRow row(4);
        row.SetInt32(0, 1);
        row.SetString(1, "Verso");
        row.SetInt64(2, 32);
        row.SetInt64(3, 6942);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify initial record
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    {
        fluss::GenericRow key(4);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt32(0), 1);
        EXPECT_EQ(result.GetString(1), "Verso");
        EXPECT_EQ(result.GetInt64(2), 32);
        EXPECT_EQ(result.GetInt64(3), 6942);
    }

    // Create partial update writer using column indices: 0 (id) and 3 (score)
    auto partial_upsert = table.NewUpsert();
    partial_upsert.PartialUpdateByIndex({0, 3});
    fluss::UpsertWriter partial_writer;
    ASSERT_OK(partial_upsert.CreateWriter(partial_writer));

    // Update only the score column (await acknowledgment)
    {
        fluss::GenericRow partial_row(4);
        partial_row.SetInt32(0, 1);
        partial_row.SetNull(1);  // not in partial update
        partial_row.SetNull(2);  // not in partial update
        partial_row.SetInt64(3, 420);
        fluss::WriteResult wr;
        ASSERT_OK(partial_writer.Upsert(partial_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify partial update - name and age should remain unchanged
    {
        fluss::GenericRow key(4);
        key.SetInt32(0, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(result.GetInt32(0), 1) << "id should remain 1";
        EXPECT_EQ(result.GetString(1), "Verso") << "name should remain unchanged";
        EXPECT_EQ(result.GetInt64(2), 32) << "age should remain unchanged";
        EXPECT_EQ(result.GetInt64(3), 420) << "score should be updated to 420";
    }

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

TEST_F(KvTableTest, PartitionedTableUpsertAndLookup) {
    auto& adm = admin();
    auto& conn = connection();

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

    // Create a partitioned KV table with region as partition key
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("region", fluss::DataType::String())
                      .AddColumn("user_id", fluss::DataType::Int())
                      .AddColumn("name", fluss::DataType::String())
                      .AddColumn("score", fluss::DataType::BigInt())
                      .SetPrimaryKeys({"region", "user_id"})
                      .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", "APAC"});

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

    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    // Insert records with different partitions
    struct TestData {
        std::string region;
        int32_t user_id;
        std::string name;
        int64_t score;
    };
    std::vector<TestData> test_data = {{"US", 1, "Gustave", 100}, {"US", 2, "Lune", 200},
                                       {"EU", 1, "Sciel", 150},   {"EU", 2, "Maelle", 250},
                                       {"APAC", 1, "Noco", 300}};

    for (const auto& d : test_data) {
        fluss::GenericRow row(4);
        row.SetString(0, d.region);
        row.SetInt32(1, d.user_id);
        row.SetString(2, d.name);
        row.SetInt64(3, d.score);
        ASSERT_OK(upsert_writer.Upsert(row));
    }
    ASSERT_OK(upsert_writer.Flush());

    // Create lookuper
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    // Lookup records
    for (const auto& d : test_data) {
        fluss::GenericRow key(4);
        key.SetString(0, d.region);
        key.SetInt32(1, d.user_id);

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());

        EXPECT_EQ(std::string(result.GetString(0)), d.region) << "region mismatch";
        EXPECT_EQ(result.GetInt32(1), d.user_id) << "user_id mismatch";
        EXPECT_EQ(std::string(result.GetString(2)), d.name) << "name mismatch";
        EXPECT_EQ(result.GetInt64(3), d.score) << "score mismatch";
    }

    // Update within a partition (await acknowledgment)
    {
        fluss::GenericRow updated_row(4);
        updated_row.SetString(0, "US");
        updated_row.SetInt32(1, 1);
        updated_row.SetString(2, "Gustave Updated");
        updated_row.SetInt64(3, 999);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(updated_row, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify the update
    {
        fluss::GenericRow key(4);
        key.SetString(0, "US");
        key.SetInt32(1, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(std::string(result.GetString(2)), "Gustave Updated");
        EXPECT_EQ(result.GetInt64(3), 999);
    }

    // Lookup in non-existent partition should return not found
    {
        fluss::GenericRow key(4);
        key.SetString(0, "UNKNOWN_REGION");
        key.SetInt32(1, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_FALSE(result.Found()) << "Lookup in non-existent partition should return not found";
    }

    // Delete a record within a partition (await acknowledgment)
    {
        fluss::GenericRow delete_key(4);
        delete_key.SetString(0, "EU");
        delete_key.SetInt32(1, 1);
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Delete(delete_key, wr));
        ASSERT_OK(wr.Wait());
    }

    // Verify deletion
    {
        fluss::GenericRow key(4);
        key.SetString(0, "EU");
        key.SetInt32(1, 1);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_FALSE(result.Found()) << "Deleted record should not exist";
    }

    // Verify other records in same partition still exist
    {
        fluss::GenericRow key(4);
        key.SetString(0, "EU");
        key.SetInt32(1, 2);
        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());
        EXPECT_EQ(std::string(result.GetString(2)), "Maelle");
    }

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

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

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

    // Create a table with all supported datatypes
    auto schema = fluss::Schema::NewBuilder()
                      .AddColumn("pk_int", fluss::DataType::Int())
                      .AddColumn("col_boolean", fluss::DataType::Boolean())
                      .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_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(20))
                      .SetPrimaryKeys({"pk_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));

    auto table_upsert = table.NewUpsert();
    fluss::UpsertWriter upsert_writer;
    ASSERT_OK(table_upsert.CreateWriter(upsert_writer));

    // Test data
    int32_t pk_int = 1;
    bool col_boolean = true;
    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;
    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::FromMillis(1769163227123);      // 2026-01-23 10:13:47.123
    auto col_timestamp_ltz = fluss::Timestamp::FromMillis(1769163227123);
    std::vector<uint8_t> col_bytes = {'b', 'i', 'n', 'a', 'r', 'y', ' ', 'd', 'a', 't', 'a'};
    std::vector<uint8_t> col_binary = {'f', 'i', 'x', 'e', 'd', ' ', 'b', 'i', 'n', 'a',
                                       'r', 'y', ' ', 'd', 'a', 't', 'a', '!', '!', '!'};

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

    // Lookup the record
    fluss::Lookuper lookuper;
    ASSERT_OK(table.NewLookup().CreateLookuper(lookuper));

    {
        fluss::GenericRow key(17);
        key.SetInt32(0, pk_int);

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());

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

        auto [bytes_ptr, bytes_len] = result.GetBytes(15);
        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] = result.GetBytes(16);
        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";
    }

    // Test with null values for nullable columns
    {
        fluss::GenericRow row_with_nulls(17);
        row_with_nulls.SetInt32(0, 2);  // pk_int = 2
        for (size_t i = 1; i < 17; ++i) {
            row_with_nulls.SetNull(i);
        }
        fluss::WriteResult wr;
        ASSERT_OK(upsert_writer.Upsert(row_with_nulls, wr));
        ASSERT_OK(wr.Wait());
    }

    // Lookup row with nulls
    {
        fluss::GenericRow key(17);
        key.SetInt32(0, 2);

        fluss::LookupResult result;
        ASSERT_OK(lookuper.Lookup(key, result));
        ASSERT_TRUE(result.Found());

        EXPECT_EQ(result.GetInt32(0), 2) << "pk_int mismatch";
        for (size_t i = 1; i < 17; ++i) {
            EXPECT_TRUE(result.IsNull(i)) << "column " << i << " should be null";
        }
    }

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