[python] Python bindings - add log filter pushdown

fluss-rust/bindings/python/example/log_table.py

@@ -84,7 +84,10 @@ async def _run(conn):
         pa.field("salary", pa.decimal128(10, 2)),
     ]
     schema = pa.schema(fields)
-    table_descriptor = fluss.TableDescriptor(fluss.Schema(schema))
+    # Statistics let the server prune batches for a pushed-down filter.
+    table_descriptor = fluss.TableDescriptor(
+        fluss.Schema(schema), properties={"table.statistics.columns": "*"}
+    )
 
     admin = conn.get_admin()
     table_path = fluss.TablePath("fluss", "example_log_table")
@@ -245,6 +248,7 @@ async def _run(conn):
     await _scan_batch(table, num_buckets)
     await _scan_records(table, num_buckets)
     await _projection(table, num_buckets)
+    await _filter_pushdown(table, num_buckets)
     await _limit_scan(table, num_buckets)
     await _context_manager_demo(conn, table_path)
 
@@ -384,6 +388,37 @@ async def _projection(table, num_buckets):
     print(f"Projected columns: {list(df_named.columns)}")
 
 
+async def _filter_pushdown(table, num_buckets):
+    print("\n--- Filter pushdown (server-side batch pruning) ---")
+
+    async def _collect(scanner, want):
+        rows = []
+        deadline = asyncio.get_running_loop().time() + 10
+        while len(rows) < want and asyncio.get_running_loop().time() < deadline:
+            rows.extend(r.row["id"] for r in await scanner.poll(500))
+        return rows
+
+    unfiltered = await table.new_scan().create_log_scanner()
+    unfiltered.subscribe_buckets({i: fluss.EARLIEST_OFFSET for i in range(num_buckets)})
+    all_ids = await _collect(unfiltered, EXPECTED_ROWS)
+    expected = sorted(i for i in all_ids if i >= 2)
+
+    # Comparisons on fluss.col(...) build a predicate; & and | combine them.
+    predicate = (fluss.col("id") >= 2) & fluss.col("name").is_not_null()
+    scanner = await table.new_scan().filter(predicate).create_log_scanner()
+    scanner.subscribe_buckets({i: fluss.EARLIEST_OFFSET for i in range(num_buckets)})
+    returned = await _collect(scanner, len(expected))
+
+    # A batch is dropped only when its statistics cannot match, so the scan
+    # returns a superset and the filter still has to be applied to the rows.
+    matched = sorted(i for i in returned if i >= 2)
+    assert matched == expected, f"Filtered scan lost rows: {matched} != {expected}"
+    print(
+        f"Filtered scan returned {len(returned)} row(s), "
+        f"{len(matched)} matching ids {matched}"
+    )
+
+
 async def _limit_scan(table, num_buckets):
     print("\n--- Limit scan: one-shot bounded BatchScanner (per bucket) ---")
     table_id = table.get_table_info().table_id

fluss-rust/bindings/python/fluss/__init__.pyi

@@ -20,6 +20,7 @@
 from enum import IntEnum
 from types import TracebackType
 from typing import (
+    Any,
     AsyncIterator,
     Dict,
     Iterator,
@@ -466,6 +467,109 @@ class DatabaseInfo:
     def modified_time(self) -> int: ...
     def __repr__(self) -> str: ...
 
+@final
+class Predicate:
+    """A filter expression for a log scan.
+
+    Build one from :func:`col`, then combine with ``&`` and ``|``.
+
+    Example:
+        ```python
+        from fluss import col
+
+        hot = (col("id") >= 200) & col("name").starts_with("high")
+        scanner = await table.new_scan().filter(hot).create_log_scanner()
+        ```
+    """
+
+    def and_(self, other: Predicate) -> Predicate:
+        """Match rows satisfying both predicates."""
+        ...
+    def or_(self, other: Predicate) -> Predicate:
+        """Match rows satisfying either predicate."""
+        ...
+    def __and__(self, other: Predicate, /) -> Predicate: ...
+    def __or__(self, other: Predicate, /) -> Predicate: ...
+    def __repr__(self) -> str: ...
+
+@final
+class ColumnRef:
+    """A column reference used to build a :class:`Predicate`.
+
+    Comparison operators build predicates, so ``col("id") >= 200`` and
+    ``col("id").greater_or_equal(200)`` are equivalent.
+
+    Literals are plain Python values: ``bool``, ``int``, ``float``, ``str``,
+    ``bytes``, ``decimal.Decimal``,
+    ``datetime.date``, ``datetime.time`` and ``datetime.datetime``. A naive
+    datetime is a wall clock and filters a TIMESTAMP column, an aware one a
+    TIMESTAMP_LTZ column. Nulls are matched with ``is_null()`` and
+    ``is_not_null()`` rather than a comparison against ``None``.
+    """
+
+    def __new__(cls, name: str) -> ColumnRef: ...
+    @property
+    def name(self) -> str:
+        """The column name this reference points at."""
+        ...
+    def __eq__(self, value: Any, /) -> Predicate: ...  # type: ignore[override]
+    def __ne__(self, value: Any, /) -> Predicate: ...  # type: ignore[override]
+    def __lt__(self, value: Any, /) -> Predicate: ...
+    def __le__(self, value: Any, /) -> Predicate: ...
+    def __gt__(self, value: Any, /) -> Predicate: ...
+    def __ge__(self, value: Any, /) -> Predicate: ...
+    def equal(self, value: Any) -> Predicate:
+        """Match rows where the column equals ``value``."""
+        ...
+    def not_equal(self, value: Any) -> Predicate:
+        """Match rows where the column differs from ``value``."""
+        ...
+    def less_than(self, value: Any) -> Predicate:
+        """Match rows where the column is less than ``value``."""
+        ...
+    def less_or_equal(self, value: Any) -> Predicate:
+        """Match rows where the column is less than or equal to ``value``."""
+        ...
+    def greater_than(self, value: Any) -> Predicate:
+        """Match rows where the column is greater than ``value``."""
+        ...
+    def greater_or_equal(self, value: Any) -> Predicate:
+        """Match rows where the column is greater than or equal to ``value``."""
+        ...
+    def is_null(self) -> Predicate:
+        """Match rows where the column is null."""
+        ...
+    def is_not_null(self) -> Predicate:
+        """Match rows where the column is not null."""
+        ...
+    def starts_with(self, prefix: str) -> Predicate:
+        """Match rows where the string column starts with ``prefix``."""
+        ...
+    def ends_with(self, suffix: str) -> Predicate:
+        """Match rows where the string column ends with ``suffix``."""
+        ...
+    def contains(self, infix: str) -> Predicate:
+        """Match rows where the string column contains ``infix``."""
+        ...
+    def is_in(self, values: List[Any]) -> Predicate:
+        """Match rows where the column equals any of ``values``."""
+        ...
+    def not_in(self, values: List[Any]) -> Predicate:
+        """Match rows where the column equals none of ``values``."""
+        ...
+    def __repr__(self) -> str: ...
+
+def col(name: str) -> ColumnRef:
+    """Reference a column by name when building a filter.
+
+    Args:
+        name: Column name as declared in the table schema.
+
+    Returns:
+        A ColumnRef that builds predicates via comparison operators.
+    """
+    ...
+
 @final
 class TableScan:
     """Builder for creating log scanners with flexible configuration.
@@ -524,6 +628,20 @@ class TableScan:
         Args:
             n: The maximum number of rows to scan. Must be positive.
 
+        Returns:
+            Self for method chaining.
+        """
+        ...
+    def filter(self, predicate: Predicate) -> "TableScan":
+        """Push a filter down to the server for batch pruning.
+
+        Pruning is conservative: a returned batch may still contain
+        non-matching rows, so re-apply the predicate on the results. Only
+        Arrow log scans prune; a filter combined with ``limit()`` is rejected.
+
+        Args:
+            predicate: Predicate built from :func:`col`.
+
         Returns:
             Self for method chaining.
         """

fluss-rust/bindings/python/src/lib.rs

@@ -27,6 +27,7 @@ mod connection;
 mod error;
 mod lookup;
 mod metadata;
+mod predicate;
 mod table;
 mod upsert;
 mod utils;
@@ -38,6 +39,7 @@ pub use connection::*;
 pub use error::*;
 pub use lookup::*;
 pub use metadata::*;
+pub use predicate::*;
 pub use table::*;
 pub use upsert::*;
 pub use utils::*;
@@ -110,6 +112,8 @@ fn _fluss(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<FlussAdmin>()?;
     m.add_class::<FlussTable>()?;
     m.add_class::<TableScan>()?;
+    m.add_class::<Predicate>()?;
+    m.add_class::<ColumnRef>()?;
     m.add_class::<TableAppend>()?;
     m.add_class::<TableUpsert>()?;
     m.add_class::<TableLookup>()?;
@@ -126,6 +130,7 @@ fn _fluss(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<ChangeType>()?;
     m.add_class::<ScanRecord>()?;
     m.add_class::<ScanRecords>()?;
+    m.add_function(wrap_pyfunction!(predicate::col, m)?)?;
     m.add_class::<RecordBatch>()?;
     m.add_class::<PartitionInfo>()?;
     m.add_class::<ServerNode>()?;