[c++] Add bounded RecordBatch log reader APIs

fluss-rust/bindings/cpp/README.md

@@ -33,10 +33,71 @@ 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);
+
+while (true) {
+    fluss::RecordBatchReadResult result;
+    reader.NextBatch(1000, result);
+    if (result.status == fluss::BoundedReadStatus::TimedOut) {
+        continue;  // Check query cancellation before retrying.
+    }
+    if (result.status == fluss::BoundedReadStatus::Finished) {
+        break;
+    }
+    process(result.batch->GetArrowRecordBatch());
+}
+```
+
+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()` is available when materializing the complete bounded result is
+preferred. `NextBatch()` reports timeout separately from completion so engines can periodically
+check cancellation.
 
 ## 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,47 @@ 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;
+    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));
+
+        int64_t bounded_row_count = 0;
+        while (true) {
+            fluss::RecordBatchReadResult result;
+            check("bounded_next_batch", bounded_reader.NextBatch(1000, result));
+            if (result.status == fluss::BoundedReadStatus::TimedOut) {
+                continue;
+            }
+            if (result.status == fluss::BoundedReadStatus::Finished) {
+                break;
+            }
+            bounded_row_count += result.batch->NumRows();
+            std::cout << "  bucket=" << result.batch->GetBucketId()
+                      << " base_offset=" << result.batch->GetBaseOffset()
+                      << " last_offset=" << result.batch->GetLastOffset()
+                      << " rows=" << result.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 +420,30 @@ 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));
+
+    while (true) {
+        fluss::RecordBatchReadResult result;
+        check("timestamp_next_batch", timestamp_reader.NextBatch(1000, result));
+        if (result.status == fluss::BoundedReadStatus::TimedOut) {
+            continue;
+        }
+        if (result.status == fluss::BoundedReadStatus::Finished) {
+            break;
+        }
+        std::cout << "Timestamp range batch: bucket=" << result.batch->GetBucketId()
+                  << " rows=" << result.batch->NumRows() << std::endl;
+    }
+
     // 9) Batch subscribe
     std::cout << "\n=== Batch Subscribe Example ===" << std::endl;
     fluss::LogScanner batch_scanner;
@@ -427,7 +487,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 +496,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 +512,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 +524,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;

fluss-rust/bindings/cpp/include/fluss.hpp

@@ -47,6 +47,7 @@ struct Table;
 struct AppendWriter;
 struct WriteResult;
 struct LogScanner;
+struct RecordBatchLogReader;
 struct BatchScanner;
 struct UpsertWriter;
 struct Lookuper;
@@ -1235,6 +1236,39 @@ struct PartitionBucketSubscription {
     int64_t offset;
 };
 
+/// Stopping offset for one bucket subscribed on a record-batch log scanner.
+struct ReaderStopOffset {
+    TableBucket bucket;
+    int64_t offset;
+};
+
+/// One bounded log range. Records are returned for
+/// [starting_offset, stopping_offset).
+struct RecordBatchLogReadRange {
+    TableBucket bucket;
+    int64_t starting_offset;
+    int64_t stopping_offset;
+};
+
+/// A log timestamp range in epoch milliseconds. Each requested bucket resolves
+/// the two timestamps to offsets before the reader is created.
+struct TimestampRange {
+    int64_t starting_timestamp_ms;
+    int64_t stopping_timestamp_ms;
+};
+
+/// Outcome of a bounded record-batch read.
+enum class BoundedReadStatus {
+    BatchAvailable = 0,
+    TimedOut = 1,
+    Finished = 2,
+};
+
+struct RecordBatchReadResult {
+    BoundedReadStatus status{BoundedReadStatus::TimedOut};
+    std::unique_ptr<ArrowRecordBatch> batch;
+};
+
 struct LakeSnapshot {
     int64_t snapshot_id;
     std::vector<BucketOffset> bucket_offsets;
@@ -1338,6 +1372,8 @@ class Lookuper;
 class PrefixLookuper;
 class WriteResult;
 class LogScanner;
+class RecordBatchLogScanner;
+class RecordBatchLogReader;
 class BatchScanner;
 class Admin;
 class Table;
@@ -1502,6 +1538,7 @@ class Admin {
                          const std::string* partition_name = nullptr);
 
     friend class Connection;
+    friend class LogScanner;
     Admin(ffi::Admin* admin) noexcept;
 
     void Destroy() noexcept;
@@ -1632,8 +1669,26 @@ class TableScan {
     /// path carries no per-record change types; read a primary-key table's
     /// changelog with `CreateLogScanner()` instead. Requires the ARROW log
     /// format.
+    Result CreateRecordBatchLogScanner(RecordBatchLogScanner& out);
+
+    /// Legacy overload. Prefer the strongly typed RecordBatchLogScanner.
     Result CreateRecordBatchLogScanner(LogScanner& out);
 
+    /// Creates a bounded reader directly from per-bucket offset ranges.
+    ///
+    /// This is the preferred API for query engines: it subscribes every bucket
+    /// at its starting offset, installs the corresponding stopping offset, and
+    /// transfers scanner ownership to the returned reader.
+    Result CreateRecordBatchLogReader(const std::vector<RecordBatchLogReadRange>& ranges,
+                                      RecordBatchLogReader& out);
+
+    /// Creates a bounded reader for a timestamp range over requested buckets.
+    ///
+    /// The timestamps are resolved independently for every requested bucket,
+    /// then read with [starting_offset, stopping_offset) semantics.
+    Result CreateRecordBatchLogReader(Admin& admin, const std::vector<TableBucket>& buckets,
+                                      const TimestampRange& range, RecordBatchLogReader& out);
+
     Result CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out);
 
    private:
@@ -1642,6 +1697,9 @@ class TableScan {
 
     std::vector<size_t> ResolveNameProjection() const;
     Result DoCreateScanner(LogScanner& out, bool is_record_batch);
+    Result ResolveTimestampRanges(Admin& admin, const std::vector<TableBucket>& buckets,
+                                  const TimestampRange& range,
+                                  std::vector<RecordBatchLogReadRange>& out) const;
 
     ffi::Table* table_{nullptr};
     std::vector<size_t> projection_;
@@ -1796,12 +1854,96 @@ class LogScanner {
    private:
     friend class Table;
     friend class TableScan;
+    friend class RecordBatchLogScanner;
     LogScanner(ffi::LogScanner* scanner) noexcept;
 
     void Destroy() noexcept;
+
+    /// Creates a bounded reader using the latest offsets observed during this call.
+    /// Subscribe the record-batch scanner at the desired starting offsets before
+    /// calling this method.
+    Result CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out);
+
+    /// Creates a bounded reader using explicit stopping offsets.
+    /// Starting offsets come from the scanner subscriptions. Every stopping
+    /// offset must correspond to a bucket already subscribed on this scanner.
+    Result CreateRecordBatchLogReaderUntilOffsets(const std::vector<ReaderStopOffset>& offsets,
+                                                  RecordBatchLogReader& out);
+
     ffi::LogScanner* scanner_{nullptr};
 };
 
+/// Strongly typed Arrow record-batch log scanner.
+///
+/// Use this type for unbounded batch polling, or move it into a bounded reader.
+class RecordBatchLogScanner {
+   public:
+    RecordBatchLogScanner() noexcept;
+    ~RecordBatchLogScanner() noexcept;
+
+    RecordBatchLogScanner(const RecordBatchLogScanner&) = delete;
+    RecordBatchLogScanner& operator=(const RecordBatchLogScanner&) = delete;
+    RecordBatchLogScanner(RecordBatchLogScanner&& other) noexcept;
+    RecordBatchLogScanner& operator=(RecordBatchLogScanner&& other) noexcept;
+
+    bool Available() const;
+
+    Result Subscribe(int32_t bucket_id, int64_t start_offset);
+    Result Subscribe(const std::vector<BucketSubscription>& bucket_offsets);
+    Result SubscribePartitionBuckets(int64_t partition_id, int32_t bucket_id, int64_t start_offset);
+    Result SubscribePartitionBuckets(const std::vector<PartitionBucketSubscription>& subscriptions);
+    Result Unsubscribe(int32_t bucket_id);
+    Result UnsubscribePartition(int64_t partition_id, int32_t bucket_id);
+    Result Poll(int64_t timeout_ms, ArrowRecordBatches& out);
+
+    /// Transfers this scanner into a reader bounded by the latest offsets
+    /// observed during the call. The scanner becomes unavailable on success.
+    Result CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out) &&;
+
+    /// Transfers this scanner into a reader with explicit stopping offsets.
+    /// The scanner becomes unavailable on success.
+    Result CreateRecordBatchLogReaderUntilOffsets(const std::vector<ReaderStopOffset>& offsets,
+                                                  RecordBatchLogReader& out) &&;
+
+   private:
+    friend class TableScan;
+    explicit RecordBatchLogScanner(ffi::LogScanner* scanner) noexcept;
+
+    LogScanner scanner_;
+};
+
+/// Bounded Arrow batch reader created from a subscribed record-batch log scanner.
+/// Only one reader or polling operation may consume the scanner at a time.
+class RecordBatchLogReader {
+   public:
+    RecordBatchLogReader() noexcept;
+    ~RecordBatchLogReader() noexcept;
+
+    RecordBatchLogReader(const RecordBatchLogReader&) = delete;
+    RecordBatchLogReader& operator=(const RecordBatchLogReader&) = delete;
+    RecordBatchLogReader(RecordBatchLogReader&& other) noexcept;
+    RecordBatchLogReader& operator=(RecordBatchLogReader&& other) noexcept;
+
+    bool Available() const;
+
+    /// Waits up to timeout_ms for the next batch.
+    /// TimedOut leaves the reader valid for a later retry; Finished means every
+    /// subscribed bucket reached its stopping offset.
+    Result NextBatch(int64_t timeout_ms, RecordBatchReadResult& out);
+
+    /// Drains all remaining batches until every stopping offset is reached.
+    Result CollectAllBatches(ArrowRecordBatches& out);
+
+   private:
+    friend class LogScanner;
+    friend class RecordBatchLogScanner;
+    friend class TableScan;
+    explicit RecordBatchLogReader(ffi::RecordBatchLogReader* reader) noexcept;
+
+    void Destroy() noexcept;
+    ffi::RecordBatchLogReader* reader_{nullptr};
+};
+
 // One-shot bounded scan of a single bucket, from TableScan::CreateBucketBatchScanner.
 class BatchScanner {
    public: