You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/08/27 13:38:39 UTC
[PR] [python] Create Python bindings for log filter pushdown [fluss]
fresh-borzoni opened a new pull request, #4127:
URL: https://github.com/apache/fluss/pull/4127
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #4127:
URL: https://github.com/apache/fluss/pull/4127
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #4127:
URL: https://github.com/apache/fluss/pull/4127#discussion_r3883219522
##########
fluss-rust/bindings/python/src/predicate.rs:
##########
@@ -0,0 +1,340 @@
+// 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, 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 {
Review Comment:
should we need a __bool__ as well? if we have something like `(id >= 1) or (id >= 200)`, will it pushes ids that are >=200? maybe add a test for this case? i tried, and seems returned based on id>=1 only
##########
fluss-rust/bindings/python/src/predicate.rs:
##########
@@ -0,0 +1,340 @@
+// 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, 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)
+ }
+
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Predicate> {
+ 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<Bound<'_, PyAny>>) -> PyResult<Predicate> {
+ 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<Bound<'_, PyAny>>) -> PyResult<Predicate> {
+ Ok(Predicate {
+ inner: self.column().not_in(literals_from_py(&values)?),
+ })
+ }
+
+ 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.
+///
+/// Args:
+/// name: Column name as declared in the table schema.
+#[pyfunction]
+pub fn col(name: String) -> ColumnRef {
+ ColumnRef::new(name)
+}
+
+fn literals_from_py(values: &[Bound<'_, PyAny>]) -> PyResult<Vec<Literal>> {
+ 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<Literal> {
+ if value.is_none() {
+ return Ok(Literal::Null);
+ }
+ if let Ok(flag) = value.cast_exact::<PyBool>() {
+ return Ok(Literal::Bool(flag.is_true()));
+ }
+ if let Ok(datetime) = value.cast::<PyDateTime>() {
+ return timestamp_literal(datetime);
+ }
+ if let Ok(date) = value.cast::<PyDate>() {
+ return date_literal(date);
+ }
+ if let Ok(time) = value.cast::<PyTime>() {
+ 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);
+ }
+ if let Ok(integer) = value.extract::<i64>() {
Review Comment:
if extract i64 fails, i think for a python int, extract f64 will still work... will this be a bug actually?
##########
fluss-rust/bindings/python/src/table.rs:
##########
@@ -656,6 +679,19 @@ fn resolve_projection_indices(
}
/// Apply projection to table scan
+/// Applies the filter, if one was set.
Review Comment:
i guess comment above should move to the next function? xD
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Create Python bindings for log filter pushdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #4127:
URL: https://github.com/apache/fluss/pull/4127#issuecomment-5440800388
@leekeiabstraction @charlesdong1991 @naivedogger PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #4127:
URL: https://github.com/apache/fluss/pull/4127#discussion_r3876408433
##########
fluss-rust/bindings/python/src/predicate.rs:
##########
@@ -0,0 +1,340 @@
+// 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, 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 {
Review Comment:
Curious why there are two flavours of and/or functions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4127:
URL: https://github.com/apache/fluss/pull/4127#discussion_r3883562370
##########
fluss-rust/bindings/python/src/table.rs:
##########
@@ -656,6 +679,19 @@ fn resolve_projection_indices(
}
/// Apply projection to table scan
+/// Applies the filter, if one was set.
Review Comment:
lol, yes 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #4127:
URL: https://github.com/apache/fluss/pull/4127#issuecomment-5457047640
@charlesdong1991 Addressed comments, PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [python] Python bindings - add log filter pushdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4127:
URL: https://github.com/apache/fluss/pull/4127#discussion_r3877174164
##########
fluss-rust/bindings/python/src/predicate.rs:
##########
@@ -0,0 +1,340 @@
+// 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, 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 {
Review Comment:
& and | are the primary form, since and/or are keywords and can't be overloaded. The named ones just mirror .and() in the Rust client and .And() in C++, the way polars pairs Expr.and_() with &.
We can leave one form, just felt it's more intuitive this way :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org