diff --git a/fluss-rust/bindings/cpp/README.md b/fluss-rust/bindings/cpp/README.md
index fa6cac26f5..4a160ef5b0 100644
--- a/fluss-rust/bindings/cpp/README.md
+++ b/fluss-rust/bindings/cpp/README.md
@@ -33,6 +33,33 @@ 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::LogScanner 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. Filter pushdown requires
+the Arrow log format and does not apply to `CreateBucketBatchScanner()`.
 
 ## TODO
 
diff --git a/fluss-rust/bindings/cpp/include/fluss.hpp b/fluss-rust/bindings/cpp/include/fluss.hpp
index 945b447ed0..54f225c9e8 100644
--- a/fluss-rust/bindings/cpp/include/fluss.hpp
+++ b/fluss-rust/bindings/cpp/include/fluss.hpp
@@ -261,6 +261,112 @@ 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);
+    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:
+    enum class Kind : int32_t {
+        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,
+    };
+
+    explicit PredicateLiteral(Kind kind);
+
+    Kind kind_;
+    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(int32_t 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,
@@ -1620,6 +1726,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.
@@ -1651,6 +1763,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 9bc12bbe8a..c450bddba9 100644
--- a/fluss-rust/bindings/cpp/src/lib.rs
+++ b/fluss-rust/bindings/cpp/src/lib.rs
@@ -110,6 +110,30 @@ mod ffi {
         child_count: u32,
     }
 
+    // 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: i32,
+        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: i32,
+        function: i32,
+        field: String,
+        literals: Vec<FfiPredicateLiteral>,
+        child_count: u32,
+    }
+
     struct FfiColumn {
         name: String,
         comment: String,
@@ -395,7 +419,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>,
@@ -1486,6 +1515,224 @@ impl Admin {
     }
 }
 
+const FFI_PREDICATE_NODE_LEAF: i32 = 0;
+const FFI_PREDICATE_NODE_COMPOUND: i32 = 1;
+
+const FFI_LITERAL_NULL: i32 = 0;
+const FFI_LITERAL_BOOLEAN: i32 = 1;
+const FFI_LITERAL_INT32: i32 = 2;
+const FFI_LITERAL_INT64: i32 = 3;
+const FFI_LITERAL_FLOAT32: i32 = 4;
+const FFI_LITERAL_FLOAT64: i32 = 5;
+const FFI_LITERAL_STRING: i32 = 6;
+const FFI_LITERAL_BYTES: i32 = 7;
+const FFI_LITERAL_DECIMAL: i32 = 8;
+const FFI_LITERAL_DATE: i32 = 9;
+const FFI_LITERAL_TIME: i32 = 10;
+const FFI_LITERAL_TIMESTAMP_NTZ: i32 = 11;
+const FFI_LITERAL_TIMESTAMP_LTZ: i32 = 12;
+
+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_PREDICATE_NODE_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.function {
+                0 => fcore::predicate::LeafFunction::Equal,
+                1 => fcore::predicate::LeafFunction::NotEqual,
+                2 => fcore::predicate::LeafFunction::LessThan,
+                3 => fcore::predicate::LeafFunction::LessOrEqual,
+                4 => fcore::predicate::LeafFunction::GreaterThan,
+                5 => fcore::predicate::LeafFunction::GreaterOrEqual,
+                6 => fcore::predicate::LeafFunction::IsNull,
+                7 => fcore::predicate::LeafFunction::IsNotNull,
+                8 => fcore::predicate::LeafFunction::StartsWith,
+                9 => fcore::predicate::LeafFunction::Contains,
+                10 => fcore::predicate::LeafFunction::EndsWith,
+                11 => fcore::predicate::LeafFunction::In,
+                12 => fcore::predicate::LeafFunction::NotIn,
+                other => {
+                    return Err(format!(
+                        "Predicate leaf at node {node_index} has unknown function {other}"
+                    ));
+                }
+            };
+            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_PREDICATE_NODE_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.function {
+                0 => fcore::predicate::CompoundFunction::And,
+                1 => fcore::predicate::CompoundFunction::Or,
+                other => {
+                    return Err(format!(
+                        "Predicate compound node {node_index} has unknown function {other}"
+                    ));
+                }
+            };
+            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}"
+        )),
+    }
+}
+
+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_LITERAL_NULL => Ok(Literal::Null),
+        FFI_LITERAL_BOOLEAN => Ok(Literal::Bool(literal.boolean_value)),
+        FFI_LITERAL_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_LITERAL_INT64 => Ok(Literal::Int64(literal.integer_value)),
+        FFI_LITERAL_FLOAT32 => Ok(Literal::Float32(literal.floating_value as f32)),
+        FFI_LITERAL_FLOAT64 => Ok(Literal::Float64(literal.floating_value)),
+        FFI_LITERAL_STRING => Ok(Literal::String(literal.string_value.to_string())),
+        FFI_LITERAL_BYTES => Ok(Literal::Bytes(literal.bytes_value.clone())),
+        FFI_LITERAL_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_LITERAL_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_LITERAL_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_LITERAL_TIMESTAMP_NTZ => {
+            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_LITERAL_TIMESTAMP_LTZ => {
+            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}")),
+    }
+}
+
 // Table implementation
 unsafe fn delete_table(table: *mut Table) {
     if !table.is_null() {
@@ -1542,10 +1789,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 941f0f046d..666b6802da 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,182 @@ int Date::Day() const {
     return tm.tm_mday;
 }
 
+PredicateLiteral::PredicateLiteral(bool value) : kind_(Kind::Boolean), boolean_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(int32_t value) : kind_(Kind::Int32), integer_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(int64_t value) : kind_(Kind::Int64), integer_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(float value) : kind_(Kind::Float32), floating_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(double value) : kind_(Kind::Float64), floating_value_(value) {}
+
+PredicateLiteral::PredicateLiteral(const char* value) : PredicateLiteral(std::string(value)) {}
+
+PredicateLiteral::PredicateLiteral(std::string value)
+    : kind_(Kind::String), string_value_(std::move(value)) {}
+
+PredicateLiteral::PredicateLiteral(std::vector<uint8_t> value)
+    : kind_(Kind::Bytes), bytes_value_(std::move(value)) {}
+
+PredicateLiteral::PredicateLiteral(Date value)
+    : kind_(Kind::Date), integer_value_(value.days_since_epoch) {}
+
+PredicateLiteral::PredicateLiteral(Time value)
+    : kind_(Kind::Time), integer_value_(value.millis_since_midnight) {}
+
+PredicateLiteral::PredicateLiteral(Kind kind) : kind_(kind) {}
+
+PredicateLiteral PredicateLiteral::Null() { return PredicateLiteral(Kind::Null); }
+
+PredicateLiteral PredicateLiteral::Decimal(std::string value) {
+    PredicateLiteral literal(Kind::Decimal);
+    literal.string_value_ = std::move(value);
+    return literal;
+}
+
+PredicateLiteral PredicateLiteral::TimestampNtz(Timestamp value) {
+    PredicateLiteral literal(Kind::TimestampNtz);
+    literal.timestamp_value_ = value;
+    return literal;
+}
+
+PredicateLiteral PredicateLiteral::TimestampLtz(Timestamp value) {
+    PredicateLiteral literal(Kind::TimestampLtz);
+    literal.timestamp_value_ = value;
+    return literal;
+}
+
+namespace {
+
+enum class PredicateLeafFunction : int32_t {
+    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,
+};
+
+enum class PredicateCompoundFunction : int32_t {
+    And = 0,
+    Or = 1,
+};
+
+}  // namespace
+
+struct Predicate::Node {
+    enum class Type : int32_t {
+        Leaf = 0,
+        Compound = 1,
+    };
+
+    Type type;
+    int32_t function;
+    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->type = Node::Type::Compound;
+    node->function = static_cast<int32_t>(PredicateCompoundFunction::And);
+    if (root_->type == Node::Type::Compound && root_->function == node->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->type = Node::Type::Compound;
+    node->function = static_cast<int32_t>(PredicateCompoundFunction::Or);
+    if (root_->type == Node::Type::Compound && root_->function == node->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(int32_t function, std::vector<PredicateLiteral> literals) const {
+    auto node = std::make_shared<Predicate::Node>();
+    node->type = Predicate::Node::Type::Leaf;
+    node->function = function;
+    node->field = name_;
+    node->literals = std::move(literals);
+    return Predicate(std::move(node));
+}
+
+Predicate ColumnRef::Equal(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::Equal), {std::move(value)});
+}
+
+Predicate ColumnRef::NotEqual(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::NotEqual), {std::move(value)});
+}
+
+Predicate ColumnRef::LessThan(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::LessThan), {std::move(value)});
+}
+
+Predicate ColumnRef::LessOrEqual(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::LessOrEqual), {std::move(value)});
+}
+
+Predicate ColumnRef::GreaterThan(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::GreaterThan), {std::move(value)});
+}
+
+Predicate ColumnRef::GreaterOrEqual(PredicateLiteral value) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::GreaterOrEqual), {std::move(value)});
+}
+
+Predicate ColumnRef::IsNull() const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::IsNull), {});
+}
+
+Predicate ColumnRef::IsNotNull() const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::IsNotNull), {});
+}
+
+Predicate ColumnRef::StartsWith(std::string prefix) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::StartsWith),
+                {PredicateLiteral(std::move(prefix))});
+}
+
+Predicate ColumnRef::Contains(std::string infix) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::Contains),
+                {PredicateLiteral(std::move(infix))});
+}
+
+Predicate ColumnRef::EndsWith(std::string suffix) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::EndsWith),
+                {PredicateLiteral(std::move(suffix))});
+}
+
+Predicate ColumnRef::In(std::vector<PredicateLiteral> values) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::In), std::move(values));
+}
+
+Predicate ColumnRef::NotIn(std::vector<PredicateLiteral> values) const {
+    return Leaf(static_cast<int32_t>(PredicateLeafFunction::NotIn), std::move(values));
+}
+
 // NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
 #define CHECK_INNER(name)                                                                 \
     do {                                                                                  \
@@ -1222,6 +1399,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;
@@ -1264,7 +1446,42 @@ 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 = static_cast<int32_t>(node->type);
+                ffi_node.function = node->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 = static_cast<int32_t>(literal.kind_);
+                    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);
@@ -1285,6 +1502,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 ef29d9c93a..284df12964 100644
--- a/fluss-rust/bindings/cpp/test/test_log_table.cpp
+++ b/fluss-rust/bindings/cpp/test/test_log_table.cpp
@@ -357,6 +357,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.
@@ -704,6 +712,80 @@ 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::LogScanner 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::LogScanner 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, AllSupportedDatatypes) {
     auto& adm = admin();
     auto& conn = connection();