Support ListOffset/SubscribeBatch/DropTable for cpp bindings by zhaohaidao · Pull Request #100 · apache/fluss-rust

zhaohaidao · 2025-12-20T12:37:32Z

Purpose

Linked issue: close #101

Brief change log

Tests

API and Format

Documentation

zhaohaidao · 2025-12-20T12:43:10Z

@luoyuxia Hi, yuxia, I added ListOffset/SubscribeBatch/Droptable apis for cpp bindings to serve internal cpp users.
PTAL if u have time.

Copilot

Pull request overview

This pull request adds three new API methods to the C++ bindings for Fluss: ListOffsets, SubscribeBatch, and DropTable. It also enhances the ScanRecord structure to include bucket_id information, allowing consumers to identify which bucket each scanned record belongs to.

Key Changes:

Added ListOffsets API to query bucket offsets by earliest, latest, or timestamp-based criteria
Added SubscribeBatch API to enable subscribing to multiple buckets in a single call
Added DropTable API for table deletion with optional ignore-if-not-exists flag

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
`crates/fluss/src/client/table/scanner.rs`	Implements `subscribe_batch` method for subscribing to multiple buckets at once
`bindings/cpp/src/types.rs`	Updates `core_scan_records_to_ffi` to extract and include `bucket_id` in scan records
`bindings/cpp/src/table.cpp`	Implements C++ wrapper for `SubscribeBatch` method
`bindings/cpp/src/lib.rs`	Adds FFI bridge structures and implements Rust-side logic for all three new APIs
`bindings/cpp/src/ffi_converter.hpp`	Updates `from_ffi_scan_record` to handle `bucket_id` field
`bindings/cpp/src/admin.cpp`	Implements C++ wrappers for `DropTable` and `ListOffsets` admin methods
`bindings/cpp/include/fluss.hpp`	Declares new APIs, structs (`OffsetQuery`, `BucketSubscription`), and enums (`OffsetSpec`)
`bindings/cpp/examples/example.cpp`	Demonstrates usage of all new APIs with comprehensive examples including timestamp-based queries

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-20T13:00:25Z

+        let offset_spec = match offset_query.offset_type {
+            0 => OffsetSpec::Earliest,
+            1 => OffsetSpec::Latest,
+            2 => OffsetSpec::Timestamp(offset_query.timestamp),
+            _ => {
+                return ffi::FfiListOffsetsResult {
+                    result: err_result(1, format!("Invalid offset_type: {}", offset_query.offset_type)),
+                    bucket_offsets: vec![],
+                };
+            }
+        };


The offset_type matching logic uses hardcoded integer values (0, 1, 2) without documentation or named constants. Consider defining these as constants in the FFI module or adding documentation comments to clarify the mapping between values and their meanings (Earliest=0, Latest=1, Timestamp=2).

Copilot · 2025-12-20T13:00:25Z

+
+        let result = RUNTIME.block_on(async { 
+            self.inner.subscribe_batch(bucket_offsets).await 


Remove trailing whitespace from this line for consistency with code style.

Suggested change

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

Copilot · 2025-12-20T13:00:25Z

+
+        let result = RUNTIME.block_on(async { 
+            self.inner.subscribe_batch(bucket_offsets).await 


Remove trailing whitespace from this line for consistency with code style.

Suggested change

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

Copilot · 2025-12-20T13:00:26Z

+    ffi_query.timestamp = offset_query.timestamp;
+
+    auto ffi_result = admin_->list_offsets(ffi_path, std::move(rust_bucket_ids), ffi_query);
+


Remove trailing whitespace from this line for consistency with code style.

Copilot · 2025-12-20T13:00:26Z

@@ -63,6 +63,24 @@ enum class DatumType {
    Bytes = 7,
 };



The constants EARLIEST_OFFSET and LATEST_OFFSET are defined but not used in the codebase shown. Consider documenting their intended use or removing them if they are not needed. These constants could be useful for users of the API to pass to Subscribe methods, but their purpose should be documented.

Suggested change

// Special offset sentinel values exposed as part of the public API.

// These can be used by callers that work directly with numeric offsets

// (for example, when subscribing to a stream) to request:

// - EARLIEST_OFFSET: start from the beginning of the log

// - LATEST_OFFSET: start from the most recent (tail) offset.

// For higher-level C++ APIs, prefer using OffsetQuery::Earliest() /

// OffsetQuery::Latest() instead of these raw constants.

Copilot · 2025-12-20T13:00:26Z

+
+        let mut scan_bucket_offsets = HashMap::new();
+        for (bucket_id, offset) in bucket_offsets {
+            let table_bucket = TableBucket::new(self.table_id, bucket_id);
+            scan_bucket_offsets.insert(table_bucket, offset);
+        }
+


Remove trailing whitespace from this line for consistency with code style.

Suggested change

let mut scan_bucket_offsets = HashMap::new();

for (bucket_id, offset) in bucket_offsets {

let table_bucket = TableBucket::new(self.table_id, bucket_id);

scan_bucket_offsets.insert(table_bucket, offset);

}

let mut scan_bucket_offsets = HashMap::new();

for (bucket_id, offset) in bucket_offsets {

let table_bucket = TableBucket::new(self.table_id, bucket_id);

scan_bucket_offsets.insert(table_bucket, offset);

}

Copilot · 2025-12-20T13:00:27Z

+
+        let mut scan_bucket_offsets = HashMap::new();
+        for (bucket_id, offset) in bucket_offsets {
+            let table_bucket = TableBucket::new(self.table_id, bucket_id);
+            scan_bucket_offsets.insert(table_bucket, offset);
+        }
+


Remove trailing whitespace from this line for consistency with code style.

Suggested change

let mut scan_bucket_offsets = HashMap::new();

for (bucket_id, offset) in bucket_offsets {

let table_bucket = TableBucket::new(self.table_id, bucket_id);

scan_bucket_offsets.insert(table_bucket, offset);

}

let mut scan_bucket_offsets = HashMap::new();

for (bucket_id, offset) in bucket_offsets {

let table_bucket = TableBucket::new(self.table_id, bucket_id);

scan_bucket_offsets.insert(table_bucket, offset);

}

Copilot · 2025-12-20T13:00:27Z

+    }
+
+    auto ffi_path = utils::to_ffi_table_path(table_path);
+


Remove trailing whitespace from this line for consistency with code style.

Copilot · 2025-12-20T13:00:27Z

+        use std::collections::HashMap;
+        let mut bucket_offsets = HashMap::new();
+        for sub in subscriptions {
+            bucket_offsets.insert(sub.bucket_id, sub.offset);
+        }
+
+        let result = RUNTIME.block_on(async { 
+            self.inner.subscribe_batch(bucket_offsets).await 


Consider adding validation for empty subscriptions vector. When subscriptions is empty, subscribe_batch will make a call that might not be meaningful. Consider returning early with an appropriate error or success result.

Suggested change

use std::collections::HashMap;

let mut bucket_offsets = HashMap::new();

for sub in subscriptions {

bucket_offsets.insert(sub.bucket_id, sub.offset);

}

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

if subscriptions.is_empty() {

return ok_result();

}

use std::collections::HashMap;

let mut bucket_offsets = HashMap::new();

for sub in subscriptions {

bucket_offsets.insert(sub.bucket_id, sub.offset);

}

let result = RUNTIME.block_on(async {

self.inner.subscribe_batch(bucket_offsets).await

maybe we can optimize it in rust core

luoyuxia

@zhaohaidao Thanks for the pr. Left minor comment. Depend on you to take or not. Otherwise, LGTM

luoyuxia · 2025-12-21T04:10:24Z

 };

+constexpr int64_t EARLIEST_OFFSET = -2;
+constexpr int64_t LATEST_OFFSET = -1;


I'm curious about LATEST_OFFSET is for what. I can't find it in java code base.

Sometimes, lagged consumption can impact business operations. In such cases, the business may choose to resolve the issue by reverting to the latest consumption.

luoyuxia · 2025-12-21T04:22:51Z

+        use std::collections::HashMap;
+        let mut bucket_offsets = HashMap::new();
+        for sub in subscriptions {
+            bucket_offsets.insert(sub.bucket_id, sub.offset);
+        }
+
+        let result = RUNTIME.block_on(async { 
+            self.inner.subscribe_batch(bucket_offsets).await 


maybe we can optimize it in rust core

zhaohaidao · 2025-12-21T06:57:34Z

@luoyuxia Hi, yuxia, Comments are addressed. PTAL if u have time.

luoyuxia

Thanks. LGTM

…pache#100) --------- Co-authored-by: 赵海源 <zhaohaiyuan@xiaohongshu.com>

support more cpp api

fe0e670

zhaohaidao changed the title ~~(WIP)Support more cpp api~~ Support ListOffset/SubscribeBatch/DropTable for cpp bindings Dec 20, 2025

luoyuxia requested a review from Copilot December 20, 2025 12:56

Copilot started reviewing on behalf of luoyuxia December 20, 2025 12:57 View session

Copilot AI reviewed Dec 20, 2025

View reviewed changes

luoyuxia reviewed Dec 21, 2025

View reviewed changes

赵海源 added 2 commits December 21, 2025 14:42

address comments

93caae3

address comments

4f1e83a

luoyuxia approved these changes Dec 21, 2025

View reviewed changes

luoyuxia merged commit 0c8a37f into apache:main Dec 21, 2025
13 checks passed

zhaohaidao added a commit to zhaohaidao/fluss-rust that referenced this pull request Jan 21, 2026

feat: support ListOffset/SubscribeBatch/DropTable for cpp bindings (a…

d6a4503

…pache#100) --------- Co-authored-by: 赵海源 <zhaohaiyuan@xiaohongshu.com>


		let result = RUNTIME.block_on(async {
		self.inner.subscribe_batch(bucket_offsets).await

		ffi_query.timestamp = offset_query.timestamp;

		auto ffi_result = admin_->list_offsets(ffi_path, std::move(rust_bucket_ids), ffi_query);

+// Special offset sentinel values exposed as part of the public API.
+// These can be used by callers that work directly with numeric offsets
+// (for example, when subscribing to a stream) to request:
+//   - EARLIEST_OFFSET: start from the beginning of the log
+//   - LATEST_OFFSET:   start from the most recent (tail) offset.
+// For higher-level C++ APIs, prefer using OffsetQuery::Earliest() /
+// OffsetQuery::Latest() instead of these raw constants.

Conversation

zhaohaidao commented Dec 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

zhaohaidao commented Dec 20, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 20, 2025

Choose a reason for hiding this comment

Uh oh!

luoyuxia Dec 21, 2025

Choose a reason for hiding this comment

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

luoyuxia Dec 21, 2025

Choose a reason for hiding this comment

Uh oh!

zhaohaidao Dec 21, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

luoyuxia Dec 21, 2025

Choose a reason for hiding this comment

Uh oh!

zhaohaidao commented Dec 21, 2025

Uh oh!

luoyuxia left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zhaohaidao commented Dec 20, 2025 •

edited

Loading