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 d41db48f02 [python] Evaluate Ray full-text prefilters on workers
(#10584)
d41db48f02 is described below
commit d41db48f02da55f9806dc40bc7f2342e290eae5e
Author: chaoyang <[email protected]>
AuthorDate: Fri Oct 16 13:01:23 2026 +0800
[python] Evaluate Ray full-text prefilters on workers (#10584)
---
paimon-python/pypaimon/ray/full_text_search.py | 65 ++++++++++++++-----
paimon-python/pypaimon/ray/vector_search.py | 4 +-
.../pypaimon/table/source/full_text_read.py | 34 ++++++----
.../pypaimon/tests/ray_full_text_search_test.py | 72 +++++++++++++++++++++-
4 files changed, 143 insertions(+), 32 deletions(-)
diff --git a/paimon-python/pypaimon/ray/full_text_search.py
b/paimon-python/pypaimon/ray/full_text_search.py
index a9d54e9e85..29114a0c36 100644
--- a/paimon-python/pypaimon/ray/full_text_search.py
+++ b/paimon-python/pypaimon/ray/full_text_search.py
@@ -21,10 +21,14 @@ from concurrent.futures import ThreadPoolExecutor,
as_completed
from contextlib import closing
from threading import Event
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
-from pypaimon.ray.vector_search import _execution_options, _map_tasks,
_require_ray, _scores
-from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead
+from pypaimon.ray.vector_search import _execution_options, _map_tasks,
_require_ray, _scores, _shared_scalar_filters
+from pypaimon.table.source.full_text_read import (
+ DataEvolutionFullTextRead, _index_type, _raw_row_ranges,
_split_search_splits,
+)
from pypaimon.table.source import global_index_live_row_filter
+from pypaimon.utils.range import Range
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
@@ -48,21 +52,26 @@ class _RayFullTextRead(DataEvolutionFullTextRead):
self._remote_args = remote_args
def _worker_reader(self):
- return DataEvolutionFullTextRead(
- self._table, self._limit, self._text_columns, self._query,
+ reader = DataEvolutionFullTextRead(
+ self._table.copy({"global-index.enabled": "false"}), self._limit,
self._text_columns, self._query,
self._partition_filter, self._filter)
+ reader._snapshot = self._snapshot
+ return reader
+
+ def read(self, splits):
+ # Keep bitmap construction and candidate data reads off the driver.
+ index_splits, raw_splits = _split_search_splits(splits)
+ indexed = self._eval_column_query({self._text_columns[0].name:
index_splits}, None)
+ raw = self._read_raw_search(_raw_row_ranges(raw_splits),
_index_type(index_splits))
+ return indexed.or_(raw).top_k(self._limit)
def _eval_column_query(self, splits_by_column, live_rows):
- items = []
- for split in splits_by_column.get(self._text_columns[0].name, []):
- include = global_index_live_row_filter.for_range(
- live_rows, split.row_range_start, split.row_range_end)
- if include is not None and include.is_empty():
- continue
- items.append((split, None if include is None else
include.serialize()))
+ items = splits_by_column.get(self._text_columns[0].name, [])
merged = {}
- with closing(_map_tasks(_search_index, self._worker_reader(), items,
- self._concurrency, self._remote_args, True))
as tasks:
+ with _shared_scalar_filters(items, self._filter, self._remote_args,
+ _prepare_scalar_filter) as prepare,
closing(_map_tasks(
+ _search_index, self._worker_reader(),
items,
+ self._concurrency, self._remote_args,
True, prepare)) as tasks:
for _, scores in tasks:
# Duplicate row IDs keep the first planned shard's score.
for row_id, score in scores.items():
@@ -81,16 +90,40 @@ class _RayFullTextRead(DataEvolutionFullTextRead):
return DictBasedScoredIndexResult(scores)
-def _search_index(reader, item):
- split, include = item
- include = None if include is None else RoaringBitmap64.deserialize(include)
+def _search_index(reader, split, scalar_filter=None):
+ live = global_index_live_row_filter.live_rows(
+ reader._table, reader._partition_filter, reader._snapshot,
+ row_ranges=[Range(split.row_range_start, split.row_range_end)])
+ if scalar_filter is None:
+ include = reader._matched_indexed_rows([split], live)
+ else:
+ bitmap, exact = scalar_filter
+ result = None if bitmap is None else GlobalIndexResult.create(
+ RoaringBitmap64.deserialize(bitmap), is_exact=exact)
+ include = reader._matched_indexed_rows([split], live, result)
+ if include is not None and include.is_empty():
+ return {}
return _scores(reader._eval(split.row_range_start, split.row_range_end,
split.full_text_index_files, include).result())
+def _prepare_scalar_filter(reader, split):
+ result = reader._scalar_index_result(split.scalar_index_files,
reader._snapshot)
+ return (None, False) if result is None else (result.results().serialize(),
result.is_exact())
+
+
def _search_raw(reader, item):
ranges, index_type, include = item
include = None if include is None else RoaringBitmap64.deserialize(include)
+ if reader._filter is not None:
+ live = global_index_live_row_filter.live_rows(
+ reader._table, reader._partition_filter, reader._snapshot,
row_ranges=ranges)
+ candidates = GlobalIndexResult.from_ranges(ranges).results()
+ if live is not None:
+ candidates = RoaringBitmap64.and_(candidates, live)
+ if include is not None:
+ candidates = RoaringBitmap64.and_(candidates, include)
+ include = reader._matching_rows(candidates)
return _scores(reader._read_raw_search(ranges, index_type, include))
diff --git a/paimon-python/pypaimon/ray/vector_search.py
b/paimon-python/pypaimon/ray/vector_search.py
index 4e7239bfd6..fcc6a0e8c7 100644
--- a/paimon-python/pypaimon/ray/vector_search.py
+++ b/paimon-python/pypaimon/ray/vector_search.py
@@ -165,7 +165,7 @@ def _prepare_scalar_filter(context, split):
@contextmanager
-def _shared_scalar_filters(splits, predicate, remote_args):
+def _shared_scalar_filters(splits, predicate, remote_args, worker=None):
"""Evaluate identical scalar-index inputs once, without fetching masks on
the driver."""
import ray
@@ -175,7 +175,7 @@ def _shared_scalar_filters(splits, predicate, remote_args):
counts = Counter(key(split) for split in splits) if predicate is not None
else {}
shared = {files for files, count in counts.items() if files and count > 1}
pending = {}
- remote = ray.remote(_prepare_scalar_filter).options(**remote_args) if
shared else None
+ remote = ray.remote(worker or
_prepare_scalar_filter).options(**remote_args) if shared else None
def prepare(context_ref, split):
files = key(split) if shared else None
diff --git a/paimon-python/pypaimon/table/source/full_text_read.py
b/paimon-python/pypaimon/table/source/full_text_read.py
index 022e448c71..21cc5a5cf9 100644
--- a/paimon-python/pypaimon/table/source/full_text_read.py
+++ b/paimon-python/pypaimon/table/source/full_text_read.py
@@ -41,6 +41,8 @@ from pypaimon.table.source.full_text_scan import
FullTextScanPlan
from pypaimon.utils.range import Range
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+_UNSET = object()
+
class FullTextRead(ABC):
"""Full-text read to read index files."""
@@ -111,12 +113,11 @@ class DataEvolutionFullTextRead(FullTextRead):
table = self._table.copy({'scalar-index.search-mode': 'full'})
return matching_rows(table, self._filter, candidates,
self._partition_filter)
- def _matched_indexed_rows(self, splits, live_rows):
+ def _matched_indexed_rows(self, splits, live_rows, scalar_result=_UNSET):
if self._filter is None:
return live_rows
from pypaimon.common.options.core_options import GlobalIndexSearchMode
from pypaimon.globalindex.data_evolution_global_index_coverage import
DataEvolutionGlobalIndexCoverage
- from pypaimon.globalindex.data_evolution_global_index_scanner import
DataEvolutionGlobalIndexScanner
from pypaimon.snapshot.time_travel_util import TimeTravelUtil
from pypaimon.table.source.global_index_row_filter import is_exact
@@ -140,17 +141,11 @@ class DataEvolutionFullTextRead(FullTextRead):
decided = RoaringBitmap64.remove_all(covered, unindexed)
matched = RoaringBitmap64()
if not decided.is_empty():
- scanner = DataEvolutionGlobalIndexScanner.create(self._table,
index_files=scalar_files,
-
snapshot=snapshot, partition_filter=self._partition_filter)
- evaluation = None
- if scanner is not None:
- try:
- evaluation = scanner.scan_with_coverage(self._filter)
- finally:
- scanner.close()
- if evaluation is not None:
- candidates = RoaringBitmap64.and_(evaluation.result.results(),
decided)
- if is_exact(self._filter, evaluation.result):
+ if scalar_result is _UNSET:
+ scalar_result = self._scalar_index_result(scalar_files,
snapshot)
+ if scalar_result is not None:
+ candidates = RoaringBitmap64.and_(scalar_result.results(),
decided)
+ if is_exact(self._filter, scalar_result):
matched = candidates
elif
self._table.options.global_index_filter_refine_from_data():
matched = self._matching_rows(candidates)
@@ -158,6 +153,19 @@ class DataEvolutionFullTextRead(FullTextRead):
unindexed = RoaringBitmap64.or_(unindexed, decided)
return RoaringBitmap64.or_(matched, self._matching_rows(unindexed))
+ def _scalar_index_result(self, scalar_files, snapshot):
+ from pypaimon.globalindex.data_evolution_global_index_scanner import
DataEvolutionGlobalIndexScanner
+
+ scanner = DataEvolutionGlobalIndexScanner.create(
+ self._table, index_files=scalar_files, snapshot=snapshot,
partition_filter=self._partition_filter)
+ if scanner is None:
+ return None
+ try:
+ evaluation = scanner.scan_with_coverage(self._filter)
+ return None if evaluation is None else evaluation.result
+ finally:
+ scanner.close()
+
def _eval_column_query(
self,
splits_by_column: Dict[str, List[IndexFullTextSearchSplit]],
diff --git a/paimon-python/pypaimon/tests/ray_full_text_search_test.py
b/paimon-python/pypaimon/tests/ray_full_text_search_test.py
index 057662e6b8..0c7f1bcf44 100644
--- a/paimon-python/pypaimon/tests/ray_full_text_search_test.py
+++ b/paimon-python/pypaimon/tests/ray_full_text_search_test.py
@@ -91,5 +91,75 @@ def test_multiple_index_shards_and_worker_failure(docs,
ray_cluster):
assert query.to_list(execution="ray") == query.to_list()
-def _fail_index(reader, item):
+def _fail_index(reader, item, scalar_filter=None):
raise RuntimeError("injected full-text failure")
+
+
[email protected]("scalar_index", [None, "btree", "bitmap"])
+def test_prefilter_and_deletion_vector_reads_stay_off_driver(docs,
ray_cluster, scalar_index):
+ from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead
+ from pypaimon.table.source import global_index_live_row_filter
+
+ if scalar_index:
+ docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("label", scalar_index)
+ text_fixtures.append_rows(docs)
+ docs.delete("id = 1")
+ query = docs.search("paimon", column="text", pre_filter="label LIKE
'%target%'").select(
+ ["id"]).with_score().order_by_score()
+ expected = query.to_list()
+ # Workers load their own module globals; these guards apply to the driver.
+ with patch.object(DataEvolutionFullTextRead, "_matched_indexed_rows",
+ side_effect=AssertionError("driver scalar evaluation")),
\
+ patch.object(DataEvolutionFullTextRead, "_matching_rows",
+ side_effect=AssertionError("driver candidate data
read")), \
+ patch.object(global_index_live_row_filter, "live_rows",
+ side_effect=AssertionError("driver deletion-vector
read")):
+ assert query.to_list(execution="ray", concurrency=2) == expected
+
+
+def test_shared_scalar_indexes_are_evaluated_once(tmp_path, ray_cluster):
+ import pyarrow as pa
+
+ docs = pm.connect(options={"warehouse": str(tmp_path /
"warehouse")}).create_table(
+ "docs", schema=pa.schema([("id", pa.int64()), ("text", pa.string()),
("label", pa.string())]),
+ options={"file.format": "parquet", "write.native.enabled": "false",
"commit.native.enabled": "false",
+ "full-text-index.search-mode": "full"})
+ docs.add([{"id": i, "text": "paimon " * (i + 1), "label": "keep" if i % 2
else "drop"}
+ for i in range(6)])
+ table = docs.raw_table.copy({"deletion-vectors.enabled": "false"})
+ table.create_global_index("label", "btree")
+ table.create_global_index("text", "full-text",
options={"global-index.row-count-per-shard": "2"})
+ query = docs.search("paimon", column="text", pre_filter="label =
'keep'").select(
+ ["id"]).with_score().order_by_score()
+ expected = query.to_list()
+ calls_path = str(tmp_path / "filter-calls")
+ original = module._prepare_scalar_filter
+
+ def record(reader, split):
+ with open(calls_path, "a") as calls:
+ calls.write("filter\n")
+ return original(reader, split)
+
+ with patch.object(module, "_prepare_scalar_filter", record):
+ assert query.to_list(execution="ray", concurrency=2) == expected
+ with open(calls_path) as calls:
+ assert calls.read().splitlines() == ["filter"]
+
+
+def test_worker_filters_use_planned_snapshot(docs, ray_cluster):
+ query = docs.search("paimon", column="text", pre_filter="id >=
0").select(["id"]).order_by_score()
+ expected = query.to_list()
+ original = module._map_tasks
+ committed = []
+
+ def commit_before_dispatch(*args):
+ if not committed:
+ docs.delete("id = 0")
+ committed.append(True)
+ return original(*args)
+
+ with patch.object(module, "_map_tasks", commit_before_dispatch):
+ assert query.to_list(execution="ray", concurrency=2) == expected
+ assert committed == [True]
+ assert query.to_list(execution="ray") == query.to_list()
+ assert query.to_list() != expected
