This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 8e3ed756 [c++] Limit batch scanner (#614)
8e3ed756 is described below
commit 8e3ed75610ddc282dfe38fde128340c35a62eeac
Author: Anton Borisov <[email protected]>
AuthorDate: Sun Jun 14 01:40:26 2026 +0100
[c++] Limit batch scanner (#614)
* [cpp] Limit scan
* add test
---
bindings/cpp/include/fluss.hpp | 39 ++++
bindings/cpp/src/lib.rs | 129 +++++++++++--
bindings/cpp/src/table.cpp | 143 +++++++++++---
bindings/cpp/test/test_kv_table.cpp | 70 +++++++
bindings/cpp/test/test_log_table.cpp | 206 +++++++++++++++++++++
crates/fluss/src/client/table/batch_scanner.rs | 32 +++-
website/docs/user-guide/cpp/api-reference.md | 11 ++
website/docs/user-guide/cpp/example/log-tables.md | 20 ++
.../user-guide/cpp/example/primary-key-tables.md | 17 ++
9 files changed, 628 insertions(+), 39 deletions(-)
diff --git a/bindings/cpp/include/fluss.hpp b/bindings/cpp/include/fluss.hpp
index c27b039b..16dc04ee 100644
--- a/bindings/cpp/include/fluss.hpp
+++ b/bindings/cpp/include/fluss.hpp
@@ -45,6 +45,7 @@ struct Table;
struct AppendWriter;
struct WriteResult;
struct LogScanner;
+struct BatchScanner;
struct UpsertWriter;
struct Lookuper;
struct PrefixLookuper;
@@ -1155,6 +1156,11 @@ class ScanRecords {
std::shared_ptr<const detail::ScanData> data_;
};
+namespace detail {
+// Defined in table.cpp; builds ArrowRecordBatch wrappers from FFI Arrow
batches.
+struct ArrowBatchImporter;
+} // namespace detail
+
class ArrowRecordBatch {
public:
std::shared_ptr<arrow::RecordBatch> GetArrowRecordBatch() const { return
batch_; }
@@ -1173,6 +1179,7 @@ class ArrowRecordBatch {
private:
friend class LogScanner;
+ friend struct detail::ArrowBatchImporter;
explicit ArrowRecordBatch(std::shared_ptr<arrow::RecordBatch> batch,
int64_t table_id,
int64_t partition_id, int32_t bucket_id,
int64_t base_offset) noexcept;
@@ -1346,6 +1353,7 @@ class Lookuper;
class PrefixLookuper;
class WriteResult;
class LogScanner;
+class BatchScanner;
class Admin;
class Table;
class TableAppend;
@@ -1619,9 +1627,13 @@ class TableScan {
TableScan& ProjectByIndex(std::vector<size_t> column_indices);
TableScan& ProjectByName(std::vector<std::string> column_names);
+ TableScan& Limit(int32_t row_number);
+
Result CreateLogScanner(LogScanner& out);
Result CreateRecordBatchLogScanner(LogScanner& out);
+ Result CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner&
out);
+
private:
friend class Table;
explicit TableScan(ffi::Table* table) noexcept;
@@ -1632,6 +1644,7 @@ class TableScan {
ffi::Table* table_{nullptr};
std::vector<size_t> projection_;
std::vector<std::string> name_projection_;
+ std::optional<int32_t> limit_;
};
class WriteResult {
@@ -1787,4 +1800,30 @@ class LogScanner {
ffi::LogScanner* scanner_{nullptr};
};
+// One-shot bounded scan of a single bucket, from
TableScan::CreateBucketBatchScanner.
+class BatchScanner {
+ public:
+ BatchScanner() noexcept;
+ ~BatchScanner() noexcept;
+
+ BatchScanner(const BatchScanner&) = delete;
+ BatchScanner& operator=(const BatchScanner&) = delete;
+ BatchScanner(BatchScanner&& other) noexcept;
+ BatchScanner& operator=(BatchScanner&& other) noexcept;
+
+ bool Available() const;
+ const TableBucket& Bucket() const { return bucket_; }
+
+ Result NextBatch(ArrowRecordBatches& out);
+ Result CollectAllBatches(ArrowRecordBatches& out);
+
+ private:
+ friend class TableScan;
+ explicit BatchScanner(ffi::BatchScanner* scanner) noexcept;
+
+ void Destroy() noexcept;
+ ffi::BatchScanner* scanner_{nullptr};
+ TableBucket bucket_{};
+};
+
} // namespace fluss
diff --git a/bindings/cpp/src/lib.rs b/bindings/cpp/src/lib.rs
index b5417ae4..25589783 100644
--- a/bindings/cpp/src/lib.rs
+++ b/bindings/cpp/src/lib.rs
@@ -18,7 +18,7 @@
mod types;
use std::str::FromStr;
-use std::sync::{Arc, LazyLock};
+use std::sync::{Arc, LazyLock, Mutex};
use std::time::Duration;
use fluss as fcore;
@@ -286,6 +286,7 @@ mod ffi {
type AppendWriter;
type WriteResult;
type LogScanner;
+ type BatchScanner;
type UpsertWriter;
type Lookuper;
type PrefixLookuper;
@@ -379,6 +380,14 @@ mod ffi {
unsafe fn delete_table(table: *mut Table);
fn new_append_writer(self: &Table) -> FfiPtrResult;
fn create_scanner(self: &Table, column_indices: Vec<usize>, batch:
bool) -> FfiPtrResult;
+ fn create_bucket_batch_scanner(
+ self: &Table,
+ column_indices: Vec<usize>,
+ limit: i32,
+ table_id: i64,
+ partition_id: i64,
+ bucket_id: i32,
+ ) -> FfiPtrResult;
fn get_table_info_from_table(self: &Table) -> FfiTableInfo;
fn get_table_path(self: &Table) -> FfiTablePath;
fn has_primary_key(self: &Table) -> bool;
@@ -735,6 +744,11 @@ mod ffi {
fn poll_record_batch(self: &LogScanner, timeout_ms: i64) ->
FfiArrowRecordBatchesResult;
fn free_arrow_ffi_structures(array_ptr: usize, schema_ptr: usize);
+ // BatchScanner
+ unsafe fn delete_batch_scanner(scanner: *mut BatchScanner);
+ fn next_batch(self: &BatchScanner) -> FfiArrowRecordBatchesResult;
+ fn collect_all_batches(self: &BatchScanner) ->
FfiArrowRecordBatchesResult;
+
// ScanResultInner accessors
fn sv_has_error(self: &ScanResultInner) -> bool;
fn sv_error_code(self: &ScanResultInner) -> i32;
@@ -978,6 +992,10 @@ pub struct LogScanner {
projected_columns: Vec<fcore::metadata::Column>,
}
+pub struct BatchScanner {
+ inner: Mutex<fcore::client::LimitBatchScanner>,
+}
+
pub struct UpsertWriter {
inner: fcore::client::UpsertWriter,
table_info: fcore::metadata::TableInfo,
@@ -1053,6 +1071,26 @@ fn err_ptr_from_core(e: &fcore::error::Error) ->
ffi::FfiPtrResult {
}
}
+fn empty_arrow_batches_result(result: ffi::FfiResult) ->
ffi::FfiArrowRecordBatchesResult {
+ ffi::FfiArrowRecordBatchesResult {
+ result,
+ arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] },
+ }
+}
+
+/// Wrap the result of `core_scan_batches_to_ffi` in an
`FfiArrowRecordBatchesResult`.
+fn arrow_batches_result(
+ converted: Result<ffi::FfiArrowRecordBatches, String>,
+) -> ffi::FfiArrowRecordBatchesResult {
+ match converted {
+ Ok(arrow_batches) => ffi::FfiArrowRecordBatchesResult {
+ result: ok_result(),
+ arrow_batches,
+ },
+ Err(e) => empty_arrow_batches_result(client_err(e)),
+ }
+}
+
// Connection implementation
fn new_connection(config: &ffi::FfiConfig) -> ffi::FfiPtrResult {
let assigner_type = match config
@@ -1691,6 +1729,49 @@ impl Table {
})
}
+ fn create_bucket_batch_scanner(
+ &self,
+ column_indices: Vec<usize>,
+ limit: i32,
+ table_id: i64,
+ partition_id: i64,
+ bucket_id: i32,
+ ) -> ffi::FfiPtrResult {
+ RUNTIME.block_on(async {
+ let fluss_table = self.fluss_table();
+ let scan = fluss_table.new_scan();
+
+ let scan = if column_indices.is_empty() {
+ scan
+ } else {
+ match scan.project(&column_indices) {
+ Ok(s) => s,
+ Err(e) => return err_ptr_from_core(&e),
+ }
+ };
+
+ let scan = match scan.limit(limit) {
+ Ok(s) => s,
+ Err(e) => return err_ptr_from_core(&e),
+ };
+
+ // A negative partition id encodes "no partition" across the FFI
boundary.
+ let partition = (partition_id >= 0).then_some(partition_id);
+ let bucket =
+ fcore::metadata::TableBucket::new_with_partition(table_id,
partition, bucket_id);
+
+ let scanner = match scan.create_bucket_batch_scanner(bucket) {
+ Ok(s) => s,
+ Err(e) => return err_ptr_from_core(&e),
+ };
+
+ let ptr = Box::into_raw(Box::new(BatchScanner {
+ inner: Mutex::new(scanner),
+ }));
+ ok_ptr(ptr as usize)
+ })
+ }
+
fn get_table_info_from_table(&self) -> ffi::FfiTableInfo {
types::core_table_info_to_ffi(&self.table_info)
}
@@ -2248,20 +2329,38 @@ impl LogScanner {
let result = RUNTIME.block_on(async { inner_batch.poll(timeout).await
});
match result {
- Ok(batches) => match types::core_scan_batches_to_ffi(&batches) {
- Ok(arrow_batches) => ffi::FfiArrowRecordBatchesResult {
- result: ok_result(),
- arrow_batches,
- },
- Err(e) => ffi::FfiArrowRecordBatchesResult {
- result: client_err(e),
- arrow_batches: ffi::FfiArrowRecordBatches { batches:
vec![] },
- },
- },
- Err(e) => ffi::FfiArrowRecordBatchesResult {
- result: err_from_core_error(&e),
- arrow_batches: ffi::FfiArrowRecordBatches { batches: vec![] },
- },
+ 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
+unsafe fn delete_batch_scanner(scanner: *mut BatchScanner) {
+ if !scanner.is_null() {
+ unsafe {
+ drop(Box::from_raw(scanner));
+ }
+ }
+}
+
+impl BatchScanner {
+ fn next_batch(&self) -> ffi::FfiArrowRecordBatchesResult {
+ let mut scanner = self.inner.lock().unwrap();
+ match RUNTIME.block_on(scanner.next_batch()) {
+ Ok(Some(batch)) =>
arrow_batches_result(types::core_scan_batches_to_ffi(
+ std::slice::from_ref(&batch),
+ )),
+ Ok(None) => empty_arrow_batches_result(ok_result()),
+ Err(e) => empty_arrow_batches_result(err_from_core_error(&e)),
+ }
+ }
+
+ fn collect_all_batches(&self) -> ffi::FfiArrowRecordBatchesResult {
+ let mut scanner = self.inner.lock().unwrap();
+ match RUNTIME.block_on(scanner.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)),
}
}
}
diff --git a/bindings/cpp/src/table.cpp b/bindings/cpp/src/table.cpp
index 9b5d087e..9f4d9e46 100644
--- a/bindings/cpp/src/table.cpp
+++ b/bindings/cpp/src/table.cpp
@@ -1267,6 +1267,10 @@ Result TableScan::DoCreateScanner(LogScanner& out, bool
is_record_batch) {
if (table_ == nullptr) {
return utils::make_client_error("Table not available");
}
+ if (limit_.has_value()) {
+ return utils::make_client_error(
+ "Log scanners don't support limit pushdown; use
CreateBucketBatchScanner()");
+ }
try {
auto resolved_indices = !name_projection_.empty() ?
ResolveNameProjection() : projection_;
@@ -1286,6 +1290,41 @@ Result TableScan::DoCreateScanner(LogScanner& out, bool
is_record_batch) {
}
}
+TableScan& TableScan::Limit(int32_t row_number) {
+ limit_ = row_number;
+ return *this;
+}
+
+Result TableScan::CreateBucketBatchScanner(const TableBucket& bucket,
BatchScanner& out) {
+ if (table_ == nullptr) {
+ return utils::make_client_error("Table not available");
+ }
+ if (!limit_.has_value()) {
+ return utils::make_client_error(
+ "CreateBucketBatchScanner requires a limit set via Limit()");
+ }
+
+ try {
+ auto resolved_indices = !name_projection_.empty() ?
ResolveNameProjection() : projection_;
+ rust::Vec<size_t> rust_indices;
+ for (size_t idx : resolved_indices) {
+ rust_indices.push_back(idx);
+ }
+ int64_t partition_id = bucket.partition_id.value_or(-1);
+ auto ffi_result = table_->create_bucket_batch_scanner(
+ std::move(rust_indices), *limit_, bucket.table_id, partition_id,
bucket.bucket_id);
+ auto result = utils::from_ffi_result(ffi_result.result);
+ if (result.Ok()) {
+ out.scanner_ = utils::ptr_from_ffi<ffi::BatchScanner>(ffi_result);
+ out.bucket_ = bucket;
+ }
+ return result;
+ } catch (const std::exception& e) {
+ // ResolveNameProjection() may throw
+ return utils::make_client_error(e.what());
+ }
+}
+
// ============================================================================
// WriteResult
// ============================================================================
@@ -1791,6 +1830,34 @@ int64_t ArrowRecordBatch::GetLastOffset() const {
return this->base_offset_ + this->NumRows() - 1;
}
+namespace detail {
+// Imports FFI Arrow batches (C Data Interface) into C++ ArrowRecordBatch
+// wrappers. A friend of ArrowRecordBatch so it can build the wrapper's private
+// ctor; shared by LogScanner::PollRecordBatch and BatchScanner.
+struct ArrowBatchImporter {
+ static Result Import(ffi::FfiArrowRecordBatches& src, ArrowRecordBatches&
out) {
+ out.batches.clear();
+ for (const auto& ffi_batch : src.batches) {
+ auto* c_array = reinterpret_cast<struct
ArrowArray*>(ffi_batch.array_ptr);
+ auto* c_schema = reinterpret_cast<struct
ArrowSchema*>(ffi_batch.schema_ptr);
+
+ auto import_result = arrow::ImportRecordBatch(c_array, c_schema);
+ // Free the Rust-allocated container structures whether or not the
+ // import succeeded, to avoid leaks.
+ ffi::free_arrow_ffi_structures(ffi_batch.array_ptr,
ffi_batch.schema_ptr);
+ if (!import_result.ok()) {
+ return utils::make_client_error("Failed to import Arrow record
batch: " +
+
import_result.status().ToString());
+ }
+ out.batches.push_back(std::unique_ptr<ArrowRecordBatch>(new
ArrowRecordBatch(
+ import_result.ValueOrDie(), ffi_batch.table_id,
ffi_batch.partition_id,
+ ffi_batch.bucket_id, ffi_batch.base_offset)));
+ }
+ return utils::make_ok();
+ }
+};
+} // namespace detail
+
Result LogScanner::PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches&
out) {
if (!Available()) {
return utils::make_client_error("LogScanner not available");
@@ -1801,35 +1868,65 @@ Result LogScanner::PollRecordBatch(int64_t timeout_ms,
ArrowRecordBatches& out)
if (!result.Ok()) {
return result;
}
+ return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out);
+}
+
+// ============================================================================
+// BatchScanner
+// ============================================================================
+
+BatchScanner::BatchScanner() noexcept = default;
+
+BatchScanner::BatchScanner(ffi::BatchScanner* scanner) noexcept :
scanner_(scanner) {}
+
+BatchScanner::~BatchScanner() noexcept { Destroy(); }
+
+void BatchScanner::Destroy() noexcept {
+ if (scanner_) {
+ ffi::delete_batch_scanner(scanner_);
+ scanner_ = nullptr;
+ }
+}
- // Convert the FFI Arrow record batches to C++ ArrowRecordBatch objects
- out.batches.clear();
- for (const auto& ffi_batch : ffi_result.arrow_batches.batches) {
- auto* c_array = reinterpret_cast<struct
ArrowArray*>(ffi_batch.array_ptr);
- auto* c_schema = reinterpret_cast<struct
ArrowSchema*>(ffi_batch.schema_ptr);
+BatchScanner::BatchScanner(BatchScanner&& other) noexcept
+ : scanner_(other.scanner_), bucket_(other.bucket_) {
+ other.scanner_ = nullptr;
+}
- auto import_result = arrow::ImportRecordBatch(c_array, c_schema);
- if (import_result.ok()) {
- auto batch_ptr = import_result.ValueOrDie();
- auto batch_wrapper = std::unique_ptr<ArrowRecordBatch>(new
ArrowRecordBatch(
- std::move(batch_ptr), ffi_batch.table_id,
ffi_batch.partition_id,
- ffi_batch.bucket_id, ffi_batch.base_offset));
- out.batches.push_back(std::move(batch_wrapper));
+BatchScanner& BatchScanner::operator=(BatchScanner&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ scanner_ = other.scanner_;
+ bucket_ = other.bucket_;
+ other.scanner_ = nullptr;
+ }
+ return *this;
+}
- // Free the container structures that were allocated in Rust after
successful import
- ffi::free_arrow_ffi_structures(ffi_batch.array_ptr,
ffi_batch.schema_ptr);
- } else {
- // Import failed, free the container structures to avoid leaks and
return error
- ffi::free_arrow_ffi_structures(ffi_batch.array_ptr,
ffi_batch.schema_ptr);
+bool BatchScanner::Available() const { return scanner_ != nullptr; }
- // Return an error indicating that the import failed
- std::string error_msg =
- "Failed to import Arrow record batch: " +
import_result.status().ToString();
- return utils::make_client_error(error_msg);
- }
+Result BatchScanner::NextBatch(ArrowRecordBatches& out) {
+ if (!Available()) {
+ return utils::make_client_error("BatchScanner not available");
+ }
+ auto ffi_result = scanner_->next_batch();
+ auto result = utils::from_ffi_result(ffi_result.result);
+ if (!result.Ok()) {
+ return result;
}
+ return detail::ArrowBatchImporter::Import(ffi_result.arrow_batches, out);
+}
- return utils::make_ok();
+Result BatchScanner::CollectAllBatches(ArrowRecordBatches& out) {
+ if (!Available()) {
+ return utils::make_client_error("BatchScanner not available");
+ }
+ auto ffi_result = scanner_->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);
}
} // namespace fluss
diff --git a/bindings/cpp/test/test_kv_table.cpp
b/bindings/cpp/test/test_kv_table.cpp
index 2ef7e60f..d3d81e98 100644
--- a/bindings/cpp/test/test_kv_table.cpp
+++ b/bindings/cpp/test/test_kv_table.cpp
@@ -155,6 +155,76 @@ TEST_F(KvTableTest, UpsertDeleteAndLookup) {
ASSERT_OK(adm.DropTable(table_path, false));
}
+TEST_F(KvTableTest, LimitScan) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_limit_scan_pk_cpp");
+
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("id", fluss::DataType::Int())
+ .AddColumn("name", fluss::DataType::String())
+ .SetPrimaryKeys({"id"})
+ .Build();
+
+ auto table_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketCount(1)
+ .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_upsert = table.NewUpsert();
+ fluss::UpsertWriter upsert_writer;
+ ASSERT_OK(table_upsert.CreateWriter(upsert_writer));
+ std::vector<std::pair<int32_t, std::string>> rows = {
+ {1, "Verso"}, {2, "Noco"}, {3, "Esquie"}, {4, "Aline"}, {5,
"Gustave"}};
+ for (const auto& [id, name] : rows) {
+ fluss::GenericRow row(2);
+ row.SetInt32(0, id);
+ row.SetString(1, name);
+ ASSERT_OK(upsert_writer.Upsert(row));
+ }
+ ASSERT_OK(upsert_writer.Flush());
+
+ int64_t table_id = table.GetTableInfo().table_id;
+ fluss::TableBucket bucket{table_id, 0};
+
+ fluss::BatchScanner scanner;
+ ASSERT_OK(table.NewScan().Limit(3).CreateBucketBatchScanner(bucket,
scanner));
+ EXPECT_TRUE(scanner.Bucket() == bucket);
+ fluss::ArrowRecordBatches first;
+ ASSERT_OK(scanner.NextBatch(first));
+ ASSERT_FALSE(first.Empty()) << "first NextBatch should return a batch";
+ int64_t scanned = 0;
+ for (const auto& b : first) {
+ EXPECT_EQ(b->GetBucketId(), 0);
+ scanned += b->NumRows();
+ }
+ EXPECT_GT(scanned, 0);
+ EXPECT_LE(scanned, 3);
+
+ fluss::ArrowRecordBatches spent;
+ ASSERT_OK(scanner.NextBatch(spent));
+ EXPECT_TRUE(spent.Empty()) << "scanner must be spent after one batch";
+
+ fluss::BatchScanner scanner2;
+ ASSERT_OK(table.NewScan().Limit(100).CreateBucketBatchScanner(bucket,
scanner2));
+ fluss::ArrowRecordBatches all;
+ ASSERT_OK(scanner2.CollectAllBatches(all));
+ int64_t total = 0;
+ for (const auto& b : all) {
+ total += b->NumRows();
+ }
+ EXPECT_EQ(total, 5);
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
TEST_F(KvTableTest, LookupWithNestedArrayArrayView) {
auto& adm = admin();
auto& conn = connection();
diff --git a/bindings/cpp/test/test_log_table.cpp
b/bindings/cpp/test/test_log_table.cpp
index 5678e4bb..cae0bcb3 100644
--- a/bindings/cpp/test/test_log_table.cpp
+++ b/bindings/cpp/test/test_log_table.cpp
@@ -171,6 +171,212 @@ TEST_F(LogTableTest, AppendRecordBatchAndScan) {
ASSERT_OK(adm.DropTable(table_path, false));
}
+TEST_F(LogTableTest, LimitScan) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_limit_scan_cpp");
+
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("c1", fluss::DataType::Int())
+ .AddColumn("c2", fluss::DataType::String())
+ .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();
+ c1.AppendValues({1, 2, 3, 4, 5}).ok();
+ auto c2 = arrow::StringBuilder();
+ c2.AppendValues({"a", "b", "c", "d", "e"}).ok();
+ auto batch = arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("c1", arrow::int32()),
arrow::field("c2", arrow::utf8())}),
+ 5, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie()});
+ ASSERT_OK(append_writer.AppendArrowBatch(batch));
+ }
+ ASSERT_OK(append_writer.Flush());
+
+ int64_t table_id = table.GetTableInfo().table_id;
+ fluss::TableBucket bucket{table_id, 0};
+
+ fluss::BatchScanner scanner;
+ ASSERT_OK(table.NewScan().Limit(3).CreateBucketBatchScanner(bucket,
scanner));
+ EXPECT_TRUE(scanner.Bucket() == bucket);
+
+ fluss::ArrowRecordBatches first;
+ ASSERT_OK(scanner.NextBatch(first));
+ ASSERT_FALSE(first.Empty()) << "first NextBatch should return a batch";
+ int64_t rows = 0;
+ for (const auto& b : first) {
+ EXPECT_EQ(b->GetTableId(), table_id);
+ EXPECT_EQ(b->GetBucketId(), 0);
+ rows += b->NumRows();
+ }
+ // The server may return fewer rows than the limit, but never more.
+ EXPECT_GT(rows, 0);
+ EXPECT_LE(rows, 3);
+
+ fluss::ArrowRecordBatches spent;
+ ASSERT_OK(scanner.NextBatch(spent));
+ EXPECT_TRUE(spent.Empty()) << "scanner must be spent after one batch";
+
+ fluss::BatchScanner scanner2;
+ ASSERT_OK(table.NewScan().Limit(10).CreateBucketBatchScanner(bucket,
scanner2));
+ fluss::ArrowRecordBatches all;
+ ASSERT_OK(scanner2.CollectAllBatches(all));
+ int64_t total = 0;
+ for (const auto& b : all) {
+ total += b->NumRows();
+ }
+ EXPECT_EQ(total, 5) << "limit 10 over a 5-row bucket returns all rows";
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(LogTableTest, LimitScanProjection) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_limit_scan_projection_cpp");
+
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("c1", fluss::DataType::Int())
+ .AddColumn("c2", fluss::DataType::String())
+ .AddColumn("c3", fluss::DataType::BigInt())
+ .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();
+ c1.AppendValues({1, 2}).ok();
+ auto c2 = arrow::StringBuilder();
+ c2.AppendValues({"a", "b"}).ok();
+ auto c3 = arrow::Int64Builder();
+ c3.AppendValues({100, 200}).ok();
+ auto batch = arrow::RecordBatch::Make(
+ arrow::schema({arrow::field("c1", arrow::int32()),
arrow::field("c2", arrow::utf8()),
+ arrow::field("c3", arrow::int64())}),
+ 2, {c1.Finish().ValueOrDie(), c2.Finish().ValueOrDie(),
c3.Finish().ValueOrDie()});
+ ASSERT_OK(append_writer.AppendArrowBatch(batch));
+ }
+ ASSERT_OK(append_writer.Flush());
+
+ fluss::TableBucket bucket{table.GetTableInfo().table_id, 0};
+
+ // Projecting [c1, c3] skips the middle c2 string column.
+ fluss::BatchScanner scanner;
+ ASSERT_OK(
+ table.NewScan().ProjectByIndex({0,
2}).Limit(10).CreateBucketBatchScanner(bucket, scanner));
+ fluss::ArrowRecordBatches all;
+ ASSERT_OK(scanner.CollectAllBatches(all));
+ ASSERT_EQ(all.Size(), 1u);
+
+ auto rb = all[0]->GetArrowRecordBatch();
+ ASSERT_EQ(rb->num_columns(), 2);
+ EXPECT_EQ(rb->schema()->field(0)->name(), "c1");
+ EXPECT_EQ(rb->schema()->field(1)->name(), "c3");
+ ASSERT_EQ(rb->num_rows(), 2);
+ auto c1a = std::static_pointer_cast<arrow::Int32Array>(rb->column(0));
+ auto c3a = std::static_pointer_cast<arrow::Int64Array>(rb->column(1));
+ EXPECT_EQ(c1a->Value(0), 1);
+ EXPECT_EQ(c1a->Value(1), 2);
+ EXPECT_EQ(c3a->Value(0), 100);
+ EXPECT_EQ(c3a->Value(1), 200);
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(LogTableTest, LimitScanErrors) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_limit_scan_errors_cpp");
+
+ auto schema = fluss::Schema::NewBuilder().AddColumn("c1",
fluss::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));
+ int64_t table_id = table.GetTableInfo().table_id;
+
+ {
+ fluss::BatchScanner s;
+ EXPECT_FALSE(
+
table.NewScan().CreateBucketBatchScanner(fluss::TableBucket{table_id, 0},
s).Ok());
+ }
+ for (int32_t bad : {0, -5}) {
+ fluss::BatchScanner s;
+ EXPECT_FALSE(table.NewScan()
+ .Limit(bad)
+
.CreateBucketBatchScanner(fluss::TableBucket{table_id, 0}, s)
+ .Ok());
+ }
+ for (const auto& bad :
+ {fluss::TableBucket{table_id + 9999, 0}, fluss::TableBucket{table_id,
99}}) {
+ fluss::BatchScanner s;
+ EXPECT_FALSE(table.NewScan().Limit(1).CreateBucketBatchScanner(bad,
s).Ok());
+ }
+ {
+ fluss::LogScanner s;
+ EXPECT_FALSE(table.NewScan().Limit(5).CreateLogScanner(s).Ok());
+ fluss::LogScanner s2;
+
EXPECT_FALSE(table.NewScan().Limit(5).CreateRecordBatchLogScanner(s2).Ok());
+ }
+ ASSERT_OK(adm.DropTable(table_path, false));
+
+ // A non-ARROW (INDEXED) log table rejects a limit scan.
+ fluss::TablePath indexed_path("fluss", "test_limit_scan_indexed_cpp");
+ auto indexed_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketCount(1)
+ .SetBucketKeys({"c1"})
+ .SetLogFormat("INDEXED")
+ .SetProperty("table.replication.factor", "1")
+ .Build();
+ fluss_test::CreateTable(adm, indexed_path, indexed_descriptor);
+ fluss::Table indexed_table;
+ ASSERT_OK(conn.GetTable(indexed_path, indexed_table));
+ {
+ fluss::BatchScanner s;
+ fluss::TableBucket b{indexed_table.GetTableInfo().table_id, 0};
+
EXPECT_FALSE(indexed_table.NewScan().Limit(1).CreateBucketBatchScanner(b,
s).Ok());
+ }
+ ASSERT_OK(adm.DropTable(indexed_path, false));
+}
+
TEST_F(LogTableTest, ListOffsets) {
auto& adm = admin();
auto& conn = connection();
diff --git a/crates/fluss/src/client/table/batch_scanner.rs
b/crates/fluss/src/client/table/batch_scanner.rs
index 5d0cf0c6..090a7428 100644
--- a/crates/fluss/src/client/table/batch_scanner.rs
+++ b/crates/fluss/src/client/table/batch_scanner.rs
@@ -410,7 +410,7 @@ mod tests {
use crate::row::binary::BinaryWriter;
use crate::row::compacted::CompactedRowWriter;
use crate::test_utils::build_table_info_with_columns;
- use arrow::array::{Array, Int32Array, Int64Array};
+ use arrow::array::{Array, Int32Array, Int64Array, StringArray};
fn build_two_col_table_info() -> TableInfo {
build_table_info_with_columns(
@@ -570,6 +570,36 @@ mod tests {
assert_eq!((c3.value(0), c3.value(1)), (100, 200));
}
+ #[test]
+ fn decode_log_batch_non_prefix_projection_reorders_columns() {
+ let table_info = build_two_col_table_info();
+ let raw = build_log_records(&table_info, 0, &[(1, "alice"), (2,
"bob")]);
+
+ // `[name, id]`: reversed, so neither a leading prefix nor in source
order.
+ let (batch, _) = decode_log_batch(&table_info, Some(&[1usize,
0usize]), raw, usize::MAX)
+ .expect("decode reordered projection");
+
+ assert_eq!(batch.num_columns(), 2);
+ assert_eq!(batch.schema().field(0).name(), "name");
+ assert_eq!(batch.schema().field(1).name(), "id");
+
+ let names = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(names.value(0), "alice");
+ assert_eq!(names.value(1), "bob");
+
+ let ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ids.value(0), 1);
+ assert_eq!(ids.value(1), 2);
+ }
+
#[test]
fn decode_log_batch_truncates_to_last_limit_rows() {
let table_info = build_two_col_table_info();
diff --git a/website/docs/user-guide/cpp/api-reference.md
b/website/docs/user-guide/cpp/api-reference.md
index 60a43eb4..f2977114 100644
--- a/website/docs/user-guide/cpp/api-reference.md
+++ b/website/docs/user-guide/cpp/api-reference.md
@@ -148,8 +148,10 @@ Complete API reference for the Fluss C++ client.
|----------------------------------------------------------------------|-----------------------------------------------|
| `ProjectByIndex(std::vector<size_t> column_indices) -> TableScan&` |
Project columns by index |
| `ProjectByName(std::vector<std::string> 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 |
| `CreateRecordBatchLogScanner(LogScanner& out) -> Result` |
Create an Arrow RecordBatch-based log scanner |
+| `CreateBucketBatchScanner(const TableBucket& bucket, BatchScanner& out) ->
Result` | Bounded scan of one bucket (requires `Limit`) |
## `AppendWriter`
@@ -202,6 +204,15 @@ Performs prefix (bucket-key) lookups, returning all rows
whose primary key start
| `Poll(int64_t timeout_ms, ScanRecords& out) -> Result`
| Poll individual records |
| `PollRecordBatch(int64_t timeout_ms, ArrowRecordBatches& out) -> Result`
| Poll Arrow RecordBatches |
+## `BatchScanner`
+
+One-shot bounded scan of a single bucket. Obtained from
`TableScan::CreateBucketBatchScanner()`. The scan runs on the first call below,
yields its single batch once, then is spent.
+
+| Method | Description
|
+|--------------------------------------------------------------|-----------------------------------------------------|
+| `NextBatch(ArrowRecordBatches& out) -> Result` | Run the scan;
returns the batch once, then empty |
+| `CollectAllBatches(ArrowRecordBatches& out) -> Result` | Drain the
scanner into all of its batches |
+
## `GenericRow`
`GenericRow` is a **write-only** row used for append, upsert, delete, and
lookup key construction. For reading field values from scan or lookup results,
see [`RowView`](#rowview) and [`LookupResult`](#lookupresult).
diff --git a/website/docs/user-guide/cpp/example/log-tables.md
b/website/docs/user-guide/cpp/example/log-tables.md
index 0125a4ce..ca243f5d 100644
--- a/website/docs/user-guide/cpp/example/log-tables.md
+++ b/website/docs/user-guide/cpp/example/log-tables.md
@@ -159,3 +159,23 @@ table.NewScan().ProjectByName({"event_id",
"timestamp"}).CreateLogScanner(name_p
fluss::LogScanner projected_arrow_scanner;
table.NewScan().ProjectByIndex({0,
2}).CreateRecordBatchLogScanner(projected_arrow_scanner);
```
+
+## Limit Scan
+
+For a bounded read of up to `n` rows from a single bucket, use a batch scanner
instead of subscribing. It issues one request; `NextBatch` yields the batch
once, then reports empty.
+
+```cpp
+int64_t table_id = table.GetTableInfo().table_id;
+fluss::TableBucket bucket{table_id, 0};
+
+fluss::BatchScanner scanner;
+table.NewScan().Limit(10).CreateBucketBatchScanner(bucket, scanner);
+
+fluss::ArrowRecordBatches batches;
+scanner.NextBatch(batches); // or CollectAllBatches(batches)
+for (const auto& batch : batches) {
+ std::cout << "rows: " << batch->NumRows() << std::endl;
+}
+```
+
+The limit applies per bucket; scan each bucket to cover a multi-bucket table.
diff --git a/website/docs/user-guide/cpp/example/primary-key-tables.md
b/website/docs/user-guide/cpp/example/primary-key-tables.md
index f26b5477..f0dfad90 100644
--- a/website/docs/user-guide/cpp/example/primary-key-tables.md
+++ b/website/docs/user-guide/cpp/example/primary-key-tables.md
@@ -130,3 +130,20 @@ if (result.Found()) {
std::cout << "Not found" << std::endl;
}
```
+
+## Limit Scan
+
+To read up to `n` rows of a bucket's current state without supplying keys, use
a batch scanner. The server returns the deduplicated current rows as Arrow
batches, convenient for previews or analytics.
+
+```cpp
+int64_t table_id = table.GetTableInfo().table_id;
+fluss::TableBucket bucket{table_id, 0};
+
+fluss::BatchScanner scanner;
+table.NewScan().Limit(10).CreateBucketBatchScanner(bucket, scanner);
+
+fluss::ArrowRecordBatches batches;
+scanner.CollectAllBatches(batches);
+```
+
+The limit applies per bucket; scan each bucket to cover a multi-bucket table.
