Skip to content

Commit 81e9494

Browse files
fresh-borzonifresh-borzoni
authored
[feat] Support subscribing to primary-key table changelog (#639)
1 parent 0dae316 commit 81e9494

20 files changed

Lines changed: 1417 additions & 98 deletions

File tree

‎bindings/cpp/CMakeLists.txt‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,12 @@ target_link_libraries(fluss_cpp_kv_example PRIVATE Arrow::arrow_shared)
236236
target_compile_definitions(fluss_cpp_kv_example PRIVATE ARROW_FOUND)
237237
target_include_directories(fluss_cpp_kv_example PUBLIC ${CPP_INCLUDE_DIR})
238238

239+
add_executable(fluss_cpp_kv_changelog_example examples/kv_changelog_example.cpp)
240+
target_link_libraries(fluss_cpp_kv_changelog_example PRIVATE fluss_cpp)
241+
target_link_libraries(fluss_cpp_kv_changelog_example PRIVATE Arrow::arrow_shared)
242+
target_compile_definitions(fluss_cpp_kv_changelog_example PRIVATE ARROW_FOUND)
243+
target_include_directories(fluss_cpp_kv_changelog_example PUBLIC ${CPP_INCLUDE_DIR})
244+
239245
if (CARGO_TARGET_DIR)
240246
set_target_properties(fluss_cpp
241247
PROPERTIES ADDITIONAL_CLEAN_FILES "${CARGO_TARGET_DIR}"

‎bindings/cpp/examples/kv_changelog_example.cpp‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
// Primary-key table changelog (CDC) example: subscribe to a KV table's
19+
// changelog and print each row-level change as a +I / -U / +U / -D event.
20+
21+
#include <iostream>
22+
#include <string>
23+
24+
#include "fluss.hpp"
25+
26+
static void check(const char* step, const fluss::Result& r) {
27+
if (!r.Ok()) {
28+
std::cerr << step << " failed: code=" << r.error_code << " msg=" << r.error_message
29+
<< std::endl;
30+
std::exit(1);
31+
}
32+
}
33+
34+
static const char* change_symbol(fluss::ChangeType ct) {
35+
switch (ct) {
36+
case fluss::ChangeType::AppendOnly:
37+
return "+A";
38+
case fluss::ChangeType::Insert:
39+
return "+I";
40+
case fluss::ChangeType::UpdateBefore:
41+
return "-U";
42+
case fluss::ChangeType::UpdateAfter:
43+
return "+U";
44+
case fluss::ChangeType::Delete:
45+
return "-D";
46+
}
47+
return "?";
48+
}
49+
50+
int main() {
51+
fluss::Configuration config;
52+
config.bootstrap_servers = "127.0.0.1:9123";
53+
54+
fluss::Connection conn;
55+
check("create", fluss::Connection::Create(config, conn));
56+
57+
fluss::Admin admin;
58+
check("get_admin", conn.GetAdmin(admin));
59+
60+
fluss::TablePath table_path("fluss", "kv_changelog_cpp");
61+
admin.DropTable(table_path, true);
62+
63+
// A single bucket keeps the changelog on one bucket and in order, which
64+
// makes the CDC output easy to follow.
65+
auto schema = fluss::Schema::NewBuilder()
66+
.AddColumn("id", fluss::DataType::Int())
67+
.AddColumn("name", fluss::DataType::String())
68+
.SetPrimaryKeys({"id"})
69+
.Build();
70+
71+
auto descriptor = fluss::TableDescriptor::NewBuilder()
72+
.SetSchema(schema)
73+
.SetBucketCount(1)
74+
.SetComment("cpp kv changelog example")
75+
.Build();
76+
77+
check("create_table", admin.CreateTable(table_path, descriptor, false));
78+
std::cout << "Created PK table: " << table_path.ToString() << std::endl;
79+
80+
fluss::Table table;
81+
check("get_table", conn.GetTable(table_path, table));
82+
83+
fluss::UpsertWriter writer;
84+
check("new_upsert_writer", table.NewUpsert().CreateWriter(writer));
85+
86+
// Insert three keys (+I), update one (-U / +U) and delete one (-D).
87+
for (const auto& kv : {std::pair<int32_t, const char*>{1, "alice"},
88+
{2, "bob"}, {3, "carol"}}) {
89+
fluss::GenericRow row(2);
90+
row.SetInt32(0, kv.first);
91+
row.SetString(1, kv.second);
92+
check("upsert", writer.Upsert(row));
93+
}
94+
{
95+
fluss::GenericRow row(2);
96+
row.SetInt32(0, 2);
97+
row.SetString(1, "bob-v2");
98+
check("update", writer.Upsert(row));
99+
}
100+
{
101+
fluss::GenericRow del(2);
102+
del.SetInt32(0, 3);
103+
check("delete", writer.Delete(del));
104+
}
105+
check("flush", writer.Flush());
106+
107+
// Subscribe from the start of the changelog and print each CDC event until
108+
// we reach the end of the log.
109+
auto table_scan = table.NewScan();
110+
fluss::LogScanner log_scanner;
111+
check("create_log_scanner", table_scan.CreateLogScanner(log_scanner));
112+
check("subscribe", log_scanner.Subscribe(0, fluss::EARLIEST_OFFSET));
113+
114+
std::cout << "Changelog (change_type id name):" << std::endl;
115+
while (true) {
116+
fluss::ScanRecords records;
117+
check("poll", log_scanner.Poll(3000, records));
118+
if (records.IsEmpty()) {
119+
break;
120+
}
121+
for (auto rec : records) {
122+
std::cout << " " << change_symbol(rec.change_type) << " " << rec.row.GetInt32(0) << " "
123+
<< rec.row.GetString(1) << std::endl;
124+
}
125+
}
126+
127+
check("drop_table", admin.DropTable(table_path, true));
128+
std::cout << "\nKV changelog example completed successfully!" << std::endl;
129+
return 0;
130+
}

‎bindings/cpp/include/fluss.hpp‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1630,7 +1630,21 @@ class TableScan {
16301630

16311631
TableScan& Limit(int32_t row_number);
16321632

1633+
/// Creates a record-mode log scanner, polled for individual `ScanRecord`s.
1634+
///
1635+
/// Works on log tables and on primary-key (KV) tables. For a primary-key
1636+
/// table this subscribes to its CDC changelog: each `ScanRecord` carries a
1637+
/// `ChangeType` -- `+I` (insert), `-U` (update-before), `+U` (update-after)
1638+
/// or `-D` (delete). A log table yields `+A` (append-only). Requires the
1639+
/// ARROW log format.
16331640
Result CreateLogScanner(LogScanner& out);
1641+
1642+
/// Creates a batch-mode log scanner that yields Arrow record batches.
1643+
///
1644+
/// Log tables only. Primary-key tables are rejected because the Arrow batch
1645+
/// path carries no per-record change types; read a primary-key table's
1646+
/// changelog with `CreateLogScanner()` instead. Requires the ARROW log
1647+
/// format.
16341648
Result CreateRecordBatchLogScanner(LogScanner& out);
16351649

16361650
Result CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out);

‎bindings/cpp/test/test_kv_changelog.cpp‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
// Integration tests for primary-key (KV) table changelog (CDC) scanning.
21+
// Mirrors crates/fluss/tests/integration/kv_changelog.rs.
22+
23+
#include <gtest/gtest.h>
24+
25+
#include <algorithm>
26+
#include <string>
27+
#include <utility>
28+
#include <vector>
29+
30+
#include "test_utils.h"
31+
32+
class KvChangelogTest : public ::testing::Test {
33+
protected:
34+
fluss::Admin& admin() { return fluss_test::FlussTestEnvironment::Instance()->GetAdmin(); }
35+
36+
fluss::Connection& connection() {
37+
return fluss_test::FlussTestEnvironment::Instance()->GetConnection();
38+
}
39+
};
40+
41+
// A record-mode scanner over a PK table yields its CDC changelog. With the
42+
// default FULL changelog image: inserting a new key emits +I, overwriting an
43+
// existing key emits -U (old image) then +U (new image), and a delete emits -D
44+
// (old image). A single bucket keeps the offsets contiguous.
45+
TEST_F(KvChangelogTest, SubscribeKvTableChangelog) {
46+
auto& adm = admin();
47+
auto& conn = connection();
48+
49+
fluss::TablePath table_path("fluss", "test_kv_changelog_cpp");
50+
51+
auto schema = fluss::Schema::NewBuilder()
52+
.AddColumn("id", fluss::DataType::Int())
53+
.AddColumn("name", fluss::DataType::String())
54+
.SetPrimaryKeys({"id"})
55+
.Build();
56+
57+
auto table_descriptor = fluss::TableDescriptor::NewBuilder()
58+
.SetSchema(schema)
59+
.SetBucketCount(1)
60+
.SetProperty("table.replication.factor", "1")
61+
.Build();
62+
63+
fluss_test::CreateTable(adm, table_path, table_descriptor);
64+
65+
fluss::Table table;
66+
ASSERT_OK(conn.GetTable(table_path, table));
67+
68+
auto table_upsert = table.NewUpsert();
69+
fluss::UpsertWriter upsert_writer;
70+
ASSERT_OK(table_upsert.CreateWriter(upsert_writer));
71+
72+
// Await each write so the changelog offsets are produced in a fixed order.
73+
{ // +I (1, alice)
74+
fluss::GenericRow row(2);
75+
row.SetInt32(0, 1);
76+
row.SetString(1, "alice");
77+
fluss::WriteResult wr;
78+
ASSERT_OK(upsert_writer.Upsert(row, wr));
79+
ASSERT_OK(wr.Wait());
80+
}
81+
{ // +I (2, bob)
82+
fluss::GenericRow row(2);
83+
row.SetInt32(0, 2);
84+
row.SetString(1, "bob");
85+
fluss::WriteResult wr;
86+
ASSERT_OK(upsert_writer.Upsert(row, wr));
87+
ASSERT_OK(wr.Wait());
88+
}
89+
{ // overwrite id=1 -> -U (1, alice) then +U (1, alice2)
90+
fluss::GenericRow row(2);
91+
row.SetInt32(0, 1);
92+
row.SetString(1, "alice2");
93+
fluss::WriteResult wr;
94+
ASSERT_OK(upsert_writer.Upsert(row, wr));
95+
ASSERT_OK(wr.Wait());
96+
}
97+
{ // -D (2, bob)
98+
fluss::GenericRow del(2);
99+
del.SetInt32(0, 2);
100+
fluss::WriteResult wr;
101+
ASSERT_OK(upsert_writer.Delete(del, wr));
102+
ASSERT_OK(wr.Wait());
103+
}
104+
105+
auto table_scan = table.NewScan();
106+
fluss::LogScanner log_scanner;
107+
ASSERT_OK(table_scan.CreateLogScanner(log_scanner));
108+
ASSERT_OK(log_scanner.Subscribe(0, fluss::EARLIEST_OFFSET));
109+
110+
struct Decoded {
111+
int64_t offset;
112+
fluss::ChangeType change_type;
113+
int32_t id;
114+
std::string name;
115+
};
116+
117+
std::vector<Decoded> records;
118+
fluss_test::PollRecords(
119+
log_scanner, 5,
120+
[](const fluss::ScanRecord& rec) {
121+
return Decoded{rec.offset, rec.change_type, rec.row.GetInt32(0),
122+
std::string(rec.row.GetString(1))};
123+
},
124+
records);
125+
126+
ASSERT_EQ(records.size(), 5u);
127+
std::sort(records.begin(), records.end(),
128+
[](const Decoded& a, const Decoded& b) { return a.offset < b.offset; });
129+
130+
const std::vector<fluss::ChangeType> expected_types = {
131+
fluss::ChangeType::Insert, fluss::ChangeType::Insert, fluss::ChangeType::UpdateBefore,
132+
fluss::ChangeType::UpdateAfter, fluss::ChangeType::Delete};
133+
const std::vector<std::pair<int32_t, std::string>> expected_rows = {
134+
{1, "alice"}, // +I
135+
{2, "bob"}, // +I
136+
{1, "alice"}, // -U (old image)
137+
{1, "alice2"}, // +U (new image)
138+
{2, "bob"}, // -D (old image)
139+
};
140+
141+
for (size_t i = 0; i < records.size(); ++i) {
142+
EXPECT_EQ(records[i].offset, static_cast<int64_t>(i));
143+
EXPECT_EQ(static_cast<int>(records[i].change_type), static_cast<int>(expected_types[i]))
144+
<< "change_type mismatch at " << i;
145+
EXPECT_EQ(records[i].id, expected_rows[i].first) << "id mismatch at " << i;
146+
EXPECT_EQ(records[i].name, expected_rows[i].second) << "name mismatch at " << i;
147+
}
148+
149+
ASSERT_OK(adm.DropTable(table_path, false));
150+
}
151+
152+
// The Arrow batch scanner carries no per-record change types, so it rejects
153+
// primary-key tables (mirrors the core / Java restriction).
154+
TEST_F(KvChangelogTest, RecordBatchScannerRejectsPrimaryKey) {
155+
auto& adm = admin();
156+
auto& conn = connection();
157+
158+
fluss::TablePath table_path("fluss", "test_kv_changelog_batch_reject_cpp");
159+
160+
auto schema = fluss::Schema::NewBuilder()
161+
.AddColumn("id", fluss::DataType::Int())
162+
.AddColumn("name", fluss::DataType::String())
163+
.SetPrimaryKeys({"id"})
164+
.Build();
165+
166+
auto table_descriptor = fluss::TableDescriptor::NewBuilder()
167+
.SetSchema(schema)
168+
.SetProperty("table.replication.factor", "1")
169+
.Build();
170+
171+
fluss_test::CreateTable(adm, table_path, table_descriptor);
172+
173+
fluss::Table table;
174+
ASSERT_OK(conn.GetTable(table_path, table));
175+
176+
auto table_scan = table.NewScan();
177+
fluss::LogScanner batch_scanner;
178+
auto result = table_scan.CreateRecordBatchLogScanner(batch_scanner);
179+
EXPECT_FALSE(result.Ok()) << "batch scanner should reject a primary-key table";
180+
181+
ASSERT_OK(adm.DropTable(table_path, false));
182+
}

0 commit comments

Comments
 (0)