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 63638ecc [c++] add prefix lookup (#606)
63638ecc is described below
commit 63638eccf292fa1e71ac750c2e82f283af8349a7
Author: Anton Borisov <[email protected]>
AuthorDate: Mon Jun 8 00:11:17 2026 +0100
[c++] add prefix lookup (#606)
* [c++] add prefix lookup
* reword comment
---
bindings/cpp/include/fluss.hpp | 157 ++++++++
bindings/cpp/src/lib.rs | 410 +++++++++++++++++++++
bindings/cpp/src/table.cpp | 216 +++++++++++
bindings/cpp/test/test_kv_table.cpp | 249 +++++++++++++
website/docs/user-guide/cpp/api-reference.md | 19 +
.../docs/user-guide/cpp/example/prefix-lookup.md | 107 ++++++
6 files changed, 1158 insertions(+)
diff --git a/bindings/cpp/include/fluss.hpp b/bindings/cpp/include/fluss.hpp
index d019b427..c27b039b 100644
--- a/bindings/cpp/include/fluss.hpp
+++ b/bindings/cpp/include/fluss.hpp
@@ -47,9 +47,11 @@ struct WriteResult;
struct LogScanner;
struct UpsertWriter;
struct Lookuper;
+struct PrefixLookuper;
struct ScanResultInner;
struct GenericRowInner;
struct LookupResultInner;
+struct PrefixLookupResultInner;
struct ArrayWriterInner;
struct ArrayViewInner;
} // namespace ffi
@@ -610,6 +612,19 @@ struct ScanData {
ScanData(const ScanData&) = delete;
ScanData& operator=(const ScanData&) = delete;
};
+
+/// Backing store for a prefix lookup result; mirrors ScanData.
+struct PrefixData {
+ ffi::PrefixLookupResultInner* raw;
+ ColumnMap columns;
+
+ PrefixData(ffi::PrefixLookupResultInner* r, ColumnMap cols)
+ : raw(r), columns(std::move(cols)) {}
+ ~PrefixData();
+
+ PrefixData(const PrefixData&) = delete;
+ PrefixData& operator=(const PrefixData&) = delete;
+};
} // namespace detail
/**
@@ -650,6 +665,7 @@ class ArrayView {
private:
friend class RowView;
friend class LookupResult;
+ friend class PrefixRowView;
explicit ArrayView(ffi::ArrayViewInner* inner) : inner_(inner) {}
void Destroy() noexcept;
ffi::ArrayViewInner* inner_{nullptr};
@@ -780,6 +796,7 @@ class GenericRow {
friend class AppendWriter;
friend class UpsertWriter;
friend class Lookuper;
+ friend class PrefixLookuper;
using ColumnInfo = detail::ColumnInfo;
using ColumnMap = detail::ColumnMap;
@@ -894,6 +911,118 @@ class RowView : public detail::NamedGetters<RowView> {
size_t rec_idx_;
};
+/// Read-only view over one row of a prefix lookup result.
+///
+/// Like RowView, but backed by PrefixData (a prefix lookup result) rather than
+/// scan data. Shares ownership of the underlying handle via reference
counting,
+/// so it can safely outlive the PrefixLookupResult that produced it.
+class PrefixRowView : public detail::NamedGetters<PrefixRowView> {
+ friend struct detail::NamedGetters<PrefixRowView>;
+
+ public:
+ PrefixRowView(std::shared_ptr<const detail::PrefixData> data, size_t
rec_idx)
+ : data_(std::move(data)), rec_idx_(rec_idx) {}
+
+ // ── Index-based getters ──────────────────────────────────────────
+ size_t FieldCount() const;
+ TypeId GetType(size_t idx) const;
+ bool IsNull(size_t idx) const;
+ bool GetBool(size_t idx) const;
+ int32_t GetInt32(size_t idx) const;
+ int64_t GetInt64(size_t idx) const;
+ float GetFloat32(size_t idx) const;
+ double GetFloat64(size_t idx) const;
+ std::string_view GetString(size_t idx) const;
+ std::pair<const uint8_t*, size_t> GetBytes(size_t idx) const;
+ fluss::Date GetDate(size_t idx) const;
+ fluss::Time GetTime(size_t idx) const;
+ fluss::Timestamp GetTimestamp(size_t idx) const;
+ bool IsDecimal(size_t idx) const;
+ std::string GetDecimalString(size_t idx) const;
+
+ // ── Array getters ────────────────────────────────────────────────
+ size_t GetArraySize(size_t idx) const;
+ TypeId GetArrayElementType(size_t idx) const;
+ bool IsArrayElementNull(size_t idx, size_t element) const;
+ bool GetArrayBool(size_t idx, size_t element) const;
+ int32_t GetArrayInt32(size_t idx, size_t element) const;
+ int64_t GetArrayInt64(size_t idx, size_t element) const;
+ float GetArrayFloat32(size_t idx, size_t element) const;
+ double GetArrayFloat64(size_t idx, size_t element) const;
+ std::string GetArrayString(size_t idx, size_t element) const;
+ std::vector<uint8_t> GetArrayBytes(size_t idx, size_t element) const;
+ fluss::Date GetArrayDate(size_t idx, size_t element) const;
+ fluss::Time GetArrayTime(size_t idx, size_t element) const;
+ fluss::Timestamp GetArrayTimestamp(size_t idx, size_t element) const;
+ std::string GetArrayDecimalString(size_t idx, size_t element) const;
+ ArrayView GetArrayView(size_t idx) const;
+
+ // Name-based getters inherited from detail::NamedGetters<PrefixRowView>
+ using detail::NamedGetters<PrefixRowView>::IsNull;
+ using detail::NamedGetters<PrefixRowView>::GetBool;
+ using detail::NamedGetters<PrefixRowView>::GetInt32;
+ using detail::NamedGetters<PrefixRowView>::GetInt64;
+ using detail::NamedGetters<PrefixRowView>::GetFloat32;
+ using detail::NamedGetters<PrefixRowView>::GetFloat64;
+ using detail::NamedGetters<PrefixRowView>::GetString;
+ using detail::NamedGetters<PrefixRowView>::GetBytes;
+ using detail::NamedGetters<PrefixRowView>::GetDate;
+ using detail::NamedGetters<PrefixRowView>::GetTime;
+ using detail::NamedGetters<PrefixRowView>::GetTimestamp;
+ using detail::NamedGetters<PrefixRowView>::GetDecimalString;
+ using detail::NamedGetters<PrefixRowView>::GetArraySize;
+ using detail::NamedGetters<PrefixRowView>::GetArrayElementType;
+ using detail::NamedGetters<PrefixRowView>::IsArrayElementNull;
+ using detail::NamedGetters<PrefixRowView>::GetArrayBool;
+ using detail::NamedGetters<PrefixRowView>::GetArrayInt32;
+ using detail::NamedGetters<PrefixRowView>::GetArrayInt64;
+ using detail::NamedGetters<PrefixRowView>::GetArrayFloat32;
+ using detail::NamedGetters<PrefixRowView>::GetArrayFloat64;
+ using detail::NamedGetters<PrefixRowView>::GetArrayString;
+ using detail::NamedGetters<PrefixRowView>::GetArrayBytes;
+ using detail::NamedGetters<PrefixRowView>::GetArrayDate;
+ using detail::NamedGetters<PrefixRowView>::GetArrayTime;
+ using detail::NamedGetters<PrefixRowView>::GetArrayTimestamp;
+ using detail::NamedGetters<PrefixRowView>::GetArrayDecimalString;
+ using detail::NamedGetters<PrefixRowView>::GetArrayView;
+
+ private:
+ size_t Resolve(const std::string& name) const {
+ if (!data_) {
+ throw std::runtime_error("PrefixRowView: name-based access not
available");
+ }
+ return detail::ResolveColumn(data_->columns, name);
+ }
+ std::shared_ptr<const detail::PrefixData> data_;
+ size_t rec_idx_;
+};
+
+/// Read-only result of a prefix lookup: zero-or-more matched rows, via
Size()/GetRow(i).
+class PrefixLookupResult {
+ public:
+ PrefixLookupResult() = default;
+
+ /// Number of matched rows.
+ size_t Size() const;
+ bool IsEmpty() const { return Size() == 0; }
+
+ /// Returns a view over the row at `index`. Throws std::out_of_range if
+ /// `index >= Size()`.
+ PrefixRowView GetRow(size_t index) const {
+ if (!data_) {
+ throw std::logic_error("PrefixLookupResult: not available
(moved-from or null)");
+ }
+ if (index >= Size()) {
+ throw std::out_of_range("PrefixLookupResult::GetRow: index out of
range");
+ }
+ return PrefixRowView(data_, index);
+ }
+
+ private:
+ friend class PrefixLookuper;
+ std::shared_ptr<const detail::PrefixData> data_;
+};
+
/// Identifies a specific bucket, optionally within a partition.
struct TableBucket {
int64_t table_id;
@@ -1214,6 +1343,7 @@ class LookupResult : public
detail::NamedGetters<LookupResult> {
class AppendWriter;
class UpsertWriter;
class Lookuper;
+class PrefixLookuper;
class WriteResult;
class LogScanner;
class Admin;
@@ -1401,6 +1531,10 @@ class Table {
TableLookup NewLookup();
TableScan NewScan();
+ /// Creates a prefix lookuper. `lookup_columns` must be the table's
bucket-key
+ /// columns (a strict prefix of the primary key); fails otherwise.
+ Result NewPrefixLookup(std::vector<std::string> lookup_columns,
PrefixLookuper& out);
+
TableInfo GetTableInfo() const;
TablePath GetTablePath() const;
bool HasPrimaryKey() const;
@@ -1600,6 +1734,29 @@ class Lookuper {
ffi::Lookuper* lookuper_{nullptr};
};
+/// Performs bucket-key prefix lookups; create via Table::NewPrefixLookup().
Move-only.
+class PrefixLookuper {
+ public:
+ PrefixLookuper() noexcept;
+ ~PrefixLookuper() noexcept;
+
+ PrefixLookuper(const PrefixLookuper&) = delete;
+ PrefixLookuper& operator=(const PrefixLookuper&) = delete;
+ PrefixLookuper(PrefixLookuper&& other) noexcept;
+ PrefixLookuper& operator=(PrefixLookuper&& other) noexcept;
+
+ bool Available() const;
+
+ /// Looks up all rows matching the prefix columns set on `prefix_row`.
+ Result PrefixLookup(const GenericRow& prefix_row, PrefixLookupResult& out);
+
+ private:
+ friend class Table;
+ PrefixLookuper(ffi::PrefixLookuper* lookuper) noexcept;
+ void Destroy() noexcept;
+ ffi::PrefixLookuper* lookuper_{nullptr};
+};
+
class LogScanner {
public:
LogScanner() noexcept;
diff --git a/bindings/cpp/src/lib.rs b/bindings/cpp/src/lib.rs
index b9d5a03e..b5417ae4 100644
--- a/bindings/cpp/src/lib.rs
+++ b/bindings/cpp/src/lib.rs
@@ -23,7 +23,10 @@ use std::time::Duration;
use fluss as fcore;
use fluss::PartitionId;
+use fluss::client::PrefixKeyLookuper;
use fluss::error::Error;
+use fluss::metadata::{Column, TableInfo};
+use fluss::row::{Datum, GenericRow};
use fluss::rpc::FlussError as CoreFlussError;
static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
@@ -285,11 +288,13 @@ mod ffi {
type LogScanner;
type UpsertWriter;
type Lookuper;
+ type PrefixLookuper;
// Opaque types for optimized FFI
type ScanResultInner;
type GenericRowInner;
type LookupResultInner;
+ type PrefixLookupResultInner;
type ArrayWriterInner;
type ArrayViewInner;
@@ -379,6 +384,7 @@ mod ffi {
fn has_primary_key(self: &Table) -> bool;
fn create_upsert_writer(self: &Table, column_indices: Vec<usize>) ->
FfiPtrResult;
fn new_lookuper(self: &Table) -> FfiPtrResult;
+ fn new_prefix_lookuper(self: &Table, lookup_column_names: Vec<String>)
-> FfiPtrResult;
// GenericRowInner — opaque row for writes
fn new_generic_row(field_count: usize) -> Box<GenericRowInner>;
@@ -541,6 +547,152 @@ mod ffi {
fn lv_get_array_view(self: &LookupResultInner, field: usize)
-> Result<Box<ArrayViewInner>>;
+ // PrefixLookuper
+ unsafe fn delete_prefix_lookuper(lookuper: *mut PrefixLookuper);
+ fn prefix_lookup(
+ self: &mut PrefixLookuper,
+ prefix_row: &GenericRowInner,
+ ) -> Box<PrefixLookupResultInner>;
+
+ // PrefixLookupResultInner accessors — like LookupResultInner but
indexed
+ // by record, since a prefix lookup returns zero-or-more rows.
+ fn plv_has_error(self: &PrefixLookupResultInner) -> bool;
+ fn plv_error_code(self: &PrefixLookupResultInner) -> i32;
+ fn plv_error_message(self: &PrefixLookupResultInner) -> &str;
+ fn plv_row_count(self: &PrefixLookupResultInner) -> usize;
+ fn plv_field_count(self: &PrefixLookupResultInner) -> usize;
+ fn plv_column_name(self: &PrefixLookupResultInner, field: usize) ->
Result<&str>;
+ fn plv_column_type(self: &PrefixLookupResultInner, field: usize) ->
Result<i32>;
+ fn plv_is_null(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<bool>;
+ fn plv_get_bool(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<bool>;
+ fn plv_get_i32(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<i32>;
+ fn plv_get_i64(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<i64>;
+ fn plv_get_f32(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<f32>;
+ fn plv_get_f64(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<f64>;
+ fn plv_get_str(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<&str>;
+ fn plv_get_bytes(self: &PrefixLookupResultInner, rec: usize, field:
usize)
+ -> Result<&[u8]>;
+ fn plv_get_date_days(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<i32>;
+ fn plv_get_time_millis(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<i32>;
+ fn plv_get_ts_millis(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<i64>;
+ fn plv_get_ts_nanos(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<i32>;
+ fn plv_is_ts_ltz(self: &PrefixLookupResultInner, rec: usize, field:
usize) -> Result<bool>;
+ fn plv_get_decimal_str(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<String>;
+
+ fn plv_get_array_size(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<usize>;
+ fn plv_get_array_is_null(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<bool>;
+ fn plv_get_array_bool(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<bool>;
+ fn plv_get_array_i32(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i32>;
+ fn plv_get_array_i64(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i64>;
+ fn plv_get_array_f32(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<f32>;
+ fn plv_get_array_f64(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<f64>;
+ fn plv_get_array_str(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<String>;
+ fn plv_get_array_bytes(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<Vec<u8>>;
+ fn plv_get_array_date_days(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i32>;
+ fn plv_get_array_time_millis(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i32>;
+ fn plv_get_array_ts_millis(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i64>;
+ fn plv_get_array_ts_nanos(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<i32>;
+ fn plv_get_array_decimal_str(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ element: usize,
+ ) -> Result<String>;
+ fn plv_get_array_element_type(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<i32>;
+ fn plv_get_array_view(
+ self: &PrefixLookupResultInner,
+ rec: usize,
+ field: usize,
+ ) -> Result<Box<ArrayViewInner>>;
+
// ArrayViewInner — opaque recursive array reader for C++ bindings
fn av_size(self: &ArrayViewInner) -> usize;
fn av_element_type_id(self: &ArrayViewInner) -> i32;
@@ -836,6 +988,13 @@ pub struct Lookuper {
table_info: fcore::metadata::TableInfo,
}
+pub struct PrefixLookuper {
+ inner: PrefixKeyLookuper,
+ table_info: TableInfo,
+ /// Full-schema indices of the lookup columns, used to compact the input
row.
+ lookup_column_indices: Vec<usize>,
+}
+
/// Error code for client-side errors that did not originate from the server
API protocol.
/// Must be non-zero so that CPP `Result::Ok()` (which checks `error_code ==
0`) correctly
/// detects client-side errors as failures. The value -2 is outside the server
API error
@@ -1595,6 +1754,38 @@ impl Table {
}));
ok_ptr(ptr as usize)
}
+
+ fn new_prefix_lookuper(&self, lookup_column_names: Vec<String>) ->
ffi::FfiPtrResult {
+ let _enter = RUNTIME.enter();
+
+ let table_lookup = match self.fluss_table().new_lookup() {
+ Ok(l) => l,
+ Err(e) => return err_ptr_from_core(&e),
+ };
+
+ // `create_lookuper` validates the column names (rejecting unknown
ones),
+ // so by the time it succeeds every name resolves to an index.
+ let row_type = self.table_info.row_type();
+ let lookup_column_indices: Vec<usize> = lookup_column_names
+ .iter()
+ .filter_map(|name| row_type.get_field_index(name))
+ .collect();
+
+ let lookuper = match table_lookup
+ .lookup_by(lookup_column_names)
+ .create_lookuper()
+ {
+ Ok(l) => l,
+ Err(e) => return err_ptr_from_core(&e),
+ };
+
+ let ptr = Box::into_raw(Box::new(PrefixLookuper {
+ inner: lookuper,
+ table_info: self.table_info.clone(),
+ lookup_column_indices,
+ }));
+ ok_ptr(ptr as usize)
+ }
}
// AppendWriter implementation
@@ -1837,6 +2028,85 @@ impl Lookuper {
}
}
+// PrefixLookuper implementation
+unsafe fn delete_prefix_lookuper(lookuper: *mut PrefixLookuper) {
+ if !lookuper.is_null() {
+ unsafe {
+ drop(Box::from_raw(lookuper));
+ }
+ }
+}
+
+impl PrefixLookuper {
+ /// Compact a sparse input row (prefix columns set at their schema
positions)
+ /// into the dense, lookup-column-ordered row the core prefix encoder
expects.
+ fn dense_prefix_row<'a>(&self, mut row: GenericRow<'a>) -> GenericRow<'a> {
+ let mut dense = GenericRow::new(self.lookup_column_indices.len());
+ for (dense_idx, &schema_idx) in
self.lookup_column_indices.iter().enumerate() {
+ if schema_idx < row.values.len() {
+ dense.values[dense_idx] =
+ std::mem::replace(&mut row.values[schema_idx],
Datum::Null);
+ }
+ }
+ dense
+ }
+
+ fn prefix_lookup(&mut self, prefix_row: &GenericRowInner) ->
Box<PrefixLookupResultInner> {
+ let schema = self.table_info.get_schema();
+ let generic_row = match types::resolve_row_types(&prefix_row.row,
Some(schema)) {
+ Ok(r) => self.dense_prefix_row(r),
+ Err(e) => {
+ return Box::new(PrefixLookupResultInner::from_error(
+ CLIENT_ERROR_CODE,
+ e.to_string(),
+ ));
+ }
+ };
+
+ let lookup_result = match
RUNTIME.block_on(self.inner.lookup(&generic_row)) {
+ Ok(r) => r,
+ Err(e) => {
+ let ffi_err = err_from_core_error(&e);
+ return Box::new(PrefixLookupResultInner::from_error(
+ ffi_err.error_code,
+ ffi_err.error_message,
+ ));
+ }
+ };
+
+ let lookup_rows = match lookup_result.get_rows() {
+ Ok(rows) => rows,
+ Err(e) => {
+ let ffi_err = err_from_core_error(&e);
+ return Box::new(PrefixLookupResultInner::from_error(
+ ffi_err.error_code,
+ ffi_err.error_message,
+ ));
+ }
+ };
+
+ let mut rows = Vec::with_capacity(lookup_rows.len());
+ for row in &lookup_rows {
+ match types::compacted_row_to_owned(row, &self.table_info) {
+ Ok(owned_row) => rows.push(owned_row),
+ Err(e) => {
+ return Box::new(PrefixLookupResultInner::from_error(
+ CLIENT_ERROR_CODE,
+ e.to_string(),
+ ));
+ }
+ }
+ }
+
+ let columns = self.table_info.get_schema().columns().to_vec();
+ Box::new(PrefixLookupResultInner {
+ error: None,
+ rows,
+ columns,
+ })
+ }
+}
+
// LogScanner implementation
unsafe fn delete_log_scanner(scanner: *mut LogScanner) {
if !scanner.is_null() {
@@ -3246,6 +3516,146 @@ impl LookupResultInner {
}
}
+// ============================================================================
+// Opaque types: PrefixLookupResultInner (prefix lookup read path)
+//
+// Like LookupResultInner but holds 0..n rows; accessors take a record index.
+// ============================================================================
+
+pub struct PrefixLookupResultInner {
+ error: Option<(i32, String)>,
+ rows: Vec<GenericRow<'static>>,
+ columns: Vec<Column>,
+}
+
+macro_rules! plv_array_element_getters {
+ ($( $method:ident, $reader_fn:ident, $ret:ty; )+) => {
+ $(
+ fn $method(&self, rec: usize, field: usize, element: usize) ->
Result<$ret, String> {
+ let r = self.plv_row(rec)?;
+ row_reader::$reader_fn(r, &self.columns, field, element)
+ }
+ )+
+ };
+}
+
+impl PrefixLookupResultInner {
+ fn from_error(code: i32, msg: String) -> Self {
+ Self {
+ error: Some((code, msg)),
+ rows: Vec::new(),
+ columns: Vec::new(),
+ }
+ }
+
+ fn plv_has_error(&self) -> bool {
+ self.error.is_some()
+ }
+
+ fn plv_error_code(&self) -> i32 {
+ self.error.as_ref().map_or(0, |e| e.0)
+ }
+
+ fn plv_error_message(&self) -> &str {
+ self.error.as_ref().map_or("", |e| e.1.as_str())
+ }
+
+ fn plv_row_count(&self) -> usize {
+ self.rows.len()
+ }
+
+ fn plv_field_count(&self) -> usize {
+ self.columns.len()
+ }
+
+ fn plv_column_type(&self, field: usize) -> Result<i32, String> {
+ row_reader::column_type(&self.columns, field)
+ }
+
+ fn plv_column_name(&self, field: usize) -> Result<&str, String> {
+ row_reader::column_name(&self.columns, field)
+ }
+
+ fn plv_row(&self, rec: usize) -> Result<&GenericRow<'static>, String> {
+ self.rows
+ .get(rec)
+ .ok_or_else(|| format!("record index {rec} out of range ({}
rows)", self.rows.len()))
+ }
+
+ // Field accessors — delegate to shared row_reader helpers.
+ fn plv_is_null(&self, rec: usize, field: usize) -> Result<bool, String> {
+ row_reader::is_null(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_bool(&self, rec: usize, field: usize) -> Result<bool, String> {
+ row_reader::get_bool(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_i32(&self, rec: usize, field: usize) -> Result<i32, String> {
+ row_reader::get_i32(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_i64(&self, rec: usize, field: usize) -> Result<i64, String> {
+ row_reader::get_i64(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_f32(&self, rec: usize, field: usize) -> Result<f32, String> {
+ row_reader::get_f32(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_f64(&self, rec: usize, field: usize) -> Result<f64, String> {
+ row_reader::get_f64(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_str(&self, rec: usize, field: usize) -> Result<&str, String> {
+ row_reader::get_str(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_bytes(&self, rec: usize, field: usize) -> Result<&[u8], String>
{
+ row_reader::get_bytes(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_date_days(&self, rec: usize, field: usize) -> Result<i32,
String> {
+ row_reader::get_date_days(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_time_millis(&self, rec: usize, field: usize) -> Result<i32,
String> {
+ row_reader::get_time_millis(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_ts_millis(&self, rec: usize, field: usize) -> Result<i64,
String> {
+ row_reader::get_ts_millis(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_ts_nanos(&self, rec: usize, field: usize) -> Result<i32,
String> {
+ row_reader::get_ts_nanos(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_is_ts_ltz(&self, _rec: usize, field: usize) -> Result<bool, String>
{
+ row_reader::is_ts_ltz(&self.columns, field)
+ }
+ fn plv_get_decimal_str(&self, rec: usize, field: usize) -> Result<String,
String> {
+ row_reader::get_decimal_str(self.plv_row(rec)?, &self.columns, field)
+ }
+ fn plv_get_array_size(&self, rec: usize, field: usize) -> Result<usize,
String> {
+ row_reader::get_array_size(self.plv_row(rec)?, &self.columns, field)
+ }
+ plv_array_element_getters! {
+ plv_get_array_is_null, get_array_is_null, bool;
+ plv_get_array_bool, get_array_bool, bool;
+ plv_get_array_i32, get_array_i32, i32;
+ plv_get_array_i64, get_array_i64, i64;
+ plv_get_array_f32, get_array_f32, f32;
+ plv_get_array_f64, get_array_f64, f64;
+ plv_get_array_str, get_array_str, String;
+ plv_get_array_bytes, get_array_bytes, Vec<u8>;
+ plv_get_array_date_days, get_array_date_days, i32;
+ plv_get_array_time_millis, get_array_time_millis, i32;
+ plv_get_array_ts_millis, get_array_ts_millis, i64;
+ plv_get_array_ts_nanos, get_array_ts_nanos, i32;
+ plv_get_array_decimal_str, get_array_decimal_str, String;
+ }
+ fn plv_get_array_element_type(&self, _rec: usize, field: usize) ->
Result<i32, String> {
+ row_reader::get_array_element_type_id(&self.columns, field)
+ }
+ fn plv_get_array_view(&self, rec: usize, field: usize) ->
Result<Box<ArrayViewInner>, String> {
+ let r = self.plv_row(rec)?;
+ let (arr, elem) = row_reader::get_array_and_elem_type(r,
&self.columns, field)?;
+ Ok(Box::new(ArrayViewInner {
+ array: arr,
+ element_type: elem.clone(),
+ }))
+ }
+}
+
// ============================================================================
// Opaque types: ArrayViewInner (recursive array reader)
//
diff --git a/bindings/cpp/src/table.cpp b/bindings/cpp/src/table.cpp
index f389f7ac..9b5d087e 100644
--- a/bindings/cpp/src/table.cpp
+++ b/bindings/cpp/src/table.cpp
@@ -431,6 +431,12 @@ detail::ScanData::~ScanData() {
}
}
+detail::PrefixData::~PrefixData() {
+ if (raw) {
+ rust::Box<ffi::PrefixLookupResultInner>::from_raw(raw);
+ }
+}
+
// ============================================================================
// RowView — zero-copy read-only row view for scan results
// ============================================================================
@@ -587,6 +593,137 @@ ArrayView RowView::GetArrayView(size_t idx) const {
return ArrayView(box.into_raw());
}
+// ============================================================================
+// PrefixLookupResult / PrefixRowView — read path for prefix lookups
+// ============================================================================
+
+size_t PrefixLookupResult::Size() const { return data_ ?
data_->raw->plv_row_count() : 0; }
+
+size_t PrefixRowView::FieldCount() const { return data_ ?
data_->raw->plv_field_count() : 0; }
+
+TypeId PrefixRowView::GetType(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return static_cast<TypeId>(data_->raw->plv_column_type(idx));
+}
+
+bool PrefixRowView::IsNull(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_is_null(rec_idx_, idx);
+}
+bool PrefixRowView::GetBool(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_bool(rec_idx_, idx);
+}
+int32_t PrefixRowView::GetInt32(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_i32(rec_idx_, idx);
+}
+int64_t PrefixRowView::GetInt64(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_i64(rec_idx_, idx);
+}
+float PrefixRowView::GetFloat32(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_f32(rec_idx_, idx);
+}
+double PrefixRowView::GetFloat64(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_f64(rec_idx_, idx);
+}
+std::string_view PrefixRowView::GetString(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ auto s = data_->raw->plv_get_str(rec_idx_, idx);
+ return std::string_view(s.data(), s.size());
+}
+std::pair<const uint8_t*, size_t> PrefixRowView::GetBytes(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ auto bytes = data_->raw->plv_get_bytes(rec_idx_, idx);
+ return {bytes.data(), bytes.size()};
+}
+Date PrefixRowView::GetDate(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return Date{data_->raw->plv_get_date_days(rec_idx_, idx)};
+}
+Time PrefixRowView::GetTime(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return Time{data_->raw->plv_get_time_millis(rec_idx_, idx)};
+}
+Timestamp PrefixRowView::GetTimestamp(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return Timestamp{data_->raw->plv_get_ts_millis(rec_idx_, idx),
+ data_->raw->plv_get_ts_nanos(rec_idx_, idx)};
+}
+bool PrefixRowView::IsDecimal(size_t idx) const { return GetType(idx) ==
TypeId::Decimal; }
+std::string PrefixRowView::GetDecimalString(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return std::string(data_->raw->plv_get_decimal_str(rec_idx_, idx));
+}
+
+size_t PrefixRowView::GetArraySize(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_size(rec_idx_, idx);
+}
+TypeId PrefixRowView::GetArrayElementType(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ return
static_cast<TypeId>(data_->raw->plv_get_array_element_type(rec_idx_, idx));
+}
+bool PrefixRowView::IsArrayElementNull(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_is_null(rec_idx_, idx, element);
+}
+bool PrefixRowView::GetArrayBool(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_bool(rec_idx_, idx, element);
+}
+int32_t PrefixRowView::GetArrayInt32(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_i32(rec_idx_, idx, element);
+}
+int64_t PrefixRowView::GetArrayInt64(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_i64(rec_idx_, idx, element);
+}
+float PrefixRowView::GetArrayFloat32(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_f32(rec_idx_, idx, element);
+}
+double PrefixRowView::GetArrayFloat64(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return data_->raw->plv_get_array_f64(rec_idx_, idx, element);
+}
+std::string PrefixRowView::GetArrayString(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return std::string(data_->raw->plv_get_array_str(rec_idx_, idx, element));
+}
+std::vector<uint8_t> PrefixRowView::GetArrayBytes(size_t idx, size_t element)
const {
+ CHECK_DATA("PrefixRowView");
+ auto rv = data_->raw->plv_get_array_bytes(rec_idx_, idx, element);
+ return {rv.data(), rv.data() + rv.size()};
+}
+fluss::Date PrefixRowView::GetArrayDate(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return fluss::Date{data_->raw->plv_get_array_date_days(rec_idx_, idx,
element)};
+}
+fluss::Time PrefixRowView::GetArrayTime(size_t idx, size_t element) const {
+ CHECK_DATA("PrefixRowView");
+ return fluss::Time{data_->raw->plv_get_array_time_millis(rec_idx_, idx,
element)};
+}
+fluss::Timestamp PrefixRowView::GetArrayTimestamp(size_t idx, size_t element)
const {
+ CHECK_DATA("PrefixRowView");
+ auto millis = data_->raw->plv_get_array_ts_millis(rec_idx_, idx, element);
+ auto nanos = data_->raw->plv_get_array_ts_nanos(rec_idx_, idx, element);
+ return fluss::Timestamp{millis, nanos};
+}
+std::string PrefixRowView::GetArrayDecimalString(size_t idx, size_t element)
const {
+ CHECK_DATA("PrefixRowView");
+ return std::string(data_->raw->plv_get_array_decimal_str(rec_idx_, idx,
element));
+}
+ArrayView PrefixRowView::GetArrayView(size_t idx) const {
+ CHECK_DATA("PrefixRowView");
+ auto box = data_->raw->plv_get_array_view(rec_idx_, idx);
+ return ArrayView(box.into_raw());
+}
+
// ============================================================================
// ScanRecords — backed by opaque Rust ScanResultInner
// ============================================================================
@@ -914,6 +1051,25 @@ TableLookup Table::NewLookup() { return
TableLookup(table_); }
TableScan Table::NewScan() { return TableScan(table_); }
+Result Table::NewPrefixLookup(std::vector<std::string> lookup_columns,
PrefixLookuper& out) {
+ if (table_ == nullptr) {
+ return utils::make_client_error("Table not available");
+ }
+
+ rust::Vec<rust::String> cols;
+ cols.reserve(lookup_columns.size());
+ for (const auto& c : lookup_columns) {
+ cols.push_back(rust::String(c));
+ }
+
+ auto ffi_result = table_->new_prefix_lookuper(std::move(cols));
+ auto result = utils::from_ffi_result(ffi_result.result);
+ if (result.Ok()) {
+ out =
PrefixLookuper(utils::ptr_from_ffi<ffi::PrefixLookuper>(ffi_result));
+ }
+ return result;
+}
+
const std::shared_ptr<GenericRow::ColumnMap>& Table::GetColumnMap() const {
if (!column_map_ && Available()) {
auto info = GetTableInfo();
@@ -1405,6 +1561,66 @@ Result Lookuper::Lookup(const GenericRow& pk_row,
LookupResult& out) {
return utils::make_ok();
}
+// ============================================================================
+// PrefixLookuper
+// ============================================================================
+
+PrefixLookuper::PrefixLookuper() noexcept = default;
+
+PrefixLookuper::PrefixLookuper(ffi::PrefixLookuper* lookuper) noexcept :
lookuper_(lookuper) {}
+
+PrefixLookuper::~PrefixLookuper() noexcept { Destroy(); }
+
+void PrefixLookuper::Destroy() noexcept {
+ if (lookuper_) {
+ ffi::delete_prefix_lookuper(lookuper_);
+ lookuper_ = nullptr;
+ }
+}
+
+PrefixLookuper::PrefixLookuper(PrefixLookuper&& other) noexcept :
lookuper_(other.lookuper_) {
+ other.lookuper_ = nullptr;
+}
+
+PrefixLookuper& PrefixLookuper::operator=(PrefixLookuper&& other) noexcept {
+ if (this != &other) {
+ Destroy();
+ lookuper_ = other.lookuper_;
+ other.lookuper_ = nullptr;
+ }
+ return *this;
+}
+
+bool PrefixLookuper::Available() const { return lookuper_ != nullptr; }
+
+Result PrefixLookuper::PrefixLookup(const GenericRow& prefix_row,
PrefixLookupResult& out) {
+ if (!Available()) {
+ return utils::make_client_error("PrefixLookuper not available");
+ }
+ if (!prefix_row.Available()) {
+ return utils::make_client_error("GenericRow not available");
+ }
+
+ auto result_box = lookuper_->prefix_lookup(*prefix_row.inner_);
+ if (result_box->plv_has_error()) {
+ return utils::make_error(result_box->plv_error_code(),
+ std::string(result_box->plv_error_message()));
+ }
+
+ // Take ownership of the FFI box first (~PrefixData calls from_raw), so the
+ // column-map loop below can't leak it if a string/map allocation throws.
+ // The map is built eagerly and shared by all PrefixRowViews.
+ auto data = std::make_shared<detail::PrefixData>(result_box.into_raw(),
detail::ColumnMap{});
+ auto col_count = data->raw->plv_field_count();
+ for (size_t i = 0; i < col_count; ++i) {
+ auto name = data->raw->plv_column_name(i);
+ data->columns[std::string(name.data(), name.size())] = {
+ i, static_cast<TypeId>(data->raw->plv_column_type(i))};
+ }
+ out.data_ = std::move(data);
+ return utils::make_ok();
+}
+
// ============================================================================
// LogScanner
// ============================================================================
diff --git a/bindings/cpp/test/test_kv_table.cpp
b/bindings/cpp/test/test_kv_table.cpp
index 5cc8f79d..2ef7e60f 100644
--- a/bindings/cpp/test/test_kv_table.cpp
+++ b/bindings/cpp/test/test_kv_table.cpp
@@ -410,6 +410,255 @@ TEST_F(KvTableTest, CompositePrimaryKeys) {
ASSERT_OK(adm.DropTable(table_path, false));
}
+TEST_F(KvTableTest, PrefixLookupByBucketKey) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_prefix_lookup_cpp");
+
+ // Bucket key (a, b) is a strict prefix of the PK (a, b, c), enabling
prefix lookup on (a, b).
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("a", fluss::DataType::Int())
+ .AddColumn("b", fluss::DataType::String())
+ .AddColumn("c", fluss::DataType::BigInt())
+ .AddColumn("d", fluss::DataType::String())
+ .SetPrimaryKeys({"a", "b", "c"})
+ .Build();
+
+ auto table_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketKeys({"a", "b"})
+ .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));
+
+ struct TestData {
+ int32_t a;
+ std::string b;
+ int64_t c;
+ std::string d;
+ };
+ std::vector<TestData> test_data = {
+ {1, "aaaaaaaaa", 1, "value1"},
+ {1, "aaaaaaaaa", 2, "value2"},
+ {1, "aaaaaaaaa", 3, "value3"},
+ {2, "aaaaaaaaa", 4, "value4"},
+ };
+
+ for (const auto& row_data : test_data) {
+ auto row = table.NewRow();
+ row.Set("a", row_data.a);
+ row.Set("b", row_data.b);
+ row.Set("c", row_data.c);
+ row.Set("d", row_data.d);
+ ASSERT_OK(upsert_writer.Upsert(row));
+ }
+ ASSERT_OK(upsert_writer.Flush());
+
+ fluss::PrefixLookuper prefix_lookuper;
+ ASSERT_OK(table.NewPrefixLookup({"a", "b"}, prefix_lookuper));
+
+ // Prefix (1, "aaaaaaaaa") matches 3 rows, returned in primary-key order.
+ {
+ auto key = table.NewRow();
+ key.Set("a", 1);
+ key.Set("b", "aaaaaaaaa");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ ASSERT_EQ(result.Size(), 3u);
+ for (size_t i = 0; i < result.Size(); ++i) {
+ auto row = result.GetRow(i);
+ EXPECT_EQ(row.GetInt32("a"), 1);
+ EXPECT_EQ(row.GetString("b"), "aaaaaaaaa");
+ EXPECT_EQ(row.GetInt64("c"), static_cast<int64_t>(i) + 1);
+ EXPECT_EQ(row.GetString("d"), "value" + std::to_string(i + 1));
+ }
+ }
+
+ // Prefix (2, "aaaaaaaaa") matches a single row.
+ {
+ auto key = table.NewRow();
+ key.Set("a", 2);
+ key.Set("b", "aaaaaaaaa");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ ASSERT_EQ(result.Size(), 1u);
+ auto row = result.GetRow(0);
+ EXPECT_EQ(row.GetInt64("c"), 4);
+ EXPECT_EQ(row.GetString("d"), "value4");
+ }
+
+ // A prefix with no matching rows yields an empty result.
+ {
+ auto key = table.NewRow();
+ key.Set("a", 3);
+ key.Set("b", "a");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ EXPECT_TRUE(result.IsEmpty());
+ }
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(KvTableTest, PrefixLookupValidationErrors) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_prefix_lookup_validation_cpp");
+
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("a", fluss::DataType::Int())
+ .AddColumn("b", fluss::DataType::String())
+ .AddColumn("c", fluss::DataType::BigInt())
+ .SetPrimaryKeys({"a", "b", "c"})
+ .Build();
+
+ auto table_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketKeys({"a", "b"})
+ .SetProperty("table.replication.factor", "1")
+ .Build();
+
+ fluss_test::CreateTable(adm, table_path, table_descriptor);
+
+ fluss::Table table;
+ ASSERT_OK(conn.GetTable(table_path, table));
+
+ // Lookup columns must list the bucket keys (a, b) in order.
+ {
+ fluss::PrefixLookuper lookuper;
+ auto result = table.NewPrefixLookup({"b", "a"}, lookuper);
+ ASSERT_FALSE(result.Ok());
+ EXPECT_EQ(result.error_code, fluss::ErrorCode::CLIENT_ERROR);
+ EXPECT_NE(result.error_message.find("must contain all bucket keys"),
std::string::npos);
+ }
+
+ // Lookup columns must equal the bucket keys, not a longer prefix of the
PK.
+ {
+ fluss::PrefixLookuper lookuper;
+ auto result = table.NewPrefixLookup({"a", "b", "c"}, lookuper);
+ ASSERT_FALSE(result.Ok());
+ EXPECT_EQ(result.error_code, fluss::ErrorCode::CLIENT_ERROR);
+ EXPECT_NE(result.error_message.find("must contain all bucket keys"),
std::string::npos);
+ }
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(KvTableTest, PrefixLookupPartitioned) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_prefix_lookup_partitioned_cpp");
+
+ // Partitioned by region; bucket key (a, b) is a prefix of the PK minus
the partition column.
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("region", fluss::DataType::String())
+ .AddColumn("a", fluss::DataType::Int())
+ .AddColumn("b", fluss::DataType::String())
+ .AddColumn("c", fluss::DataType::BigInt())
+ .AddColumn("d", fluss::DataType::String())
+ .SetPrimaryKeys({"region", "a", "b", "c"})
+ .Build();
+
+ auto table_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetPartitionKeys({"region"})
+ .SetBucketKeys({"a", "b"})
+ .SetProperty("table.replication.factor", "1")
+ .Build();
+
+ fluss_test::CreateTable(adm, table_path, table_descriptor);
+ fluss_test::CreatePartitions(adm, table_path, "region", {"US", "EU"});
+
+ 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));
+
+ struct TestData {
+ std::string region;
+ int32_t a;
+ std::string b;
+ int64_t c;
+ std::string d;
+ };
+ std::vector<TestData> test_data = {
+ {"US", 1, "aaaaaaaaa", 1, "us-1"},
+ {"US", 1, "aaaaaaaaa", 2, "us-2"},
+ {"US", 2, "aaaaaaaaa", 3, "us-3"},
+ {"EU", 1, "aaaaaaaaa", 4, "eu-1"},
+ {"EU", 1, "bbbbbbbbb", 5, "eu-2"},
+ };
+
+ for (const auto& row_data : test_data) {
+ auto row = table.NewRow();
+ row.Set("region", row_data.region);
+ row.Set("a", row_data.a);
+ row.Set("b", row_data.b);
+ row.Set("c", row_data.c);
+ row.Set("d", row_data.d);
+ ASSERT_OK(upsert_writer.Upsert(row));
+ }
+ ASSERT_OK(upsert_writer.Flush());
+
+ fluss::PrefixLookuper prefix_lookuper;
+ ASSERT_OK(table.NewPrefixLookup({"region", "a", "b"}, prefix_lookuper));
+
+ // (US, 1, "aaaaaaaaa") matches 2 rows.
+ {
+ auto key = table.NewRow();
+ key.Set("region", "US");
+ key.Set("a", 1);
+ key.Set("b", "aaaaaaaaa");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ ASSERT_EQ(result.Size(), 2u);
+ for (size_t i = 0; i < result.Size(); ++i) {
+ auto row = result.GetRow(i);
+ EXPECT_EQ(row.GetString("region"), "US");
+ EXPECT_EQ(row.GetInt32("a"), 1);
+ EXPECT_EQ(row.GetString("b"), "aaaaaaaaa");
+ }
+ }
+
+ // (EU, 1, "bbbbbbbbb") matches a single row.
+ {
+ auto key = table.NewRow();
+ key.Set("region", "EU");
+ key.Set("a", 1);
+ key.Set("b", "bbbbbbbbb");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ ASSERT_EQ(result.Size(), 1u);
+ EXPECT_EQ(result.GetRow(0).GetString("d"), "eu-2");
+ }
+
+ // A non-existent partition yields an empty result.
+ {
+ auto key = table.NewRow();
+ key.Set("region", "APAC");
+ key.Set("a", 1);
+ key.Set("b", "aaaaaaaaa");
+ fluss::PrefixLookupResult result;
+ ASSERT_OK(prefix_lookuper.PrefixLookup(key, result));
+ EXPECT_TRUE(result.IsEmpty());
+ }
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
TEST_F(KvTableTest, PartialUpdate) {
auto& adm = admin();
auto& conn = connection();
diff --git a/website/docs/user-guide/cpp/api-reference.md
b/website/docs/user-guide/cpp/api-reference.md
index e9b94c9d..60a43eb4 100644
--- a/website/docs/user-guide/cpp/api-reference.md
+++ b/website/docs/user-guide/cpp/api-reference.md
@@ -116,6 +116,7 @@ Complete API reference for the Fluss C++ client.
| `NewAppend() -> TableAppend` | Create an append builder for log tables |
| `NewUpsert() -> TableUpsert` | Create an upsert builder for PK tables |
| `NewLookup() -> TableLookup` | Create a lookup builder for PK tables |
+| `NewPrefixLookup(std::vector<std::string> cols, PrefixLookuper& out) ->
Result` | Create a prefix (bucket-key) lookuper |
| `NewScan() -> TableScan` | Create a scan builder |
| `GetTableInfo() -> TableInfo` | Get table metadata |
| `GetTablePath() -> TablePath` | Get the table path |
@@ -180,6 +181,14 @@ Complete API reference for the Fluss C++ client.
|---------------------------------------------------------------|-----------------------------|
| `Lookup(const GenericRow& pk_row, LookupResult& out) -> Result` | Lookup a
row by primary key |
+## `PrefixLookuper`
+
+Performs prefix (bucket-key) lookups, returning all rows whose primary key
starts with the given prefix. Obtained from `Table::NewPrefixLookup()`. See the
[Prefix Lookup example](./example/prefix-lookup.md).
+
+| Method
| Description |
+|---------------------------------------------------------------------------------|-----------------------------------------------|
+| `PrefixLookup(const GenericRow& prefix_row, PrefixLookupResult& out) ->
Result` | Look up all rows matching the prefix columns |
+
## `LogScanner`
| Method
| Description |
@@ -410,6 +419,16 @@ Same array getters as
[`RowView`](#array-getters-index-based) — `GetArraySize`
| `GetTimestamp(const std::string& name) -> Timestamp` | Get timestamp by
column name |
| `GetDecimalString(const std::string& name) -> std::string` | Get decimal as
string by column name |
+## `PrefixLookupResult`
+
+Read-only result of a prefix lookup — zero or more matched rows. Each row is a
`PrefixRowView` exposing the same getters as [`RowView`](#rowview) (index- and
name-based, including arrays).
+
+| Method | Description
|
+|-----------------------------------------|---------------------------------------------------|
+| `Size() -> size_t` | Number of matched rows
|
+| `IsEmpty() -> bool` | Whether the result has no rows
|
+| `GetRow(size_t index) -> PrefixRowView` | Get the row at `index` (throws if
out of range) |
+
## `ArrowRecordBatch`
| Method | Description
|
diff --git a/website/docs/user-guide/cpp/example/prefix-lookup.md
b/website/docs/user-guide/cpp/example/prefix-lookup.md
new file mode 100644
index 00000000..e5d52015
--- /dev/null
+++ b/website/docs/user-guide/cpp/example/prefix-lookup.md
@@ -0,0 +1,107 @@
+---
+sidebar_position: 7
+---
+# Prefix Lookup
+
+Prefix lookup returns all rows whose primary key starts with a given prefix.
It is enabled by choosing a **bucket key that is a strict prefix of the primary
key** — rows sharing the same bucket-key prefix land in the same bucket, so one
bucket lookup returns them all.
+
+## Table Requirements
+
+- The table must have a primary key.
+- The bucket key must be a strict prefix of the primary key (on partitioned
tables, of the *non-partition* portion of the primary key).
+- The bucket key cannot equal the full primary key — that is a normal
primary-key lookup; use
[`Lookuper`](./primary-key-tables.md#looking-up-records) instead.
+- The columns passed to `NewPrefixLookup()` must equal `partition_keys ++
bucket_key` (in that order, if partitioned).
+
+`NewPrefixLookup()` validates these rules and returns a non-`Ok` `Result`
(with `error_code == ErrorCode::CLIENT_ERROR`) describing the violation.
+
+## Non-Partitioned Table
+
+Pick a schema where the bucket key is a prefix of the primary key:
+
+```cpp
+auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("user_id", fluss::DataType::Int())
+ .AddColumn("session_id", fluss::DataType::String())
+ .AddColumn("event_seq", fluss::DataType::BigInt())
+ .AddColumn("event_data", fluss::DataType::String())
+ .SetPrimaryKeys({"user_id", "session_id", "event_seq"})
+ .Build();
+
+// Bucket key (user_id, session_id) is a prefix of the primary key.
+auto descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketKeys({"user_id", "session_id"})
+ .SetBucketCount(3)
+ .Build();
+
+fluss::TablePath table_path("fluss", "sessions");
+admin.CreateTable(table_path, descriptor, true);
+```
+
+Create the lookuper with the prefix columns, then call `PrefixLookup`:
+
+```cpp
+fluss::Table table;
+conn.GetTable(table_path, table);
+
+fluss::PrefixLookuper prefix_lookuper;
+table.NewPrefixLookup({"user_id", "session_id"}, prefix_lookuper);
+
+auto prefix = table.NewRow();
+prefix.Set("user_id", 1);
+prefix.Set("session_id", "sess-a");
+
+fluss::PrefixLookupResult result;
+prefix_lookuper.PrefixLookup(prefix, result);
+
+for (size_t i = 0; i < result.Size(); ++i) {
+ auto row = result.GetRow(i);
+ std::cout << "seq=" << row.GetInt64("event_seq")
+ << ", data=" << row.GetString("event_data") << std::endl;
+}
+```
+
+Unlike primary-key lookup (which returns a single row via
`LookupResult::Found()`), prefix lookup returns zero or more rows via `Size()`
/ `GetRow(i)`, in primary-key order.
+
+## Partitioned Table
+
+On a partitioned table, the partition columns are stripped from the primary
key before the bucket-prefix rule is evaluated. The lookup key must still carry
the partition values so the client can route the request to the right partition
— so the columns passed to `NewPrefixLookup()` are `partition_keys ++
bucket_key`.
+
+```cpp
+auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("region", fluss::DataType::String())
+ .AddColumn("user_id", fluss::DataType::Int())
+ .AddColumn("session_id", fluss::DataType::String())
+ .AddColumn("event_seq", fluss::DataType::BigInt())
+ .AddColumn("event_data", fluss::DataType::String())
+ .SetPrimaryKeys({"region", "user_id", "session_id", "event_seq"})
+ .Build();
+
+auto descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetPartitionKeys({"region"})
+ // Bucket key (user_id, session_id) is a prefix of the PK minus the
partition column.
+ .SetBucketKeys({"user_id", "session_id"})
+ .SetBucketCount(3)
+ .Build();
+```
+
+```cpp
+fluss::PrefixLookuper prefix_lookuper;
+// lookup columns = partition keys ++ bucket key
+table.NewPrefixLookup({"region", "user_id", "session_id"}, prefix_lookuper);
+
+auto prefix = table.NewRow();
+prefix.Set("region", "US"); // partition column
+prefix.Set("user_id", 1);
+prefix.Set("session_id", "sess-a");
+
+fluss::PrefixLookupResult result;
+prefix_lookuper.PrefixLookup(prefix, result);
+
+for (size_t i = 0; i < result.Size(); ++i) {
+ auto row = result.GetRow(i);
+ std::cout << "seq=" << row.GetInt64("event_seq")
+ << ", data=" << row.GetString("event_data") << std::endl;
+}
+```
