You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "zhaohaidao (via GitHub)" <gi...@apache.org> on 2025/12/20 12:37:35 UTC
[PR] (WIP)Support more cpp api [fluss-rust]
zhaohaidao opened a new pull request, #100:
URL: https://github.com/apache/fluss-rust/pull/100
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss-rust/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #xxx
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#issuecomment-3677793296
@luoyuxia Hi, yuxia, I added ListOffset/SubscribeBatch/Droptable apis for cpp bindings to serve internal cpp users.
PTAL if u have time.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#issuecomment-3678529096
@luoyuxia Hi, yuxia, Comments are addressed. PTAL if u have time.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#discussion_r2637535765
##########
bindings/cpp/src/lib.rs:
##########
@@ -375,6 +428,56 @@ impl Admin {
},
}
}
+
+ fn list_offsets(
+ &self,
+ table_path: &ffi::FfiTablePath,
+ bucket_ids: Vec<i32>,
+ offset_query: &ffi::FfiOffsetQuery,
+ ) -> ffi::FfiListOffsetsResult {
+ use fcore::rpc::message::OffsetSpec;
+
+ let path = fcore::metadata::TablePath::new(
+ table_path.database_name.clone(),
+ table_path.table_name.clone(),
+ );
+
+ 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![],
+ };
+ }
+ };
+
Review Comment:
maybe we can optimize it in rust core
##########
bindings/cpp/src/lib.rs:
##########
@@ -511,6 +614,23 @@ impl LogScanner {
}
}
+ fn subscribe_batch(&self, subscriptions: Vec<ffi::FfiBucketSubscription>) -> ffi::FfiResult {
+ 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
Review Comment:
maybe we can optimize it in rust core
##########
bindings/cpp/include/fluss.hpp:
##########
@@ -448,6 +479,7 @@ class LogScanner {
bool Available() const;
Result Subscribe(int32_t bucket_id, int64_t start_offset);
+ Result SubscribeBatch(const std::vector<BucketSubscription>& bucket_offsets);
Review Comment:
Can `SubscribeBatch` be renamed to `Subscribe`?
##########
bindings/cpp/include/fluss.hpp:
##########
@@ -63,6 +63,24 @@ enum class DatumType {
Bytes = 7,
};
+constexpr int64_t EARLIEST_OFFSET = -2;
+constexpr int64_t LATEST_OFFSET = -1;
Review Comment:
I'm curious about `LATEST_OFFSET` is for what. I can't find it in java code base.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#discussion_r2637621191
##########
bindings/cpp/include/fluss.hpp:
##########
@@ -63,6 +63,24 @@ enum class DatumType {
Bytes = 7,
};
+constexpr int64_t EARLIEST_OFFSET = -2;
+constexpr int64_t LATEST_OFFSET = -1;
Review Comment:
Sometimes, lagged consumption can impact business operations. In such cases, the business may choose to resolve the issue by reverting to the latest consumption.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#discussion_r2637103368
##########
bindings/cpp/src/lib.rs:
##########
@@ -375,6 +428,56 @@ impl Admin {
},
}
}
+
+ fn list_offsets(
+ &self,
+ table_path: &ffi::FfiTablePath,
+ bucket_ids: Vec<i32>,
+ offset_query: &ffi::FfiOffsetQuery,
+ ) -> ffi::FfiListOffsetsResult {
+ use fcore::rpc::message::OffsetSpec;
+
+ let path = fcore::metadata::TablePath::new(
+ table_path.database_name.clone(),
+ table_path.table_name.clone(),
+ );
+
+ 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![],
+ };
+ }
+ };
+
Review Comment:
Consider adding validation for empty bucket_ids vector. When bucket_ids is empty, list_offsets will make a call that might not be meaningful or could be inefficient. Consider returning early with an appropriate error or empty result.
```suggestion
if bucket_ids.is_empty() {
return ffi::FfiListOffsetsResult {
result: ok_result(),
bucket_offsets: vec![],
};
}
```
##########
bindings/cpp/include/fluss.hpp:
##########
@@ -63,6 +63,24 @@ enum class DatumType {
Bytes = 7,
};
Review Comment:
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.
```suggestion
// 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.
```
##########
bindings/cpp/src/lib.rs:
##########
@@ -375,6 +428,56 @@ impl Admin {
},
}
}
+
+ fn list_offsets(
+ &self,
+ table_path: &ffi::FfiTablePath,
+ bucket_ids: Vec<i32>,
+ offset_query: &ffi::FfiOffsetQuery,
+ ) -> ffi::FfiListOffsetsResult {
+ use fcore::rpc::message::OffsetSpec;
+
+ let path = fcore::metadata::TablePath::new(
+ table_path.database_name.clone(),
+ table_path.table_name.clone(),
+ );
+
+ 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![],
+ };
+ }
+ };
Review Comment:
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).
##########
bindings/cpp/src/lib.rs:
##########
@@ -511,6 +614,23 @@ impl LogScanner {
}
}
+ fn subscribe_batch(&self, subscriptions: Vec<ffi::FfiBucketSubscription>) -> ffi::FfiResult {
+ 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
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
```suggestion
let result = RUNTIME.block_on(async {
self.inner.subscribe_batch(bucket_offsets).await
```
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -185,6 +185,22 @@ impl LogScanner {
Ok(())
}
+ pub async fn subscribe_batch(&self, bucket_offsets: HashMap<i32, i64>) -> Result<()> {
+ self.metadata
+ .check_and_update_table_metadata(from_ref(&self.table_path))
+ .await?;
+
+ 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);
+ }
+
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
```suggestion
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);
}
```
##########
crates/fluss/src/client/table/scanner.rs:
##########
@@ -185,6 +185,22 @@ impl LogScanner {
Ok(())
}
+ pub async fn subscribe_batch(&self, bucket_offsets: HashMap<i32, i64>) -> Result<()> {
+ self.metadata
+ .check_and_update_table_metadata(from_ref(&self.table_path))
+ .await?;
+
+ 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);
+ }
+
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
```suggestion
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);
}
```
##########
bindings/cpp/src/admin.cpp:
##########
@@ -98,4 +108,36 @@ Result Admin::GetLatestLakeSnapshot(const TablePath& table_path, LakeSnapshot& o
return result;
}
+Result Admin::ListOffsets(const TablePath& table_path,
+ const std::vector<int32_t>& bucket_ids,
+ const OffsetQuery& offset_query,
+ std::unordered_map<int32_t, int64_t>& out) {
+ if (!Available()) {
+ return utils::make_error(1, "Admin not available");
+ }
+
+ auto ffi_path = utils::to_ffi_table_path(table_path);
+
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
##########
bindings/cpp/src/admin.cpp:
##########
@@ -98,4 +108,36 @@ Result Admin::GetLatestLakeSnapshot(const TablePath& table_path, LakeSnapshot& o
return result;
}
+Result Admin::ListOffsets(const TablePath& table_path,
+ const std::vector<int32_t>& bucket_ids,
+ const OffsetQuery& offset_query,
+ std::unordered_map<int32_t, int64_t>& out) {
+ if (!Available()) {
+ return utils::make_error(1, "Admin not available");
+ }
+
+ auto ffi_path = utils::to_ffi_table_path(table_path);
+
+ rust::Vec<int32_t> rust_bucket_ids;
+ for (int32_t id : bucket_ids) {
+ rust_bucket_ids.push_back(id);
+ }
+
+ ffi::FfiOffsetQuery ffi_query;
+ ffi_query.offset_type = static_cast<int32_t>(offset_query.spec);
+ ffi_query.timestamp = offset_query.timestamp;
+
+ auto ffi_result = admin_->list_offsets(ffi_path, std::move(rust_bucket_ids), ffi_query);
+
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
##########
bindings/cpp/src/lib.rs:
##########
@@ -511,6 +614,23 @@ impl LogScanner {
}
}
+ fn subscribe_batch(&self, subscriptions: Vec<ffi::FfiBucketSubscription>) -> ffi::FfiResult {
+ 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
Review Comment:
Remove trailing whitespace from this line for consistency with code style.
```suggestion
let result = RUNTIME.block_on(async {
self.inner.subscribe_batch(bucket_offsets).await
```
##########
bindings/cpp/src/lib.rs:
##########
@@ -511,6 +614,23 @@ impl LogScanner {
}
}
+ fn subscribe_batch(&self, subscriptions: Vec<ffi::FfiBucketSubscription>) -> ffi::FfiResult {
+ 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
Review Comment:
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.
```suggestion
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
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #100:
URL: https://github.com/apache/fluss-rust/pull/100#discussion_r2637632658
##########
bindings/cpp/src/lib.rs:
##########
@@ -375,6 +428,56 @@ impl Admin {
},
}
}
+
+ fn list_offsets(
+ &self,
+ table_path: &ffi::FfiTablePath,
+ bucket_ids: Vec<i32>,
+ offset_query: &ffi::FfiOffsetQuery,
+ ) -> ffi::FfiListOffsetsResult {
+ use fcore::rpc::message::OffsetSpec;
+
+ let path = fcore::metadata::TablePath::new(
+ table_path.database_name.clone(),
+ table_path.table_name.clone(),
+ );
+
+ 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![],
+ };
+ }
+ };
+
Review Comment:
make sense
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support ListOffset/SubscribeBatch/DropTable for cpp bindings [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #100:
URL: https://github.com/apache/fluss-rust/pull/100
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org