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 cd39bd7056 [python] Fix union global index reader semantics (#8742)
cd39bd7056 is described below
commit cd39bd70567eb27909142523c0b4faace2c8b540
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 20 13:37:00 2026 +0800
[python] Fix union global index reader semantics (#8742)
- propagate unsupported scalar results across every child of
`UnionGlobalIndexReader`
- make the default batch-vector fan-out non-blocking and preserve
asynchronous failures
- close every union child before propagating the first close failure
- add focused regression tests for scalar, batch-vector, and close
semantics
---
.../pypaimon/globalindex/global_index_reader.py | 35 ++++-
.../globalindex/union_global_index_reader.py | 61 +++++---
.../tests/union_global_index_reader_test.py | 160 +++++++++++++++++++++
3 files changed, 228 insertions(+), 28 deletions(-)
diff --git a/paimon-python/pypaimon/globalindex/global_index_reader.py
b/paimon-python/pypaimon/globalindex/global_index_reader.py
index d3915583ca..2a808cc34c 100644
--- a/paimon-python/pypaimon/globalindex/global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/global_index_reader.py
@@ -17,6 +17,7 @@
"""Global index reader interface."""
+import threading
from abc import ABC, abstractmethod
from concurrent.futures import Future
from typing import List, Optional
@@ -52,6 +53,32 @@ def _map_future(source, transform):
return result
+def _collect_futures(futures):
+ """Collect futures in input order without blocking the caller."""
+ result = Future()
+ if not futures:
+ result.set_result([])
+ return result
+
+ remaining = [len(futures)]
+ lock = threading.Lock()
+
+ def on_done(_):
+ with lock:
+ remaining[0] -= 1
+ is_last = remaining[0] == 0
+ if not is_last:
+ return
+ try:
+ result.set_result([future.result() for future in futures])
+ except Exception as e:
+ result.set_exception(e)
+
+ for future in futures:
+ future.add_done_callback(on_done)
+ return result
+
+
class GlobalIndexReader(ABC):
"""Index reader for global index. All visit methods return
Future[Optional[GlobalIndexResult]]."""
@@ -61,14 +88,10 @@ class GlobalIndexReader(ABC):
def visit_batch_vector_search(
self, batch_vector_search: 'BatchVectorSearch'
) -> 'Future[List[Optional[GlobalIndexResult]]]':
- """Default: fan out to single-vector search; result ``i`` maps to
``vectors[i]``.
-
- Blocks per future (fine while readers return completed futures); an
- async reader should override.
- """
+ """Fan out asynchronously; result ``i`` maps to ``vectors[i]``."""
singles = [self.visit_vector_search(batch_vector_search.for_index(i))
for i in range(batch_vector_search.vector_count)]
- return _completed_future([f.result() for f in singles])
+ return _collect_futures(singles)
def visit_full_text_search(self, full_text_search: 'FullTextSearch') ->
'Future[Optional[GlobalIndexResult]]':
raise NotImplementedError("Full-text search not supported by this
reader")
diff --git a/paimon-python/pypaimon/globalindex/union_global_index_reader.py
b/paimon-python/pypaimon/globalindex/union_global_index_reader.py
index 40d5e57386..8230acd84f 100644
--- a/paimon-python/pypaimon/globalindex/union_global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/union_global_index_reader.py
@@ -34,8 +34,13 @@ class UnionGlobalIndexReader(GlobalIndexReader):
def __init__(self, readers: List[GlobalIndexReader]):
self._readers = readers
- def _union_futures(self, visitor: Callable[[GlobalIndexReader],
'Future[Optional[GlobalIndexResult]]']
- ) -> 'Future[Optional[GlobalIndexResult]]':
+ def _union_futures(
+ self,
+ visitor: Callable[
+ [GlobalIndexReader], 'Future[Optional[GlobalIndexResult]]'
+ ],
+ propagate_unsupported: bool = False,
+ ) -> 'Future[Optional[GlobalIndexResult]]':
futures = [visitor(reader) for reader in self._readers]
if not futures:
@@ -51,9 +56,12 @@ class UnionGlobalIndexReader(GlobalIndexReader):
if remaining[0] == 0:
try:
result: Optional[GlobalIndexResult] = None
- for f in futures:
- current = f.result()
+ results = [f.result() for f in futures]
+ for current in results:
if current is None:
+ if propagate_unsupported:
+ all_done.set_result(None)
+ return
continue
if result is None:
result = current
@@ -68,6 +76,11 @@ class UnionGlobalIndexReader(GlobalIndexReader):
return all_done
+ def _union_scalar_futures(
+ self, visitor: Callable[[GlobalIndexReader],
'Future[Optional[GlobalIndexResult]]']
+ ) -> 'Future[Optional[GlobalIndexResult]]':
+ return self._union_futures(visitor, propagate_unsupported=True)
+
# ---- vector / full-text search ----------------------------------------
def visit_vector_search(self, vector_search) ->
'Future[Optional[GlobalIndexResult]]':
@@ -79,56 +92,60 @@ class UnionGlobalIndexReader(GlobalIndexReader):
# ---- scalar predicates (every reader sees the visit) ------------------
def visit_equal(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_equal(field_ref, literal))
+ return self._union_scalar_futures(lambda r: r.visit_equal(field_ref,
literal))
def visit_not_equal(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_not_equal(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_not_equal(field_ref, literal))
def visit_less_than(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_less_than(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_less_than(field_ref, literal))
def visit_less_or_equal(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_less_or_equal(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_less_or_equal(field_ref, literal))
def visit_greater_than(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_greater_than(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_greater_than(field_ref, literal))
def visit_greater_or_equal(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r:
r.visit_greater_or_equal(field_ref, literal))
+ return self._union_scalar_futures(lambda r:
r.visit_greater_or_equal(field_ref, literal))
def visit_is_null(self, field_ref: FieldRef) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_is_null(field_ref))
+ return self._union_scalar_futures(lambda r: r.visit_is_null(field_ref))
def visit_is_not_null(self, field_ref: FieldRef) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_is_not_null(field_ref))
+ return self._union_scalar_futures(lambda r:
r.visit_is_not_null(field_ref))
def visit_in(self, field_ref: FieldRef, literals) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_in(field_ref, literals))
+ return self._union_scalar_futures(lambda r: r.visit_in(field_ref,
literals))
def visit_not_in(self, field_ref: FieldRef, literals) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_not_in(field_ref,
literals))
+ return self._union_scalar_futures(lambda r: r.visit_not_in(field_ref,
literals))
def visit_starts_with(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_starts_with(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_starts_with(field_ref, literal))
def visit_ends_with(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_ends_with(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_ends_with(field_ref, literal))
def visit_contains(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_contains(field_ref,
literal))
+ return self._union_scalar_futures(lambda r:
r.visit_contains(field_ref, literal))
def visit_like(self, field_ref: FieldRef, literal) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_like(field_ref, literal))
+ return self._union_scalar_futures(lambda r: r.visit_like(field_ref,
literal))
def visit_between(self, field_ref: FieldRef, from_v, to_v) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_between(field_ref,
from_v, to_v))
+ return self._union_scalar_futures(lambda r: r.visit_between(field_ref,
from_v, to_v))
def visit_not_between(self, field_ref: FieldRef, from_v, to_v) ->
'Future[Optional[GlobalIndexResult]]':
- return self._union_futures(lambda r: r.visit_not_between(field_ref,
from_v, to_v))
+ return self._union_scalar_futures(lambda r:
r.visit_not_between(field_ref, from_v, to_v))
def close(self) -> None:
+ first_error: Optional[Exception] = None
for reader in self._readers:
try:
reader.close()
- except Exception:
- pass
+ except Exception as e:
+ if first_error is None:
+ first_error = e
+ if first_error is not None:
+ raise first_error
diff --git a/paimon-python/pypaimon/tests/union_global_index_reader_test.py
b/paimon-python/pypaimon/tests/union_global_index_reader_test.py
new file mode 100644
index 0000000000..935bf45f31
--- /dev/null
+++ b/paimon-python/pypaimon/tests/union_global_index_reader_test.py
@@ -0,0 +1,160 @@
+# 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.
+
+import unittest
+from concurrent.futures import Future, ThreadPoolExecutor
+from unittest.mock import Mock
+
+from pypaimon.globalindex.batch_vector_search import BatchVectorSearch
+from pypaimon.globalindex.global_index_reader import (
+ FieldRef,
+ GlobalIndexReader,
+ _completed_future,
+)
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.union_global_index_reader import
UnionGlobalIndexReader
+from pypaimon.utils.range import Range
+
+
+class _EqualReader(GlobalIndexReader):
+
+ def __init__(self, result):
+ self._result = result
+
+ def visit_equal(self, field_ref, literal):
+ return _completed_future(self._result)
+
+ def close(self):
+ pass
+
+
+class _VectorReader(GlobalIndexReader):
+
+ def __init__(self, result_future):
+ self._result_future = result_future
+
+ def visit_vector_search(self, vector_search):
+ return self._result_future
+
+ def close(self):
+ pass
+
+
+class UnionGlobalIndexReaderTest(unittest.TestCase):
+
+ def test_scalar_unsupported_is_propagated(self):
+ supported = GlobalIndexResult.from_range(Range(1, 2))
+ reader = UnionGlobalIndexReader([_EqualReader(supported),
_EqualReader(None)])
+
+ result = reader.visit_equal(FieldRef(0, "f0", "INT"), 1).result()
+
+ self.assertIsNone(result)
+
+ def test_scalar_child_failure_is_not_masked_by_unsupported(self):
+ field_ref = FieldRef(0, "f0", "INT")
+ unsupported_reader = Mock(spec=GlobalIndexReader)
+ unsupported_reader.visit_equal.return_value = _completed_future(None)
+ failed_future = Future()
+ failed_future.set_exception(RuntimeError("child failure"))
+ failed_reader = Mock(spec=GlobalIndexReader)
+ failed_reader.visit_equal.return_value = failed_future
+
+ for readers in [
+ [unsupported_reader, failed_reader],
+ [failed_reader, unsupported_reader],
+ ]:
+ with self.subTest(readers=readers):
+ reader = UnionGlobalIndexReader(readers)
+
+ with self.assertRaisesRegex(RuntimeError, "child failure"):
+ reader.visit_equal(field_ref, 1).result()
+
+ def test_all_scalar_visitors_propagate_unsupported(self):
+ field_ref = FieldRef(0, "f0", "INT")
+ cases = [
+ ("visit_not_equal", (field_ref, 1)),
+ ("visit_less_than", (field_ref, 1)),
+ ("visit_less_or_equal", (field_ref, 1)),
+ ("visit_greater_than", (field_ref, 1)),
+ ("visit_greater_or_equal", (field_ref, 1)),
+ ("visit_is_null", (field_ref,)),
+ ("visit_is_not_null", (field_ref,)),
+ ("visit_in", (field_ref, [1, 2])),
+ ("visit_not_in", (field_ref, [1, 2])),
+ ("visit_starts_with", (field_ref, "1")),
+ ("visit_ends_with", (field_ref, "1")),
+ ("visit_contains", (field_ref, "1")),
+ ("visit_like", (field_ref, "1%")),
+ ("visit_between", (field_ref, 1, 2)),
+ ("visit_not_between", (field_ref, 1, 2)),
+ ]
+
+ for method_name, args in cases:
+ with self.subTest(method=method_name):
+ supported_reader = Mock(spec=GlobalIndexReader)
+ unsupported_reader = Mock(spec=GlobalIndexReader)
+ supported_result = GlobalIndexResult.from_range(Range(1, 2))
+ getattr(supported_reader, method_name).return_value =
_completed_future(
+ supported_result)
+ getattr(unsupported_reader, method_name).return_value =
_completed_future(None)
+ reader = UnionGlobalIndexReader([supported_reader,
unsupported_reader])
+
+ result = getattr(reader, method_name)(*args).result()
+
+ self.assertIsNone(result)
+
+ def test_batch_vector_search_returns_before_children_complete(self):
+ child_result = Future()
+ reader = UnionGlobalIndexReader([_VectorReader(child_result)])
+ search = BatchVectorSearch(vectors=[[1.0]], limit=1, field_name="f0")
+ executor = ThreadPoolExecutor(max_workers=1)
+ invocation = executor.submit(reader.visit_batch_vector_search, search)
+
+ try:
+ result_future = invocation.result(timeout=1)
+ self.assertFalse(result_future.done())
+ finally:
+ child_result.set_result(None)
+ executor.shutdown()
+
+ self.assertEqual([None], result_future.result(timeout=1))
+
+ def test_batch_vector_search_propagates_failure_asynchronously(self):
+ child_result = Future()
+ reader = UnionGlobalIndexReader([_VectorReader(child_result)])
+ search = BatchVectorSearch(vectors=[[1.0]], limit=1, field_name="f0")
+
+ result_future = reader.visit_batch_vector_search(search)
+ child_result.set_exception(RuntimeError("async failure"))
+
+ with self.assertRaisesRegex(RuntimeError, "async failure"):
+ result_future.result(timeout=1)
+
+ def test_close_propagates_failure_after_closing_all_readers(self):
+ failing_reader = Mock(spec=GlobalIndexReader)
+ failing_reader.close.side_effect = OSError("close failed")
+ remaining_reader = Mock(spec=GlobalIndexReader)
+ reader = UnionGlobalIndexReader([failing_reader, remaining_reader])
+
+ with self.assertRaisesRegex(OSError, "close failed"):
+ reader.close()
+
+ remaining_reader.close.assert_called_once_with()
+
+
+if __name__ == "__main__":
+ unittest.main()
