This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new c040921b Support BLOB and row-id predicates in Python native reads
(#906)
c040921b is described below
commit c040921b745132b1abaaec518d430ea4b590593a
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 21 21:31:37 2026 +0800
Support BLOB and row-id predicates in Python native reads (#906)
---
bindings/python/src/predicate.rs | 26 ++++++++++++--
bindings/python/src/read.rs | 24 +++++++++++--
bindings/python/tests/test_read.py | 67 +++++++++++++++++++++++++++++++++++++
crates/paimon/src/spec/predicate.rs | 18 ++++++++++
4 files changed, 131 insertions(+), 4 deletions(-)
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index 2a505a61..45898335 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -47,7 +47,7 @@ use pyo3::types::{
/// - `Decimal` accepts a `decimal.Decimal` or an `int`, rescaled losslessly to
/// the column's scale; anything needing rounding, exceeding the column's
/// precision, non-finite, or a binary `float` is rejected.
-/// - `Binary`/`VarBinary` accept Python `bytes` or `bytearray`, copied without
+/// - `Binary`/`VarBinary`/`Blob` accept Python `bytes` or `bytearray`, copied
without
/// decoding or changing their length.
/// - Complex types are not supported yet and raise
/// `NotImplementedError`.
@@ -82,7 +82,7 @@ pub(crate) fn py_to_datum(value: &Bound<'_, PyAny>,
data_type: &DataType) -> PyR
.map_err(|_| PyValueError::new_err("expected a str literal for
String field"))?;
Ok(Datum::String(s.to_str()?.to_string()))
}
- DataType::Binary(_) | DataType::VarBinary(_) => {
+ DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) => {
let bytes = if let Ok(bytes) = value.cast::<PyBytes>() {
bytes.as_bytes().to_vec()
} else if let Ok(bytes) = value.cast::<PyByteArray>() {
@@ -693,6 +693,28 @@ mod tests {
});
}
+ #[test]
+ fn blob_leaf_accepts_binary_literals() {
+ Python::attach(|py| {
+ let fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(paimon::spec::BlobType::new()),
+ )];
+ let dict = PyDict::new(py);
+ dict.set_item("method", "equal").unwrap();
+ dict.set_item("field", "payload").unwrap();
+ dict.set_item("literals", vec![PyBytes::new(py, &[0, 255, 0])])
+ .unwrap();
+ let predicate = dict_to_predicate(&dict, &fields, true).unwrap();
+ assert!(matches!(predicate, Predicate::Leaf { literals, .. }
+ if literals == vec![Datum::Bytes(vec![0, 255, 0])]));
+
+ dict.set_item("literals", vec!["not bytes"]).unwrap();
+ assert!(dict_to_predicate(&dict, &fields, true).is_err());
+ });
+ }
+
#[test]
fn unsupported_operator_not_raises_not_implemented() {
Python::attach(|py| {
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 120fdbf2..f3c00ca4 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -22,7 +22,9 @@ use std::sync::{Arc, Mutex};
use arrow::pyarrow::ToPyArrow;
use arrow::record_batch::RecordBatch;
use futures::TryStreamExt;
-use paimon::spec::{DataField, DataType, Predicate, RowType};
+use paimon::spec::{
+ BigIntType, DataField, DataType, Predicate, RowType, ROW_ID_FIELD_ID,
ROW_ID_FIELD_NAME,
+};
use paimon::table::{ArrowRecordBatchStream, DataSplit, IncrementalScanMode,
RowRange, Table};
use paimon_datafusion::runtime::runtime;
use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
@@ -376,7 +378,25 @@ impl PyReadBuilder {
mut slf: PyRefMut<'py, Self>,
predicate: &Bound<'_, PyDict>,
) -> PyResult<PyRefMut<'py, Self>> {
- let filter = dict_to_predicate(predicate, slf.table.schema().fields(),
slf.case_sensitive)?;
+ let mut fields = slf.table.schema().fields().to_vec();
+ // _ROW_ID is synthesized during read and absent from the table schema.
+ // Appending it preserves every physical column's original predicate
index.
+ // A real `_row_id` column takes precedence in case-insensitive mode.
+ let has_row_id_field = fields.iter().any(|field| {
+ if slf.case_sensitive {
+ field.name() == ROW_ID_FIELD_NAME
+ } else {
+ field.name().eq_ignore_ascii_case(ROW_ID_FIELD_NAME)
+ }
+ });
+ if !has_row_id_field {
+ fields.push(DataField::new(
+ ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME.to_string(),
+ DataType::BigInt(BigIntType::with_nullable(true)),
+ ));
+ }
+ let filter = dict_to_predicate(predicate, &fields,
slf.case_sensitive)?;
slf.filter = Some(filter);
Ok(slf)
}
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index a6240546..0892c245 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -204,6 +204,49 @@ def test_with_row_ranges():
table.new_read_builder().with_row_ranges([(2, 1)])
[email protected]("data_evolution", [False, True])
[email protected]("case_sensitive", [False, True])
+def test_row_id_filter_with_projection_and_data_predicate(data_evolution,
case_sensitive):
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.rid")
+ options = "'row-tracking.enabled' = 'true'"
+ if data_evolution:
+ options += ", 'data-evolution.enabled' = 'true'"
+ ctx.sql(f"CREATE TABLE paimon.rid.t (id INT, name STRING) WITH
({options})")
+ ctx.sql("INSERT INTO paimon.rid.t (id, name) VALUES (1, 'a'), (2,
'b'), (3, 'c')")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("rid.t")
+ builder = table.new_read_builder().with_case_sensitive(case_sensitive)
+ builder.with_projection(["name", "_ROW_ID"])
+ row_id_field = "_ROW_ID" if case_sensitive else "_row_id"
+ builder.with_filter({"method": "and", "children": [
+ {"method": "equal", "field": row_id_field, "literals": [1]},
+ {"method": "equal", "field": "id", "literals": [2]},
+ ]})
+ rows = pa.Table.from_batches(builder.new_read().read(
+ builder.new_scan().plan().splits()))
+ assert rows.to_pylist() == [{"name": "b", "_ROW_ID": 1}]
+
+
[email protected]("predicate_field", ["_row_id", "_ROW_ID"])
+def
test_case_insensitive_filter_uses_real_lowercase_row_id_column(predicate_field):
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.realrowid")
+ ctx.sql("CREATE TABLE paimon.realrowid.t (_row_id BIGINT, id INT)")
+ ctx.sql("INSERT INTO paimon.realrowid.t VALUES (11, 1), (22, 2)")
+ table = PaimonCatalog({"warehouse":
warehouse}).get_table("realrowid.t")
+ builder = table.new_read_builder().with_case_sensitive(False)
+ builder.with_projection(["_row_id", "id"])
+ builder.with_filter({
+ "method": "equal", "field": predicate_field, "literals": [22],
+ })
+ batches = builder.new_read().read(builder.new_scan().plan().splits())
+ assert pa.Table.from_batches(batches).to_pylist() == [{"_row_id": 22,
"id": 2}]
+
+
def test_row_tracking_append_row_ranges_keep_global_row_ids():
with tempfile.TemporaryDirectory() as warehouse:
ctx = SQLContext()
@@ -586,6 +629,30 @@ def
test_filter_binary_literal_rejects_other_types(literal):
{"method": "equal", "field": "payload", "literals": [literal]})
+def test_blob_predicate_with_limit_reads_matching_payload():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.blobpred")
+ ctx.sql("""CREATE TABLE paimon.blobpred.t (id INT, payload BLOB) WITH (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true')""")
+ ctx.sql("""INSERT INTO paimon.blobpred.t (id, payload) VALUES
+ (1, X'01'), (2, X'02'), (3, X'03')""")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("blobpred.t")
+ builder = table.new_read_builder().with_limit(1).with_filter({
+ "method": "equal", "field": "payload", "literals": [b"\x02"],
+ })
+ batches = builder.new_read().read(builder.new_scan().plan().splits())
+ assert pa.Table.from_batches(batches).to_pylist() == [
+ {"id": 2, "payload": b"\x02"}]
+
+ with pytest.raises(ValueError, match="bytes or bytearray"):
+ table.new_read_builder().with_filter({
+ "method": "equal", "field": "payload", "literals": ["02"],
+ })
+
+
def test_empty_plan_preserves_snapshot_id():
with tempfile.TemporaryDirectory() as warehouse:
ctx = SQLContext()
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 82a93eac..dd437fdf 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -1563,6 +1563,7 @@ fn validate_datum_matches_type(datum: &Datum, data_type:
&DataType) -> Result<()
| (Datum::Decimal { .. }, DataType::Decimal(_))
| (Datum::Bytes(_), DataType::Binary(_))
| (Datum::Bytes(_), DataType::VarBinary(_))
+ | (Datum::Bytes(_), DataType::Blob(_))
| (Datum::Variant { .. }, DataType::Variant(_))
);
if !ok {
@@ -2082,6 +2083,23 @@ mod tests {
}
}
+ #[test]
+ fn test_builder_accepts_blob_bytes_without_accepting_wrong_literal_type() {
+ let fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let pb = PredicateBuilder::new(&fields);
+ let bytes = vec![0, 255, 0];
+ let predicate = pb.equal("payload",
Datum::Bytes(bytes.clone())).unwrap();
+ assert!(matches!(predicate, Predicate::Leaf { literals, .. }
+ if literals == vec![Datum::Bytes(bytes)]));
+ assert!(pb
+ .equal("payload", Datum::String("wrong".to_string()))
+ .is_err());
+ }
+
#[test]
fn test_builder_comparison_ops() {
let pb = PredicateBuilder::new(&test_fields());
