You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/08/27 11:24:53 UTC
(fluss) branch main updated: [c++] Create C++ bindings for log filter pushdown (#4036)
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.git
The following commit(s) were added to refs/heads/main by this push:
new 8ae5b0717 [c++] Create C++ bindings for log filter pushdown (#4036)
8ae5b0717 is described below
commit 8ae5b0717e62a839256f16fe8fbec3e365c6b16b
Author: naivedogger <ya...@gmail.com>
AuthorDate: Thu Aug 27 19:24:48 2026 +0800
[c++] Create C++ bindings for log filter pushdown (#4036)
* [c++] Add predicate pushdown bindings
* [c++] Use filter pushdown terminology
* [c++] Simplify filter pushdown bindings
* [c++] Extend filter literal integration coverage
---
fluss-rust/bindings/cpp/README.md | 30 ++
fluss-rust/bindings/cpp/include/fluss.hpp | 120 ++++++++
fluss-rust/bindings/cpp/src/lib.rs | 312 ++++++++++++++++++++-
fluss-rust/bindings/cpp/src/table.cpp | 201 ++++++++++++-
fluss-rust/bindings/cpp/test/test_log_table.cpp | 186 ++++++++++++
fluss-rust/bindings/cpp/test/test_utils.h | 16 +-
.../website/docs/user-guide/cpp/api-reference.md | 24 ++
.../docs/user-guide/cpp/example/log-tables.md | 32 ++-
8 files changed, 914 insertions(+), 7 deletions(-)
diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md
index de8ef86b1..35c2ecd27 100644
--- a/fluss-rust/bindings/cpp/README.md
+++ b/fluss-rust/bindings/cpp/README.md
@@ -33,6 +33,36 @@ 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.
+## Log filter pushdown
+
+`TableScan::Filter()` pushes a predicate to Arrow log scans for server-side
+RecordBatch pruning:
+
+```cpp
+fluss::RecordBatchLogScanner scanner;
+auto predicate =
+ fluss::Col("amount")
+ .GreaterOrEqual(100)
+ .And(fluss::Col("region").In({"CN", "SG"}));
+
+auto result = table.NewScan()
+ .Filter(std::move(predicate))
+ .ProjectByName({"order_id", "amount"})
+ .CreateRecordBatchLogScanner(scanner);
+```
+
+Supported expressions include comparisons, `IS NULL` / `IS NOT NULL`, string
+prefix/infix/suffix matching, `IN` / `NOT IN`, and `AND` / `OR`. Scalar
+literals include booleans, integers, floating-point values, strings, bytes,
+decimals, dates, times, and timestamps.
+
+Pushdown is conservative: Fluss skips only whole RecordBatches whose statistics
+prove that they cannot match. Returned batches may still contain non-matching
+rows, so callers must evaluate the predicate again. Configure
+`table.statistics.columns` for the referenced columns; batches without usable
+statistics are retained. Filter pushdown requires the Arrow log format and does
+not apply to `CreateBucketBatchScanner()`.
+
## Examples and Documentation
- [examples/example.cpp](examples/example.cpp) demonstrates log-table writes, continuous scans,
diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp
index 3b8d27604..08f054ddc 100644
--- a/fluss-rust/bindings/cpp/include/fluss.hpp
+++ b/fluss-rust/bindings/cpp/include/fluss.hpp
@@ -27,6 +27,7 @@
#include <stdexcept>
#include <string>
#include <string_view>
+#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>
@@ -59,6 +60,8 @@ struct PrefixLookupResultInner;
struct ArrayWriterInner;
struct MapWriterInner;
struct ValueInner;
+enum class FfiPredicateLiteralType : int32_t;
+enum class FfiPredicateLeafFunction : int32_t;
} // namespace ffi
/// Named constants for Fluss API error codes.
@@ -262,6 +265,116 @@ struct Timestamp {
}
};
+/// Scalar literal used by a log-scan predicate.
+///
+/// Integer literals are coerced to the scanned column's integer type by the
+/// Rust client, with range checks. Decimal and timestamp literals use the
+/// explicit factories below to preserve their Fluss logical type.
+class PredicateLiteral {
+ public:
+ PredicateLiteral(bool value);
+ PredicateLiteral(int32_t value);
+ PredicateLiteral(int64_t value);
+ template <typename T, std::enable_if_t<std::is_integral_v<std::decay_t<T>> &&
+ !std::is_same_v<std::decay_t<T>, bool> &&
+ !std::is_same_v<std::decay_t<T>, int32_t> &&
+ !std::is_same_v<std::decay_t<T>, int64_t>,
+ int> = 0>
+ PredicateLiteral(T value) : PredicateLiteral(ToInt64(value)) {}
+ PredicateLiteral(float value);
+ PredicateLiteral(double value);
+ PredicateLiteral(const char* value);
+ PredicateLiteral(std::string value);
+ PredicateLiteral(std::vector<uint8_t> value);
+ PredicateLiteral(Date value);
+ PredicateLiteral(Time value);
+
+ static PredicateLiteral Null();
+ static PredicateLiteral Decimal(std::string value);
+ static PredicateLiteral TimestampNtz(Timestamp value);
+ static PredicateLiteral TimestampLtz(Timestamp value);
+
+ private:
+ explicit PredicateLiteral(ffi::FfiPredicateLiteralType literal_type);
+
+ template <typename T>
+ static int64_t ToInt64(T value) {
+ using ValueType = std::decay_t<T>;
+ static_assert(sizeof(ValueType) <= sizeof(int64_t), "Integer literal is wider than INT64");
+ if constexpr (std::is_unsigned_v<ValueType>) {
+ if (value >
+ static_cast<std::make_unsigned_t<int64_t>>(std::numeric_limits<int64_t>::max())) {
+ throw std::out_of_range("Unsigned predicate literal does not fit INT64");
+ }
+ }
+ return static_cast<int64_t>(value);
+ }
+
+ ffi::FfiPredicateLiteralType literal_type_;
+ bool boolean_value_{false};
+ int64_t integer_value_{0};
+ double floating_value_{0};
+ std::string string_value_;
+ std::vector<uint8_t> bytes_value_;
+ Timestamp timestamp_value_;
+
+ friend class Predicate;
+ friend class TableScan;
+};
+
+/// Filter expression for server-side Arrow log RecordBatch pruning.
+///
+/// Filter pushdown is conservative: a returned RecordBatch may still
+/// contain non-matching rows, so callers must evaluate the predicate again.
+class Predicate {
+ public:
+ Predicate(const Predicate&) = default;
+ Predicate& operator=(const Predicate&) = default;
+ Predicate(Predicate&&) noexcept = default;
+ Predicate& operator=(Predicate&&) noexcept = default;
+
+ Predicate And(Predicate other) const;
+ Predicate Or(Predicate other) const;
+
+ private:
+ struct Node;
+
+ explicit Predicate(std::shared_ptr<const Node> root);
+
+ std::shared_ptr<const Node> root_;
+
+ friend class ColumnRef;
+ friend class TableScan;
+};
+
+/// Column reference used to build a Predicate.
+class ColumnRef {
+ public:
+ explicit ColumnRef(std::string name) : name_(std::move(name)) {}
+
+ Predicate Equal(PredicateLiteral value) const;
+ Predicate NotEqual(PredicateLiteral value) const;
+ Predicate LessThan(PredicateLiteral value) const;
+ Predicate LessOrEqual(PredicateLiteral value) const;
+ Predicate GreaterThan(PredicateLiteral value) const;
+ Predicate GreaterOrEqual(PredicateLiteral value) const;
+ Predicate IsNull() const;
+ Predicate IsNotNull() const;
+ Predicate StartsWith(std::string prefix) const;
+ Predicate Contains(std::string infix) const;
+ Predicate EndsWith(std::string suffix) const;
+ Predicate In(std::vector<PredicateLiteral> values) const;
+ Predicate NotIn(std::vector<PredicateLiteral> values) const;
+
+ private:
+ Predicate Leaf(ffi::FfiPredicateLeafFunction function,
+ std::vector<PredicateLiteral> literals) const;
+
+ std::string name_;
+};
+
+inline ColumnRef Col(std::string name) { return ColumnRef(std::move(name)); }
+
enum class ChangeType {
AppendOnly = 0,
Insert = 1,
@@ -1664,6 +1777,12 @@ class TableScan {
TableScan& ProjectByIndex(std::vector<size_t> column_indices);
TableScan& ProjectByName(std::vector<std::string> column_names);
+ /// Pushes a predicate down for conservative server-side RecordBatch pruning.
+ ///
+ /// Only Arrow log scans support this. Returned batches may contain
+ /// non-matching rows and must be filtered again by the caller.
+ TableScan& Filter(Predicate predicate);
+
TableScan& Limit(int32_t row_number);
/// Creates a record-mode log scanner, polled for individual `ScanRecord`s.
@@ -1713,6 +1832,7 @@ class TableScan {
ffi::Table* table_{nullptr};
std::vector<size_t> projection_;
std::vector<std::string> name_projection_;
+ std::optional<Predicate> predicate_;
std::optional<int32_t> limit_;
};
diff --git a/fluss-rust/bindings/cpp/src/lib.rs b/fluss-rust/bindings/cpp/src/lib.rs
index 6f4f824f1..704644f26 100644
--- a/fluss-rust/bindings/cpp/src/lib.rs
+++ b/fluss-rust/bindings/cpp/src/lib.rs
@@ -110,6 +110,77 @@ mod ffi {
child_count: u32,
}
+ #[repr(i32)]
+ enum FfiPredicateNodeType {
+ Leaf = 0,
+ Compound = 1,
+ }
+
+ #[repr(i32)]
+ enum FfiPredicateLeafFunction {
+ Equal = 0,
+ NotEqual = 1,
+ LessThan = 2,
+ LessOrEqual = 3,
+ GreaterThan = 4,
+ GreaterOrEqual = 5,
+ IsNull = 6,
+ IsNotNull = 7,
+ StartsWith = 8,
+ Contains = 9,
+ EndsWith = 10,
+ In = 11,
+ NotIn = 12,
+ }
+
+ #[repr(i32)]
+ enum FfiPredicateCompoundFunction {
+ And = 0,
+ Or = 1,
+ }
+
+ #[repr(i32)]
+ enum FfiPredicateLiteralType {
+ Null = 0,
+ Boolean = 1,
+ Int32 = 2,
+ Int64 = 3,
+ Float32 = 4,
+ Float64 = 5,
+ String = 6,
+ Bytes = 7,
+ Decimal = 8,
+ Date = 9,
+ Time = 10,
+ TimestampNtz = 11,
+ TimestampLtz = 12,
+ }
+
+ // One scalar literal in a predicate leaf. `literal_type` is private to the
+ // C++ binding and decoded into fluss::predicate::Literal before the scan is
+ // created.
+ struct FfiPredicateLiteral {
+ literal_type: FfiPredicateLiteralType,
+ boolean_value: bool,
+ integer_value: i64,
+ floating_value: f64,
+ string_value: String,
+ bytes_value: Vec<u8>,
+ timestamp_millis: i64,
+ timestamp_nanos: i32,
+ }
+
+ // Predicate tree serialized in preorder. A leaf has child_count == 0 and
+ // carries field/literals; a compound node is followed by child_count nodes.
+ struct FfiPredicateNode {
+ node_type: FfiPredicateNodeType,
+ leaf_function: FfiPredicateLeafFunction,
+ compound_function: FfiPredicateCompoundFunction,
+ field: String,
+ literals: Vec<FfiPredicateLiteral>,
+ child_count: u32,
+ }
+
struct FfiColumn {
name: String,
comment: String,
@@ -426,7 +497,12 @@ mod ffi {
// Table
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_scanner(
+ self: &Table,
+ column_indices: Vec<usize>,
+ predicate_nodes: Vec<FfiPredicateNode>,
+ batch: bool,
+ ) -> FfiPtrResult;
fn create_bucket_batch_scanner(
self: &Table,
column_indices: Vec<usize>,
@@ -1577,6 +1653,224 @@ impl Admin {
}
}
+fn predicate_from_ffi_nodes(
+ nodes: &[ffi::FfiPredicateNode],
+ row_type: &fcore::metadata::RowType,
+) -> Result<Option<fcore::predicate::Predicate>, String> {
+ if nodes.is_empty() {
+ return Ok(None);
+ }
+
+ let mut next = 0;
+ let predicate = predicate_from_ffi_node(nodes, &mut next, row_type)?;
+ if next != nodes.len() {
+ return Err(format!(
+ "Predicate contains {} trailing node(s)",
+ nodes.len() - next
+ ));
+ }
+ Ok(Some(predicate))
+}
+
+fn predicate_from_ffi_node(
+ nodes: &[ffi::FfiPredicateNode],
+ next: &mut usize,
+ row_type: &fcore::metadata::RowType,
+) -> Result<fcore::predicate::Predicate, String> {
+ let node_index = *next;
+ let node = nodes
+ .get(node_index)
+ .ok_or_else(|| "Predicate tree ended before all children were decoded".to_string())?;
+ *next += 1;
+
+ match node.node_type {
+ ffi::FfiPredicateNodeType::Leaf => {
+ if node.child_count != 0 {
+ return Err(format!(
+ "Predicate leaf at node {node_index} has {} child nodes",
+ node.child_count
+ ));
+ }
+ if node.field.is_empty() {
+ return Err(format!(
+ "Predicate leaf at node {node_index} has an empty field name"
+ ));
+ }
+
+ let field = row_type
+ .fields()
+ .iter()
+ .find(|field| field.name() == node.field)
+ .ok_or_else(|| {
+ format!(
+ "Filter column '{}' does not exist in the table schema",
+ node.field
+ )
+ })?;
+ let function = match node.leaf_function {
+ ffi::FfiPredicateLeafFunction::Equal => fcore::predicate::LeafFunction::Equal,
+ ffi::FfiPredicateLeafFunction::NotEqual => fcore::predicate::LeafFunction::NotEqual,
+ ffi::FfiPredicateLeafFunction::LessThan => fcore::predicate::LeafFunction::LessThan,
+ ffi::FfiPredicateLeafFunction::LessOrEqual => {
+ fcore::predicate::LeafFunction::LessOrEqual
+ }
+ ffi::FfiPredicateLeafFunction::GreaterThan => {
+ fcore::predicate::LeafFunction::GreaterThan
+ }
+ ffi::FfiPredicateLeafFunction::GreaterOrEqual => {
+ fcore::predicate::LeafFunction::GreaterOrEqual
+ }
+ ffi::FfiPredicateLeafFunction::IsNull => fcore::predicate::LeafFunction::IsNull,
+ ffi::FfiPredicateLeafFunction::IsNotNull => {
+ fcore::predicate::LeafFunction::IsNotNull
+ }
+ ffi::FfiPredicateLeafFunction::StartsWith => {
+ fcore::predicate::LeafFunction::StartsWith
+ }
+ ffi::FfiPredicateLeafFunction::Contains => fcore::predicate::LeafFunction::Contains,
+ ffi::FfiPredicateLeafFunction::EndsWith => fcore::predicate::LeafFunction::EndsWith,
+ ffi::FfiPredicateLeafFunction::In => fcore::predicate::LeafFunction::In,
+ ffi::FfiPredicateLeafFunction::NotIn => fcore::predicate::LeafFunction::NotIn,
+ other => {
+ return Err(format!(
+ "Predicate leaf at node {node_index} has unknown function {}",
+ other.repr
+ ));
+ }
+ };
+ let literals = node
+ .literals
+ .iter()
+ .map(|literal| predicate_literal_from_ffi(literal, field))
+ .collect::<Result<Vec<_>, _>>()?;
+
+ Ok(fcore::predicate::Predicate::Leaf {
+ field: node.field.to_string(),
+ function,
+ literals,
+ })
+ }
+ ffi::FfiPredicateNodeType::Compound => {
+ if !node.field.is_empty() || !node.literals.is_empty() {
+ return Err(format!(
+ "Predicate compound node {node_index} unexpectedly carries leaf data"
+ ));
+ }
+ let function = match node.compound_function {
+ ffi::FfiPredicateCompoundFunction::And => fcore::predicate::CompoundFunction::And,
+ ffi::FfiPredicateCompoundFunction::Or => fcore::predicate::CompoundFunction::Or,
+ other => {
+ return Err(format!(
+ "Predicate compound node {node_index} has unknown function {}",
+ other.repr
+ ));
+ }
+ };
+ let mut children = Vec::with_capacity(node.child_count as usize);
+ for _ in 0..node.child_count {
+ children.push(predicate_from_ffi_node(nodes, next, row_type)?);
+ }
+ Ok(fcore::predicate::Predicate::Compound { function, children })
+ }
+ other => Err(format!(
+ "Predicate node {node_index} has unknown node type {}",
+ other.repr
+ )),
+ }
+}
+
+fn predicate_literal_from_ffi(
+ literal: &ffi::FfiPredicateLiteral,
+ field: &fcore::metadata::DataField,
+) -> Result<fcore::predicate::Literal, String> {
+ use fcore::predicate::Literal;
+
+ match literal.literal_type {
+ ffi::FfiPredicateLiteralType::Null => Ok(Literal::Null),
+ ffi::FfiPredicateLiteralType::Boolean => Ok(Literal::Bool(literal.boolean_value)),
+ ffi::FfiPredicateLiteralType::Int32 => {
+ let value = i32::try_from(literal.integer_value).map_err(|_| {
+ format!(
+ "Filter literal {} does not fit INT32",
+ literal.integer_value
+ )
+ })?;
+ Ok(Literal::Int32(value))
+ }
+ ffi::FfiPredicateLiteralType::Int64 => Ok(Literal::Int64(literal.integer_value)),
+ ffi::FfiPredicateLiteralType::Float32 => {
+ Ok(Literal::Float32(literal.floating_value as f32))
+ }
+ ffi::FfiPredicateLiteralType::Float64 => Ok(Literal::Float64(literal.floating_value)),
+ ffi::FfiPredicateLiteralType::String => {
+ Ok(Literal::String(literal.string_value.to_string()))
+ }
+ ffi::FfiPredicateLiteralType::Bytes => Ok(Literal::Bytes(literal.bytes_value.clone())),
+ ffi::FfiPredicateLiteralType::Decimal => {
+ let decimal_type = match field.data_type() {
+ fcore::metadata::DataType::Decimal(decimal_type) => decimal_type,
+ other => {
+ return Err(format!(
+ "Decimal predicate literal cannot be used with column '{}' of type {other}",
+ field.name()
+ ));
+ }
+ };
+ let value = bigdecimal::BigDecimal::from_str(&literal.string_value)
+ .map_err(|e| format!("Invalid decimal predicate literal: {e}"))?;
+ let decimal = fcore::row::Decimal::from_big_decimal(
+ value.clone(),
+ decimal_type.precision(),
+ decimal_type.scale(),
+ )
+ .map_err(|e| e.to_string())?;
+ if decimal.to_big_decimal() != value {
+ return Err(format!(
+ "Decimal predicate literal '{}' cannot be represented exactly by column '{}'",
+ literal.string_value,
+ field.name()
+ ));
+ }
+ Ok(Literal::Decimal(decimal))
+ }
+ ffi::FfiPredicateLiteralType::Date => {
+ let days = i32::try_from(literal.integer_value).map_err(|_| {
+ format!(
+ "Date predicate literal {} does not fit INT32",
+ literal.integer_value
+ )
+ })?;
+ Ok(Literal::Date(days))
+ }
+ ffi::FfiPredicateLiteralType::Time => {
+ let millis = i32::try_from(literal.integer_value).map_err(|_| {
+ format!(
+ "Time predicate literal {} does not fit INT32",
+ literal.integer_value
+ )
+ })?;
+ Ok(Literal::Time(millis))
+ }
+ ffi::FfiPredicateLiteralType::TimestampNtz => {
+ let timestamp = fcore::row::TimestampNtz::from_millis_nanos(
+ literal.timestamp_millis,
+ literal.timestamp_nanos,
+ )
+ .map_err(|e| e.to_string())?;
+ Ok(Literal::TimestampNtz(timestamp))
+ }
+ ffi::FfiPredicateLiteralType::TimestampLtz => {
+ let timestamp = fcore::row::TimestampLtz::from_millis_nanos(
+ literal.timestamp_millis,
+ literal.timestamp_nanos,
+ )
+ .map_err(|e| e.to_string())?;
+ Ok(Literal::TimestampLtz(timestamp))
+ }
+ other => Err(format!("Unknown predicate literal type {}", other.repr)),
+ }
+}
+
// Table implementation
unsafe fn delete_table(table: *mut Table) {
if !table.is_null() {
@@ -1633,10 +1927,24 @@ impl Table {
ok_ptr(ptr as usize)
}
- fn create_scanner(&self, column_indices: Vec<usize>, batch: bool) -> ffi::FfiPtrResult {
+ fn create_scanner(
+ &self,
+ column_indices: Vec<usize>,
+ predicate_nodes: Vec<ffi::FfiPredicateNode>,
+ batch: bool,
+ ) -> ffi::FfiPtrResult {
RUNTIME.block_on(async {
let fluss_table = self.fluss_table();
let scan = fluss_table.new_scan();
+ let scan =
+ match predicate_from_ffi_nodes(&predicate_nodes, self.table_info.get_row_type()) {
+ Ok(Some(predicate)) => match scan.filter(predicate) {
+ Ok(scan) => scan,
+ Err(e) => return err_ptr_from_core(&e),
+ },
+ Ok(None) => scan,
+ Err(e) => return client_err_ptr(e),
+ };
let (projected_columns, scan) = if column_indices.is_empty() {
(self.table_info.get_schema().columns().to_vec(), scan)
diff --git a/fluss-rust/bindings/cpp/src/table.cpp b/fluss-rust/bindings/cpp/src/table.cpp
index 5d21fa799..033414f06 100644
--- a/fluss-rust/bindings/cpp/src/table.cpp
+++ b/fluss-rust/bindings/cpp/src/table.cpp
@@ -21,6 +21,7 @@
#include <cassert>
#include <ctime>
+#include <functional>
#include "ffi_converter.hpp"
#include "fluss.hpp"
@@ -82,6 +83,160 @@ int Date::Day() const {
return tm.tm_mday;
}
+PredicateLiteral::PredicateLiteral(bool value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Boolean), boolean_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(int32_t value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Int32), integer_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(int64_t value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Int64), integer_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(float value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Float32), floating_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(double value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Float64), floating_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(const char* value) : PredicateLiteral(std::string(value)) {}
+
+PredicateLiteral::PredicateLiteral(std::string value)
+ : literal_type_(ffi::FfiPredicateLiteralType::String), string_value_(std::move(value)) {}
+
+PredicateLiteral::PredicateLiteral(std::vector<uint8_t> value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Bytes), bytes_value_(std::move(value)) {}
+
+PredicateLiteral::PredicateLiteral(Date value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Date), integer_value_(value.days_since_epoch) {}
+
+PredicateLiteral::PredicateLiteral(Time value)
+ : literal_type_(ffi::FfiPredicateLiteralType::Time),
+ integer_value_(value.millis_since_midnight) {}
+
+PredicateLiteral::PredicateLiteral(ffi::FfiPredicateLiteralType literal_type)
+ : literal_type_(literal_type) {}
+
+PredicateLiteral PredicateLiteral::Null() {
+ return PredicateLiteral(ffi::FfiPredicateLiteralType::Null);
+}
+
+PredicateLiteral PredicateLiteral::Decimal(std::string value) {
+ PredicateLiteral literal(ffi::FfiPredicateLiteralType::Decimal);
+ literal.string_value_ = std::move(value);
+ return literal;
+}
+
+PredicateLiteral PredicateLiteral::TimestampNtz(Timestamp value) {
+ PredicateLiteral literal(ffi::FfiPredicateLiteralType::TimestampNtz);
+ literal.timestamp_value_ = value;
+ return literal;
+}
+
+PredicateLiteral PredicateLiteral::TimestampLtz(Timestamp value) {
+ PredicateLiteral literal(ffi::FfiPredicateLiteralType::TimestampLtz);
+ literal.timestamp_value_ = value;
+ return literal;
+}
+
+struct Predicate::Node {
+ ffi::FfiPredicateNodeType node_type{ffi::FfiPredicateNodeType::Leaf};
+ ffi::FfiPredicateLeafFunction leaf_function{ffi::FfiPredicateLeafFunction::Equal};
+ ffi::FfiPredicateCompoundFunction compound_function{ffi::FfiPredicateCompoundFunction::And};
+ std::string field;
+ std::vector<PredicateLiteral> literals;
+ std::vector<std::shared_ptr<const Node>> children;
+};
+
+Predicate::Predicate(std::shared_ptr<const Node> root) : root_(std::move(root)) {}
+
+Predicate Predicate::And(Predicate other) const {
+ auto node = std::make_shared<Node>();
+ node->node_type = ffi::FfiPredicateNodeType::Compound;
+ node->compound_function = ffi::FfiPredicateCompoundFunction::And;
+ if (root_->node_type == ffi::FfiPredicateNodeType::Compound &&
+ root_->compound_function == node->compound_function) {
+ node->children = root_->children;
+ } else {
+ node->children.push_back(root_);
+ }
+ node->children.push_back(std::move(other.root_));
+ return Predicate(std::move(node));
+}
+
+Predicate Predicate::Or(Predicate other) const {
+ auto node = std::make_shared<Node>();
+ node->node_type = ffi::FfiPredicateNodeType::Compound;
+ node->compound_function = ffi::FfiPredicateCompoundFunction::Or;
+ if (root_->node_type == ffi::FfiPredicateNodeType::Compound &&
+ root_->compound_function == node->compound_function) {
+ node->children = root_->children;
+ } else {
+ node->children.push_back(root_);
+ }
+ node->children.push_back(std::move(other.root_));
+ return Predicate(std::move(node));
+}
+
+Predicate ColumnRef::Leaf(ffi::FfiPredicateLeafFunction function,
+ std::vector<PredicateLiteral> literals) const {
+ auto node = std::make_shared<Predicate::Node>();
+ node->node_type = ffi::FfiPredicateNodeType::Leaf;
+ node->leaf_function = function;
+ node->field = name_;
+ node->literals = std::move(literals);
+ return Predicate(std::move(node));
+}
+
+Predicate ColumnRef::Equal(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::Equal, {std::move(value)});
+}
+
+Predicate ColumnRef::NotEqual(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::NotEqual, {std::move(value)});
+}
+
+Predicate ColumnRef::LessThan(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::LessThan, {std::move(value)});
+}
+
+Predicate ColumnRef::LessOrEqual(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::LessOrEqual, {std::move(value)});
+}
+
+Predicate ColumnRef::GreaterThan(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::GreaterThan, {std::move(value)});
+}
+
+Predicate ColumnRef::GreaterOrEqual(PredicateLiteral value) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::GreaterOrEqual, {std::move(value)});
+}
+
+Predicate ColumnRef::IsNull() const { return Leaf(ffi::FfiPredicateLeafFunction::IsNull, {}); }
+
+Predicate ColumnRef::IsNotNull() const {
+ return Leaf(ffi::FfiPredicateLeafFunction::IsNotNull, {});
+}
+
+Predicate ColumnRef::StartsWith(std::string prefix) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::StartsWith, {PredicateLiteral(std::move(prefix))});
+}
+
+Predicate ColumnRef::Contains(std::string infix) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::Contains, {PredicateLiteral(std::move(infix))});
+}
+
+Predicate ColumnRef::EndsWith(std::string suffix) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::EndsWith, {PredicateLiteral(std::move(suffix))});
+}
+
+Predicate ColumnRef::In(std::vector<PredicateLiteral> values) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::In, std::move(values));
+}
+
+Predicate ColumnRef::NotIn(std::vector<PredicateLiteral> values) const {
+ return Leaf(ffi::FfiPredicateLeafFunction::NotIn, std::move(values));
+}
+
// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
#define CHECK_INNER(name) \
do { \
@@ -1222,6 +1377,11 @@ TableScan& TableScan::ProjectByName(std::vector<std::string> column_names) {
return *this;
}
+TableScan& TableScan::Filter(Predicate predicate) {
+ predicate_ = std::move(predicate);
+ return *this;
+}
+
std::vector<size_t> TableScan::ResolveNameProjection() const {
auto ffi_info = table_->get_table_info_from_table();
const auto& columns = ffi_info.schema.columns;
@@ -1274,7 +1434,43 @@ Result TableScan::DoCreateScanner(LogScanner& out, bool is_record_batch) {
for (size_t idx : resolved_indices) {
rust_indices.push_back(idx);
}
- auto ffi_result = table_->create_scanner(std::move(rust_indices), is_record_batch);
+
+ rust::Vec<ffi::FfiPredicateNode> rust_predicate;
+ if (predicate_.has_value()) {
+ std::function<void(const std::shared_ptr<const Predicate::Node>&)> append_node;
+ append_node = [&](const std::shared_ptr<const Predicate::Node>& node) {
+ ffi::FfiPredicateNode ffi_node;
+ ffi_node.node_type = node->node_type;
+ ffi_node.leaf_function = node->leaf_function;
+ ffi_node.compound_function = node->compound_function;
+ ffi_node.field = rust::String(node->field);
+ ffi_node.child_count = static_cast<uint32_t>(node->children.size());
+
+ for (const auto& literal : node->literals) {
+ ffi::FfiPredicateLiteral ffi_literal;
+ ffi_literal.literal_type = literal.literal_type_;
+ ffi_literal.boolean_value = literal.boolean_value_;
+ ffi_literal.integer_value = literal.integer_value_;
+ ffi_literal.floating_value = literal.floating_value_;
+ ffi_literal.string_value = rust::String(literal.string_value_);
+ for (uint8_t value : literal.bytes_value_) {
+ ffi_literal.bytes_value.push_back(value);
+ }
+ ffi_literal.timestamp_millis = literal.timestamp_value_.epoch_millis;
+ ffi_literal.timestamp_nanos = literal.timestamp_value_.nano_of_millisecond;
+ ffi_node.literals.push_back(std::move(ffi_literal));
+ }
+
+ rust_predicate.push_back(std::move(ffi_node));
+ for (const auto& child : node->children) {
+ append_node(child);
+ }
+ };
+ append_node(predicate_->root_);
+ }
+
+ auto ffi_result = table_->create_scanner(std::move(rust_indices), std::move(rust_predicate),
+ is_record_batch);
auto result = utils::from_ffi_result(ffi_result.result);
if (result.Ok()) {
out.scanner_ = utils::ptr_from_ffi<ffi::LogScanner>(ffi_result);
@@ -1317,6 +1513,9 @@ Result TableScan::CreateBucketBatchScanner(const TableBucket& bucket, BatchScann
if (table_ == nullptr) {
return utils::make_client_error("Table not available");
}
+ if (predicate_.has_value()) {
+ return utils::make_client_error("CreateBucketBatchScanner doesn't support filter pushdown");
+ }
if (!limit_.has_value()) {
return utils::make_client_error(
"CreateBucketBatchScanner requires a limit set via Limit()");
diff --git a/fluss-rust/bindings/cpp/test/test_log_table.cpp b/fluss-rust/bindings/cpp/test/test_log_table.cpp
index 9477f308f..5f12ecf28 100644
--- a/fluss-rust/bindings/cpp/test/test_log_table.cpp
+++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp
@@ -645,6 +645,14 @@ TEST_F(LogTableTest, LimitScanErrors) {
fluss::LogScanner s2;
EXPECT_FALSE(table.NewScan().Limit(5).CreateRecordBatchLogScanner(s2).Ok());
}
+ {
+ fluss::BatchScanner s;
+ EXPECT_FALSE(table.NewScan()
+ .Filter(fluss::Col("c1").GreaterThan(0))
+ .Limit(1)
+ .CreateBucketBatchScanner(fluss::TableBucket{table_id, 0}, s)
+ .Ok());
+ }
ASSERT_OK(adm.DropTable(table_path, false));
// A non-ARROW (INDEXED) log table rejects a limit scan.
@@ -992,6 +1000,184 @@ TEST_F(LogTableTest, TestPollBatches) {
ASSERT_OK(adm.DropTable(table_path, false));
}
+TEST_F(LogTableTest, FilterPushdownWithProjection) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_filter_pushdown_cpp");
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("id", DataType::Int())
+ .AddColumn("name", DataType::String())
+ .Build();
+ auto table_descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketCount(1)
+ .SetBucketKeys({"id"})
+ .SetProperty("table.replication.factor", "1")
+ .SetProperty("table.statistics.columns", "id,name")
+ .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));
+
+ auto make_batch = [](std::vector<int32_t> ids, std::vector<std::string> names) {
+ arrow::Int32Builder id_builder;
+ id_builder.AppendValues(ids).ok();
+ arrow::StringBuilder name_builder;
+ name_builder.AppendValues(names).ok();
+ return arrow::RecordBatch::Make(
+ arrow::schema(
+ {arrow::field("id", arrow::int32()), arrow::field("name", arrow::utf8())}),
+ static_cast<int64_t>(ids.size()),
+ {id_builder.Finish().ValueOrDie(), name_builder.Finish().ValueOrDie()});
+ };
+
+ ASSERT_OK(writer.AppendArrowBatch(make_batch({1, 2}, {"low-1", "low-2"})));
+ ASSERT_OK(writer.Flush());
+ ASSERT_OK(writer.AppendArrowBatch(make_batch({6, 7}, {"high-6", "high-7"})));
+ ASSERT_OK(writer.Flush());
+
+ fluss::RecordBatchLogScanner scanner;
+ ASSERT_OK(
+ table.NewScan()
+ .Filter(fluss::Col("id").GreaterThan(5).And(fluss::Col("name").StartsWith("high")))
+ .ProjectByName({"name"})
+ .CreateRecordBatchLogScanner(scanner));
+ ASSERT_OK(scanner.Subscribe(0, fluss::EARLIEST_OFFSET));
+
+ auto extract_names = [](const fluss::ArrowRecordBatches& batches) {
+ std::vector<std::string> names;
+ for (const auto& batch : batches) {
+ auto array = std::static_pointer_cast<arrow::StringArray>(
+ batch->GetArrowRecordBatch()->column(0));
+ for (int64_t i = 0; i < array->length(); ++i) {
+ names.push_back(array->GetString(i));
+ }
+ }
+ return names;
+ };
+
+ std::vector<std::string> names;
+ fluss_test::PollRecordBatches(scanner, 2, extract_names, names);
+ EXPECT_EQ(names, (std::vector<std::string>{"high-6", "high-7"}));
+
+ fluss::RecordBatchLogScanner invalid_scanner;
+ auto invalid_result = table.NewScan()
+ .Filter(fluss::Col("missing").Equal(1))
+ .CreateRecordBatchLogScanner(invalid_scanner);
+ EXPECT_FALSE(invalid_result.Ok());
+ EXPECT_NE(invalid_result.error_message.find("missing"), std::string::npos);
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
+TEST_F(LogTableTest, FilterPushdownLiteralTypes) {
+ auto& adm = admin();
+ auto& conn = connection();
+
+ fluss::TablePath table_path("fluss", "test_filter_pushdown_literals_cpp");
+ auto schema = fluss::Schema::NewBuilder()
+ .AddColumn("id", DataType::Int())
+ .AddColumn("amount", DataType::Decimal(10, 2))
+ .AddColumn("event_time", DataType::Timestamp(9))
+ .AddColumn("event_time_ltz", DataType::TimestampLtz(9))
+ .Build();
+ auto table_descriptor =
+ fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetBucketCount(1)
+ .SetBucketKeys({"id"})
+ .SetProperty("table.replication.factor", "1")
+ .SetProperty("table.statistics.columns", "id,amount,event_time,event_time_ltz")
+ .Build();
+ fluss_test::CreateTable(adm, table_path, table_descriptor);
+
+ fluss::Table table;
+ ASSERT_OK(conn.GetTable(table_path, table));
+
+ const auto boundary = fluss::Timestamp::FromMillisNanos(1769163227123, 456000);
+ const auto before_boundary = fluss::Timestamp::FromMillisNanos(1769163227000, 123000);
+ const auto after_boundary = fluss::Timestamp::FromMillisNanos(1769163228000, 789000);
+
+ fluss::AppendWriter writer;
+ ASSERT_OK(table.NewAppend().CreateWriter(writer));
+ {
+ fluss::GenericRow row(4);
+ row.SetInt32(0, 1);
+ row.SetDecimal(1, "10.00");
+ row.SetTimestampNtz(2, before_boundary);
+ row.SetTimestampLtz(3, after_boundary);
+ ASSERT_OK(writer.Append(row));
+ ASSERT_OK(writer.Flush());
+ }
+ {
+ fluss::GenericRow row(4);
+ row.SetInt32(0, 2);
+ row.SetDecimal(1, "12.34");
+ row.SetTimestampNtz(2, after_boundary);
+ row.SetTimestampLtz(3, before_boundary);
+ ASSERT_OK(writer.Append(row));
+ ASSERT_OK(writer.Flush());
+ }
+
+ auto extract_ids = [](const fluss::ArrowRecordBatches& batches) {
+ std::vector<int32_t> ids;
+ for (const auto& batch : batches) {
+ auto array = std::static_pointer_cast<arrow::Int32Array>(
+ batch->GetArrowRecordBatch()->column(0));
+ for (int64_t i = 0; i < array->length(); ++i) {
+ ids.push_back(array->Value(i));
+ }
+ }
+ return ids;
+ };
+
+ auto expect_only_second_row = [&](fluss::Predicate predicate) {
+ fluss::RecordBatchLogScanner scanner;
+ ASSERT_OK(table.NewScan()
+ .Filter(std::move(predicate))
+ .ProjectByName({"id"})
+ .CreateRecordBatchLogScanner(scanner));
+ ASSERT_OK(scanner.Subscribe(0, fluss::EARLIEST_OFFSET));
+
+ std::vector<int32_t> ids;
+ fluss_test::PollRecordBatches(scanner, 1, extract_ids, ids);
+ EXPECT_EQ(ids, (std::vector<int32_t>{2}));
+ };
+
+ expect_only_second_row(fluss::Col("amount").Equal(fluss::PredicateLiteral::Decimal("12.34")));
+ expect_only_second_row(
+ fluss::Col("event_time").GreaterOrEqual(fluss::PredicateLiteral::TimestampNtz(boundary)));
+ expect_only_second_row(
+ fluss::Col("event_time_ltz").LessOrEqual(fluss::PredicateLiteral::TimestampLtz(boundary)));
+ expect_only_second_row(fluss::Col("id").Equal(2L).Or(fluss::Col("id").Equal(2u)));
+
+ fluss::RecordBatchLogScanner decimal_scanner;
+ auto decimal_result =
+ table.NewScan()
+ .Filter(fluss::Col("amount").Equal(fluss::PredicateLiteral::Decimal("12.345")))
+ .CreateRecordBatchLogScanner(decimal_scanner);
+ EXPECT_FALSE(decimal_result.Ok());
+ EXPECT_NE(decimal_result.error_message.find("cannot be represented exactly"),
+ std::string::npos);
+
+ fluss::RecordBatchLogScanner timestamp_scanner;
+ auto timestamp_result =
+ table.NewScan()
+ .Filter(
+ fluss::Col("event_time_ltz").Equal(fluss::PredicateLiteral::TimestampNtz(boundary)))
+ .CreateRecordBatchLogScanner(timestamp_scanner);
+ EXPECT_FALSE(timestamp_result.Ok());
+ EXPECT_NE(timestamp_result.error_message.find("does not match"), std::string::npos);
+
+ EXPECT_THROW(fluss::Col("id").Equal(std::numeric_limits<uint64_t>::max()), std::out_of_range);
+
+ ASSERT_OK(adm.DropTable(table_path, false));
+}
+
TEST_F(LogTableTest, AllSupportedDatatypes) {
auto& adm = admin();
auto& conn = connection();
diff --git a/fluss-rust/bindings/cpp/test/test_utils.h b/fluss-rust/bindings/cpp/test/test_utils.h
index 5d40afbaf..9549c2140 100644
--- a/fluss-rust/bindings/cpp/test/test_utils.h
+++ b/fluss-rust/bindings/cpp/test/test_utils.h
@@ -224,13 +224,23 @@ void PollRecords(fluss::LogScanner& scanner, size_t expected_count, ExtractFn ex
}
}
-template <typename T, typename ExtractFn>
-void PollRecordBatches(fluss::LogScanner& scanner, size_t expected_count, ExtractFn extract_fn,
+inline fluss::Result PollRecordBatch(fluss::LogScanner& scanner, int64_t timeout_ms,
+ fluss::ArrowRecordBatches& out) {
+ return scanner.PollRecordBatch(timeout_ms, out);
+}
+
+inline fluss::Result PollRecordBatch(fluss::RecordBatchLogScanner& scanner, int64_t timeout_ms,
+ fluss::ArrowRecordBatches& out) {
+ return scanner.Poll(timeout_ms, out);
+}
+
+template <typename Scanner, typename T, typename ExtractFn>
+void PollRecordBatches(Scanner& scanner, size_t expected_count, ExtractFn extract_fn,
std::vector<T>& out) {
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
while (out.size() < expected_count && std::chrono::steady_clock::now() < deadline) {
fluss::ArrowRecordBatches batches;
- ASSERT_OK(scanner.PollRecordBatch(1000, batches));
+ ASSERT_OK(PollRecordBatch(scanner, 1000, batches));
auto items = extract_fn(batches);
out.insert(out.end(), items.begin(), items.end());
}
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 6ac8c6657..91fe29a8d 100644
--- a/fluss-rust/website/docs/user-guide/cpp/api-reference.md
+++ b/fluss-rust/website/docs/user-guide/cpp/api-reference.md
@@ -151,6 +151,7 @@ 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 |
+| `Filter(Predicate predicate) -> TableScan&` | Push a predicate down for conservative server-side RecordBatch pruning |
| `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(RecordBatchLogScanner& out) -> Result` | Create a strongly typed Arrow RecordBatch scanner |
@@ -158,6 +159,29 @@ Complete API reference for the Fluss C++ client.
| `CreateRecordBatchLogReader(Admin& admin, const std::vector<TableBucket>& 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`) |
+## Filter predicates
+
+Use `Col(name)` to build a predicate and combine expressions with `And()` or `Or()`.
+
+| `ColumnRef` method | Description |
+|---------------------------------------------------------|--------------------------------|
+| `Equal(value)` / `NotEqual(value)` | Equality comparisons |
+| `LessThan(value)` / `LessOrEqual(value)` | Less-than comparisons |
+| `GreaterThan(value)` / `GreaterOrEqual(value)` | Greater-than comparisons |
+| `IsNull()` / `IsNotNull()` | Null checks |
+| `StartsWith(prefix)` / `Contains(infix)` / `EndsWith(suffix)` | String matching |
+| `In(values)` / `NotIn(values)` | Set membership |
+
+Booleans, integers, floating-point values, strings, bytes, `Date`, and `Time` convert to
+`PredicateLiteral` directly. Use `PredicateLiteral::Decimal()`,
+`PredicateLiteral::TimestampNtz()`, and `PredicateLiteral::TimestampLtz()` when the Fluss logical
+type must be explicit.
+
+Filter pushdown skips only whole RecordBatches whose statistics prove that they cannot match.
+Returned batches may still contain non-matching rows and must be filtered again by the caller.
+Configure `table.statistics.columns` for referenced columns; batches without usable statistics
+are retained. Filters apply to Arrow log scans and are rejected by `CreateBucketBatchScanner()`.
+
## `AppendWriter`
| Method | Description |
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 956c0dbbd..82c171d46 100644
--- a/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md
+++ b/fluss-rust/website/docs/user-guide/cpp/example/log-tables.md
@@ -144,6 +144,36 @@ for (size_t i = 0; i < batches.Size(); ++i) {
}
```
+## Filter Pushdown
+
+Configure statistics for columns used by filters:
+
+```cpp
+auto descriptor = fluss::TableDescriptor::NewBuilder()
+ .SetSchema(schema)
+ .SetProperty("table.statistics.columns", "event_id,event_type")
+ .Build();
+```
+
+Then attach a predicate to the scan:
+
+```cpp
+auto predicate =
+ fluss::Col("event_id")
+ .GreaterOrEqual(100)
+ .And(fluss::Col("event_type").StartsWith("user_"));
+
+fluss::RecordBatchLogScanner scanner;
+auto result = table.NewScan()
+ .Filter(std::move(predicate))
+ .ProjectByName({"event_id", "event_type"})
+ .CreateRecordBatchLogScanner(scanner);
+```
+
+Fluss uses RecordBatch statistics to skip batches that cannot match. It does not filter individual
+rows, so consumers must evaluate the predicate again on returned batches. Batches without usable
+statistics are retained conservatively.
+
## Bounded Arrow RecordBatch Reading
Use `RecordBatchLogReader` when the scan should finish after reaching a fixed offset for every
@@ -291,7 +321,7 @@ fluss::LogScanner name_projected_scanner;
table.NewScan().ProjectByName({"event_id", "timestamp"}).CreateLogScanner(name_projected_scanner);
// Arrow RecordBatch with projection
-fluss::LogScanner projected_arrow_scanner;
+fluss::RecordBatchLogScanner projected_arrow_scanner;
table.NewScan().ProjectByIndex({0, 2}).CreateRecordBatchLogScanner(projected_arrow_scanner);
```