fresh-borzoni commented on code in PR #611:
URL: https://github.com/apache/fluss-rust/pull/611#discussion_r3442225804
##########
bindings/cpp/test/test_kv_table.cpp:
##########
@@ -212,24 +214,776 @@ TEST_F(KvTableTest, LookupWithNestedArrayArrayView) {
fluss::LookupResult result;
ASSERT_OK(lookuper.Lookup(key, result));
ASSERT_TRUE(result.Found());
- EXPECT_EQ(result.GetArraySize("matrix"), 2u);
- EXPECT_EQ(result.GetArrayElementType("matrix"), fluss::TypeId::Array);
-
- auto outer = result.GetArrayView("matrix");
+ auto outer = result.GetValue("matrix");
+ EXPECT_EQ(outer.Type(), fluss::TypeId::Array);
ASSERT_EQ(outer.Size(), 2u);
- EXPECT_EQ(outer.ElementType(), fluss::TypeId::Array);
- auto first = outer.GetArray(0);
+ auto first = outer.At(0);
+ EXPECT_EQ(first.Type(), fluss::TypeId::Array);
ASSERT_EQ(first.Size(), 2u);
- EXPECT_EQ(first.ElementType(), fluss::TypeId::Int);
- EXPECT_EQ(first.GetInt32(0), 11);
- EXPECT_EQ(first.GetInt32(1), 12);
+ EXPECT_EQ(first.At(0).GetInt32(), 11);
+ EXPECT_EQ(first.At(1).GetInt32(), 12);
- auto second = outer.GetArray(1);
+ auto second = outer.At(1);
ASSERT_EQ(second.Size(), 2u);
- EXPECT_EQ(second.ElementType(), fluss::TypeId::Int);
- EXPECT_EQ(second.GetInt32(0), 21);
- EXPECT_EQ(second.GetInt32(1), 22);
+ 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())),
+ })),
+ });
+ 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));
+ }
+
+ 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");
+ 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");
+ 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.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);
+ }
+
+ // Row 2 (id=2) — every compound column NULL.
+ {
+ auto row = table.NewRow();
+ row.SetInt32(0, 2);
+ for (size_t i = 1; i <= 9; ++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 (size_t i = 1; i <= 9; ++i) {
+ EXPECT_TRUE(result2.IsNull(i)) << "column " << i << " should be
null";
+ }
+ }
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(KvTableTest, LookupWithValueHandle) {
Review Comment:
Leftover, removed it and moved some checks in the matrix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
