[c++] Add bounded RecordBatch log reader APIs

fluss-rust/bindings/cpp/README.md

@@ -33,10 +33,92 @@ bazel build //...
 `ci.sh` defaults to optimized builds via `-c opt` (override with `BAZEL_BUILD_FLAGS` if needed).
 See [ci.sh](ci.sh) for the CI build sequence.
 
+## Examples and Documentation
+
+- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes, continuous scans,
+  bounded Arrow record-batch scans, projections, and offset queries.
+- [examples/admin_example.cpp](examples/admin_example.cpp) demonstrates database, table,
+  partition, and cluster administration.
+- [examples/kv_example.cpp](examples/kv_example.cpp) and
+  [examples/kv_changelog_example.cpp](examples/kv_changelog_example.cpp) demonstrate
+  primary-key table access.
+- The website documentation includes the
+  [C++ API reference](../../website/docs/user-guide/cpp/api-reference.md) and
+  [log-table examples](../../website/docs/user-guide/cpp/example/log-tables.md).
+
+For a bounded log scan, pass the per-bucket offset ranges directly to `TableScan`. The returned
+reader yields one Arrow batch at a time until every `[starting_offset, stopping_offset)` range
+is complete:
+
+```cpp
+auto info = table.GetTableInfo();
+std::vector<int32_t> bucket_ids;
+for (int32_t bucket_id = 0; bucket_id < info.num_buckets; ++bucket_id) {
+    bucket_ids.push_back(bucket_id);
+}
+
+std::unordered_map<int32_t, int64_t> latest_offsets;
+admin.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets);
+
+std::vector<fluss::RecordBatchLogReadRange> ranges;
+for (int32_t bucket_id : bucket_ids) {
+    ranges.push_back(
+        {fluss::TableBucket{info.table_id, bucket_id}, 0, latest_offsets.at(bucket_id)});
+}
+
+fluss::RecordBatchLogReader reader;
+table.NewScan().CreateRecordBatchLogReader(ranges, reader);
+
+const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(30);
+bool finished = false;
+while (std::chrono::steady_clock::now() < deadline) {
+    fluss::RecordBatchReadResult result;
+    auto read_result = reader.NextBatch(1000, result);
+    if (!read_result.Ok()) {
+        // Bail out on unretriable failures (auth, invalid table, ...); the
+        // reader's status field is only meaningful when `Ok()` is true.
+        if (!read_result.IsRetriable()) {
+            throw std::runtime_error(read_result.error_message);
+        }
+        continue;
+    }
+    if (result.status == fluss::BoundedReadStatus::TimedOut) {
+        continue;
+    }
+    if (result.status == fluss::BoundedReadStatus::Finished) {
+        finished = true;
+        break;
+    }
+    process(result.batch->GetArrowRecordBatch());
+}
+if (!finished) {
+    throw std::runtime_error("Bounded read exceeded its execution deadline");
+}
+```
+
+Timestamp-bounded reads use the same iterator after resolving the timestamps independently for
+each bucket:
+
+```cpp
+fluss::RecordBatchLogReader timestamp_reader;
+table.NewScan().CreateRecordBatchLogReader(
+    admin, table_buckets,
+    fluss::TimestampRange{starting_timestamp_ms, stopping_timestamp_ms}, timestamp_reader);
+```
+
+`CollectAllBatches(timeout_ms, out)` is available when materializing the complete bounded result
+is preferred. `timeout_ms` is the total execution budget for the whole call, so callers should
+normally pass the query's remaining execution time and call the method once. It appends complete
+batches to `out` as they arrive. If the budget expires before every stopping offset is reached, it
+stops collecting and returns a retriable `REQUEST_TIME_OUT`; `out` may contain a partial result,
+and only an `Ok()` result means the bounded result is complete. The timeout is checked between
+complete Arrow batches and never splits a batch already being returned. The reader remains valid
+after timeout if a caller has an explicit resume policy, but unconditional retry is not the
+intended usage. `NextBatch()` remains the per-poll API for engines that need to check cancellation
+between batches.
 
 ## TODO
 
-- [] How to introduce fluss-cpp in your own project, https://github.com/apache/opendal/blob/main/bindings/cpp/README.md is a good reference
+- [ ] How to introduce fluss-cpp in your own project, https://github.com/apache/opendal/blob/main/bindings/cpp/README.md is a good reference
 - [ ] Add CMake/Bazel install and packaging instructions.
-- [ ] Document API usage and minimal example in this README.
-- [ ] Add more C++ examples (log scan, upsert, etc.).
+- [ ] Add more C++ examples (upsert, partitioned bounded scans, etc.).

fluss-rust/bindings/cpp/examples/example.cpp

@@ -369,11 +369,51 @@ int main() {
         std::cout << "  Bucket " << bucket_id << ": offset=" << offset << std::endl;
     }
 
+    // 8.1) Bounded Arrow record batch scan with explicit stopping offsets
+    std::cout << "\n=== Bounded Arrow Record Batch Scan ===" << std::endl;
+    constexpr int64_t kBoundedReadTimeoutMs = 30000;
+    std::vector<fluss::RecordBatchLogReadRange> bounded_ranges;
+    for (int32_t bucket_id : all_bucket_ids) {
+        const int64_t starting_offset = earliest_offsets.at(bucket_id);
+        const int64_t stopping_offset = latest_offsets.at(bucket_id);
+        bounded_ranges.push_back(
+            {fluss::TableBucket{info.table_id, bucket_id}, starting_offset, stopping_offset});
+    }
+
+    if (!bounded_ranges.empty()) {
+        fluss::RecordBatchLogReader bounded_reader;
+        check("create_bounded_reader",
+              table.NewScan().CreateRecordBatchLogReader(bounded_ranges, bounded_reader));
+
+        fluss::ArrowRecordBatches bounded_batches;
+        auto collect_result =
+            bounded_reader.CollectAllBatches(kBoundedReadTimeoutMs, bounded_batches);
+        if (!collect_result.Ok()) {
+            std::cerr << "collect_bounded_batches failed after collecting "
+                      << bounded_batches.Size()
+                      << " partial batches: code=" << collect_result.error_code
+                      << " msg=" << collect_result.error_message << std::endl;
+            return 1;
+        }
+
+        int64_t bounded_row_count = 0;
+        for (const auto& batch : bounded_batches) {
+            bounded_row_count += batch->NumRows();
+            std::cout << "  bucket=" << batch->GetBucketId()
+                      << " base_offset=" << batch->GetBaseOffset()
+                      << " last_offset=" << batch->GetLastOffset()
+                      << " rows=" << batch->NumRows() << std::endl;
+        }
+        std::cout << "Bounded scan completed with " << bounded_row_count << " rows" << std::endl;
+    }
+
     auto now = std::chrono::system_clock::now();
     auto one_hour_ago = now - std::chrono::hours(1);
     auto timestamp_ms =
         std::chrono::duration_cast<std::chrono::milliseconds>(one_hour_ago.time_since_epoch())
             .count();
+    auto now_ms =
+        std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
 
     std::unordered_map<int32_t, int64_t> timestamp_offsets;
     check("list_timestamp_offsets",
@@ -384,6 +424,32 @@ int main() {
         std::cout << "  Bucket " << bucket_id << ": offset=" << offset << std::endl;
     }
 
+    // 8.2) Bounded Arrow record batch scan by timestamp range
+    std::vector<fluss::TableBucket> table_buckets;
+    for (int32_t bucket_id : all_bucket_ids) {
+        table_buckets.push_back({info.table_id, bucket_id});
+    }
+
+    fluss::RecordBatchLogReader timestamp_reader;
+    check("create_timestamp_reader",
+          table.NewScan().CreateRecordBatchLogReader(
+              admin, table_buckets, fluss::TimestampRange{timestamp_ms, now_ms}, timestamp_reader));
+
+    fluss::ArrowRecordBatches timestamp_batches;
+    auto timestamp_collect_result =
+        timestamp_reader.CollectAllBatches(kBoundedReadTimeoutMs, timestamp_batches);
+    if (!timestamp_collect_result.Ok()) {
+        std::cerr << "collect_timestamp_batches failed after collecting "
+                  << timestamp_batches.Size()
+                  << " partial batches: code=" << timestamp_collect_result.error_code
+                  << " msg=" << timestamp_collect_result.error_message << std::endl;
+        return 1;
+    }
+    for (const auto& batch : timestamp_batches) {
+        std::cout << "Timestamp range batch: bucket=" << batch->GetBucketId()
+                  << " rows=" << batch->NumRows() << std::endl;
+    }
+
     // 9) Batch subscribe
     std::cout << "\n=== Batch Subscribe Example ===" << std::endl;
     fluss::LogScanner batch_scanner;
@@ -427,7 +493,7 @@ int main() {
     // 10) Arrow record batch polling
     std::cout << "\n=== Testing Arrow Record Batch Polling ===" << std::endl;
 
-    fluss::LogScanner arrow_scanner;
+    fluss::RecordBatchLogScanner arrow_scanner;
     check("new_record_batch_log_scanner",
           table.NewScan().CreateRecordBatchLogScanner(arrow_scanner));
 
@@ -436,7 +502,7 @@ int main() {
     }
 
     fluss::ArrowRecordBatches arrow_batches;
-    check("poll_record_batch", arrow_scanner.PollRecordBatch(5000, arrow_batches));
+    check("poll_record_batch", arrow_scanner.Poll(5000, arrow_batches));
 
     std::cout << "Polled " << arrow_batches.Size() << " Arrow record batches" << std::endl;
     for (size_t i = 0; i < arrow_batches.Size(); ++i) {
@@ -452,7 +518,7 @@ int main() {
     // 11) Arrow record batch polling with projection
     std::cout << "\n=== Testing Arrow Record Batch Polling with Projection ===" << std::endl;
 
-    fluss::LogScanner projected_arrow_scanner;
+    fluss::RecordBatchLogScanner projected_arrow_scanner;
     check("new_record_batch_log_scanner_with_projection",
           table.NewScan()
               .ProjectByIndex(projected_columns)
@@ -464,7 +530,7 @@ int main() {
 
     fluss::ArrowRecordBatches projected_arrow_batches;
     check("poll_projected_record_batch",
-          projected_arrow_scanner.PollRecordBatch(5000, projected_arrow_batches));
+          projected_arrow_scanner.Poll(5000, projected_arrow_batches));
 
     std::cout << "Polled " << projected_arrow_batches.Size() << " projected Arrow record batches"
               << std::endl;