diff --git a/fluss-rust/bindings/python/example/log_table.py b/fluss-rust/bindings/python/example/log_table.py index 018d056b2e6..f3260148634 100644 --- a/fluss-rust/bindings/python/example/log_table.py +++ b/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 diff --git a/fluss-rust/bindings/python/fluss/__init__.pyi b/fluss-rust/bindings/python/fluss/__init__.pyi index 01245d6ecfa..6e3b79adf6e 100644 --- a/fluss-rust/bindings/python/fluss/__init__.pyi +++ b/fluss-rust/bindings/python/fluss/__init__.pyi @@ -20,7 +20,9 @@ from enum import IntEnum from types import TracebackType from typing import ( + Any, AsyncIterator, + NoReturn, Dict, Iterator, List, @@ -466,6 +468,116 @@ 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 __bool__(self) -> NoReturn: + """Always raises: use ``&`` and ``|``, since ``and``/``or`` would + silently return one side.""" + ... + 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 __bool__(self) -> NoReturn: + """Always raises, like :meth:`Predicate.__bool__`.""" + ... + 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 +636,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. """ diff --git a/fluss-rust/bindings/python/src/lib.rs b/fluss-rust/bindings/python/src/lib.rs index 011f0656fce..f884f127a25 100644 --- a/fluss-rust/bindings/python/src/lib.rs +++ b/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::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; @@ -126,6 +130,7 @@ fn _fluss(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_function(wrap_pyfunction!(predicate::col, m)?)?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/fluss-rust/bindings/python/src/predicate.rs b/fluss-rust/bindings/python/src/predicate.rs new file mode 100644 index 00000000000..5972f2b9262 --- /dev/null +++ b/fluss-rust/bindings/python/src/predicate.rs @@ -0,0 +1,351 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Predicates for log-scan filter pushdown, mirroring `fluss::predicate`. + +use crate::error::FlussError; +use fcore::predicate::Literal; +use fluss as fcore; +use pyo3::exceptions::PyTypeError; +use pyo3::prelude::*; +use pyo3::pyclass::CompareOp; +use pyo3::types::{ + PyBool, PyBytes, PyDate, PyDateTime, PyDelta, PyDeltaAccess, PyInt, PyTime, PyTimeAccess, + PyTzInfoAccess, +}; + +/// Python's ordinal for 1970-01-01, the DATE epoch. +const UNIX_EPOCH_ORDINAL: i64 = 719_163; + +/// A filter expression for a log scan, combined with `&` and `|`. +#[pyclass(module = "fluss", from_py_object)] +#[derive(Clone)] +pub struct Predicate { + inner: fcore::predicate::Predicate, +} + +impl Predicate { + pub(crate) fn to_core(&self) -> fcore::predicate::Predicate { + self.inner.clone() + } +} + +#[pymethods] +impl Predicate { + /// Match rows satisfying both predicates. + fn and_(&self, other: &Predicate) -> Predicate { + Predicate { + inner: self.inner.clone().and(other.inner.clone()), + } + } + + /// Match rows satisfying either predicate. + fn or_(&self, other: &Predicate) -> Predicate { + Predicate { + inner: self.inner.clone().or(other.inner.clone()), + } + } + + fn __and__(&self, other: &Predicate) -> Predicate { + self.and_(other) + } + + fn __or__(&self, other: &Predicate) -> Predicate { + self.or_(other) + } + + /// `and`/`or` would silently return one side, so refuse a truth value. + fn __bool__(&self) -> PyResult { + Err(PyTypeError::new_err( + "A Predicate has no truth value; combine them with & and |, not and/or", + )) + } + + fn __repr__(&self) -> String { + format!("Predicate({:?})", self.inner) + } +} + +/// A column reference, where `col("id") >= 200` and +/// `col("id").greater_or_equal(200)` build the same [`Predicate`]. +#[pyclass(module = "fluss")] +pub struct ColumnRef { + name: String, +} + +#[pymethods] +impl ColumnRef { + #[new] + fn new(name: String) -> Self { + ColumnRef { name } + } + + /// The column name. + #[getter] + fn name(&self) -> &str { + &self.name + } + + fn __richcmp__(&self, other: &Bound<'_, PyAny>, op: CompareOp) -> PyResult { + let literal = literal_from_py(other)?; + let column = self.column(); + let inner = match op { + CompareOp::Eq => column.eq(literal), + CompareOp::Ne => column.ne(literal), + CompareOp::Lt => column.lt(literal), + CompareOp::Le => column.le(literal), + CompareOp::Gt => column.gt(literal), + CompareOp::Ge => column.ge(literal), + }; + Ok(Predicate { inner }) + } + + /// Match rows where the column equals `value`. + fn equal(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().eq(literal_from_py(value)?), + }) + } + + /// Match rows where the column differs from `value`. + fn not_equal(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().ne(literal_from_py(value)?), + }) + } + + /// Match rows where the column is less than `value`. + fn less_than(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().lt(literal_from_py(value)?), + }) + } + + /// Match rows where the column is less than or equal to `value`. + fn less_or_equal(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().le(literal_from_py(value)?), + }) + } + + /// Match rows where the column is greater than `value`. + fn greater_than(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().gt(literal_from_py(value)?), + }) + } + + /// Match rows where the column is greater than or equal to `value`. + fn greater_or_equal(&self, value: &Bound<'_, PyAny>) -> PyResult { + Ok(Predicate { + inner: self.column().ge(literal_from_py(value)?), + }) + } + + /// Match rows where the column is null. + fn is_null(&self) -> Predicate { + Predicate { + inner: self.column().is_null(), + } + } + + /// Match rows where the column is not null. + fn is_not_null(&self) -> Predicate { + Predicate { + inner: self.column().is_not_null(), + } + } + + /// Match rows where the string column starts with `prefix`. + fn starts_with(&self, prefix: String) -> Predicate { + Predicate { + inner: self.column().starts_with(prefix), + } + } + + /// Match rows where the string column ends with `suffix`. + fn ends_with(&self, suffix: String) -> Predicate { + Predicate { + inner: self.column().ends_with(suffix), + } + } + + /// Match rows where the string column contains `infix`. + fn contains(&self, infix: String) -> Predicate { + Predicate { + inner: self.column().contains(infix), + } + } + + /// Match rows where the column equals any of `values`. + fn is_in(&self, values: Vec>) -> PyResult { + Ok(Predicate { + inner: self.column().is_in(literals_from_py(&values)?), + }) + } + + /// Match rows where the column equals none of `values`. + fn not_in(&self, values: Vec>) -> PyResult { + Ok(Predicate { + inner: self.column().not_in(literals_from_py(&values)?), + }) + } + + fn __bool__(&self) -> PyResult { + Err(PyTypeError::new_err( + "A ColumnRef has no truth value; compare it first, then combine with & and |", + )) + } + + fn __repr__(&self) -> String { + format!("ColumnRef({})", self.name) + } +} + +impl ColumnRef { + fn column(&self) -> fcore::predicate::ColumnRef { + fcore::predicate::col(self.name.clone()) + } +} + +/// Reference a column by name when building a filter. +#[pyfunction] +pub fn col(name: String) -> ColumnRef { + ColumnRef::new(name) +} + +fn literals_from_py(values: &[Bound<'_, PyAny>]) -> PyResult> { + values.iter().map(literal_from_py).collect() +} + +/// `bool` before `int` and `datetime` before `date`: each is a subclass of the other. +fn literal_from_py(value: &Bound<'_, PyAny>) -> PyResult { + if value.is_none() { + return Ok(Literal::Null); + } + if let Ok(flag) = value.cast_exact::() { + return Ok(Literal::Bool(flag.is_true())); + } + if let Ok(datetime) = value.cast::() { + return timestamp_literal(datetime); + } + if let Ok(date) = value.cast::() { + return date_literal(date); + } + if let Ok(time) = value.cast::() { + return Ok(Literal::Time(time_millis_of_day(time))); + } + let decimal_type = value.py().import("decimal")?.getattr("Decimal")?; + if value.is_instance(&decimal_type)? { + return decimal_literal(value); + } + match value.extract::() { + Ok(integer) => return Ok(Literal::Int64(integer)), + // Falling through would turn an out-of-range int into a float literal. + Err(err) if value.is_instance_of::() => return Err(err), + Err(_) => {} + } + if let Ok(floating) = value.extract::() { + return Ok(Literal::Float64(floating)); + } + if let Ok(text) = value.extract::() { + return Ok(Literal::String(text)); + } + if let Ok(bytes) = value.cast::() { + return Ok(Literal::Bytes(bytes.as_bytes().to_vec())); + } + + Err(PyTypeError::new_err(format!( + "Unsupported filter literal of type '{}'. Supported: bool, int, float, \ + str, bytes, decimal.Decimal, datetime.date, datetime.time, \ + datetime.datetime; use is_null()/is_not_null() for nulls", + value + .get_type() + .name() + .map(|name| name.to_string()) + .unwrap_or_else(|_| "unknown".to_string()) + ))) +} + +/// Kept at the caller's scale; the scan rescales it and rejects an inexact one. +fn decimal_literal(value: &Bound<'_, PyAny>) -> PyResult { + let text = value.str()?.to_string(); + let big_decimal: bigdecimal::BigDecimal = text + .parse() + .map_err(|e| FlussError::new_err(format!("Invalid decimal literal '{text}': {e}")))?; + let scale = i64::max(big_decimal.fractional_digit_count(), 0) as u32; + let digits = big_decimal.digits() as u32; + let precision = u32::max(digits, scale).max(1); + let decimal = fcore::row::Decimal::from_big_decimal(big_decimal, precision, scale) + .map_err(|e| FlussError::from_core_error(&e))?; + Ok(Literal::Decimal(decimal)) +} + +/// Naive datetimes filter TIMESTAMP, aware ones TIMESTAMP_LTZ. +/// +/// Subtracting a matching epoch, not `datetime.timestamp()`, which reads a +/// naive value as local time and would shift the literal by the UTC offset. +fn timestamp_literal(value: &Bound<'_, PyDateTime>) -> PyResult { + let datetime_module = value.py().import("datetime")?; + let datetime_type = datetime_module.getattr("datetime")?; + let aware = value.get_tzinfo().is_some(); + let epoch = if aware { + let utc = datetime_module.getattr("timezone")?.getattr("utc")?; + datetime_type.call1((1970, 1, 1, 0, 0, 0, 0, utc))? + } else { + datetime_type.call1((1970, 1, 1))? + }; + + let delta = value.sub(epoch)?; + let delta = delta.cast::()?; + // A timedelta carries the sign in `days`, so this holds before 1970 too. + let micros = delta.get_days() as i64 * 86_400_000_000 + + delta.get_seconds() as i64 * 1_000_000 + + delta.get_microseconds() as i64; + let millis = micros.div_euclid(1_000); + let nanos = (micros.rem_euclid(1_000) * 1_000) as i32; + + if aware { + let timestamp = fcore::row::TimestampLtz::from_millis_nanos(millis, nanos) + .map_err(|e| FlussError::from_core_error(&e))?; + Ok(Literal::TimestampLtz(timestamp)) + } else { + let timestamp = fcore::row::TimestampNtz::from_millis_nanos(millis, nanos) + .map_err(|e| FlussError::from_core_error(&e))?; + Ok(Literal::TimestampNtz(timestamp)) + } +} + +/// Fluss stores DATE as days since the Unix epoch. +fn date_literal(value: &Bound<'_, PyDate>) -> PyResult { + let ordinal: i64 = value.call_method0("toordinal")?.extract()?; + let days = i32::try_from(ordinal - UNIX_EPOCH_ORDINAL).map_err(|_| { + FlussError::new_err(format!( + "Date literal {value} is out of range for a DATE column" + )) + })?; + Ok(Literal::Date(days)) +} + +/// Fluss stores TIME as milliseconds of day. +fn time_millis_of_day(value: &Bound<'_, PyTime>) -> i32 { + let hours = value.get_hour() as i32; + let minutes = value.get_minute() as i32; + let seconds = value.get_second() as i32; + let micros = value.get_microsecond() as i32; + ((hours * 3_600 + minutes * 60 + seconds) * 1_000) + micros / 1_000 +} diff --git a/fluss-rust/bindings/python/src/table.rs b/fluss-rust/bindings/python/src/table.rs index ac050ff960f..db66dbaae45 100644 --- a/fluss-rust/bindings/python/src/table.rs +++ b/fluss-rust/bindings/python/src/table.rs @@ -431,6 +431,7 @@ pub struct TableScan { projection: Option, fixed_schema: bool, limit: Option, + filter: Option, } /// Scanner type for internal use @@ -495,6 +496,21 @@ impl TableScan { Ok(slf) } + /// Push a filter down for server-side batch pruning. + /// + /// A returned batch may still contain non-matching rows, so re-apply the + /// predicate on the results. + /// + /// Args: + /// predicate: Predicate built from `fluss.col(...)`. + /// + /// Returns: + /// Self for method chaining. + pub fn filter(mut slf: PyRefMut<'_, Self>, predicate: Predicate) -> PyRefMut<'_, Self> { + slf.filter = Some(predicate); + slf + } + /// Create a one-shot bounded scanner over a single bucket. /// /// Requires a limit set via `limit()`; the scan runs on the first @@ -517,9 +533,12 @@ impl TableScan { let projection = self.projection.clone(); let projection_indices = resolve_projection_indices(&projection, &self.table_info)?; - let scan = apply_projection(table.new_scan(), projection)? - .limit(limit) - .map_err(|e| FlussError::from_core_error(&e))?; + let scan = apply_filter( + apply_projection(table.new_scan(), projection)?, + &self.filter, + )? + .limit(limit) + .map_err(|e| FlussError::from_core_error(&e))?; let scanner = scan .create_bucket_batch_scanner(bucket.to_core()) .map_err(|e| FlussError::from_core_error(&e))?; @@ -581,13 +600,17 @@ impl TableScan { let table_info = self.table_info.clone(); let projection = self.projection.clone(); let fixed_schema = self.fixed_schema; + let filter = self.filter.clone(); future_into_py(py, async move { let fluss_table = fcore::client::FlussTable::new(&conn, metadata, table_info.clone()); let projection_indices = resolve_projection_indices(&projection, &table_info)?; - let table_scan = apply_projection(fluss_table.new_scan(), projection)? - .with_fixed_schema(fixed_schema); + let table_scan = apply_filter( + apply_projection(fluss_table.new_scan(), projection)?, + &filter, + )? + .with_fixed_schema(fixed_schema); let admin = conn .get_admin() @@ -655,6 +678,19 @@ fn resolve_projection_indices( } } +/// Applies the filter, if one was set. +fn apply_filter<'a>( + table_scan: fcore::client::TableScan<'a>, + filter: &Option, +) -> PyResult> { + match filter { + Some(predicate) => table_scan + .filter(predicate.to_core()) + .map_err(|e| FlussError::from_core_error(&e)), + None => Ok(table_scan), + } +} + /// Apply projection to table scan fn apply_projection( table_scan: fcore::client::TableScan, @@ -717,6 +753,7 @@ impl FlussTable { projection: None, fixed_schema: false, limit: None, + filter: None, } } diff --git a/fluss-rust/bindings/python/test/test_log_table.py b/fluss-rust/bindings/python/test/test_log_table.py index b6bee545a39..f812e16b390 100644 --- a/fluss-rust/bindings/python/test/test_log_table.py +++ b/fluss-rust/bindings/python/test/test_log_table.py @@ -21,6 +21,8 @@ """ import asyncio +import datetime +import decimal import time import pyarrow as pa @@ -1625,3 +1627,500 @@ async def test_append_arrow_batch_complex_types(connection, admin): ] await admin.drop_table(table_path, ignore_if_not_exists=False) + + +def _stats_descriptor(schema): + """A single-bucket log table with statistics, keeping segments local. + + Only server-side reads prune, and the client downloads remote segments. + """ + return fluss.TableDescriptor( + schema, + bucket_count=1, + properties={ + "table.statistics.columns": "*", + "table.log.tiered.local-segments": "100", + }, + ) + + +async def _filtered_ids(table, predicate, expected_count, column="id", project=None): + """Scans with `predicate` pushed down and returns the ids that came back. + + Expecting nothing still waits for a fetch, or the assertion passes whether + the batch was pruned or merely slow. + """ + scan = table.new_scan() + if project is not None: + scan = scan.project_by_name(project) + scanner = await scan.filter(predicate).create_log_scanner() + scanner.subscribe_buckets({0: 0}) + if expected_count == 0: + records = await scanner.poll(3000) + else: + records = await _poll_records(scanner, expected_count=expected_count) + return sorted(record.row[column] for record in records) + + +async def test_filter_pushdown_prunes_non_matching_batches(connection, admin): + """Batches whose statistics cannot match the filter are pruned server-side.""" + table_path = fluss.TablePath("fluss", "py_test_filter_pushdown_prune") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("name", pa.string())] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + # Three wire batches with disjoint id ranges, so a batch either fully + # matches the filter or cannot match. + for base in (1, 100, 200): + ids = list(range(base, base + 5)) + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [pa.array(ids, type=pa.int32()), pa.array([f"v{i}" for i in ids])], + schema=arrow_schema, + ) + ) + await writer.flush() + + ids = await _filtered_ids(table, fluss.col("id") >= 200, expected_count=5) + assert ids == [200, 201, 202, 203, 204], ( + "only the batch overlapping the filter should be fetched; " + "the two disjoint batches must be pruned server-side" + ) + + # The filter resolves against the full row type, so filtering on a column + # excluded from the projection must still prune. + scanner = await ( + table.new_scan() + .project_by_name(["name"]) + .filter(fluss.col("id") >= 200) + .create_log_scanner() + ) + scanner.subscribe_buckets({0: 0}) + records = await _poll_records(scanner, expected_count=5) + names = sorted(record.row["name"] for record in records) + assert names == ["v200", "v201", "v202", "v203", "v204"] + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_with_overlapping_statistics_returns_whole_batch( + connection, admin +): + """Pruning is batch-granular, so an overlapping batch comes back whole.""" + table_path = fluss.TablePath("fluss", "py_test_filter_overlapping") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("name", pa.string())] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + for ids in ([1, 2, 3, 4, 5], [1, 6, 3, 8, 2]): + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [pa.array(ids, type=pa.int32()), pa.array([f"v{i}" for i in ids])], + schema=arrow_schema, + ) + ) + await writer.flush() + + ids = await _filtered_ids(table, fluss.col("id") > 5, expected_count=5) + assert ids == [1, 2, 3, 6, 8], ( + "the mixed batch (min=1, max=8) is returned whole as a superset, " + "while the all-low batch is pruned" + ) + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_pushdown_prunes_row_appended_batches(connection, admin): + """Row appends go through the row-to-Arrow builder and still carry statistics.""" + table_path = fluss.TablePath("fluss", "py_test_filter_row_append") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("name", pa.string())] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + # Flush between the groups so each becomes its own wire batch. + for base, prefix in ((1, "low"), (200, "v")): + for i in range(base, base + 5): + writer.append({"id": i, "name": f"{prefix}{i}"}) + await writer.flush() + + ids = await _filtered_ids(table, fluss.col("id") >= 200, expected_count=5) + assert ids == [200, 201, 202, 203, 204] + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_without_statistics_returns_all_rows(connection, admin): + """Without statistics the server cannot prune, so the scan is a superset.""" + table_path = fluss.TablePath("fluss", "py_test_filter_no_statistics") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("name", pa.string())] + ) + schema = fluss.Schema(arrow_schema) + descriptor = fluss.TableDescriptor( + schema, + bucket_count=1, + properties={"table.log.tiered.local-segments": "100"}, + ) + await admin.create_table(table_path, descriptor, ignore_if_exists=False) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + for ids in ([1, 2, 3], [100, 101, 102]): + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [pa.array(ids, type=pa.int32()), pa.array([f"v{i}" for i in ids])], + schema=arrow_schema, + ) + ) + await writer.flush() + + ids = await _filtered_ids(table, fluss.col("id") >= 100, expected_count=6) + assert ids == [1, 2, 3, 100, 101, 102] + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_pushdown_prunes_across_column_types(connection, admin): + """Both decimal widths, both timestamp precisions, TIME(0) which Arrow stores + as seconds, string bounds too long to inline, and null counts.""" + table_path = fluss.TablePath("fluss", "py_test_filter_column_types") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [ + pa.field("id", pa.int32()), + pa.field("name", pa.string()), + pa.field("price", pa.decimal128(10, 2)), + pa.field("big", pa.decimal128(22, 5)), + pa.field("ts3", pa.timestamp("ms")), + pa.field("ts6", pa.timestamp("us")), + pa.field("t", pa.time32("s")), + pa.field("opt", pa.int32()), + ] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + + def batch(ids, name_prefix, price, big, millis, micros, seconds, opt): + return pa.RecordBatch.from_arrays( + [ + pa.array(ids, type=pa.int32()), + # Names exceed seven bytes so the string bounds spill out of + # the statistics row's inline slot. + pa.array([f"{name_prefix}-{i}" for i in ids]), + pa.array([price] * len(ids), type=pa.decimal128(10, 2)), + pa.array([big] * len(ids), type=pa.decimal128(22, 5)), + pa.array([millis] * len(ids), type=pa.timestamp("ms")), + pa.array([micros] * len(ids), type=pa.timestamp("us")), + pa.array([seconds] * len(ids), type=pa.time32("s")), + pa.array([opt] * len(ids), type=pa.int32()), + ], + schema=arrow_schema, + ) + + low_ids = [1, 2, 3] + high_ids = [101, 102, 103] + writer.write_arrow_batch( + batch( + low_ids, + "aaaaaaaaaaaa", + decimal.Decimal("10.01"), + decimal.Decimal("1.00001"), + datetime.datetime(2020, 1, 1, 0, 0, 0), + datetime.datetime(2020, 1, 1, 0, 0, 0, 123456), + datetime.time(9, 0, 0), + None, + ) + ) + await writer.flush() + writer.write_arrow_batch( + batch( + high_ids, + "zzzzzzzzzzzz", + decimal.Decimal("990.01"), + decimal.Decimal("9000000.00001"), + datetime.datetime(2030, 1, 1, 0, 0, 0), + datetime.datetime(2030, 1, 1, 0, 0, 0, 456789), + datetime.time(20, 0, 0), + 7, + ) + ) + await writer.flush() + + cases = [ + ("int", fluss.col("id") > 50, high_ids), + ("string", fluss.col("name") > "mmmmmmmmmmmm", high_ids), + ("compact decimal", fluss.col("price") > decimal.Decimal("500.00"), high_ids), + ("wide decimal", fluss.col("big") > decimal.Decimal("100.00000"), high_ids), + ( + "timestamp(3)", + fluss.col("ts3") > datetime.datetime(2025, 1, 1, 0, 0, 0), + high_ids, + ), + ( + "timestamp(6)", + fluss.col("ts6") > datetime.datetime(2025, 1, 1, 0, 0, 0, 500000), + high_ids, + ), + ("time(0)", fluss.col("t") > datetime.time(12, 0, 0), high_ids), + ("is_not_null", fluss.col("opt").is_not_null(), high_ids), + ("is_null", fluss.col("opt").is_null(), low_ids), + ] + + for label, predicate, expected in cases: + ids = await _filtered_ids(table, predicate, expected_count=len(expected)) + assert ids == expected, f"{label} filter should prune the other batch" + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_rejects_invalid_predicates(connection, admin): + """A filter is validated against the schema when the scanner is created.""" + table_path = fluss.TablePath("fluss", "py_test_filter_invalid") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("price", pa.decimal128(10, 2))] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + table = await connection.get_table(table_path) + + # Unknown column. + with pytest.raises(Exception, match="missing"): + await table.new_scan().filter(fluss.col("missing") == 1).create_log_scanner() + + # A literal the column's scale cannot hold exactly would move the bound. + with pytest.raises(Exception, match="does not fit"): + await ( + table.new_scan() + .filter(fluss.col("price") == decimal.Decimal("12.345")) + .create_log_scanner() + ) + + # A literal of the wrong type for the column. + with pytest.raises(Exception, match="does not match"): + await table.new_scan().filter(fluss.col("id") == "abc").create_log_scanner() + + # Unsupported Python literal, rejected while building the predicate. + with pytest.raises(TypeError, match="Unsupported filter literal"): + fluss.col("id") == object() + + # Limit pushdown and filter pushdown target different scanners. + with pytest.raises(Exception, match="limit"): + await table.new_scan().filter(fluss.col("id") > 1).limit(1).create_log_scanner() + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +@pytest.fixture +def non_utc_timezone(monkeypatch): + """Runs a test in a non-UTC zone, where a local-time misread would show.""" + monkeypatch.setenv("TZ", "Europe/Berlin") + time.tzset() + yield + monkeypatch.undo() + time.tzset() + + +async def test_filter_timestamp_literals_are_wall_clock( + connection, admin, non_utc_timezone +): + """A naive datetime is a wall clock, not local time.""" + table_path = fluss.TablePath("fluss", "py_test_filter_timestamp_wall_clock") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("ts", pa.timestamp("ms"))] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + written = datetime.datetime(2030, 1, 1, 12, 0, 0) + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [ + pa.array([1], type=pa.int32()), + pa.array([written], type=pa.timestamp("ms")), + ], + schema=arrow_schema, + ) + ) + await writer.flush() + + # A literal shifted by the local offset would keep this batch. + after = await _filtered_ids( + table, fluss.col("ts") > written + datetime.timedelta(seconds=1), 0 + ) + assert after == [], f"a bound after the row must prune it, got {after}" + + before = await _filtered_ids( + table, fluss.col("ts") > written - datetime.timedelta(seconds=1), 1 + ) + assert before == [1], f"a bound before the row must keep it, got {before}" + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_set_and_string_predicates(connection, admin): + """is_in, not_in and the string predicates reach the server.""" + table_path = fluss.TablePath("fluss", "py_test_filter_set_predicates") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("name", pa.string())] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + for ids, prefix in (([1, 2, 3], "low"), ([101, 102, 103], "high")): + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [ + pa.array(ids, type=pa.int32()), + pa.array([f"{prefix}-{i}" for i in ids]), + ], + schema=arrow_schema, + ) + ) + await writer.flush() + + assert await _filtered_ids(table, fluss.col("id").is_in([101, 102, 103]), 3) == [ + 101, + 102, + 103, + ] + # min/max cannot refute a NOT IN set, so both batches survive. + assert await _filtered_ids(table, fluss.col("id").not_in([1, 2, 3]), 6) == [ + 1, + 2, + 3, + 101, + 102, + 103, + ] + assert await _filtered_ids(table, fluss.col("name").starts_with("high"), 3) == [ + 101, + 102, + 103, + ] + # ends_with tells nothing about min/max, so neither batch can be pruned. + assert await _filtered_ids(table, fluss.col("name").ends_with("-102"), 6) == [ + 1, + 2, + 3, + 101, + 102, + 103, + ] + # An empty set matches nothing. + assert await _filtered_ids(table, fluss.col("id").is_in([]), 0) == [] + # Either branch of an OR keeps a batch. + both = fluss.col("id").is_in([1]) | fluss.col("id").is_in([101]) + assert await _filtered_ids(table, both, 6) == [1, 2, 3, 101, 102, 103] + + # Null comparisons go through is_null(), not `== None`. + with pytest.raises(Exception, match="is_null"): + await table.new_scan().filter(fluss.col("id") == None).create_log_scanner() # noqa: E711 + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +async def test_filter_float_literals(connection, admin): + """A double literal is rejected unless it narrows to the column exactly.""" + table_path = fluss.TablePath("fluss", "py_test_filter_float") + await admin.drop_table(table_path, ignore_if_not_exists=True) + + arrow_schema = pa.schema( + [pa.field("id", pa.int32()), pa.field("score", pa.float32())] + ) + schema = fluss.Schema(arrow_schema) + await admin.create_table( + table_path, _stats_descriptor(schema), ignore_if_exists=False + ) + + table = await connection.get_table(table_path) + writer = table.new_append().create_writer() + for ids, score in (([1, 2], 1.5), ([101, 102], 900.5)): + writer.write_arrow_batch( + pa.RecordBatch.from_arrays( + [ + pa.array(ids, type=pa.int32()), + pa.array([score] * len(ids), type=pa.float32()), + ], + schema=arrow_schema, + ) + ) + await writer.flush() + + assert await _filtered_ids(table, fluss.col("score") > 500.25, 2) == [101, 102] + + # 0.1 has no exact single-precision form, so the bound would move. + with pytest.raises(Exception, match="cannot be represented exactly"): + await table.new_scan().filter(fluss.col("score") > 0.1).create_log_scanner() + + await admin.drop_table(table_path, ignore_if_not_exists=False) + + +def test_predicate_rejects_python_boolean_operators(): + """`and`/`or` would silently return one side, so a predicate has no truth value.""" + low = fluss.col("id") >= 1 + high = fluss.col("id") >= 200 + + for build in (lambda: low and high, lambda: low or high, lambda: bool(low)): + with pytest.raises(TypeError, match="no truth value"): + build() + + with pytest.raises(TypeError, match="no truth value"): + bool(fluss.col("id")) + + # The operators are what actually combine them. + assert "Compound" in repr(low & high) + assert "Compound" in repr(low | high) + + +def test_predicate_rejects_out_of_range_integer_literal(): + """An int too large for INT64 must not silently become a float literal.""" + with pytest.raises(OverflowError): + fluss.col("id") > 2**70 diff --git a/website/docs/apis/python/api-reference.md b/website/docs/apis/python/api-reference.md index 54d8af665fc..de78c357c2f 100644 --- a/website/docs/apis/python/api-reference.md +++ b/website/docs/apis/python/api-reference.md @@ -97,10 +97,53 @@ Supports `async with` statement (async context manager). | `.project(indices) -> TableScan` | Project columns by index | | `.project_by_name(names) -> TableScan` | Project columns by name | | `.limit(n) -> TableScan` | Set a positive row limit (enables `create_bucket_batch_scanner`; rejected by log scanners) | +| `.filter(predicate) -> TableScan` | Push a filter down for server-side batch pruning (see [`col`](#col)) | | `await .create_log_scanner() -> LogScanner` | Create record-based scanner (for `poll()`); on a primary-key table, subscribes to its CDC changelog (per-record `change_type`) | | `await .create_record_batch_log_scanner() -> LogScanner` | Create batch-based scanner (for `poll_arrow()`, `to_arrow()`, etc.); log tables only — no per-record change types | | `.create_bucket_batch_scanner(bucket) -> BatchScanner` | Bounded scan of one bucket (requires `limit`; runs on first `next_batch()`) | +## `col` + +`fluss.col(name)` returns a `ColumnRef` used to build a filter for +`TableScan.filter()`. Pruning is conservative: a returned batch may still hold +non-matching rows, so apply the predicate again on the results. + +| Method | Description | +|-------------------------------------------|-----------------------------------------| +| `col(name) >= value` (`>`, `<`, `<=`, `==`, `!=`) | Comparison predicate | +| `.equal(v)`, `.not_equal(v)` | Same as `==` and `!=` | +| `.less_than(v)`, `.less_or_equal(v)` | Same as `<` and `<=` | +| `.greater_than(v)`, `.greater_or_equal(v)`| Same as `>` and `>=` | +| `.is_null()`, `.is_not_null()` | Prune on the batch's null count | +| `.starts_with(p)`, `.ends_with(s)`, `.contains(i)` | String predicates | +| `.is_in(values)`, `.not_in(values)` | Set membership | + +Predicates combine with `&` and `|` (or `and_()` and `or_()`). + +Literals are plain Python values, and the column's type decides how they are +encoded, so `col("id") >= 200` works for any integer column and an out-of-range +literal is rejected when the scanner is created: + +| Python type | Fluss column types | +|----------------------|-----------------------------------------------------| +| `bool` | BOOLEAN | +| `int` | TINYINT, SMALLINT, INT, BIGINT (range-checked) | +| `float` | DOUBLE, and FLOAT when exactly representable | +| `str` | CHAR, STRING | +| `bytes` | BINARY, BYTES | +| `None` | not comparable; use `.is_null()` / `.is_not_null()` | +| `decimal.Decimal` | DECIMAL, rejected if the column's scale cannot hold it exactly | +| `datetime.date` | DATE | +| `datetime.time` | TIME | +| `datetime.datetime` | TIMESTAMP when naive, TIMESTAMP_LTZ when tz-aware | + +```python +from fluss import col + +hot = (col("id") >= 200) & col("name").starts_with("high") +scanner = await table.new_scan().filter(hot).create_log_scanner() +``` + ## `TableAppend` Builder for creating an `AppendWriter`. Obtain via `FlussTable.new_append()`. diff --git a/website/docs/apis/python/example/log-tables.md b/website/docs/apis/python/example/log-tables.md index 9f45639440b..ea95a26f16f 100644 --- a/website/docs/apis/python/example/log-tables.md +++ b/website/docs/apis/python/example/log-tables.md @@ -128,6 +128,50 @@ scanner = await table.new_scan().project([0, 2]).create_record_batch_log_scanner scanner = await table.new_scan().project_by_name(["id", "score"]).create_record_batch_log_scanner() ``` +## Filter Pushdown + +A filter is pushed to the server, which uses each batch's statistics to skip +batches that cannot match. Enable statistics on the table first: + +```python +descriptor = fluss.TableDescriptor( + fluss.Schema(schema), + properties={"table.statistics.columns": "*"}, +) +``` + +Then build a predicate with `fluss.col(...)` and pass it to the scan: + +```python +from fluss import col + +predicate = (col("id") >= 200) & col("name").starts_with("high") +scanner = await table.new_scan().filter(predicate).create_log_scanner() +scanner.subscribe_buckets({0: fluss.EARLIEST_OFFSET}) + +records = await scanner.poll(1000) +# Pruning is batch-granular, so a kept batch can still contain rows that do +# not match. Re-apply the predicate on the rows you receive. +matching = [r.row for r in records if r.row["id"] >= 200] +``` + +Filters work with projection, including on a column that is not projected: + +```python +scanner = await ( + table.new_scan() + .project_by_name(["name"]) + .filter(col("id") >= 200) + .create_log_scanner() +) +``` + +Null counts drive `is_null()` and `is_not_null()` (a comparison against `None` +is rejected), and `decimal.Decimal`, +`datetime.date`, `datetime.time` and `datetime.datetime` filter their matching +column types. See the [`col` reference](../api-reference.md#col) for the full +literal mapping. + ## Limit Scan For a bounded read of up to `n` rows from a single bucket, use a batch scanner instead of subscribing. It issues one request; poll it with `next_batch()` until it returns `None`.