Skip to content

[c++] Add bounded RecordBatch log reader APIs - #3988

Open
naivedogger wants to merge 6 commits into
apache:mainfrom
naivedogger:feature/cpp-bounded-log-reader
Open

[c++] Add bounded RecordBatch log reader APIs#3988
naivedogger wants to merge 6 commits into
apache:mainfrom
naivedogger:feature/cpp-bounded-log-reader

Conversation

@naivedogger

@naivedogger naivedogger commented Aug 14, 2026

Copy link
Copy Markdown
Contributor

Purpose

Linked issue: close #3987

Expose bounded RecordBatchLogReader APIs through the C++ binding.

API example

fluss::RecordBatchLogReader reader;
check("create_bounded_reader",
      table.NewScan().CreateRecordBatchLogReader(
          admin,
          assigned_buckets,
          fluss::TimestampRange{
              starting_timestamp_ms,
              stopping_timestamp_ms,
          },
          reader));

while (true) {
    fluss::RecordBatchReadResult result;
    check("next_batch", reader.NextBatch(1000, result));

    if (result.status == fluss::BoundedReadStatus::TimedOut) {
        continue;
    }
    if (result.status == fluss::BoundedReadStatus::Finished) {
        break;
    }

    process(result.batch->GetArrowRecordBatch());
}

The API also supports:

  • explicit starting and stopping offsets for each bucket;
  • starting offsets configured through scanner subscriptions;
  • stopping at explicit or latest offsets;
  • partitioned and non-partitioned tables.

TimedOut allows callers to check cancellation or deadlines before retrying. Finished means all assigned buckets have reached their stopping offsets.

Tests

  • cargo test -p fluss-rs --lib client::table::reader
  • cargo test -p fluss-cpp --lib
  • cargo clippy -p fluss-rs -p fluss-cpp --lib -- -D warnings
  • cargo fmt --check
  • CMake C++ build
  • Docker-backed C++ integration tests for explicit offsets, latest offsets,
    timestamp ranges, and partitioned tables

API and Format

This PR adds backward-compatible public C++ APIs.

Documentation

Updated the C++ README, API reference, log table guide, and example program.

@naivedogger

Copy link
Copy Markdown
Contributor Author

@fresh-borzoni @charlesdong1991 @leekeiabstraction Appreciate a review here when you have time 🙏

@leekeiabstraction leekeiabstraction left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TY for the PR, added a couple of comments. PTAL.

auto ffi_result = reader_->record_batch_log_reader_next_batch(timeout_ms);
auto result = utils::from_ffi_result(ffi_result.result);
if (!result.Ok()) {
return result;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When FFI call fails, we return with out.status still set to TimedOut, this will cause caller following documentation to retry forever on unretriable errors.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. If the FFI call fails, NextBatch() now leaves out as Finished with no batch instead of TimedOut. This prevents callers from mistaking an actual error for a timeout and retrying forever. Callers should still check the returned Result before using out.

Comment on lines +1934 to +1935
/// Drains all remaining batches until every stopping offset is reached.
Result CollectAllBatches(ArrowRecordBatches& out);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't have a timeout arg, what happens if it is an error such as a tablet server outage? Will this hang caller forever?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed. CollectAllBatches(timeout_ms, out) now uses timeout_ms as the time limit for the entire collection. If the timeout expires before all stopping offsets are reached, it stops and returns REQUEST_TIME_OUT; any complete batches already collected remain in out. Only Ok() means the result is complete. The documentation, examples, and tests have been updated accordingly.

@naivedogger

Copy link
Copy Markdown
Contributor Author

Thanks @leekeiabstraction for the thorough review! Both comments have been addressed and the updates have been pushed. PTAL when you have a chance.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[c++] Support bounded RecordBatch log reads

2 participants