This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new c1e5ed1ae3 [python] Support string filters on system tables (#10631)
c1e5ed1ae3 is described below
commit c1e5ed1ae3a531dda79eca4e5debd34da0f0c923
Author: QuakeWang <[email protected]>
AuthorDate: Sun Oct 18 16:48:18 2026 +0800
[python] Support string filters on system tables (#10631)
---
.../pypaimon/table/system/system_table_read.py | 92 +++++--
.../pypaimon/tests/system/files_table_test.py | 14 ++
.../pypaimon/tests/system/options_table_test.py | 24 ++
.../tests/system/system_string_filter_test.py | 278 +++++++++++++++++++++
.../pypaimon/tests/system/system_table_test.py | 13 +-
5 files changed, 395 insertions(+), 26 deletions(-)
diff --git a/paimon-python/pypaimon/table/system/system_table_read.py
b/paimon-python/pypaimon/table/system/system_table_read.py
index e7f4c81736..ece7312d6b 100644
--- a/paimon-python/pypaimon/table/system/system_table_read.py
+++ b/paimon-python/pypaimon/table/system/system_table_read.py
@@ -25,26 +25,25 @@ tables produced by :class:`SystemSplit` — no manifest
reads, no data
files.
"""
+import re
from typing import Any, Iterator, List, Optional, TYPE_CHECKING
import pandas
import pyarrow
-from pypaimon.common.predicate import Predicate
+from pypaimon.common.predicate import Like, Predicate
from pypaimon.read.split import Split
from pypaimon.schema.data_types import (
AtomicType, DataField, PyarrowFieldParser)
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.system.system_table_scan import SystemSplit
if TYPE_CHECKING: # pragma: no cover - type-only import
from pypaimon.table.system.system_table import SystemTable
-_PREDICATE_NOT_SUPPORTED = (
- "string-match predicate pushdown (starts_with / ends_with / contains / "
- "like) on system tables is not supported yet; filter the resulting "
- "PyArrow / pandas table on the client side"
-)
+_PREDICATE_NOT_SUPPORTED = "This predicate is not supported on system tables"
+_STRING_PREDICATES = ("startsWith", "endsWith", "contains", "like")
# Value comparisons (equal / greater-than / is-in / ...) have no PyArrow
# compute kernel for nested columns such as ``$files.write_cols`` or
@@ -102,7 +101,8 @@ class SystemTableRead:
return con
def _materialise(self, splits: List[Split]) -> pyarrow.Table:
- from pypaimon.read.push_down_utils import
predicate_supports_arrow_filter
+ from pypaimon.read.push_down_utils import (
+ predicate_field_names, predicate_supports_arrow_filter,
rewrite_predicate_indices)
predicate = self.predicate
if predicate is not None and not isinstance(predicate, Predicate):
@@ -111,19 +111,23 @@ class SystemTableRead:
# arrow_filter dereferences ``.method`` and would otherwise leak
# an internal AttributeError.
raise NotImplementedError(_PREDICATE_NOT_SUPPORTED)
- if predicate is not None and not predicate_supports_arrow_filter(
+ has_string_filter = self._validate_filter(predicate) if predicate is
not None else False
+ if predicate is not None and not has_string_filter and not
predicate_supports_arrow_filter(
predicate
):
raise NotImplementedError(_PREDICATE_NOT_SUPPORTED)
- if predicate is not None:
- self._reject_non_scalar_filter_fields(predicate)
if not splits:
return self._empty_table()
- arrow_predicate = (
- self.predicate.to_arrow() if self.predicate is not None else None
- )
+ row_predicate = None
+ arrow_predicate = None
+ if has_string_filter:
+ predicate_fields = predicate_field_names(predicate)
+ filter_fields = [f for f in self.system_table.row_type().fields if
f.name in predicate_fields]
+ row_predicate = rewrite_predicate_indices(predicate, filter_fields)
+ elif predicate is not None:
+ arrow_predicate = predicate.to_arrow()
projected_names = [f.name for f in self.read_type]
slices: List[pyarrow.Table] = []
@@ -136,7 +140,9 @@ class SystemTableRead:
arrow_table = split.arrow_table()
# Filter on the full schema before projection so a predicate may
# reference columns that are not part of the projection.
- if arrow_predicate is not None:
+ if row_predicate is not None:
+ arrow_table = self._filter_rows(arrow_table, row_predicate,
filter_fields)
+ elif arrow_predicate is not None:
arrow_table = arrow_table.filter(arrow_predicate)
if projected_names:
arrow_table = arrow_table.select(projected_names)
@@ -147,8 +153,45 @@ class SystemTableRead:
combined = combined.slice(0, self.limit)
return
combined.cast(PyarrowFieldParser.from_paimon_schema(self.read_type))
- def _reject_non_scalar_filter_fields(self, predicate: Predicate) -> None:
- """Reject value comparisons on nested system-table columns.
+ def _filter_rows(
+ self, arrow_table: pyarrow.Table, predicate: Predicate, fields:
List[DataField]
+ ) -> pyarrow.Table:
+ # String to_arrow() conversions may silently fall back to True or None.
+ # Evaluate the whole tree locally, including AND / OR.
+ columns = [arrow_table.column(f.name).to_pylist() for f in fields]
+ mask = pyarrow.array([
+ self._test_row(predicate, GenericRow(list(values), fields))
+ for values in zip(*columns)
+ ], type=pyarrow.bool_())
+ return arrow_table.filter(mask)
+
+ def _test_row(self, predicate: Predicate, row: GenericRow) -> bool:
+ if predicate.method in ("and", "or"):
+ results = (self._test_row(child, row) for child in
predicate.literals)
+ return all(results) if predicate.method == "and" else any(results)
+ if predicate.method in (
+ "equal", "notEqual", "lessThan", "lessOrEqual", "greaterThan",
"greaterOrEqual"
+ ) and predicate.literals[0] is None:
+ # Arrow comparisons with NULL never select a row.
+ return False
+ if predicate.method in ("between", "notBetween") and any(v is None for
v in predicate.literals):
+ value = row.get_field(predicate.index)
+ if predicate.method == "between" or value is None:
+ return False
+ # NOT BETWEEN is an OR: a non-null bound can still match.
+ lower, upper = predicate.literals
+ return ((lower is not None and value < lower)
+ or (upper is not None and value > upper))
+ if predicate.method == "like":
+ value = row.get_field(predicate.index)
+ # SQL '_' matches a newline too, as in Java's Like predicate.
+ return value is not None and re.fullmatch(
+ Like._sql_like_to_regex(predicate.literals[0]), value,
flags=re.DOTALL
+ ) is not None
+ return predicate.test(row)
+
+ def _validate_filter(self, predicate: Predicate) -> bool:
+ """Validate operands and report whether exact row evaluation is needed.
System tables expose list/map/row columns (e.g. ``$files.write_cols``,
``$table_indexes.dv_ranges``). PyArrow has no equality / comparison /
@@ -160,12 +203,15 @@ class SystemTableRead:
field_types = {
f.name: f.type for f in self.system_table.row_type().fields
}
+ has_string_filter = False
pending: List[Predicate] = [predicate]
while pending:
current = pending.pop()
if current.method in ("and", "or"):
pending.extend(current.literals or [])
continue
+ if current.method not in Predicate.testers:
+ raise ValueError("Unsupported predicate method: " +
current.method)
if current.method in _COMPARISON_NEEDS_SCALAR:
continue
field_type = field_types.get(current.field)
@@ -180,6 +226,20 @@ class SystemTableRead:
+ ") is not supported; filter the resulting PyArrow / "
+ "pandas table on the client side"
)
+ if current.method in _STRING_PREDICATES:
+ if field_type is None:
+ raise ValueError("Unknown system-table column: " +
repr(current.field))
+ if not
pyarrow.types.is_string(PyarrowFieldParser.from_paimon_type(field_type)):
+ raise NotImplementedError(
+ "String predicates require a string system-table
column: " + repr(current.field))
+ if (not current.literals or len(current.literals) != 1
+ or not isinstance(current.literals[0], str)):
+ raise ValueError("String predicates require a single
string literal")
+ if current.method == "like":
+ # Validate even when input is empty or another branch
matches.
+ Like._sql_like_to_regex(current.literals[0])
+ has_string_filter = True
+ return has_string_filter
def _empty_table(self) -> pyarrow.Table:
target_schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
diff --git a/paimon-python/pypaimon/tests/system/files_table_test.py
b/paimon-python/pypaimon/tests/system/files_table_test.py
index df15a3a59c..ad0e99fc89 100644
--- a/paimon-python/pypaimon/tests/system/files_table_test.py
+++ b/paimon-python/pypaimon/tests/system/files_table_test.py
@@ -97,6 +97,20 @@ class FilesTableTest(unittest.TestCase):
table = self.catalog.get_table("db.t$files")
self.assertIsInstance(table, FilesTable)
+ def test_file_path_suffix_filter(self):
+ self._create_partitioned_table()
+ self._write_two_partitions()
+ table = self.catalog.get_table("db.t$files")
+ expected = _read(table).column("file_path").to_pylist()
+ self.assertTrue(expected)
+ rb = table.new_read_builder()
+ pb = rb.new_predicate_builder()
+ for suffix, paths in ((".parquet", expected), (".missing", [])):
+ with self.subTest(suffix=suffix):
+ rb.with_filter(pb.endswith("file_path",
suffix)).with_projection(["file_path"])
+ result = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ self.assertEqual(result.column("file_path").to_pylist(), paths)
+
def test_schema_column_layout(self):
self._create_partitioned_table()
table = self.catalog.get_table("db.t$files")
diff --git a/paimon-python/pypaimon/tests/system/options_table_test.py
b/paimon-python/pypaimon/tests/system/options_table_test.py
index c78b79e701..64cc9ae3c8 100644
--- a/paimon-python/pypaimon/tests/system/options_table_test.py
+++ b/paimon-python/pypaimon/tests/system/options_table_test.py
@@ -83,6 +83,30 @@ class OptionsTableTest(unittest.TestCase):
self.assertEqual(["key"], arrow_table.schema.names)
self.assertIn("bucket", arrow_table.column("key").to_pylist())
+ def test_prefix_filter_before_projection_and_limit(self):
+ table = self.catalog.get_table("db.t$options")
+ rb = table.new_read_builder()
+ predicate = rb.new_predicate_builder().startswith("key", "user-")
+ rb.with_filter(predicate).with_projection(["value"]).with_limit(1)
+ result = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+ self.assertEqual(result.to_pylist(), [{"value": "alpha"}])
+
+ def test_prefix_filter_preserves_null_comparison_semantics(self):
+ table = self.catalog.get_table("db.t$options")
+ rb = table.new_read_builder()
+ pb = rb.new_predicate_builder()
+ for method in ("equal", "not_equal", "less_than", "less_or_equal",
"greater_than", "greater_or_equal"):
+ with self.subTest(method=method):
+ comparison = getattr(pb, method)("value", None)
+ rb.with_filter(comparison)
+
self.assertEqual(rb.new_read().to_arrow(rb.new_scan().plan().splits()).to_pylist(),
[])
+ rb.with_filter(pb.and_predicates([pb.startswith("key",
"user-"), comparison]))
+
self.assertEqual(rb.new_read().to_arrow(rb.new_scan().plan().splits()).to_pylist(),
[])
+ rb.with_filter(pb.or_predicates([comparison,
pb.startswith("key", "user-")]))
+
self.assertEqual(rb.new_read().to_arrow(rb.new_scan().plan().splits()).to_pylist(),
[
+ {"key": "user-flag", "value": "alpha"},
+ ])
+
if __name__ == "__main__":
unittest.main()
diff --git a/paimon-python/pypaimon/tests/system/system_string_filter_test.py
b/paimon-python/pypaimon/tests/system/system_string_filter_test.py
new file mode 100644
index 0000000000..c42453107a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/system/system_string_filter_test.py
@@ -0,0 +1,278 @@
+# 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.
+
+"""Exact string filtering on in-memory system-table metadata."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.common.predicate import Predicate
+from pypaimon.schema.data_types import AtomicType, DataField, RowType
+from pypaimon.table.system.system_table_scan import SystemSplit
+from pypaimon.tests.system.system_table_test import (
+ _ArraySystemTable, _DataSystemTable, _fake_base)
+
+
+class _StringSystemTable(_DataSystemTable):
+
+ def _build_arrow_table(self):
+ return pa.table({
+ "id": pa.array(range(1, 11), type=pa.int32()),
+ "name": pa.array([
+ "skip.orc", "data-1.parquet", "data-2.orc", "other.parquet",
+ None, "", "DATA-3.parquet", "metric_100%",
"\u6587\u4ef6.parquet", "path\\file",
+ ]),
+ })
+
+
[email protected]
+def table():
+ return _StringSystemTable(_fake_base())
+
+
+def _read(builder):
+ return builder.new_read().to_arrow(builder.new_scan().plan().splits())
+
+
[email protected]("method,pattern,expected", [
+ ("startswith", "data-", [2, 3]),
+ ("endswith", ".parquet", [2, 4, 7, 9]),
+ ("contains", "parquet", [2, 4, 7, 9]),
+ ("like", "data-_.parquet", [2]),
+])
[email protected]("combination", ["leaf", "and", "or"])
+def test_string_predicates_and_combinations(table, method, pattern, expected,
combination):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = getattr(pb, method)("name", pattern)
+ if combination == "and":
+ predicate = pb.and_predicates([predicate, pb.greater_than("id", 2)])
+ expected = [i for i in expected if i > 2]
+ elif combination == "or":
+ predicate = pb.or_predicates([pb.equal("id", 1), predicate])
+ expected = [1] + expected
+ result = _read(builder.with_filter(predicate).with_projection(["id"]))
+ assert result.column("id").to_pylist() == expected
+
+
[email protected]("method", [
+ "equal", "not_equal", "less_than", "less_or_equal", "greater_than",
"greater_or_equal",
+])
[email protected]("field", ["id", "name"])
[email protected]("combination", ["and", "or"])
+def test_mixed_null_comparisons_preserve_arrow_semantics(table, method, field,
combination):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ comparison = getattr(pb, method)(field, None)
+ assert _read(builder.with_filter(comparison)).num_rows == 0
+ predicate = getattr(pb, combination + "_predicates")([
+ comparison, pb.contains("name", "parquet"),
+ ])
+ result = _read(builder.with_filter(predicate).with_projection(["id"]))
+ assert result.column("id").to_pylist() == ([] if combination == "and" else
[2, 4, 7, 9])
+
+
[email protected]("method,literals,expected", [
+ ("between", [None, 3], []),
+ ("between", [3, None], []),
+ ("between", [None, None], []),
+ ("not_between", [None, 3], [4, 5, 6, 7, 8, 9, 10]),
+ ("not_between", [3, None], [1, 2]),
+ ("not_between", [None, None], []),
+ ("is_in", [[None, 2]], [2]),
+ ("is_in", [[None]], []),
+ ("is_not_in", [[None, 2]], []),
+])
[email protected]("combination", ["and", "or"])
+def test_mixed_null_range_and_set_comparisons(table, method, literals,
expected, combination):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ comparison = getattr(pb, method)("id", *literals)
+ assert _read(builder.with_filter(comparison)).column("id").to_pylist() ==
expected
+ predicate = getattr(pb, combination + "_predicates")([
+ comparison, pb.contains("name", "parquet"),
+ ])
+ string_ids = {2, 4, 7, 9}
+ expected = sorted(set(expected) & string_ids if combination == "and" else
set(expected) | string_ids)
+ assert _read(builder.with_filter(predicate)).column("id").to_pylist() ==
expected
+
+
[email protected]("method,pattern,expected", [
+ ("startswith", "", [1, 2, 3, 4, 6, 7, 8, 9, 10]),
+ ("endswith", "", [1, 2, 3, 4, 6, 7, 8, 9, 10]),
+ ("contains", "", [1, 2, 3, 4, 6, 7, 8, 9, 10]),
+ ("like", "%", [1, 2, 3, 4, 6, 7, 8, 9, 10]),
+ ("like", "", [6]),
+ ("contains", "%", [8]),
+ ("contains", "_", [8]),
+ ("contains", "\u6587\u4ef6", [9]),
+ ("like", r"metric\_100\%", [8]),
+ ("like", r"path\\file", [10]),
+ ("like", "%.parquet", [2, 4, 7, 9]),
+ ("like", "missing%", []),
+])
+def test_string_matching_edge_cases(table, method, pattern, expected):
+ builder = table.new_read_builder()
+ predicate = getattr(builder.new_predicate_builder(), method)("name",
pattern)
+ assert _read(builder.with_filter(predicate)).column("id").to_pylist() ==
expected
+
+
[email protected]("pattern,expected", [("a_b", [1, 2]), ("a%b", [1,
2]), ("a.b", [2])])
+def test_like_wildcards_match_newlines_and_escape_regex_characters(table,
pattern, expected):
+ rows = pa.table({"id": pa.array([1, 2], type=pa.int32()), "name": ["a\nb",
"a.b"]})
+ with patch.object(table, "_build_arrow_table", return_value=rows):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = pb.and_predicates([pb.like("name", pattern),
pb.greater_than("id", 0)])
+ assert _read(builder.with_filter(predicate)).column("id").to_pylist()
== expected
+
+
+def test_nested_predicate_never_uses_lossy_arrow_conversion(table):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = pb.and_predicates([
+ pb.or_predicates([pb.contains("name", "parquet"), pb.is_null("name")]),
+ pb.greater_than("id", 3),
+ ])
+ with patch.object(Predicate, "to_arrow",
side_effect=AssertionError("Unsafe Arrow conversion")):
+ result = _read(builder.with_filter(predicate))
+ assert result.column("id").to_pylist() == [4, 5, 7, 9]
+
+
+def test_combines_multiple_string_predicates(table):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = pb.and_predicates([
+ pb.or_predicates([pb.startswith("name", "data-"), pb.endswith("name",
".parquet")]),
+ pb.contains("name", "parquet"),
+ ])
+ assert _read(builder.with_filter(predicate)).column("id").to_pylist() ==
[2, 4, 7, 9]
+
+
+def test_does_not_convert_unrelated_columns_to_python(table):
+ fields = table.row_type().fields + [DataField(2, "event_time",
AtomicType("TIMESTAMP(6)"))]
+ # Year 10000 fits Arrow timestamps but cannot be converted to datetime.
+ rows = table._build_arrow_table().append_column(
+ "event_time", pa.array([253402300800000000] * 10,
type=pa.timestamp("us")))
+ with patch.object(table, "row_type", return_value=RowType(False, fields)):
+ with patch.object(table, "_build_arrow_table", return_value=rows):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().startswith("name",
"data-")
+ result =
_read(builder.with_filter(predicate).with_projection(["id"]))
+ assert result.to_pylist() == [{"id": 2}, {"id": 3}]
+
+
+def test_string_filter_precedes_projection_and_global_limit(table):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().endswith("name", ".parquet")
+ builder.with_filter(predicate).with_projection(["id"]).with_limit(2)
+ rows = table._build_arrow_table()
+ splits = [SystemSplit(rows.slice(0, 3)), SystemSplit(rows.slice(3))]
+ assert builder.new_read().to_arrow(splits).to_pylist() == [{"id": 2},
{"id": 4}]
+
+
+def test_projected_predicate_indices_are_rebound_without_mutation(table):
+ builder = table.new_read_builder().with_projection(["name", "id"])
+ pb = builder.new_predicate_builder()
+ string_predicate = pb.startswith("name", "data-")
+ predicate = pb.and_predicates([string_predicate, pb.equal("id", 3)])
+ assert _read(builder.with_filter(predicate)).to_pylist() == [{"name":
"data-2.orc", "id": 3}]
+ assert string_predicate.index == 0
+
+
[email protected]("empty_splits", [False, True])
+def test_string_filter_on_empty_input_keeps_projected_schema(table,
empty_splits):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().contains("name", "parquet")
+ builder.with_filter(predicate).with_projection(["id"])
+ splits = [] if empty_splits else
[SystemSplit(table._build_arrow_table().slice(0, 0))]
+ result = builder.new_read().to_arrow(splits)
+ assert result.num_rows == 0
+ assert result.schema == pa.schema([pa.field("id", pa.int32(),
nullable=False)])
+
+
[email protected]("method", ["startswith", "endswith", "contains",
"like"])
[email protected]("field", ["id", "tags"])
+def test_string_predicates_require_string_columns(method, field):
+ builder = _ArraySystemTable(_fake_base()).new_read_builder()
+ predicate = getattr(builder.new_predicate_builder(), method)(field, "a")
+ with pytest.raises(NotImplementedError, match="column"):
+ _read(builder.with_filter(predicate))
+
+
[email protected]("method", ["startswith", "endswith", "contains",
"like"])
[email protected]("pattern", [None, 1, b"data"])
+def test_string_patterns_must_be_strings_even_without_splits(table, method,
pattern):
+ builder = table.new_read_builder()
+ predicate = getattr(builder.new_predicate_builder(), method)("name",
pattern)
+ with pytest.raises(ValueError, match="string literal"):
+ builder.with_filter(predicate).new_read().to_arrow([])
+
+
[email protected]("pattern", ["bad\\", r"bad\q"])
+def test_invalid_like_escape_is_rejected_even_in_unused_branch(table, pattern):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = pb.or_predicates([pb.is_not_null("id"), pb.like("name",
pattern)])
+ with pytest.raises(ValueError, match="Invalid escape sequence"):
+ _read(builder.with_filter(predicate))
+
+
[email protected]("literals", [None, [], ["a", "b"]])
+def test_string_pattern_requires_exactly_one_literal(table, literals):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().contains("name",
"a").new_literals(literals)
+ with pytest.raises(ValueError, match="single string literal"):
+ _read(builder.with_filter(predicate))
+
+
+def test_unknown_predicate_is_rejected_in_unused_branch(table):
+ builder = table.new_read_builder()
+ pb = builder.new_predicate_builder()
+ predicate = pb.or_predicates([
+ pb.contains("name", ""), Predicate("unknown", 0, "id", []),
+ ])
+ with pytest.raises(ValueError, match="Unsupported predicate method"):
+ _read(builder.with_filter(predicate))
+
+
[email protected]("output", ["to_arrow", "to_pandas", "to_iterator",
"to_record_batch_iterator"])
+def test_string_filter_applies_to_all_outputs(table, output):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().startswith("name", "data-")
+ builder.with_filter(predicate).with_projection(["id"]).with_limit(1)
+ result = getattr(builder.new_read(),
output)(builder.new_scan().plan().splits())
+ if output == "to_pandas":
+ result = result.to_dict("records")
+ elif output == "to_iterator":
+ result = list(result)
+ elif output == "to_record_batch_iterator":
+ result = pa.Table.from_batches(list(result)).to_pylist()
+ else:
+ result = result.to_pylist()
+ assert result == [{"id": 2}]
+
+
+def test_native_result_handoff_uses_string_filter(table):
+ builder = table.new_read_builder()
+ predicate = builder.new_predicate_builder().contains("name", "parquet")
+ rows = table._build_arrow_table()
+ with
patch("pypaimon.table.system.native_system_table.try_read_native_system_table",
return_value=rows):
+ with patch.object(table, "_build_arrow_table",
side_effect=AssertionError("Metadata rebuilt")):
+ assert
_read(builder.with_filter(predicate)).column("id").to_pylist() == [2, 4, 7, 9]
diff --git a/paimon-python/pypaimon/tests/system/system_table_test.py
b/paimon-python/pypaimon/tests/system/system_table_test.py
index 2bf76319ef..7ae9a6f2a6 100644
--- a/paimon-python/pypaimon/tests/system/system_table_test.py
+++ b/paimon-python/pypaimon/tests/system/system_table_test.py
@@ -194,16 +194,9 @@ class SystemTableFilterTest(unittest.TestCase):
self.assertEqual(table.column_names, ["name"])
self.assertEqual(table.column("name").to_pylist(), ["c"])
- def test_string_match_filter_still_raises(self):
- # starts_with / ends_with / contains / like are not safe as final
- # Arrow row filters, so they still surface a clear NotImplementedError.
- pred = self._predicate_builder().contains("name", "b")
- sys_table = _DataSystemTable(_fake_base())
- rb = sys_table.new_read_builder().with_filter(pred)
- splits = rb.new_scan().plan().splits()
- read = rb.new_read()
- with self.assertRaises(NotImplementedError):
- read.to_arrow(splits)
+ def test_string_match_filter_selects_matching_rows(self):
+ table = self._read(self._predicate_builder().contains("name", "b"))
+ self.assertEqual(table.column("id").to_pylist(), [2])
def test_comparison_on_array_column_raises(self):
# PyArrow's ArrowNotImplementedError (a NotImplementedError subclass)
