leekeiabstraction commented on code in PR #611:
URL: https://github.com/apache/fluss-rust/pull/611#discussion_r3406350985
##########
bindings/cpp/test/test_log_table.cpp:
##########
Review Comment:
Do we need MapWriterOverflowDetection test case as well?
##########
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()))),
Review Comment:
Should we have coverage for small and tiny ints?
##########
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:
Seems like a lot of the checks in LookupWithValueHandle and
LookupComplexTypesMatrix are similar. Is it worth refactoring to minimise code
to maintain?
--
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]
