From c38b6275a9bc93c18e30cd717427e194fc3d4ed5 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Thu, 13 Aug 2026 19:51:25 +0800 Subject: [PATCH 1/2] [c++] Add bounded log reader --- fluss-rust/bindings/cpp/README.md | 64 +++++- fluss-rust/bindings/cpp/examples/example.cpp | 44 ++++ fluss-rust/bindings/cpp/include/fluss.hpp | 57 +++++ fluss-rust/bindings/cpp/src/lib.rs | 163 +++++++++++++++ fluss-rust/bindings/cpp/src/table.cpp | 120 +++++++++++ .../bindings/cpp/test/test_log_table.cpp | 197 ++++++++++++++++++ .../crates/fluss/src/client/table/mod.rs | 2 +- .../crates/fluss/src/client/table/reader.rs | 45 +++- .../docs/user-guide/cpp/api-reference.md | 31 +++ .../docs/user-guide/cpp/example/log-tables.md | 74 +++++++ 10 files changed, 788 insertions(+), 9 deletions(-) diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md index fa6cac26f5b..c8cd54fa871 100644 --- a/fluss-rust/bindings/cpp/README.md +++ b/fluss-rust/bindings/cpp/README.md @@ -33,10 +33,68 @@ 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, first subscribe the record-batch scanner at the desired starting +offsets, then create a `RecordBatchLogReader` with either explicit stopping offsets or the +latest offsets observed at reader creation: + +```cpp +auto info = table.GetTableInfo(); +std::vector bucket_ids; +for (int32_t bucket_id = 0; bucket_id < info.num_buckets; ++bucket_id) { + bucket_ids.push_back(bucket_id); +} + +std::unordered_map latest_offsets; +admin.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets); + +fluss::LogScanner scanner; +table.NewScan().CreateRecordBatchLogScanner(scanner); + +std::vector stopping_offsets; +for (int32_t bucket_id : bucket_ids) { + scanner.Subscribe(bucket_id, 0); + stopping_offsets.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, latest_offsets.at(bucket_id)}); +} + +fluss::RecordBatchLogReader reader; +scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader); + +while (true) { + fluss::ArrowRecordBatches batches; + fluss::BoundedReadStatus status; + reader.NextBatch(1000, batches, status); + if (status == fluss::BoundedReadStatus::TimedOut) { + continue; // Check query cancellation before retrying. + } + if (status == fluss::BoundedReadStatus::Finished) { + break; + } + for (const auto& batch : batches) { + process(batch->GetArrowRecordBatch()); + } +} +``` + +`CollectAllBatches()` is available when materializing the complete bounded result is +preferred. `NextBatch()` reports timeout separately from completion so engines can periodically +check cancellation. Use only one active reader or polling operation for a scanner at a time. ## 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.). diff --git a/fluss-rust/bindings/cpp/examples/example.cpp b/fluss-rust/bindings/cpp/examples/example.cpp index d86ee5cda72..2bd087c1d11 100644 --- a/fluss-rust/bindings/cpp/examples/example.cpp +++ b/fluss-rust/bindings/cpp/examples/example.cpp @@ -369,6 +369,50 @@ 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; + fluss::LogScanner bounded_scanner; + check("new_bounded_record_batch_scanner", + table.NewScan().CreateRecordBatchLogScanner(bounded_scanner)); + + std::vector stopping_offsets; + 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); + if (starting_offset >= stopping_offset) { + continue; + } + check("subscribe_bounded", bounded_scanner.Subscribe(bucket_id, starting_offset)); + stopping_offsets.push_back({fluss::TableBucket{info.table_id, bucket_id}, stopping_offset}); + } + + if (!stopping_offsets.empty()) { + fluss::RecordBatchLogReader bounded_reader; + check("create_bounded_reader", bounded_scanner.CreateRecordBatchLogReaderUntilOffsets( + stopping_offsets, bounded_reader)); + + int64_t bounded_row_count = 0; + while (true) { + fluss::ArrowRecordBatches bounded_batches; + fluss::BoundedReadStatus status; + check("bounded_next_batch", bounded_reader.NextBatch(1000, bounded_batches, status)); + if (status == fluss::BoundedReadStatus::TimedOut) { + continue; + } + if (status == fluss::BoundedReadStatus::Finished) { + break; + } + for (const auto& bounded_batch : bounded_batches) { + bounded_row_count += bounded_batch->NumRows(); + std::cout << " bucket=" << bounded_batch->GetBucketId() + << " base_offset=" << bounded_batch->GetBaseOffset() + << " last_offset=" << bounded_batch->GetLastOffset() + << " rows=" << bounded_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 = diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp index 935cf70853b..b32ab1c9e01 100644 --- a/fluss-rust/bindings/cpp/include/fluss.hpp +++ b/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,19 @@ struct PartitionBucketSubscription { int64_t offset; }; +/// Stopping offset for one bucket subscribed on a record-batch log scanner. +struct ReaderStopOffset { + TableBucket bucket; + int64_t offset; +}; + +/// Outcome of a bounded record-batch read. +enum class BoundedReadStatus { + BatchAvailable = 0, + TimedOut = 1, + Finished = 2, +}; + struct LakeSnapshot { int64_t snapshot_id; std::vector bucket_offsets; @@ -1338,6 +1352,7 @@ class Lookuper; class PrefixLookuper; class WriteResult; class LogScanner; +class RecordBatchLogReader; class BatchScanner; class Admin; class Table; @@ -1502,6 +1517,7 @@ class Admin { const std::string* partition_name = nullptr); friend class Connection; + friend class LogScanner; Admin(ffi::Admin* admin) noexcept; void Destroy() noexcept; @@ -1793,6 +1809,17 @@ class LogScanner { Result Poll(int64_t timeout_ms, ScanRecords& out); Result PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out); + /// 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& offsets, + RecordBatchLogReader& out); + private: friend class Table; friend class TableScan; @@ -1802,6 +1829,36 @@ class LogScanner { ffi::LogScanner* scanner_{nullptr}; }; +/// 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, ArrowRecordBatches& out, BoundedReadStatus& status); + + /// Drains all remaining batches until every stopping offset is reached. + Result CollectAllBatches(ArrowRecordBatches& out); + + private: + friend class LogScanner; + 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: diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs index 675a28061c1..a44151ad882 100644 --- a/fluss-rust/bindings/cpp/src/lib.rs +++ b/fluss-rust/bindings/cpp/src/lib.rs @@ -174,6 +174,20 @@ mod ffi { arrow_batches: FfiArrowRecordBatches, } + struct FfiBoundedReadResult { + result: FfiResult, + arrow_batches: FfiArrowRecordBatches, + status: i32, + } + + struct FfiReaderStopOffset { + table_id: i64, + has_partition_id: bool, + partition_id: i64, + bucket_id: i32, + offset: i64, + } + struct FfiLakeSnapshot { snapshot_id: i64, bucket_offsets: Vec, @@ -300,6 +314,7 @@ mod ffi { type AppendWriter; type WriteResult; type LogScanner; + type RecordBatchLogReader; type BatchScanner; type UpsertWriter; type Lookuper; @@ -699,8 +714,26 @@ mod ffi { -> FfiResult; fn poll(self: &LogScanner, timeout_ms: i64) -> Box; fn poll_record_batch(self: &LogScanner, timeout_ms: i64) -> FfiArrowRecordBatchesResult; + fn create_record_batch_log_reader_until_latest( + self: &LogScanner, + admin: &Admin, + ) -> FfiPtrResult; + fn create_record_batch_log_reader_until_offsets( + self: &LogScanner, + offsets: Vec, + ) -> FfiPtrResult; fn free_arrow_ffi_structures(array_ptr: usize, schema_ptr: usize); + // RecordBatchLogReader + unsafe fn delete_record_batch_log_reader(reader: *mut RecordBatchLogReader); + fn record_batch_log_reader_next_batch( + self: &RecordBatchLogReader, + timeout_ms: i64, + ) -> FfiBoundedReadResult; + fn record_batch_log_reader_collect_all_batches( + self: &RecordBatchLogReader, + ) -> FfiArrowRecordBatchesResult; + // BatchScanner unsafe fn delete_batch_scanner(scanner: *mut BatchScanner); fn next_batch(self: &BatchScanner) -> FfiArrowRecordBatchesResult; @@ -844,6 +877,10 @@ pub struct LogScanner { projected_columns: Vec, } +pub struct RecordBatchLogReader { + inner: Mutex, +} + pub struct BatchScanner { inner: Mutex, } @@ -943,6 +980,32 @@ fn arrow_batches_result( } } +fn bounded_read_result( + converted: Result, + status: i32, +) -> ffi::FfiBoundedReadResult { + match converted { + Ok(arrow_batches) => ffi::FfiBoundedReadResult { + result: ok_result(), + arrow_batches, + status, + }, + Err(e) => ffi::FfiBoundedReadResult { + result: client_err(e), + arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] }, + status, + }, + } +} + +fn empty_bounded_read_result(result: ffi::FfiResult, status: i32) -> ffi::FfiBoundedReadResult { + ffi::FfiBoundedReadResult { + result, + arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] }, + status, + } +} + // Connection implementation fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult { let assigner_type = match config @@ -2178,6 +2241,106 @@ impl LogScanner { Err(e) => empty_arrow_batches_result(err_from_core_error(&e)), } } + + fn create_record_batch_log_reader_until_latest(&self, admin: &Admin) -> ffi::FfiPtrResult { + let ScannerKind::Batch(ref scanner) = self.scanner else { + return client_err_ptr("Batch-based scanner not available".to_string()); + }; + + let reader_result = RUNTIME.block_on(async { + fcore::client::RecordBatchLogReader::new_until_latest( + scanner.new_shared_handle(), + admin.inner.as_ref(), + ) + .await + }); + + match reader_result { + Ok(reader) => { + let ptr = Box::into_raw(Box::new(RecordBatchLogReader { + inner: Mutex::new(reader), + })); + ok_ptr(ptr as usize) + } + Err(e) => err_ptr_from_core(&e), + } + } + + fn create_record_batch_log_reader_until_offsets( + &self, + offsets: Vec, + ) -> ffi::FfiPtrResult { + let ScannerKind::Batch(ref scanner) = self.scanner else { + return client_err_ptr("Batch-based scanner not available".to_string()); + }; + + let stopping_offsets = offsets + .into_iter() + .map(|offset| { + let partition_id = offset.has_partition_id.then_some(offset.partition_id); + let bucket = fcore::metadata::TableBucket::new_with_partition( + offset.table_id, + partition_id, + offset.bucket_id, + ); + (bucket, offset.offset) + }) + .collect(); + + match fcore::client::RecordBatchLogReader::new_until_offsets( + scanner.new_shared_handle(), + stopping_offsets, + ) { + Ok(reader) => { + let ptr = Box::into_raw(Box::new(RecordBatchLogReader { + inner: Mutex::new(reader), + })); + ok_ptr(ptr as usize) + } + Err(e) => err_ptr_from_core(&e), + } + } +} + +// RecordBatchLogReader implementation +unsafe fn delete_record_batch_log_reader(reader: *mut RecordBatchLogReader) { + if !reader.is_null() { + unsafe { + drop(Box::from_raw(reader)); + } + } +} + +impl RecordBatchLogReader { + fn record_batch_log_reader_next_batch(&self, timeout_ms: i64) -> ffi::FfiBoundedReadResult { + const BATCH_AVAILABLE: i32 = 0; + const TIMED_OUT: i32 = 1; + const FINISHED: i32 = 2; + + let mut reader = self.inner.lock().unwrap(); + let timeout = Duration::from_millis(timeout_ms.max(0) as u64); + match RUNTIME.block_on(reader.next_batch_with_timeout(timeout)) { + Ok(fcore::client::RecordBatchReadOutcome::Batch(batch)) => bounded_read_result( + types::core_scan_batches_to_ffi(std::slice::from_ref(&batch)), + BATCH_AVAILABLE, + ), + Ok(fcore::client::RecordBatchReadOutcome::TimedOut) => { + empty_bounded_read_result(ok_result(), TIMED_OUT) + } + Ok(fcore::client::RecordBatchReadOutcome::Finished) => { + empty_bounded_read_result(ok_result(), FINISHED) + } + Err(e) => empty_bounded_read_result(err_from_core_error(&e), FINISHED), + } + } + + fn record_batch_log_reader_collect_all_batches(&self) -> ffi::FfiArrowRecordBatchesResult { + let mut reader = self.inner.lock().unwrap(); + match RUNTIME.block_on(reader.collect_all_batches()) { + Ok(batches) => arrow_batches_result(types::core_scan_batches_to_ffi(&batches)), + Err(e) => empty_arrow_batches_result(err_from_core_error(&e)), + } + } } // BatchScanner implementation diff --git a/fluss-rust/bindings/cpp/src/table.cpp b/fluss-rust/bindings/cpp/src/table.cpp index 941f0f046d7..2df9760bc3a 100644 --- a/fluss-rust/bindings/cpp/src/table.cpp +++ b/fluss-rust/bindings/cpp/src/table.cpp @@ -1857,6 +1857,126 @@ Result LogScanner::PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out); } +// ============================================================================ +// RecordBatchLogReader +// ============================================================================ + +Result LogScanner::CreateRecordBatchLogReaderUntilLatest(const Admin& admin, + RecordBatchLogReader& out) { + if (!Available()) { + return utils::make_client_error("LogScanner not available"); + } + if (!admin.Available()) { + return utils::make_client_error("Admin not available"); + } + + auto ffi_result = scanner_->create_record_batch_log_reader_until_latest(*admin.admin_); + auto result = utils::from_ffi_result(ffi_result.result); + if (result.Ok()) { + out.Destroy(); + out.reader_ = utils::ptr_from_ffi(ffi_result); + } + return result; +} + +Result LogScanner::CreateRecordBatchLogReaderUntilOffsets( + const std::vector& offsets, RecordBatchLogReader& out) { + if (!Available()) { + return utils::make_client_error("LogScanner not available"); + } + + rust::Vec ffi_offsets; + for (const auto& offset : offsets) { + ffi::FfiReaderStopOffset ffi_offset; + ffi_offset.table_id = offset.bucket.table_id; + ffi_offset.has_partition_id = offset.bucket.partition_id.has_value(); + ffi_offset.partition_id = offset.bucket.partition_id.value_or(0); + ffi_offset.bucket_id = offset.bucket.bucket_id; + ffi_offset.offset = offset.offset; + ffi_offsets.push_back(ffi_offset); + } + + auto ffi_result = + scanner_->create_record_batch_log_reader_until_offsets(std::move(ffi_offsets)); + auto result = utils::from_ffi_result(ffi_result.result); + if (result.Ok()) { + out.Destroy(); + out.reader_ = utils::ptr_from_ffi(ffi_result); + } + return result; +} + +RecordBatchLogReader::RecordBatchLogReader() noexcept = default; + +RecordBatchLogReader::RecordBatchLogReader(ffi::RecordBatchLogReader* reader) noexcept + : reader_(reader) {} + +RecordBatchLogReader::~RecordBatchLogReader() noexcept { Destroy(); } + +void RecordBatchLogReader::Destroy() noexcept { + if (reader_) { + ffi::delete_record_batch_log_reader(reader_); + reader_ = nullptr; + } +} + +RecordBatchLogReader::RecordBatchLogReader(RecordBatchLogReader&& other) noexcept + : reader_(other.reader_) { + other.reader_ = nullptr; +} + +RecordBatchLogReader& RecordBatchLogReader::operator=(RecordBatchLogReader&& other) noexcept { + if (this != &other) { + Destroy(); + reader_ = other.reader_; + other.reader_ = nullptr; + } + return *this; +} + +bool RecordBatchLogReader::Available() const { return reader_ != nullptr; } + +Result RecordBatchLogReader::NextBatch(int64_t timeout_ms, ArrowRecordBatches& out, + BoundedReadStatus& status) { + if (!Available()) { + return utils::make_client_error("RecordBatchLogReader not available"); + } + + 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; + } + switch (ffi_result.status) { + case 0: + status = BoundedReadStatus::BatchAvailable; + break; + case 1: + status = BoundedReadStatus::TimedOut; + break; + case 2: + status = BoundedReadStatus::Finished; + break; + default: + return utils::make_client_error("Unknown bounded read status: " + + std::to_string(ffi_result.status)); + } + return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out); +} + +Result RecordBatchLogReader::CollectAllBatches(ArrowRecordBatches& out) { + if (!Available()) { + return utils::make_client_error("RecordBatchLogReader not available"); + } + + auto ffi_result = reader_->record_batch_log_reader_collect_all_batches(); + auto result = utils::from_ffi_result(ffi_result.result); + if (!result.Ok()) { + return result; + } + return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out); +} + // ============================================================================ // BatchScanner // ============================================================================ diff --git a/fluss-rust/bindings/cpp/test/test_log_table.cpp b/fluss-rust/bindings/cpp/test/test_log_table.cpp index ef29d9c93aa..fbbbbb1ccb3 100644 --- a/fluss-rust/bindings/cpp/test/test_log_table.cpp +++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp @@ -173,6 +173,169 @@ TEST_F(LogTableTest, AppendRecordBatchAndScan) { ASSERT_OK(adm.DropTable(table_path, false)); } +TEST_F(LogTableTest, RecordBatchLogReaderUntilOffsets) { + auto& adm = admin(); + auto& conn = connection(); + + constexpr int32_t kNumBuckets = 3; + fluss::TablePath table_path("fluss", "test_record_batch_log_reader_offsets_cpp"); + auto schema = fluss::Schema::NewBuilder() + .AddColumn("c1", DataType::Int()) + .AddColumn("c2", DataType::String()) + .Build(); + auto table_descriptor = fluss::TableDescriptor::NewBuilder() + .SetSchema(schema) + .SetBucketCount(kNumBuckets) + .SetBucketKeys({"c1"}) + .SetProperty("table.replication.factor", "1") + .Build(); + fluss_test::CreateTable(adm, table_path, table_descriptor); + + fluss::Table table; + ASSERT_OK(conn.GetTable(table_path, table)); + auto table_append = table.NewAppend(); + fluss::AppendWriter append_writer; + ASSERT_OK(table_append.CreateWriter(append_writer)); + + auto c1 = arrow::Int32Builder(); + auto c2 = arrow::StringBuilder(); + for (int32_t value = 1; value <= 60; ++value) { + ASSERT_TRUE(c1.Append(value).ok()); + ASSERT_TRUE(c2.Append("v" + std::to_string(value)).ok()); + } + auto batch = arrow::RecordBatch::Make( + arrow::schema({arrow::field("c1", arrow::int32()), arrow::field("c2", arrow::utf8())}), 60, + {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie()}); + ASSERT_OK(append_writer.AppendArrowBatch(batch)); + ASSERT_OK(append_writer.Flush()); + + fluss::LogScanner scanner; + ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(scanner)); + + const int64_t table_id = table.GetTableInfo().table_id; + std::vector bucket_ids; + for (int32_t bucket_id = 0; bucket_id < kNumBuckets; ++bucket_id) { + bucket_ids.push_back(bucket_id); + } + + std::unordered_map latest_offsets; + ASSERT_OK(adm.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets)); + ASSERT_EQ(latest_offsets.size(), bucket_ids.size()); + + std::vector stopping_offsets; + std::vector expected_rows_by_bucket(kNumBuckets); + for (int32_t bucket_id : bucket_ids) { + const int64_t stopping_offset = latest_offsets.at(bucket_id); + ASSERT_GT(stopping_offset, 1) + << "Bucket " << bucket_id << " should contain rows after starting offset 1"; + ASSERT_OK(scanner.Subscribe(bucket_id, 1)); + stopping_offsets.push_back({fluss::TableBucket{table_id, bucket_id}, stopping_offset}); + expected_rows_by_bucket[bucket_id] = stopping_offset - 1; + } + + fluss::RecordBatchLogReader reader; + ASSERT_OK(scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader)); + + fluss::ArrowRecordBatches batches; + ASSERT_OK(reader.CollectAllBatches(batches)); + + std::vector actual_rows_by_bucket(kNumBuckets); + for (const auto& bounded_batch : batches) { + const int32_t bucket_id = bounded_batch->GetBucketId(); + ASSERT_GE(bucket_id, 0); + ASSERT_LT(bucket_id, kNumBuckets); + EXPECT_GE(bounded_batch->GetBaseOffset(), 1); + EXPECT_LT(bounded_batch->GetLastOffset(), latest_offsets.at(bucket_id)); + actual_rows_by_bucket[bucket_id] += bounded_batch->NumRows(); + } + for (int32_t bucket_id : bucket_ids) { + EXPECT_EQ(actual_rows_by_bucket[bucket_id], expected_rows_by_bucket[bucket_id]) + << "Unexpected row count for bucket " << bucket_id; + } + + fluss::ArrowRecordBatches eof_batches; + fluss::BoundedReadStatus eof_status = fluss::BoundedReadStatus::BatchAvailable; + ASSERT_OK(reader.NextBatch(1000, eof_batches, eof_status)); + EXPECT_TRUE(eof_batches.Empty()); + EXPECT_EQ(eof_status, fluss::BoundedReadStatus::Finished); + + // A bounded reader whose stopping offset is not available yet should return + // TimedOut without becoming exhausted, so query engines can check cancellation + // and retry. + { + fluss::LogScanner waiting_scanner; + ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(waiting_scanner)); + const int64_t start_offset = latest_offsets.at(0); + ASSERT_OK(waiting_scanner.Subscribe(0, start_offset)); + + fluss::RecordBatchLogReader waiting_reader; + ASSERT_OK(waiting_scanner.CreateRecordBatchLogReaderUntilOffsets( + {{fluss::TableBucket{table_id, 0}, start_offset + 1}}, waiting_reader)); + + fluss::ArrowRecordBatches timeout_batches; + fluss::BoundedReadStatus timeout_status = fluss::BoundedReadStatus::Finished; + ASSERT_OK(waiting_reader.NextBatch(100, timeout_batches, timeout_status)); + EXPECT_TRUE(timeout_batches.Empty()); + EXPECT_EQ(timeout_status, fluss::BoundedReadStatus::TimedOut); + } + + ASSERT_OK(adm.DropTable(table_path, false)); +} + +TEST_F(LogTableTest, RecordBatchLogReaderUntilLatest) { + auto& adm = admin(); + auto& conn = connection(); + + fluss::TablePath table_path("fluss", "test_record_batch_log_reader_latest_cpp"); + auto schema = fluss::Schema::NewBuilder().AddColumn("c1", DataType::Int()).Build(); + auto table_descriptor = fluss::TableDescriptor::NewBuilder() + .SetSchema(schema) + .SetBucketCount(1) + .SetBucketKeys({"c1"}) + .SetProperty("table.replication.factor", "1") + .Build(); + fluss_test::CreateTable(adm, table_path, table_descriptor); + + fluss::Table table; + ASSERT_OK(conn.GetTable(table_path, table)); + auto table_append = table.NewAppend(); + fluss::AppendWriter append_writer; + ASSERT_OK(table_append.CreateWriter(append_writer)); + + auto c1 = arrow::Int32Builder(); + ASSERT_TRUE(c1.AppendValues({1, 2, 3}).ok()); + auto batch = arrow::RecordBatch::Make(arrow::schema({arrow::field("c1", arrow::int32())}), 3, + {c1.Finish().ValueOrDie()}); + ASSERT_OK(append_writer.AppendArrowBatch(batch)); + ASSERT_OK(append_writer.Flush()); + + fluss::LogScanner scanner; + ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(scanner)); + ASSERT_OK(scanner.Subscribe(0, 0)); + + fluss::RecordBatchLogReader reader; + ASSERT_OK(scanner.CreateRecordBatchLogReaderUntilLatest(adm, reader)); + + fluss::ArrowRecordBatches batches; + fluss::BoundedReadStatus status = fluss::BoundedReadStatus::TimedOut; + ASSERT_OK(reader.NextBatch(5000, batches, status)); + ASSERT_EQ(status, fluss::BoundedReadStatus::BatchAvailable); + ASSERT_EQ(batches.Size(), 1); + auto ids = + std::static_pointer_cast(batches[0]->GetArrowRecordBatch()->column(0)); + ASSERT_EQ(ids->length(), 3); + EXPECT_EQ(ids->Value(0), 1); + EXPECT_EQ(ids->Value(1), 2); + EXPECT_EQ(ids->Value(2), 3); + + fluss::ArrowRecordBatches eof_batches; + ASSERT_OK(reader.NextBatch(1000, eof_batches, status)); + EXPECT_TRUE(eof_batches.Empty()); + EXPECT_EQ(status, fluss::BoundedReadStatus::Finished); + + ASSERT_OK(adm.DropTable(table_path, false)); +} + TEST_F(LogTableTest, LimitScan) { auto& adm = admin(); auto& conn = connection(); @@ -988,6 +1151,40 @@ TEST_F(LogTableTest, PartitionedTableAppendScan) { {7, "EU", 700}, {8, "EU", 800}}; EXPECT_EQ(collected, expected); + // Test bounded record-batch reading across partition buckets. + { + fluss::Table bounded_table; + ASSERT_OK(conn.GetTable(table_path, bounded_table)); + fluss::LogScanner bounded_scanner; + ASSERT_OK(bounded_table.NewScan().CreateRecordBatchLogScanner(bounded_scanner)); + + std::vector subscriptions; + std::vector stopping_offsets; + for (const auto& pi : partition_infos) { + subscriptions.push_back({pi.partition_id, 0, 1}); + stopping_offsets.push_back( + {fluss::TableBucket{bounded_table.GetTableInfo().table_id, 0, pi.partition_id}, 3}); + } + ASSERT_OK(bounded_scanner.SubscribePartitionBuckets(subscriptions)); + + fluss::RecordBatchLogReader reader; + ASSERT_OK(bounded_scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader)); + + fluss::ArrowRecordBatches batches; + ASSERT_OK(reader.CollectAllBatches(batches)); + + std::vector ids; + for (const auto& bounded_batch : batches) { + auto id_array = std::static_pointer_cast( + bounded_batch->GetArrowRecordBatch()->column(0)); + for (int64_t row = 0; row < id_array->length(); ++row) { + ids.push_back(id_array->Value(row)); + } + } + std::sort(ids.begin(), ids.end()); + EXPECT_EQ(ids, std::vector({2, 4, 5, 7})); + } + // Test unsubscribe_partition: unsubscribe EU, should only get US data { fluss::Table unsub_table; diff --git a/fluss-rust/crates/fluss/src/client/table/mod.rs b/fluss-rust/crates/fluss/src/client/table/mod.rs index 41cbadab0b8..35e2e88c636 100644 --- a/fluss-rust/crates/fluss/src/client/table/mod.rs +++ b/fluss-rust/crates/fluss/src/client/table/mod.rs @@ -39,7 +39,7 @@ mod upsert; pub use append::{AppendWriter, TableAppend}; pub use batch_scanner::LimitBatchScanner; pub use lookup::{LookupResult, Lookuper, PrefixKeyLookuper, TableLookup, TablePrefixLookup}; -pub use reader::{RecordBatchLogReader, SyncRecordBatchLogReader}; +pub use reader::{RecordBatchLogReader, RecordBatchReadOutcome, SyncRecordBatchLogReader}; pub use remote_log::{ DEFAULT_REMOTE_FILE_DOWNLOAD_THREAD_NUM, DEFAULT_SCANNER_REMOTE_LOG_PREFETCH_NUM, }; diff --git a/fluss-rust/crates/fluss/src/client/table/reader.rs b/fluss-rust/crates/fluss/src/client/table/reader.rs index 8e46eafe48a..8c567c1c55a 100644 --- a/fluss-rust/crates/fluss/src/client/table/reader.rs +++ b/fluss-rust/crates/fluss/src/client/table/reader.rs @@ -41,10 +41,21 @@ use arrow_schema::SchemaRef; use futures::Stream; use log::warn; use std::collections::{HashMap, VecDeque}; -use std::time::Duration; +use std::time::{Duration, Instant}; const DEFAULT_POLL_TIMEOUT: Duration = Duration::from_millis(500); +/// Outcome of a bounded record-batch read with a caller-supplied timeout. +#[derive(Debug)] +pub enum RecordBatchReadOutcome { + /// A batch is available. + Batch(ScanBatch), + /// No batch became available before the timeout elapsed. + TimedOut, + /// Every subscribed bucket reached its stopping offset. + Finished, +} + /// Bounded log reader that consumes log data up to specified stopping offsets. /// /// This type wraps a [`RecordBatchLogScanner`] and adds stopping semantics: @@ -173,19 +184,43 @@ impl RecordBatchLogReader { /// Completed buckets are unsubscribed from the scanner to avoid wasting /// network traffic on data the reader will discard. pub async fn next_batch(&mut self) -> Result> { + loop { + match self.next_batch_with_timeout(DEFAULT_POLL_TIMEOUT).await? { + RecordBatchReadOutcome::Batch(batch) => return Ok(Some(batch)), + RecordBatchReadOutcome::TimedOut => continue, + RecordBatchReadOutcome::Finished => return Ok(None), + } + } + } + + /// Fetch the next [`ScanBatch`] while waiting for at most `timeout`. + /// + /// Unlike [`next_batch`](Self::next_batch), this method returns + /// [`RecordBatchReadOutcome::TimedOut`] when no data becomes available + /// before the timeout. The reader remains valid and the caller may retry. + pub async fn next_batch_with_timeout( + &mut self, + timeout: Duration, + ) -> Result { + let start = Instant::now(); loop { if let Some(batch) = self.buffer.pop_front() { - return Ok(Some(batch)); + return Ok(RecordBatchReadOutcome::Batch(batch)); } if self.stopping_offsets.is_empty() { - return Ok(None); + return Ok(RecordBatchReadOutcome::Finished); } - let scan_batches = self.scanner.poll(DEFAULT_POLL_TIMEOUT).await?; + let elapsed = start.elapsed(); + if elapsed >= timeout { + return Ok(RecordBatchReadOutcome::TimedOut); + } + + let scan_batches = self.scanner.poll(timeout - elapsed).await?; if scan_batches.is_empty() { - continue; + return Ok(RecordBatchReadOutcome::TimedOut); } let completed = diff --git a/fluss-rust/website/docs/user-guide/cpp/api-reference.md b/fluss-rust/website/docs/user-guide/cpp/api-reference.md index 621eb7ace50..8487e686c50 100644 --- a/fluss-rust/website/docs/user-guide/cpp/api-reference.md +++ b/fluss-rust/website/docs/user-guide/cpp/api-reference.md @@ -205,6 +205,37 @@ Performs prefix (bucket-key) lookups, returning all rows whose primary key start | `UnsubscribePartition(int64_t partition_id, int32_t bucket_id) -> Result` | Unsubscribe from a partition bucket | | `Poll(int64_t timeout_ms, ScanRecords& out) -> Result` | Poll individual records | | `PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) -> Result` | Poll Arrow RecordBatches | +| `CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out) -> Result` | Create a bounded reader using the latest offsets observed during the call | +| `CreateRecordBatchLogReaderUntilOffsets(const std::vector& offsets, RecordBatchLogReader& out) -> Result` | Create a bounded reader using explicit stopping offsets | + +The reader methods require a scanner created by +`TableScan::CreateRecordBatchLogScanner()`. Subscribe every bucket assigned to this reader first; +subscriptions provide the starting offsets, while `ReaderStopOffset` provides the stopping offset +for each bucket. Only one reader or polling operation may consume a scanner at a time. + +## `ReaderStopOffset` + +| Field | Type | Description | +|----------|---------------|--------------------------------------------------| +| `bucket` | `TableBucket` | A bucket already subscribed on the log scanner | +| `offset` | `int64_t` | Offset at which the bounded reader stops | + +## `RecordBatchLogReader` + +Bounded Arrow record-batch reader created from a subscribed `LogScanner`. + +| Method | Description | +|----------------------------------------------------------------------------------------------------|----------------------------------------------------------------| +| `Available() -> bool` | Check whether the reader is valid | +| `NextBatch(int64_t timeout_ms, ArrowRecordBatches& out, BoundedReadStatus& status) -> Result` | Wait up to the timeout for a batch, timeout, or completion | +| `CollectAllBatches(ArrowRecordBatches& out) -> Result` | Drain all batches until every stopping offset has been reached | + +`BoundedReadStatus` is `BatchAvailable`, `TimedOut`, or `Finished`. A timeout does not exhaust +the reader; callers may check cancellation and invoke `NextBatch()` again. + +`CreateRecordBatchLogReaderUntilLatest()` queries the latest offset of every subscribed bucket +when the reader is created. `CreateRecordBatchLogReaderUntilOffsets()` is useful when a +coordinator has already selected offsets and needs multiple workers to use the same boundary. ## `BatchScanner` diff --git a/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md b/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md index ca243f5d2ae..e74fdeaf937 100644 --- a/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md +++ b/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md @@ -144,6 +144,80 @@ for (size_t i = 0; i < batches.Size(); ++i) { } ``` +## Bounded Arrow RecordBatch Reading + +Use `RecordBatchLogReader` when the scan should finish after reaching a fixed offset for every +bucket. Starting offsets are configured by subscribing the scanner. Stopping offsets can be +queried once and passed explicitly: + +```cpp +auto info = table.GetTableInfo(); + +std::vector bucket_ids; +for (int32_t bucket_id = 0; bucket_id < info.num_buckets; ++bucket_id) { + bucket_ids.push_back(bucket_id); +} + +std::unordered_map latest_offsets; +admin.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets); + +fluss::LogScanner scanner; +table.NewScan().CreateRecordBatchLogScanner(scanner); + +std::vector stopping_offsets; +for (int32_t bucket_id : bucket_ids) { + scanner.Subscribe(bucket_id, 0); + stopping_offsets.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, latest_offsets.at(bucket_id)}); +} + +fluss::RecordBatchLogReader reader; +scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader); + +while (true) { + fluss::ArrowRecordBatches batches; + fluss::BoundedReadStatus status; + reader.NextBatch(1000, batches, status); + if (status == fluss::BoundedReadStatus::TimedOut) { + continue; // Check query cancellation before retrying. + } + if (status == fluss::BoundedReadStatus::Finished) { + break; + } + + for (const auto& batch : batches) { + std::cout << "bucket=" << batch->GetBucketId() + << " base_offset=" << batch->GetBaseOffset() + << " last_offset=" << batch->GetLastOffset() + << " rows=" << batch->NumRows() << std::endl; + } +} +``` + +`TimedOut` does not exhaust the reader. It lets a query engine periodically check cancellation +or deadlines before retrying. `Finished` means all stopping offsets have been reached. + +For the common case where the client should read everything currently available, let the +reader query the latest offsets: + +```cpp +fluss::LogScanner latest_scanner; +table.NewScan().CreateRecordBatchLogScanner(latest_scanner); +for (int32_t bucket_id : bucket_ids) { + latest_scanner.Subscribe(bucket_id, 0); +} + +fluss::RecordBatchLogReader latest_reader; +latest_scanner.CreateRecordBatchLogReaderUntilLatest(admin, latest_reader); + +fluss::ArrowRecordBatches batches; +latest_reader.CollectAllBatches(batches); +``` + +Subscribe every bucket assigned to the reader before creating it. Do not call `PollRecordBatch()` +or create a second reader concurrently on the same scanner. Destroy the reader before reusing +that scanner. + ## Column Projection ```cpp From 210b43304ef61a300f2337ed40b804ed2689229a Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 14 Aug 2026 10:50:30 +0800 Subject: [PATCH 2/2] [c++] Simplify bounded log reader API --- fluss-rust/bindings/cpp/README.md | 43 +-- fluss-rust/bindings/cpp/examples/example.cpp | 74 ++-- fluss-rust/bindings/cpp/include/fluss.hpp | 95 +++++- fluss-rust/bindings/cpp/src/lib.rs | 24 +- fluss-rust/bindings/cpp/src/table.cpp | 320 +++++++++++++++++- .../bindings/cpp/test/test_log_table.cpp | 213 +++++++++--- .../crates/fluss/src/client/table/reader.rs | 73 +++- .../docs/user-guide/cpp/api-reference.md | 72 +++- .../docs/user-guide/cpp/example/log-tables.md | 69 ++-- 9 files changed, 807 insertions(+), 176 deletions(-) diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md index c8cd54fa871..b609128118f 100644 --- a/fluss-rust/bindings/cpp/README.md +++ b/fluss-rust/bindings/cpp/README.md @@ -46,9 +46,9 @@ See [ci.sh](ci.sh) for the CI build sequence. [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, first subscribe the record-batch scanner at the desired starting -offsets, then create a `RecordBatchLogReader` with either explicit stopping offsets or the -latest offsets observed at reader creation: +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(); @@ -60,38 +60,41 @@ for (int32_t bucket_id = 0; bucket_id < info.num_buckets; ++bucket_id) { std::unordered_map latest_offsets; admin.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets); -fluss::LogScanner scanner; -table.NewScan().CreateRecordBatchLogScanner(scanner); - -std::vector stopping_offsets; +std::vector ranges; for (int32_t bucket_id : bucket_ids) { - scanner.Subscribe(bucket_id, 0); - stopping_offsets.push_back( - {fluss::TableBucket{info.table_id, bucket_id}, latest_offsets.at(bucket_id)}); + ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, 0, latest_offsets.at(bucket_id)}); } fluss::RecordBatchLogReader reader; -scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader); +table.NewScan().CreateRecordBatchLogReader(ranges, reader); while (true) { - fluss::ArrowRecordBatches batches; - fluss::BoundedReadStatus status; - reader.NextBatch(1000, batches, status); - if (status == fluss::BoundedReadStatus::TimedOut) { + fluss::RecordBatchReadResult result; + reader.NextBatch(1000, result); + if (result.status == fluss::BoundedReadStatus::TimedOut) { continue; // Check query cancellation before retrying. } - if (status == fluss::BoundedReadStatus::Finished) { + if (result.status == fluss::BoundedReadStatus::Finished) { break; } - for (const auto& batch : batches) { - process(batch->GetArrowRecordBatch()); - } + 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. Use only one active reader or polling operation for a scanner at a time. +check cancellation. ## TODO diff --git a/fluss-rust/bindings/cpp/examples/example.cpp b/fluss-rust/bindings/cpp/examples/example.cpp index 2bd087c1d11..9341619cc03 100644 --- a/fluss-rust/bindings/cpp/examples/example.cpp +++ b/fluss-rust/bindings/cpp/examples/example.cpp @@ -371,44 +371,34 @@ int main() { // 8.1) Bounded Arrow record batch scan with explicit stopping offsets std::cout << "\n=== Bounded Arrow Record Batch Scan ===" << std::endl; - fluss::LogScanner bounded_scanner; - check("new_bounded_record_batch_scanner", - table.NewScan().CreateRecordBatchLogScanner(bounded_scanner)); - - std::vector stopping_offsets; + std::vector 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); - if (starting_offset >= stopping_offset) { - continue; - } - check("subscribe_bounded", bounded_scanner.Subscribe(bucket_id, starting_offset)); - stopping_offsets.push_back({fluss::TableBucket{info.table_id, bucket_id}, stopping_offset}); + bounded_ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, starting_offset, stopping_offset}); } - if (!stopping_offsets.empty()) { + if (!bounded_ranges.empty()) { fluss::RecordBatchLogReader bounded_reader; - check("create_bounded_reader", bounded_scanner.CreateRecordBatchLogReaderUntilOffsets( - stopping_offsets, bounded_reader)); + check("create_bounded_reader", + table.NewScan().CreateRecordBatchLogReader(bounded_ranges, bounded_reader)); int64_t bounded_row_count = 0; while (true) { - fluss::ArrowRecordBatches bounded_batches; - fluss::BoundedReadStatus status; - check("bounded_next_batch", bounded_reader.NextBatch(1000, bounded_batches, status)); - if (status == fluss::BoundedReadStatus::TimedOut) { + fluss::RecordBatchReadResult result; + check("bounded_next_batch", bounded_reader.NextBatch(1000, result)); + if (result.status == fluss::BoundedReadStatus::TimedOut) { continue; } - if (status == fluss::BoundedReadStatus::Finished) { + if (result.status == fluss::BoundedReadStatus::Finished) { break; } - for (const auto& bounded_batch : bounded_batches) { - bounded_row_count += bounded_batch->NumRows(); - std::cout << " bucket=" << bounded_batch->GetBucketId() - << " base_offset=" << bounded_batch->GetBaseOffset() - << " last_offset=" << bounded_batch->GetLastOffset() - << " rows=" << bounded_batch->NumRows() << std::endl; - } + 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; } @@ -418,6 +408,8 @@ int main() { auto timestamp_ms = std::chrono::duration_cast(one_hour_ago.time_since_epoch()) .count(); + auto now_ms = + std::chrono::duration_cast(now.time_since_epoch()).count(); std::unordered_map timestamp_offsets; check("list_timestamp_offsets", @@ -428,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 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; @@ -471,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)); @@ -480,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) { @@ -496,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) @@ -508,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; diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp index b32ab1c9e01..6f472dc59f4 100644 --- a/fluss-rust/bindings/cpp/include/fluss.hpp +++ b/fluss-rust/bindings/cpp/include/fluss.hpp @@ -1242,6 +1242,21 @@ struct ReaderStopOffset { 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, @@ -1249,6 +1264,11 @@ enum class BoundedReadStatus { Finished = 2, }; +struct RecordBatchReadResult { + BoundedReadStatus status{BoundedReadStatus::TimedOut}; + std::unique_ptr batch; +}; + struct LakeSnapshot { int64_t snapshot_id; std::vector bucket_offsets; @@ -1352,6 +1372,7 @@ class Lookuper; class PrefixLookuper; class WriteResult; class LogScanner; +class RecordBatchLogScanner; class RecordBatchLogReader; class BatchScanner; class Admin; @@ -1648,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& 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& buckets, + const TimestampRange& range, RecordBatchLogReader& out); + Result CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out); private: @@ -1658,6 +1697,9 @@ class TableScan { std::vector ResolveNameProjection() const; Result DoCreateScanner(LogScanner& out, bool is_record_batch); + Result ResolveTimestampRanges(Admin& admin, const std::vector& buckets, + const TimestampRange& range, + std::vector& out) const; ffi::Table* table_{nullptr}; std::vector projection_; @@ -1809,6 +1851,14 @@ class LogScanner { Result Poll(int64_t timeout_ms, ScanRecords& out); Result PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out); + 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. @@ -1820,13 +1870,46 @@ class LogScanner { Result CreateRecordBatchLogReaderUntilOffsets(const std::vector& 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& bucket_offsets); + Result SubscribePartitionBuckets(int64_t partition_id, int32_t bucket_id, int64_t start_offset); + Result SubscribePartitionBuckets(const std::vector& 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& offsets, + RecordBatchLogReader& out) &&; + private: - friend class Table; friend class TableScan; - LogScanner(ffi::LogScanner* scanner) noexcept; + explicit RecordBatchLogScanner(ffi::LogScanner* scanner) noexcept; - void Destroy() noexcept; - ffi::LogScanner* scanner_{nullptr}; + LogScanner scanner_; }; /// Bounded Arrow batch reader created from a subscribed record-batch log scanner. @@ -1846,13 +1929,15 @@ class RecordBatchLogReader { /// 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, ArrowRecordBatches& out, BoundedReadStatus& status); + 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; diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs index a44151ad882..ea09e5441ff 100644 --- a/fluss-rust/bindings/cpp/src/lib.rs +++ b/fluss-rust/bindings/cpp/src/lib.rs @@ -2274,18 +2274,20 @@ impl LogScanner { return client_err_ptr("Batch-based scanner not available".to_string()); }; - let stopping_offsets = offsets - .into_iter() - .map(|offset| { - let partition_id = offset.has_partition_id.then_some(offset.partition_id); - let bucket = fcore::metadata::TableBucket::new_with_partition( - offset.table_id, - partition_id, - offset.bucket_id, + let mut stopping_offsets = HashMap::with_capacity(offsets.len()); + for offset in offsets { + let partition_id = offset.has_partition_id.then_some(offset.partition_id); + let bucket = fcore::metadata::TableBucket::new_with_partition( + offset.table_id, + partition_id, + offset.bucket_id, + ); + if stopping_offsets.insert(bucket, offset.offset).is_some() { + return client_err_ptr( + "Duplicate bucket in bounded reader stopping offsets".to_string(), ); - (bucket, offset.offset) - }) - .collect(); + } + } match fcore::client::RecordBatchLogReader::new_until_offsets( scanner.new_shared_handle(), diff --git a/fluss-rust/bindings/cpp/src/table.cpp b/fluss-rust/bindings/cpp/src/table.cpp index 2df9760bc3a..68b1b9c5c11 100644 --- a/fluss-rust/bindings/cpp/src/table.cpp +++ b/fluss-rust/bindings/cpp/src/table.cpp @@ -21,6 +21,9 @@ #include #include +#include +#include +#include #include "ffi_converter.hpp" #include "fluss.hpp" @@ -1245,6 +1248,16 @@ std::vector TableScan::ResolveNameProjection() const { Result TableScan::CreateLogScanner(LogScanner& out) { return DoCreateScanner(out, false); } +Result TableScan::CreateRecordBatchLogScanner(RecordBatchLogScanner& out) { + LogScanner scanner; + auto result = DoCreateScanner(scanner, true); + if (result.Ok()) { + out = RecordBatchLogScanner(scanner.scanner_); + scanner.scanner_ = nullptr; + } + return result; +} + Result TableScan::CreateRecordBatchLogScanner(LogScanner& out) { return DoCreateScanner(out, true); } @@ -1276,6 +1289,210 @@ Result TableScan::DoCreateScanner(LogScanner& out, bool is_record_batch) { } } +Result TableScan::CreateRecordBatchLogReader(const std::vector& ranges, + RecordBatchLogReader& out) { + if (table_ == nullptr) { + return utils::make_client_error("Table not available"); + } + + auto info = utils::from_ffi_table_info(table_->get_table_info_from_table()); + std::vector bucket_subscriptions; + std::vector partition_subscriptions; + std::vector stopping_offsets; + std::set> seen_buckets; + + for (const auto& range : ranges) { + if (range.bucket.table_id != info.table_id) { + return utils::make_client_error("Read range table_id does not match the scanned table"); + } + if (range.starting_offset > range.stopping_offset) { + return utils::make_client_error( + "Read range starting_offset must not exceed stopping_offset"); + } + + const int64_t partition_id = range.bucket.partition_id.value_or(-1); + if (!seen_buckets.emplace(range.bucket.table_id, partition_id, range.bucket.bucket_id) + .second) { + return utils::make_client_error("Duplicate bucket in bounded read ranges"); + } + if (info.is_partitioned != range.bucket.partition_id.has_value()) { + return utils::make_client_error( + info.is_partitioned + ? "Partitioned table read ranges must include partition_id" + : "Non-partitioned table read ranges must not include partition_id"); + } + + // Empty ranges complete immediately and do not need a subscription. + if (range.starting_offset == range.stopping_offset) { + continue; + } + + if (info.is_partitioned) { + partition_subscriptions.push_back( + {*range.bucket.partition_id, range.bucket.bucket_id, range.starting_offset}); + } else { + bucket_subscriptions.push_back({range.bucket.bucket_id, range.starting_offset}); + } + stopping_offsets.push_back({range.bucket, range.stopping_offset}); + } + + RecordBatchLogScanner scanner; + auto result = CreateRecordBatchLogScanner(scanner); + if (!result.Ok()) { + return result; + } + + if (!partition_subscriptions.empty()) { + result = scanner.SubscribePartitionBuckets(partition_subscriptions); + } else if (!bucket_subscriptions.empty()) { + result = scanner.Subscribe(bucket_subscriptions); + } + if (!result.Ok()) { + return result; + } + + return std::move(scanner).CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, out); +} + +Result TableScan::ResolveTimestampRanges(Admin& admin, const std::vector& buckets, + const TimestampRange& range, + std::vector& out) const { + if (table_ == nullptr) { + return utils::make_client_error("Table not available"); + } + if (!admin.Available()) { + return utils::make_client_error("Admin not available"); + } + if (range.starting_timestamp_ms > range.stopping_timestamp_ms) { + return utils::make_client_error( + "starting_timestamp_ms must not exceed stopping_timestamp_ms"); + } + + auto info = utils::from_ffi_table_info(table_->get_table_info_from_table()); + auto ffi_path = table_->get_table_path(); + TablePath table_path(std::string(ffi_path.database_name), std::string(ffi_path.table_name)); + std::set> seen_buckets; + out.clear(); + if (buckets.empty()) { + return utils::make_ok(); + } + + if (!info.is_partitioned) { + std::vector bucket_ids; + for (const auto& bucket : buckets) { + if (bucket.table_id != info.table_id || bucket.partition_id.has_value()) { + return utils::make_client_error( + "Timestamp range contains a bucket from another table or partition mode"); + } + if (!seen_buckets.emplace(bucket.table_id, -1, bucket.bucket_id).second) { + return utils::make_client_error("Duplicate bucket in timestamp read range"); + } + bucket_ids.push_back(bucket.bucket_id); + } + + std::unordered_map starting_offsets; + std::unordered_map stopping_offsets; + auto result = + admin.ListOffsets(table_path, bucket_ids, + OffsetSpec::Timestamp(range.starting_timestamp_ms), starting_offsets); + if (!result.Ok()) { + return result; + } + result = + admin.ListOffsets(table_path, bucket_ids, + OffsetSpec::Timestamp(range.stopping_timestamp_ms), stopping_offsets); + if (!result.Ok()) { + return result; + } + + for (const auto& bucket : buckets) { + auto start = starting_offsets.find(bucket.bucket_id); + auto stop = stopping_offsets.find(bucket.bucket_id); + if (start == starting_offsets.end() || stop == stopping_offsets.end()) { + return utils::make_client_error( + "Timestamp offset lookup did not return every requested bucket"); + } + out.push_back({bucket, start->second, stop->second}); + } + return utils::make_ok(); + } + + std::vector partition_infos; + auto result = admin.ListPartitionInfos(table_path, partition_infos); + if (!result.Ok()) { + return result; + } + std::unordered_map partition_names; + for (const auto& partition : partition_infos) { + partition_names.emplace(partition.partition_id, partition.partition_name); + } + + std::map> bucket_ids_by_partition; + for (const auto& bucket : buckets) { + if (bucket.table_id != info.table_id || !bucket.partition_id.has_value()) { + return utils::make_client_error( + "Timestamp range contains a bucket from another table or partition mode"); + } + if (!seen_buckets.emplace(bucket.table_id, *bucket.partition_id, bucket.bucket_id).second) { + return utils::make_client_error("Duplicate bucket in timestamp read range"); + } + bucket_ids_by_partition[*bucket.partition_id].push_back(bucket.bucket_id); + } + + std::map, int64_t> starting_offsets; + std::map, int64_t> stopping_offsets; + for (const auto& entry : bucket_ids_by_partition) { + const int64_t partition_id = entry.first; + const auto& bucket_ids = entry.second; + auto partition_name = partition_names.find(partition_id); + if (partition_name == partition_names.end()) { + return utils::make_client_error("Unknown partition_id in timestamp read range"); + } + + std::unordered_map partition_starts; + std::unordered_map partition_stops; + result = admin.ListPartitionOffsets(table_path, partition_name->second, bucket_ids, + OffsetSpec::Timestamp(range.starting_timestamp_ms), + partition_starts); + if (!result.Ok()) { + return result; + } + result = admin.ListPartitionOffsets(table_path, partition_name->second, bucket_ids, + OffsetSpec::Timestamp(range.stopping_timestamp_ms), + partition_stops); + if (!result.Ok()) { + return result; + } + for (int32_t bucket_id : bucket_ids) { + auto start = partition_starts.find(bucket_id); + auto stop = partition_stops.find(bucket_id); + if (start == partition_starts.end() || stop == partition_stops.end()) { + return utils::make_client_error( + "Timestamp offset lookup did not return every requested partition bucket"); + } + starting_offsets[{partition_id, bucket_id}] = start->second; + stopping_offsets[{partition_id, bucket_id}] = stop->second; + } + } + + for (const auto& bucket : buckets) { + const auto key = std::make_pair(*bucket.partition_id, bucket.bucket_id); + out.push_back({bucket, starting_offsets.at(key), stopping_offsets.at(key)}); + } + return utils::make_ok(); +} + +Result TableScan::CreateRecordBatchLogReader(Admin& admin, const std::vector& buckets, + const TimestampRange& range, + RecordBatchLogReader& out) { + std::vector ranges; + auto result = ResolveTimestampRanges(admin, buckets, range, ranges); + if (!result.Ok()) { + return result; + } + return CreateRecordBatchLogReader(ranges, out); +} + TableScan& TableScan::Limit(int32_t row_number) { limit_ = row_number; return *this; @@ -1841,6 +2058,20 @@ struct ArrowBatchImporter { } return utils::make_ok(); } + + static Result ImportOne(ffi::FfiArrowRecordBatches& src, + std::unique_ptr& out) { + ArrowRecordBatches batches; + auto result = Import(src, batches); + if (!result.Ok()) { + return result; + } + if (batches.Size() > 1) { + return utils::make_client_error("Bounded reader returned more than one record batch"); + } + out = batches.Empty() ? nullptr : std::move(batches.batches.front()); + return utils::make_ok(); + } }; } // namespace detail @@ -1857,6 +2088,72 @@ Result LogScanner::PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out); } +// ============================================================================ +// RecordBatchLogScanner +// ============================================================================ + +RecordBatchLogScanner::RecordBatchLogScanner() noexcept = default; + +RecordBatchLogScanner::RecordBatchLogScanner(ffi::LogScanner* scanner) noexcept + : scanner_(scanner) {} + +RecordBatchLogScanner::~RecordBatchLogScanner() noexcept = default; + +RecordBatchLogScanner::RecordBatchLogScanner(RecordBatchLogScanner&& other) noexcept = default; + +RecordBatchLogScanner& RecordBatchLogScanner::operator=(RecordBatchLogScanner&& other) noexcept = + default; + +bool RecordBatchLogScanner::Available() const { return scanner_.Available(); } + +Result RecordBatchLogScanner::Subscribe(int32_t bucket_id, int64_t start_offset) { + return scanner_.Subscribe(bucket_id, start_offset); +} + +Result RecordBatchLogScanner::Subscribe(const std::vector& bucket_offsets) { + return scanner_.Subscribe(bucket_offsets); +} + +Result RecordBatchLogScanner::SubscribePartitionBuckets(int64_t partition_id, int32_t bucket_id, + int64_t start_offset) { + return scanner_.SubscribePartitionBuckets(partition_id, bucket_id, start_offset); +} + +Result RecordBatchLogScanner::SubscribePartitionBuckets( + const std::vector& subscriptions) { + return scanner_.SubscribePartitionBuckets(subscriptions); +} + +Result RecordBatchLogScanner::Unsubscribe(int32_t bucket_id) { + return scanner_.Unsubscribe(bucket_id); +} + +Result RecordBatchLogScanner::UnsubscribePartition(int64_t partition_id, int32_t bucket_id) { + return scanner_.UnsubscribePartition(partition_id, bucket_id); +} + +Result RecordBatchLogScanner::Poll(int64_t timeout_ms, ArrowRecordBatches& out) { + return scanner_.PollRecordBatch(timeout_ms, out); +} + +Result RecordBatchLogScanner::CreateRecordBatchLogReaderUntilLatest(const Admin& admin, + RecordBatchLogReader& out) && { + auto result = scanner_.CreateRecordBatchLogReaderUntilLatest(admin, out); + if (result.Ok()) { + scanner_ = LogScanner(); + } + return result; +} + +Result RecordBatchLogScanner::CreateRecordBatchLogReaderUntilOffsets( + const std::vector& offsets, RecordBatchLogReader& out) && { + auto result = scanner_.CreateRecordBatchLogReaderUntilOffsets(offsets, out); + if (result.Ok()) { + scanner_ = LogScanner(); + } + return result; +} + // ============================================================================ // RecordBatchLogReader // ============================================================================ @@ -1936,12 +2233,13 @@ RecordBatchLogReader& RecordBatchLogReader::operator=(RecordBatchLogReader&& oth bool RecordBatchLogReader::Available() const { return reader_ != nullptr; } -Result RecordBatchLogReader::NextBatch(int64_t timeout_ms, ArrowRecordBatches& out, - BoundedReadStatus& status) { +Result RecordBatchLogReader::NextBatch(int64_t timeout_ms, RecordBatchReadResult& out) { if (!Available()) { return utils::make_client_error("RecordBatchLogReader not available"); } + out.status = BoundedReadStatus::TimedOut; + out.batch.reset(); auto ffi_result = reader_->record_batch_log_reader_next_batch(timeout_ms); auto result = utils::from_ffi_result(ffi_result.result); if (!result.Ok()) { @@ -1949,19 +2247,29 @@ Result RecordBatchLogReader::NextBatch(int64_t timeout_ms, ArrowRecordBatches& o } switch (ffi_result.status) { case 0: - status = BoundedReadStatus::BatchAvailable; + out.status = BoundedReadStatus::BatchAvailable; break; case 1: - status = BoundedReadStatus::TimedOut; + out.status = BoundedReadStatus::TimedOut; break; case 2: - status = BoundedReadStatus::Finished; + out.status = BoundedReadStatus::Finished; break; default: return utils::make_client_error("Unknown bounded read status: " + std::to_string(ffi_result.status)); } - return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out); + result = detail::ArrowBatchImporter::ImportOne(ffi_result.arrow_batches, out.batch); + if (!result.Ok()) { + return result; + } + if (out.status == BoundedReadStatus::BatchAvailable && !out.batch) { + return utils::make_client_error("Bounded reader reported a batch without returning one"); + } + if (out.status != BoundedReadStatus::BatchAvailable && out.batch) { + return utils::make_client_error("Bounded reader returned a batch for a terminal status"); + } + return utils::make_ok(); } Result RecordBatchLogReader::CollectAllBatches(ArrowRecordBatches& out) { diff --git a/fluss-rust/bindings/cpp/test/test_log_table.cpp b/fluss-rust/bindings/cpp/test/test_log_table.cpp index fbbbbb1ccb3..8ec73e08540 100644 --- a/fluss-rust/bindings/cpp/test/test_log_table.cpp +++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp @@ -209,9 +209,6 @@ TEST_F(LogTableTest, RecordBatchLogReaderUntilOffsets) { ASSERT_OK(append_writer.AppendArrowBatch(batch)); ASSERT_OK(append_writer.Flush()); - fluss::LogScanner scanner; - ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(scanner)); - const int64_t table_id = table.GetTableInfo().table_id; std::vector bucket_ids; for (int32_t bucket_id = 0; bucket_id < kNumBuckets; ++bucket_id) { @@ -222,61 +219,63 @@ TEST_F(LogTableTest, RecordBatchLogReaderUntilOffsets) { ASSERT_OK(adm.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets)); ASSERT_EQ(latest_offsets.size(), bucket_ids.size()); - std::vector stopping_offsets; + std::vector ranges; std::vector expected_rows_by_bucket(kNumBuckets); for (int32_t bucket_id : bucket_ids) { const int64_t stopping_offset = latest_offsets.at(bucket_id); ASSERT_GT(stopping_offset, 1) << "Bucket " << bucket_id << " should contain rows after starting offset 1"; - ASSERT_OK(scanner.Subscribe(bucket_id, 1)); - stopping_offsets.push_back({fluss::TableBucket{table_id, bucket_id}, stopping_offset}); + ranges.push_back({fluss::TableBucket{table_id, bucket_id}, 1, stopping_offset}); expected_rows_by_bucket[bucket_id] = stopping_offset - 1; } fluss::RecordBatchLogReader reader; - ASSERT_OK(scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader)); - - fluss::ArrowRecordBatches batches; - ASSERT_OK(reader.CollectAllBatches(batches)); + ASSERT_OK(table.NewScan().CreateRecordBatchLogReader(ranges, reader)); std::vector actual_rows_by_bucket(kNumBuckets); - for (const auto& bounded_batch : batches) { - const int32_t bucket_id = bounded_batch->GetBucketId(); + int timeout_count = 0; + while (true) { + fluss::RecordBatchReadResult result; + ASSERT_OK(reader.NextBatch(1000, result)); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + ASSERT_LT(++timeout_count, 10); + continue; + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + + ASSERT_NE(result.batch, nullptr); + const int32_t bucket_id = result.batch->GetBucketId(); ASSERT_GE(bucket_id, 0); ASSERT_LT(bucket_id, kNumBuckets); - EXPECT_GE(bounded_batch->GetBaseOffset(), 1); - EXPECT_LT(bounded_batch->GetLastOffset(), latest_offsets.at(bucket_id)); - actual_rows_by_bucket[bucket_id] += bounded_batch->NumRows(); + EXPECT_GE(result.batch->GetBaseOffset(), 1); + EXPECT_LT(result.batch->GetLastOffset(), latest_offsets.at(bucket_id)); + actual_rows_by_bucket[bucket_id] += result.batch->NumRows(); } for (int32_t bucket_id : bucket_ids) { EXPECT_EQ(actual_rows_by_bucket[bucket_id], expected_rows_by_bucket[bucket_id]) << "Unexpected row count for bucket " << bucket_id; } - fluss::ArrowRecordBatches eof_batches; - fluss::BoundedReadStatus eof_status = fluss::BoundedReadStatus::BatchAvailable; - ASSERT_OK(reader.NextBatch(1000, eof_batches, eof_status)); - EXPECT_TRUE(eof_batches.Empty()); - EXPECT_EQ(eof_status, fluss::BoundedReadStatus::Finished); + fluss::RecordBatchReadResult eof_result; + ASSERT_OK(reader.NextBatch(1000, eof_result)); + EXPECT_EQ(eof_result.batch, nullptr); + EXPECT_EQ(eof_result.status, fluss::BoundedReadStatus::Finished); // A bounded reader whose stopping offset is not available yet should return // TimedOut without becoming exhausted, so query engines can check cancellation // and retry. { - fluss::LogScanner waiting_scanner; - ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(waiting_scanner)); const int64_t start_offset = latest_offsets.at(0); - ASSERT_OK(waiting_scanner.Subscribe(0, start_offset)); - fluss::RecordBatchLogReader waiting_reader; - ASSERT_OK(waiting_scanner.CreateRecordBatchLogReaderUntilOffsets( - {{fluss::TableBucket{table_id, 0}, start_offset + 1}}, waiting_reader)); - - fluss::ArrowRecordBatches timeout_batches; - fluss::BoundedReadStatus timeout_status = fluss::BoundedReadStatus::Finished; - ASSERT_OK(waiting_reader.NextBatch(100, timeout_batches, timeout_status)); - EXPECT_TRUE(timeout_batches.Empty()); - EXPECT_EQ(timeout_status, fluss::BoundedReadStatus::TimedOut); + ASSERT_OK(table.NewScan().CreateRecordBatchLogReader( + {{fluss::TableBucket{table_id, 0}, start_offset, start_offset + 1}}, waiting_reader)); + + fluss::RecordBatchReadResult timeout_result; + ASSERT_OK(waiting_reader.NextBatch(100, timeout_result)); + EXPECT_EQ(timeout_result.batch, nullptr); + EXPECT_EQ(timeout_result.status, fluss::BoundedReadStatus::TimedOut); } ASSERT_OK(adm.DropTable(table_path, false)); @@ -309,29 +308,96 @@ TEST_F(LogTableTest, RecordBatchLogReaderUntilLatest) { ASSERT_OK(append_writer.AppendArrowBatch(batch)); ASSERT_OK(append_writer.Flush()); - fluss::LogScanner scanner; + fluss::RecordBatchLogScanner scanner; ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(scanner)); ASSERT_OK(scanner.Subscribe(0, 0)); fluss::RecordBatchLogReader reader; - ASSERT_OK(scanner.CreateRecordBatchLogReaderUntilLatest(adm, reader)); - - fluss::ArrowRecordBatches batches; - fluss::BoundedReadStatus status = fluss::BoundedReadStatus::TimedOut; - ASSERT_OK(reader.NextBatch(5000, batches, status)); - ASSERT_EQ(status, fluss::BoundedReadStatus::BatchAvailable); - ASSERT_EQ(batches.Size(), 1); - auto ids = - std::static_pointer_cast(batches[0]->GetArrowRecordBatch()->column(0)); + ASSERT_OK(std::move(scanner).CreateRecordBatchLogReaderUntilLatest(adm, reader)); + EXPECT_FALSE(scanner.Available()); + + fluss::RecordBatchReadResult read_result; + ASSERT_OK(reader.NextBatch(5000, read_result)); + ASSERT_EQ(read_result.status, fluss::BoundedReadStatus::BatchAvailable); + ASSERT_NE(read_result.batch, nullptr); + auto ids = std::static_pointer_cast( + read_result.batch->GetArrowRecordBatch()->column(0)); ASSERT_EQ(ids->length(), 3); EXPECT_EQ(ids->Value(0), 1); EXPECT_EQ(ids->Value(1), 2); EXPECT_EQ(ids->Value(2), 3); - fluss::ArrowRecordBatches eof_batches; - ASSERT_OK(reader.NextBatch(1000, eof_batches, status)); - EXPECT_TRUE(eof_batches.Empty()); - EXPECT_EQ(status, fluss::BoundedReadStatus::Finished); + fluss::RecordBatchReadResult eof_result; + ASSERT_OK(reader.NextBatch(1000, eof_result)); + EXPECT_EQ(eof_result.batch, nullptr); + EXPECT_EQ(eof_result.status, fluss::BoundedReadStatus::Finished); + + ASSERT_OK(adm.DropTable(table_path, false)); +} + +TEST_F(LogTableTest, RecordBatchLogReaderTimestampRange) { + auto& adm = admin(); + auto& conn = connection(); + + fluss::TablePath table_path("fluss", "test_record_batch_log_reader_timestamp_cpp"); + auto schema = fluss::Schema::NewBuilder().AddColumn("c1", DataType::Int()).Build(); + auto table_descriptor = fluss::TableDescriptor::NewBuilder() + .SetSchema(schema) + .SetBucketCount(1) + .SetBucketKeys({"c1"}) + .SetProperty("table.replication.factor", "1") + .Build(); + fluss_test::CreateTable(adm, table_path, table_descriptor); + + fluss::Table table; + ASSERT_OK(conn.GetTable(table_path, table)); + fluss::AppendWriter writer; + ASSERT_OK(table.NewAppend().CreateWriter(writer)); + + const auto starting_timestamp_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + std::this_thread::sleep_for(std::chrono::seconds(1)); + + auto c1 = arrow::Int32Builder(); + ASSERT_TRUE(c1.AppendValues({1, 2, 3}).ok()); + auto batch = arrow::RecordBatch::Make(arrow::schema({arrow::field("c1", arrow::int32())}), 3, + {c1.Finish().ValueOrDie()}); + ASSERT_OK(writer.AppendArrowBatch(batch)); + ASSERT_OK(writer.Flush()); + + std::this_thread::sleep_for(std::chrono::seconds(1)); + const auto stopping_timestamp_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + + const auto info = table.GetTableInfo(); + fluss::RecordBatchLogReader timestamp_reader; + ASSERT_OK(table.NewScan().CreateRecordBatchLogReader( + adm, {fluss::TableBucket{info.table_id, 0}}, + fluss::TimestampRange{starting_timestamp_ms, stopping_timestamp_ms}, timestamp_reader)); + + std::vector ids; + int timeout_count = 0; + while (true) { + fluss::RecordBatchReadResult result; + ASSERT_OK(timestamp_reader.NextBatch(1000, result)); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + ASSERT_LT(++timeout_count, 10); + continue; + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + + ASSERT_NE(result.batch, nullptr); + auto id_array = std::static_pointer_cast( + result.batch->GetArrowRecordBatch()->column(0)); + for (int64_t i = 0; i < id_array->length(); ++i) { + ids.push_back(id_array->Value(i)); + } + } + EXPECT_EQ(ids, std::vector({1, 2, 3})); ASSERT_OK(adm.DropTable(table_path, false)); } @@ -1053,6 +1119,11 @@ TEST_F(LogTableTest, PartitionedTableAppendScan) { fluss::AppendWriter append_writer; ASSERT_OK(table_append.CreateWriter(append_writer)); + const auto starting_timestamp_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + std::this_thread::sleep_for(std::chrono::seconds(1)); + // Append rows struct TestData { int32_t id; @@ -1110,6 +1181,11 @@ TEST_F(LogTableTest, PartitionedTableAppendScan) { } ASSERT_OK(append_writer.Flush()); + std::this_thread::sleep_for(std::chrono::seconds(1)); + const auto stopping_timestamp_ms = std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); + // Test list partition offsets std::unordered_map us_offsets; ASSERT_OK(adm.ListPartitionOffsets(table_path, "US", {0}, fluss::OffsetSpec::Latest(), @@ -1155,20 +1231,15 @@ TEST_F(LogTableTest, PartitionedTableAppendScan) { { fluss::Table bounded_table; ASSERT_OK(conn.GetTable(table_path, bounded_table)); - fluss::LogScanner bounded_scanner; - ASSERT_OK(bounded_table.NewScan().CreateRecordBatchLogScanner(bounded_scanner)); - - std::vector subscriptions; - std::vector stopping_offsets; + std::vector ranges; for (const auto& pi : partition_infos) { - subscriptions.push_back({pi.partition_id, 0, 1}); - stopping_offsets.push_back( - {fluss::TableBucket{bounded_table.GetTableInfo().table_id, 0, pi.partition_id}, 3}); + ranges.push_back( + {fluss::TableBucket{bounded_table.GetTableInfo().table_id, 0, pi.partition_id}, 1, + 3}); } - ASSERT_OK(bounded_scanner.SubscribePartitionBuckets(subscriptions)); fluss::RecordBatchLogReader reader; - ASSERT_OK(bounded_scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader)); + ASSERT_OK(bounded_table.NewScan().CreateRecordBatchLogReader(ranges, reader)); fluss::ArrowRecordBatches batches; ASSERT_OK(reader.CollectAllBatches(batches)); @@ -1185,6 +1256,36 @@ TEST_F(LogTableTest, PartitionedTableAppendScan) { EXPECT_EQ(ids, std::vector({2, 4, 5, 7})); } + // Test timestamp-bounded reading across partition buckets. + { + fluss::Table timestamp_table; + ASSERT_OK(conn.GetTable(table_path, timestamp_table)); + + std::vector buckets; + for (const auto& pi : partition_infos) { + buckets.push_back({timestamp_table.GetTableInfo().table_id, 0, pi.partition_id}); + } + + fluss::RecordBatchLogReader reader; + ASSERT_OK(timestamp_table.NewScan().CreateRecordBatchLogReader( + adm, buckets, fluss::TimestampRange{starting_timestamp_ms, stopping_timestamp_ms}, + reader)); + + fluss::ArrowRecordBatches batches; + ASSERT_OK(reader.CollectAllBatches(batches)); + + std::vector ids; + for (const auto& bounded_batch : batches) { + auto id_array = std::static_pointer_cast( + bounded_batch->GetArrowRecordBatch()->column(0)); + for (int64_t row = 0; row < id_array->length(); ++row) { + ids.push_back(id_array->Value(row)); + } + } + std::sort(ids.begin(), ids.end()); + EXPECT_EQ(ids, std::vector({1, 2, 3, 4, 5, 6, 7, 8})); + } + // Test unsubscribe_partition: unsubscribe EU, should only get US data { fluss::Table unsub_table; diff --git a/fluss-rust/crates/fluss/src/client/table/reader.rs b/fluss-rust/crates/fluss/src/client/table/reader.rs index 8c567c1c55a..81a4e849c33 100644 --- a/fluss-rust/crates/fluss/src/client/table/reader.rs +++ b/fluss-rust/crates/fluss/src/client/table/reader.rs @@ -136,18 +136,26 @@ impl RecordBatchLogReader { /// Create a reader with explicit stopping offsets per bucket. /// /// # NOTE: Every key in `stopping_offsets` **must** correspond to a bucket that is - /// currently subscribed on the `scanner`. If a stopping offset refers to a - /// bucket that will never appear in polled batches, the reader will loop - /// indefinitely waiting for data that never arrives. + /// currently subscribed on the `scanner`; construction fails otherwise. + /// Concrete subscriptions that already meet their stop point are treated + /// as empty ranges and complete immediately. /// /// Use [`new_until_latest`](Self::new_until_latest) for the common case; /// it queries the server and builds a validated stopping-offset map /// automatically. pub fn new_until_offsets( scanner: RecordBatchLogScanner, - stopping_offsets: HashMap, + mut stopping_offsets: HashMap, ) -> Result { scanner.try_set_reader_active()?; + + if let Err(error) = + validate_stopping_offsets(scanner.get_subscribed_buckets(), &mut stopping_offsets) + { + scanner.clear_reader_active(); + return Err(error); + } + let schema = scanner.schema(); Ok(Self { scanner, @@ -358,6 +366,34 @@ impl arrow::record_batch::RecordBatchReader for SyncRecordBatchLogReader { } } +fn validate_stopping_offsets( + subscriptions: Vec<(TableBucket, i64)>, + stopping_offsets: &mut HashMap, +) -> Result<()> { + let subscribed: HashMap = subscriptions.into_iter().collect(); + for bucket in stopping_offsets.keys() { + if !subscribed.contains_key(bucket) { + return Err(Error::IllegalArgument { + message: format!( + "Stopping offset for {bucket:?} has no matching scanner subscription." + ), + }); + } + } + + // A concrete subscription that already meets the stop point is an empty + // range. Remove it up front so the reader can finish without waiting for a + // server batch that may never arrive. Negative offsets are symbolic values + // such as EARLIEST_OFFSET and cannot be compared until the server resolves + // them. + stopping_offsets.retain(|bucket, stop| { + subscribed + .get(bucket) + .is_none_or(|start| *start < 0 || start < stop) + }); + Ok(()) +} + /// Query latest offsets for all subscribed buckets, handling both partitioned /// and non-partitioned tables. /// @@ -568,6 +604,35 @@ mod tests { TableBucket::new(1, id) } + #[test] + fn validate_stopping_offsets_rejects_unsubscribed_bucket() { + let mut offsets = HashMap::from([(bucket(1), 10)]); + let result = validate_stopping_offsets(vec![(bucket(0), 0)], &mut offsets); + + assert!(matches!(result, Err(Error::IllegalArgument { .. }))); + } + + #[test] + fn validate_stopping_offsets_prunes_completed_range() { + let mut offsets = HashMap::from([(bucket(0), 10), (bucket(1), 20)]); + validate_stopping_offsets(vec![(bucket(0), 10), (bucket(1), 15)], &mut offsets).unwrap(); + + assert!(!offsets.contains_key(&bucket(0))); + assert_eq!(offsets.get(&bucket(1)), Some(&20)); + } + + #[test] + fn validate_stopping_offsets_keeps_symbolic_start() { + let mut offsets = HashMap::from([(bucket(0), 0)]); + validate_stopping_offsets( + vec![(bucket(0), crate::client::EARLIEST_OFFSET)], + &mut offsets, + ) + .unwrap(); + + assert_eq!(offsets.get(&bucket(0)), Some(&0)); + } + #[test] fn filter_batch_entirely_before_stop() { let mut offsets = HashMap::from([(bucket(0), 100)]); diff --git a/fluss-rust/website/docs/user-guide/cpp/api-reference.md b/fluss-rust/website/docs/user-guide/cpp/api-reference.md index 8487e686c50..49be2cfef9c 100644 --- a/fluss-rust/website/docs/user-guide/cpp/api-reference.md +++ b/fluss-rust/website/docs/user-guide/cpp/api-reference.md @@ -152,7 +152,9 @@ Complete API reference for the Fluss C++ client. | `ProjectByName(std::vector column_names) -> TableScan&` | Project columns by name | | `Limit(int32_t row_number) -> TableScan&` | Set a positive row limit (enables `CreateBucketBatchScanner`; rejected by log scanners) | | `CreateLogScanner(LogScanner& out) -> Result` | Create a record-based log scanner; on a primary-key table, subscribes to its CDC changelog (per-record `change_type`) | -| `CreateRecordBatchLogScanner(LogScanner& out) -> Result` | Create an Arrow RecordBatch-based log scanner (log tables only — no per-record change types) | +| `CreateRecordBatchLogScanner(RecordBatchLogScanner& out) -> Result` | Create a strongly typed Arrow RecordBatch scanner | +| `CreateRecordBatchLogReader(const std::vector& ranges, RecordBatchLogReader& out) -> Result` | Create a bounded reader directly from per-bucket offset ranges | +| `CreateRecordBatchLogReader(Admin& admin, const std::vector& buckets, const TimestampRange& range, RecordBatchLogReader& out) -> Result` | Resolve a timestamp range per bucket and create a bounded reader | | `CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out) -> Result` | Bounded scan of one bucket (requires `Limit`) | ## `AppendWriter` @@ -204,14 +206,46 @@ Performs prefix (bucket-key) lookups, returning all rows whose primary key start | `Unsubscribe(int32_t bucket_id) -> Result` | Unsubscribe from a non-partitioned bucket | | `UnsubscribePartition(int64_t partition_id, int32_t bucket_id) -> Result` | Unsubscribe from a partition bucket | | `Poll(int64_t timeout_ms, ScanRecords& out) -> Result` | Poll individual records | -| `PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) -> Result` | Poll Arrow RecordBatches | -| `CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out) -> Result` | Create a bounded reader using the latest offsets observed during the call | -| `CreateRecordBatchLogReaderUntilOffsets(const std::vector& offsets, RecordBatchLogReader& out) -> Result` | Create a bounded reader using explicit stopping offsets | +| `PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) -> Result` | Legacy Arrow RecordBatch polling API | -The reader methods require a scanner created by -`TableScan::CreateRecordBatchLogScanner()`. Subscribe every bucket assigned to this reader first; -subscriptions provide the starting offsets, while `ReaderStopOffset` provides the stopping offset -for each bucket. Only one reader or polling operation may consume a scanner at a time. +## `RecordBatchLogScanner` + +Strongly typed unbounded Arrow RecordBatch scanner. Its subscribe and unsubscribe methods mirror +`LogScanner`; `Poll()` returns Arrow batches. + +| Method | Description | +|------------------------------------------------------------------------------------------------------|------------------------------------------| +| `Subscribe(int32_t bucket_id, int64_t offset) -> Result` | Subscribe to a single bucket | +| `Subscribe(const std::vector& bucket_offsets) -> Result` | Subscribe to multiple buckets | +| `SubscribePartitionBuckets(const std::vector& subscriptions) -> Result` | Subscribe to multiple partition buckets | +| `Poll(int64_t timeout_ms, ArrowRecordBatches& out) -> Result` | Poll Arrow RecordBatches | +| `CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out) && -> Result` | Move the scanner into a reader bounded by current latest offsets | +| `CreateRecordBatchLogReaderUntilOffsets(const std::vector& offsets, RecordBatchLogReader& out) && -> Result` | Move the scanner into a reader using explicit stops | + +Prefer `TableScan::CreateRecordBatchLogReader()` for query engines. The scanner-level methods are +useful when subscriptions need to be configured incrementally. + +## `RecordBatchLogReadRange` + +| Field | Type | Description | +|-------------------|---------------|-------------------------------------| +| `bucket` | `TableBucket` | Bucket assigned to this reader | +| `starting_offset` | `int64_t` | Inclusive starting offset | +| `stopping_offset` | `int64_t` | Exclusive stopping offset | + +## `TimestampRange` + +| Field | Type | Description | +|---------------------------|-----------|----------------------------------------------| +| `starting_timestamp_ms` | `int64_t` | Starting log timestamp in epoch milliseconds | +| `stopping_timestamp_ms` | `int64_t` | Stopping log timestamp in epoch milliseconds | + +## `RecordBatchReadResult` + +| Field | Type | Description | +|----------|-------------------------------------|--------------------------------------------------| +| `status` | `BoundedReadStatus` | Batch available, timed out, or finished | +| `batch` | `std::unique_ptr` | Present only when `status` is `BatchAvailable` | ## `ReaderStopOffset` @@ -222,20 +256,24 @@ for each bucket. Only one reader or polling operation may consume a scanner at a ## `RecordBatchLogReader` -Bounded Arrow record-batch reader created from a subscribed `LogScanner`. +Bounded Arrow record-batch reader. It can manage multiple buckets, each with its own stopping +offset, and returns one batch per successful `NextBatch()` call. -| Method | Description | -|----------------------------------------------------------------------------------------------------|----------------------------------------------------------------| -| `Available() -> bool` | Check whether the reader is valid | -| `NextBatch(int64_t timeout_ms, ArrowRecordBatches& out, BoundedReadStatus& status) -> Result` | Wait up to the timeout for a batch, timeout, or completion | -| `CollectAllBatches(ArrowRecordBatches& out) -> Result` | Drain all batches until every stopping offset has been reached | +| Method | Description | +|---------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------| +| `Available() -> bool` | Check whether the reader is valid | +| `NextBatch(int64_t timeout_ms, RecordBatchReadResult& out) -> Result` | Wait up to the timeout for one batch, timeout, or completion | +| `CollectAllBatches(ArrowRecordBatches& out) -> Result` | Drain all batches until every stopping offset has been reached | `BoundedReadStatus` is `BatchAvailable`, `TimedOut`, or `Finished`. A timeout does not exhaust the reader; callers may check cancellation and invoke `NextBatch()` again. -`CreateRecordBatchLogReaderUntilLatest()` queries the latest offset of every subscribed bucket -when the reader is created. `CreateRecordBatchLogReaderUntilOffsets()` is useful when a -coordinator has already selected offsets and needs multiple workers to use the same boundary. +`RecordBatchReadResult::batch` is non-null only when `status` is `BatchAvailable`. + +`TableScan::CreateRecordBatchLogReader()` accepts resolved per-bucket ranges, which is useful when +a coordinator has selected globally consistent offsets for multiple workers. The timestamp +overload resolves both timestamps with `OffsetSpec::Timestamp` for every requested bucket before +reading. ## `BatchScanner` diff --git a/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md b/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md index e74fdeaf937..e814ab97a65 100644 --- a/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md +++ b/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md @@ -123,7 +123,7 @@ scanner.Unsubscribe(1); ```cpp #include -fluss::LogScanner arrow_scanner; +fluss::RecordBatchLogScanner arrow_scanner; table.NewScan().CreateRecordBatchLogScanner(arrow_scanner); for (int b = 0; b < info.num_buckets; ++b) { @@ -131,7 +131,7 @@ for (int b = 0; b < info.num_buckets; ++b) { } fluss::ArrowRecordBatches batches; -arrow_scanner.PollRecordBatch(5000, batches); +arrow_scanner.Poll(5000, batches); for (size_t i = 0; i < batches.Size(); ++i) { const auto& batch = batches[i]; @@ -147,8 +147,7 @@ for (size_t i = 0; i < batches.Size(); ++i) { ## Bounded Arrow RecordBatch Reading Use `RecordBatchLogReader` when the scan should finish after reaching a fixed offset for every -bucket. Starting offsets are configured by subscribing the scanner. Stopping offsets can be -queried once and passed explicitly: +bucket. Query engines can pass the complete per-bucket ranges directly: ```cpp auto info = table.GetTableInfo(); @@ -161,62 +160,76 @@ for (int32_t bucket_id = 0; bucket_id < info.num_buckets; ++bucket_id) { std::unordered_map latest_offsets; admin.ListOffsets(table_path, bucket_ids, fluss::OffsetSpec::Latest(), latest_offsets); -fluss::LogScanner scanner; -table.NewScan().CreateRecordBatchLogScanner(scanner); - -std::vector stopping_offsets; +std::vector ranges; for (int32_t bucket_id : bucket_ids) { - scanner.Subscribe(bucket_id, 0); - stopping_offsets.push_back( - {fluss::TableBucket{info.table_id, bucket_id}, latest_offsets.at(bucket_id)}); + ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, 0, latest_offsets.at(bucket_id)}); } fluss::RecordBatchLogReader reader; -scanner.CreateRecordBatchLogReaderUntilOffsets(stopping_offsets, reader); +table.NewScan().CreateRecordBatchLogReader(ranges, reader); while (true) { - fluss::ArrowRecordBatches batches; - fluss::BoundedReadStatus status; - reader.NextBatch(1000, batches, status); - if (status == fluss::BoundedReadStatus::TimedOut) { + fluss::RecordBatchReadResult result; + reader.NextBatch(1000, result); + if (result.status == fluss::BoundedReadStatus::TimedOut) { continue; // Check query cancellation before retrying. } - if (status == fluss::BoundedReadStatus::Finished) { + if (result.status == fluss::BoundedReadStatus::Finished) { break; } - for (const auto& batch : batches) { - std::cout << "bucket=" << batch->GetBucketId() - << " base_offset=" << batch->GetBaseOffset() - << " last_offset=" << batch->GetLastOffset() - << " rows=" << batch->NumRows() << std::endl; - } + std::cout << "bucket=" << result.batch->GetBucketId() + << " base_offset=" << result.batch->GetBaseOffset() + << " last_offset=" << result.batch->GetLastOffset() + << " rows=" << result.batch->NumRows() << std::endl; } ``` `TimedOut` does not exhaust the reader. It lets a query engine periodically check cancellation or deadlines before retrying. `Finished` means all stopping offsets have been reached. +To read a log timestamp range, pass the assigned buckets and timestamps. Fluss resolves both +timestamps with `OffsetSpec::Timestamp` for every bucket, then uses the same bounded offset reader: + +```cpp +fluss::RecordBatchLogReader reader; +table.NewScan().CreateRecordBatchLogReader( + admin, assigned_buckets, + fluss::TimestampRange{starting_timestamp_ms, stopping_timestamp_ms}, reader); + +while (true) { + fluss::RecordBatchReadResult result; + reader.NextBatch(1000, result); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + continue; + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + process(result.batch->GetArrowRecordBatch()); +} +``` + For the common case where the client should read everything currently available, let the reader query the latest offsets: ```cpp -fluss::LogScanner latest_scanner; +fluss::RecordBatchLogScanner latest_scanner; table.NewScan().CreateRecordBatchLogScanner(latest_scanner); for (int32_t bucket_id : bucket_ids) { latest_scanner.Subscribe(bucket_id, 0); } fluss::RecordBatchLogReader latest_reader; -latest_scanner.CreateRecordBatchLogReaderUntilLatest(admin, latest_reader); +std::move(latest_scanner).CreateRecordBatchLogReaderUntilLatest(admin, latest_reader); fluss::ArrowRecordBatches batches; latest_reader.CollectAllBatches(batches); ``` -Subscribe every bucket assigned to the reader before creating it. Do not call `PollRecordBatch()` -or create a second reader concurrently on the same scanner. Destroy the reader before reusing -that scanner. +The scanner-level creation methods transfer ownership on success, so the scanner becomes +unavailable after it is moved into the reader. ## Column Projection