diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md index fa6cac26f5b..b609128118f 100644 --- a/fluss-rust/bindings/cpp/README.md +++ b/fluss-rust/bindings/cpp/README.md @@ -33,10 +33,71 @@ bazel build //... `ci.sh` defaults to optimized builds via `-c opt` (override with `BAZEL_BUILD_FLAGS` if needed). See [ci.sh](ci.sh) for the CI build sequence. +## Examples and Documentation + +- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes, continuous scans, + bounded Arrow record-batch scans, projections, and offset queries. +- [examples/admin_example.cpp](examples/admin_example.cpp) demonstrates database, table, + partition, and cluster administration. +- [examples/kv_example.cpp](examples/kv_example.cpp) and + [examples/kv_changelog_example.cpp](examples/kv_changelog_example.cpp) demonstrate + primary-key table access. +- The website documentation includes the + [C++ API reference](../../website/docs/user-guide/cpp/api-reference.md) and + [log-table examples](../../website/docs/user-guide/cpp/example/log-tables.md). + +For a bounded log scan, pass the per-bucket offset ranges directly to `TableScan`. The returned +reader yields one Arrow batch at a time until every `[starting_offset, stopping_offset)` range +is complete: + +```cpp +auto info = table.GetTableInfo(); +std::vector 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); + +std::vector ranges; +for (int32_t bucket_id : bucket_ids) { + ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, 0, latest_offsets.at(bucket_id)}); +} + +fluss::RecordBatchLogReader reader; +table.NewScan().CreateRecordBatchLogReader(ranges, reader); + +while (true) { + fluss::RecordBatchReadResult result; + reader.NextBatch(1000, result); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + continue; // Check query cancellation before retrying. + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + process(result.batch->GetArrowRecordBatch()); +} +``` + +Timestamp-bounded reads use the same iterator after resolving the timestamps independently for +each bucket: + +```cpp +fluss::RecordBatchLogReader timestamp_reader; +table.NewScan().CreateRecordBatchLogReader( + admin, table_buckets, + fluss::TimestampRange{starting_timestamp_ms, stopping_timestamp_ms}, timestamp_reader); +``` + +`CollectAllBatches()` is available when materializing the complete bounded result is +preferred. `NextBatch()` reports timeout separately from completion so engines can periodically +check cancellation. ## TODO -- [] How to introduce fluss-cpp in your own project, https://github.com/apache/opendal/blob/main/bindings/cpp/README.md is a good reference +- [ ] How to introduce fluss-cpp in your own project, https://github.com/apache/opendal/blob/main/bindings/cpp/README.md is a good reference - [ ] Add CMake/Bazel install and packaging instructions. -- [ ] Document API usage and minimal example in this README. -- [ ] Add more C++ examples (log scan, upsert, etc.). +- [ ] Add more C++ examples (upsert, partitioned bounded scans, etc.). diff --git a/fluss-rust/bindings/cpp/examples/example.cpp b/fluss-rust/bindings/cpp/examples/example.cpp index d86ee5cda72..9341619cc03 100644 --- a/fluss-rust/bindings/cpp/examples/example.cpp +++ b/fluss-rust/bindings/cpp/examples/example.cpp @@ -369,11 +369,47 @@ int main() { std::cout << " Bucket " << bucket_id << ": offset=" << offset << std::endl; } + // 8.1) Bounded Arrow record batch scan with explicit stopping offsets + std::cout << "\n=== Bounded Arrow Record Batch Scan ===" << std::endl; + std::vector bounded_ranges; + for (int32_t bucket_id : all_bucket_ids) { + const int64_t starting_offset = earliest_offsets.at(bucket_id); + const int64_t stopping_offset = latest_offsets.at(bucket_id); + bounded_ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, starting_offset, stopping_offset}); + } + + if (!bounded_ranges.empty()) { + fluss::RecordBatchLogReader bounded_reader; + check("create_bounded_reader", + table.NewScan().CreateRecordBatchLogReader(bounded_ranges, bounded_reader)); + + int64_t bounded_row_count = 0; + while (true) { + fluss::RecordBatchReadResult result; + check("bounded_next_batch", bounded_reader.NextBatch(1000, result)); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + continue; + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + bounded_row_count += result.batch->NumRows(); + std::cout << " bucket=" << result.batch->GetBucketId() + << " base_offset=" << result.batch->GetBaseOffset() + << " last_offset=" << result.batch->GetLastOffset() + << " rows=" << result.batch->NumRows() << std::endl; + } + std::cout << "Bounded scan completed with " << bounded_row_count << " rows" << std::endl; + } + auto now = std::chrono::system_clock::now(); auto one_hour_ago = now - std::chrono::hours(1); auto timestamp_ms = std::chrono::duration_cast(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", @@ -384,6 +420,30 @@ int main() { std::cout << " Bucket " << bucket_id << ": offset=" << offset << std::endl; } + // 8.2) Bounded Arrow record batch scan by timestamp range + std::vector table_buckets; + for (int32_t bucket_id : all_bucket_ids) { + table_buckets.push_back({info.table_id, bucket_id}); + } + + fluss::RecordBatchLogReader timestamp_reader; + check("create_timestamp_reader", + table.NewScan().CreateRecordBatchLogReader( + admin, table_buckets, fluss::TimestampRange{timestamp_ms, now_ms}, timestamp_reader)); + + while (true) { + fluss::RecordBatchReadResult result; + check("timestamp_next_batch", timestamp_reader.NextBatch(1000, result)); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + continue; + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + std::cout << "Timestamp range batch: bucket=" << result.batch->GetBucketId() + << " rows=" << result.batch->NumRows() << std::endl; + } + // 9) Batch subscribe std::cout << "\n=== Batch Subscribe Example ===" << std::endl; fluss::LogScanner batch_scanner; @@ -427,7 +487,7 @@ int main() { // 10) Arrow record batch polling std::cout << "\n=== Testing Arrow Record Batch Polling ===" << std::endl; - fluss::LogScanner arrow_scanner; + fluss::RecordBatchLogScanner arrow_scanner; check("new_record_batch_log_scanner", table.NewScan().CreateRecordBatchLogScanner(arrow_scanner)); @@ -436,7 +496,7 @@ int main() { } fluss::ArrowRecordBatches arrow_batches; - check("poll_record_batch", arrow_scanner.PollRecordBatch(5000, arrow_batches)); + check("poll_record_batch", arrow_scanner.Poll(5000, arrow_batches)); std::cout << "Polled " << arrow_batches.Size() << " Arrow record batches" << std::endl; for (size_t i = 0; i < arrow_batches.Size(); ++i) { @@ -452,7 +512,7 @@ int main() { // 11) Arrow record batch polling with projection std::cout << "\n=== Testing Arrow Record Batch Polling with Projection ===" << std::endl; - fluss::LogScanner projected_arrow_scanner; + fluss::RecordBatchLogScanner projected_arrow_scanner; check("new_record_batch_log_scanner_with_projection", table.NewScan() .ProjectByIndex(projected_columns) @@ -464,7 +524,7 @@ int main() { fluss::ArrowRecordBatches projected_arrow_batches; check("poll_projected_record_batch", - projected_arrow_scanner.PollRecordBatch(5000, projected_arrow_batches)); + projected_arrow_scanner.Poll(5000, projected_arrow_batches)); std::cout << "Polled " << projected_arrow_batches.Size() << " projected Arrow record batches" << std::endl; diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp index 935cf70853b..6f472dc59f4 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,39 @@ struct PartitionBucketSubscription { int64_t offset; }; +/// Stopping offset for one bucket subscribed on a record-batch log scanner. +struct ReaderStopOffset { + TableBucket bucket; + int64_t offset; +}; + +/// One bounded log range. Records are returned for +/// [starting_offset, stopping_offset). +struct RecordBatchLogReadRange { + TableBucket bucket; + int64_t starting_offset; + int64_t stopping_offset; +}; + +/// A log timestamp range in epoch milliseconds. Each requested bucket resolves +/// the two timestamps to offsets before the reader is created. +struct TimestampRange { + int64_t starting_timestamp_ms; + int64_t stopping_timestamp_ms; +}; + +/// Outcome of a bounded record-batch read. +enum class BoundedReadStatus { + BatchAvailable = 0, + TimedOut = 1, + Finished = 2, +}; + +struct RecordBatchReadResult { + BoundedReadStatus status{BoundedReadStatus::TimedOut}; + std::unique_ptr batch; +}; + struct LakeSnapshot { int64_t snapshot_id; std::vector bucket_offsets; @@ -1338,6 +1372,8 @@ class Lookuper; class PrefixLookuper; class WriteResult; class LogScanner; +class RecordBatchLogScanner; +class RecordBatchLogReader; class BatchScanner; class Admin; class Table; @@ -1502,6 +1538,7 @@ class Admin { const std::string* partition_name = nullptr); friend class Connection; + friend class LogScanner; Admin(ffi::Admin* admin) noexcept; void Destroy() noexcept; @@ -1632,8 +1669,26 @@ class TableScan { /// path carries no per-record change types; read a primary-key table's /// changelog with `CreateLogScanner()` instead. Requires the ARROW log /// format. + Result CreateRecordBatchLogScanner(RecordBatchLogScanner& out); + + /// Legacy overload. Prefer the strongly typed RecordBatchLogScanner. Result CreateRecordBatchLogScanner(LogScanner& out); + /// Creates a bounded reader directly from per-bucket offset ranges. + /// + /// This is the preferred API for query engines: it subscribes every bucket + /// at its starting offset, installs the corresponding stopping offset, and + /// transfers scanner ownership to the returned reader. + Result CreateRecordBatchLogReader(const std::vector& 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: @@ -1642,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_; @@ -1796,12 +1854,96 @@ class LogScanner { private: friend class Table; friend class TableScan; + friend class RecordBatchLogScanner; LogScanner(ffi::LogScanner* scanner) noexcept; void Destroy() noexcept; + + /// Creates a bounded reader using the latest offsets observed during this call. + /// Subscribe the record-batch scanner at the desired starting offsets before + /// calling this method. + Result CreateRecordBatchLogReaderUntilLatest(const Admin& admin, RecordBatchLogReader& out); + + /// Creates a bounded reader using explicit stopping offsets. + /// Starting offsets come from the scanner subscriptions. Every stopping + /// offset must correspond to a bucket already subscribed on this scanner. + Result CreateRecordBatchLogReaderUntilOffsets(const std::vector& 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 TableScan; + explicit RecordBatchLogScanner(ffi::LogScanner* scanner) noexcept; + + LogScanner scanner_; +}; + +/// Bounded Arrow batch reader created from a subscribed record-batch log scanner. +/// Only one reader or polling operation may consume the scanner at a time. +class RecordBatchLogReader { + public: + RecordBatchLogReader() noexcept; + ~RecordBatchLogReader() noexcept; + + RecordBatchLogReader(const RecordBatchLogReader&) = delete; + RecordBatchLogReader& operator=(const RecordBatchLogReader&) = delete; + RecordBatchLogReader(RecordBatchLogReader&& other) noexcept; + RecordBatchLogReader& operator=(RecordBatchLogReader&& other) noexcept; + + bool Available() const; + + /// Waits up to timeout_ms for the next batch. + /// TimedOut leaves the reader valid for a later retry; Finished means every + /// subscribed bucket reached its stopping offset. + Result NextBatch(int64_t timeout_ms, RecordBatchReadResult& out); + + /// Drains all remaining batches until every stopping offset is reached. + Result CollectAllBatches(ArrowRecordBatches& out); + + private: + friend class LogScanner; + friend class RecordBatchLogScanner; + friend class TableScan; + explicit RecordBatchLogReader(ffi::RecordBatchLogReader* reader) noexcept; + + void Destroy() noexcept; + ffi::RecordBatchLogReader* reader_{nullptr}; +}; + // One-shot bounded scan of a single bucket, from TableScan::CreateBucketBatchScanner. class BatchScanner { public: diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs index 675a28061c1..ea09e5441ff 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,108 @@ 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 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(), + ); + } + } + + 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..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,203 @@ 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 +// ============================================================================ + +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, 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()) { + return result; + } + switch (ffi_result.status) { + case 0: + out.status = BoundedReadStatus::BatchAvailable; + break; + case 1: + out.status = BoundedReadStatus::TimedOut; + break; + case 2: + out.status = BoundedReadStatus::Finished; + break; + default: + return utils::make_client_error("Unknown bounded read status: " + + std::to_string(ffi_result.status)); + } + 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) { + 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..8ec73e08540 100644 --- a/fluss-rust/bindings/cpp/test/test_log_table.cpp +++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp @@ -173,6 +173,235 @@ 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()); + + 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 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"; + 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(table.NewScan().CreateRecordBatchLogReader(ranges, reader)); + + std::vector actual_rows_by_bucket(kNumBuckets); + 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(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::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. + { + const int64_t start_offset = latest_offsets.at(0); + fluss::RecordBatchLogReader waiting_reader; + 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)); +} + +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::RecordBatchLogScanner scanner; + ASSERT_OK(table.NewScan().CreateRecordBatchLogScanner(scanner)); + ASSERT_OK(scanner.Subscribe(0, 0)); + + fluss::RecordBatchLogReader reader; + 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::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)); +} + TEST_F(LogTableTest, LimitScan) { auto& adm = admin(); auto& conn = connection(); @@ -890,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; @@ -947,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(), @@ -988,6 +1227,65 @@ 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)); + std::vector ranges; + for (const auto& pi : partition_infos) { + ranges.push_back( + {fluss::TableBucket{bounded_table.GetTableInfo().table_id, 0, pi.partition_id}, 1, + 3}); + } + + fluss::RecordBatchLogReader reader; + ASSERT_OK(bounded_table.NewScan().CreateRecordBatchLogReader(ranges, 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 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/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..81a4e849c33 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: @@ -125,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, @@ -173,19 +192,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 = @@ -323,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. /// @@ -533,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 621eb7ace50..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,7 +206,74 @@ 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 | +| `PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) -> Result` | Legacy Arrow RecordBatch polling API | + +## `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` + +| 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. 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, 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. + +`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 ca243f5d2ae..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]; @@ -144,6 +144,93 @@ 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. Query engines can pass the complete per-bucket ranges directly: + +```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); + +std::vector ranges; +for (int32_t bucket_id : bucket_ids) { + ranges.push_back( + {fluss::TableBucket{info.table_id, bucket_id}, 0, latest_offsets.at(bucket_id)}); +} + +fluss::RecordBatchLogReader reader; +table.NewScan().CreateRecordBatchLogReader(ranges, reader); + +while (true) { + fluss::RecordBatchReadResult result; + reader.NextBatch(1000, result); + if (result.status == fluss::BoundedReadStatus::TimedOut) { + continue; // Check query cancellation before retrying. + } + if (result.status == fluss::BoundedReadStatus::Finished) { + break; + } + + 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::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; +std::move(latest_scanner).CreateRecordBatchLogReaderUntilLatest(admin, latest_reader); + +fluss::ArrowRecordBatches batches; +latest_reader.CollectAllBatches(batches); +``` + +The scanner-level creation methods transfer ownership on success, so the scanner becomes +unavailable after it is moved into the reader. + ## Column Projection ```cpp