diff --git a/examples/ice_disk_to_arrow_memory.py b/examples/ice_disk_to_arrow_memory.py index 52df48a..e64fa51 100644 --- a/examples/ice_disk_to_arrow_memory.py +++ b/examples/ice_disk_to_arrow_memory.py @@ -77,11 +77,17 @@ def main() -> None: db = lb.Database(":memory:") conn = db.connect() if args.layout == "flat": - register_flat(conn, args.data_dir, args.node_table, args.rel_table, src_table, dst_table) + register_flat( + conn, args.data_dir, args.node_table, args.rel_table, src_table, dst_table + ) else: - register_csr(conn, args.data_dir, args.node_table, args.rel_table, src_table, dst_table) + register_csr( + conn, args.data_dir, args.node_table, args.rel_table, src_table, dst_table + ) - result = conn.execute(f"MATCH (a:{src_table})-[r:{args.rel_table}]->(b:{dst_table}) RETURN COUNT(*)") + result = conn.execute( + f"MATCH (a:{src_table})-[r:{args.rel_table}]->(b:{dst_table}) RETURN COUNT(*)" + ) print(result.get_next()[0]) diff --git a/src_cpp/numpy/numpy_scan.cpp b/src_cpp/numpy/numpy_scan.cpp index 0dc885c..c172205 100644 --- a/src_cpp/numpy/numpy_scan.cpp +++ b/src_cpp/numpy/numpy_scan.cpp @@ -172,7 +172,7 @@ void NumpyScan::scan(PandasColumnBindData* bindData, uint64_t count, uint64_t of !py::isinstance(val)) { if (val == Py_None || (py::isinstance(val) && std::isnan(PyFloat_AsDouble(val)))) { - outputVector->setNull(pos, true /* isNull */); + outputVector->setNull(i, true /* isNull */); continue; } if (!py::isinstance(val)) { diff --git a/src_py/connection.py b/src_py/connection.py index fa41b89..806f47a 100644 --- a/src_py/connection.py +++ b/src_py/connection.py @@ -1,8 +1,10 @@ from __future__ import annotations +import contextlib import inspect import json import re +import threading import uuid import warnings from typing import TYPE_CHECKING, Any @@ -27,6 +29,188 @@ from typing_extensions import Self +def _capi_value_signature(value: Any) -> tuple: + """ + Compute a stable, hashable signature for a single parameter value. + + The signature mirrors the *structural* type that ``lbug`` will infer + when binding the parameter: for a Python ``dict`` that has ``"key"`` + and ``"value"`` list fields of equal length, the binder produces a + ``MAP(K, V)`` and the signature records the element type of both + lists. Any other string-keyed ``dict`` produces a ``STRUCT`` and the + signature records each field's name and child signature. This + granularity is required because two STRUCTs with different field + names produce different C++ plans, and routing a second shape + through the first shape's cached plan would misread fields. + + The signature is independent of *value contents* (so callers passing + the same query with different numeric values still share the cache) + but is sensitive to anything that would change the binder's plan: + nesting, field names, and element type. + """ + if value is None: + return ("null",) + if isinstance(value, bool): + return ("bool",) + # ``bool`` is a subclass of ``int``; check it first. + if isinstance(value, int): + return ("int",) + if isinstance(value, float): + return ("float",) + if isinstance(value, (bytes, bytearray, memoryview)): + return ("blob",) + # ``CAPIJsonParameter`` is a frozen dataclass wrapping a JSON string. + # Treat it as a string for caching purposes. + if type(value).__name__ == "CAPIJsonParameter": + return ("string",) + if isinstance(value, str): + return ("string",) + if isinstance(value, dict): + if ( + set(value.keys()) == {"key", "value"} + and isinstance(value["key"], list) + and isinstance(value["value"], list) + and len(value["key"]) == len(value["value"]) + ): + # ``MAP`` - the binder treats the two parallel lists as the + # key and value columns. Record the per-list element type so + # that ``MAP(INT,INT)`` vs ``MAP(STRING,INT)`` get distinct + # plans. + key_sig = ( + ("list", ("null",)) + if not value["key"] + else ("list", _capi_value_signature(value["key"][0])) + ) + val_sig = ( + ("list", ("null",)) + if not value["value"] + else ("list", _capi_value_signature(value["value"][0])) + ) + return ("map", key_sig, val_sig) + if all(isinstance(k, str) for k in value): + # ``STRUCT`` - the binder infers one field per key. Field + # names matter because they appear in the plan's expressions + # and column metadata. + return ( + "struct", + tuple(sorted((k, _capi_value_signature(v)) for k, v in value.items())), + ) + return ("dict",) + if isinstance(value, (list, tuple)): + if not value: + return ("list", ("null",)) + return ("list", _capi_value_signature(value[0])) + return ("unknown", type(value).__name__) + + +def _capi_param_signature(parameters: dict[str, Any]) -> tuple: + """ + Return a hashable signature of parameter *types* (not values). + + The signature is independent of value contents but stable across + calls that pass the same kind of parameters. The keys are sorted so + that two callers that pass the same parameters in different orders + share a prepared statement. + """ + return tuple( + sorted((key, _capi_value_signature(val)) for key, val in parameters.items()) + ) + + +def _pybind_int_signature(value: int) -> tuple[str]: + if -(2**7) <= value <= 2**7 - 1: + return ("int8",) + if 0 <= value <= 2**8 - 1: + return ("uint8",) + if -(2**15) <= value <= 2**15 - 1: + return ("int16",) + if 0 <= value <= 2**16 - 1: + return ("uint16",) + if -(2**31) <= value <= 2**31 - 1: + return ("int32",) + if 0 <= value <= 2**32 - 1: + return ("uint32",) + return ("int64",) + + +def _pybind_homogeneous_list_signature( + value: list[Any] | tuple[Any, ...], +) -> tuple | None: + first_non_null = next((item for item in value if item is not None), None) + if first_non_null is None: + return ("list", ("any",)) + if not isinstance(first_non_null, (bool, int, float)): + return None + first_type = type(first_non_null) + if any(item is not None and type(item) is not first_type for item in value): + return None + if isinstance(first_non_null, bool): + return ("list", ("bool",)) + if isinstance(first_non_null, int): + return ("list", ("int64",)) + return ("list", ("double",)) + + +def _pybind_value_signature(value: Any) -> tuple: + """ + Compute the parameter type signature used by the pybind binder. + + This intentionally differs from the C-API signature for Python ints: + pybind narrows scalar ints to INT8/UINT8/INT16/... based on the value + seen during prepare. Reusing a prepared statement across those + widths corrupts later executions because updateParameter() mutates + the value without revalidating the logical type. + """ + if value is None: + return ("any",) + if isinstance(value, bool): + return ("bool",) + if isinstance(value, int): + return _pybind_int_signature(value) + if isinstance(value, float): + return ("double",) + if isinstance(value, (bytes, bytearray, memoryview)): + return ("blob",) + if isinstance(value, str): + return ("string",) + module_name = type(value).__module__ + if module_name.startswith(("pandas", "polars", "pyarrow")): + return ("pointer", module_name, type(value).__name__, id(value)) + if isinstance(value, dict): + items = list(value.items()) + if ( + len(items) == 2 + and items[0][0] == "key" + and items[1][0] == "value" + and isinstance(items[0][1], list) + and isinstance(items[1][1], list) + and len(items[0][1]) == len(items[1][1]) + ): + return ( + "map", + _pybind_value_signature(items[0][1])[1], + _pybind_value_signature(items[1][1])[1], + ) + return ( + "struct", + tuple((str(k), _pybind_value_signature(v)) for k, v in items), + ) + if isinstance(value, (list, tuple)): + homogeneous = _pybind_homogeneous_list_signature(value) + if homogeneous is not None: + return homogeneous + if not value: + return ("list", ("any",)) + return ("list", _pybind_value_signature(value[0])) + return ("unknown", type(value).__name__) + + +def _pybind_param_signature(parameters: dict[str, Any]) -> tuple: + return tuple( + sorted((key, _pybind_value_signature(val)) for key, val in parameters.items()) + ) + + class Connection: """Connection to a database.""" @@ -52,6 +236,20 @@ def __init__(self, database: Database, num_threads: int = 0): self._query_timeout_ms = 0 self._query_results: WeakSet[QueryResult] = WeakSet() self._capi_scan_tables: set[str] = set() + # Implicit prepared-statement cache, shared by the pybind and + # C-API paths. The key is (query, backend_param_signature) so that two + # callers passing the same query with *different* parameter + # shapes (e.g. MAP vs STRUCT, or two STRUCTs with different + # field names) or pybind-native integer widths get independent + # prepared statements. Each cached entry's plan stays consistent + # with its bound values, avoiding the upstream "cached plan vs + # rebound parameterMap" mismatch. + self._pybind_implicit_prepared_cache: dict[tuple[str, tuple], Any] = {} + # Serializes prepare / bind / execute on a single connection so + # that the cached entry's mutable bound state cannot be torn by + # concurrent callers (multi-threaded users or AsyncConnection's + # thread-pool). + self._prepared_cache_lock = threading.RLock() self.database._register_connection(self) self.init_connection() @@ -116,6 +314,23 @@ def close(self) -> None: for query_result in list(self._query_results): query_result.close() self._query_results.clear() + # Destroy the C++ ``lbug_prepared_statement`` held by every cached + # entry before we drop the references. ``lbug_prepared_statement`` + # does not own a Python ``__del__`` so the entries would otherwise + # leak their ``_prepared_statement`` and ``_bound_values`` C++ + # allocations. Pybind entries are safe to drop because their state + # is held in a ``shared_ptr`` that self-cleans on refcount=0. + with self._prepared_cache_lock: + for prepared in self._pybind_implicit_prepared_cache.values(): + close_fn = getattr(prepared, "close", None) + if callable(close_fn): + with contextlib.suppress(RuntimeError): + # The C++ connection may already be torn down + # (close ordering between connection and cache). + # Best-effort: the entries are about to be dropped + # anyway. + close_fn() + self._pybind_implicit_prepared_cache.clear() if self._connection is not None and not self.database.is_closed: self._connection.close() @@ -460,8 +675,39 @@ def _execute_with_pybind( return py_connection.query(query) query, parameters = self._normalize_parameters_for_pybind(query, parameters) - prepared = py_connection.prepare(query, parameters) - return py_connection.execute(prepared, parameters) + with self._prepared_cache_lock: + prepared = self._get_or_prepare_pybind_statement( + py_connection, query, parameters + ) + return py_connection.execute(prepared, parameters) + + def _get_or_prepare_pybind_statement( + self, + py_connection: Any, + query: str, + parameters: dict[str, Any], + ) -> Any: + # Caller must hold ``self._prepared_cache_lock``. + cache_key = (query, _pybind_param_signature(parameters)) + prepared = self._pybind_implicit_prepared_cache.get(cache_key) + if prepared is None: + prepared = py_connection.prepare(query, parameters) + self._pybind_implicit_prepared_cache[cache_key] = prepared + return prepared + + def _get_or_prepare_capi_statement( + self, + query: str, + parameters: dict[str, Any], + ) -> PreparedStatement: + # Caller must hold ``self._prepared_cache_lock``. + cache_key = (query, _capi_param_signature(parameters)) + cached = self._pybind_implicit_prepared_cache.get(cache_key) + if cached is not None: + return cached + prepared = self._prepare(query, parameters) + self._pybind_implicit_prepared_cache[cache_key] = prepared + return prepared def _maybe_raise_scan_unsupported_object(self, query: str) -> None: match = re.search( @@ -565,12 +811,21 @@ def execute( query, parameters = self._normalize_parameters_for_capi( query, parameters ) - prepared_statement = ( - self._prepare(query, parameters) if isinstance(query, str) else query - ) - query_result_internal = self._connection.execute( - prepared_statement._prepared_statement, parameters - ) + # Hold the lock across prepare + bind + execute so that + # concurrent callers (threads / async tasks) cannot tear the + # cached entry's mutable bound state. The C++ side has its + # own ``mtx`` around ``executeWithParams``; the lock here + # only protects the C-API ``_bound_values`` map and the + # cache itself. + with self._prepared_cache_lock: + prepared_statement = ( + self._get_or_prepare_capi_statement(query, parameters) + if isinstance(query, str) + else query + ) + query_result_internal = self._connection.execute( + prepared_statement._prepared_statement, parameters + ) if not query_result_internal.isSuccess(): raise RuntimeError(query_result_internal.getErrorMessage()) for table_name in scan_tables_to_drop: diff --git a/src_py/prepared_statement.py b/src_py/prepared_statement.py index 25efe9a..de997db 100644 --- a/src_py/prepared_statement.py +++ b/src_py/prepared_statement.py @@ -33,6 +33,27 @@ def __init__( self._prepared_statement = connection._connection.prepare(query, parameters) self._connection = connection + def close(self) -> None: + """ + Release the underlying C-API prepared statement resources. + + The C-API ``PreparedStatement`` (from ``_lbug_capi.py``) holds a + ``lbug_prepared_statement`` C struct that must be destroyed explicitly; + it is NOT garbage-collected when the Python wrapper is dropped. + The pybind variant is managed by ``shared_ptr`` and can be dropped + without an explicit call. + + ``Connection.close()`` iterates the implicit prepared-statement cache + and calls this method on every cached entry so that resources are + freed when the connection is closed. + """ + close_fn = getattr(self._prepared_statement, "close", None) + if callable(close_fn): + close_fn() + + def __del__(self) -> None: + self.close() + def is_success(self) -> bool: """ Check if the prepared statement is successfully prepared. diff --git a/test/test_fsm.py b/test/test_fsm.py index fba92e7..7b9855f 100644 --- a/test/test_fsm.py +++ b/test/test_fsm.py @@ -24,6 +24,24 @@ def get_used_page_ranges(conn, table, column=None): return used_pages +def get_used_node_data_page_ranges(conn, table): + """ + Like get_used_page_ranges, but excludes INDEX pages. + + Useful for `match ... delete` scenarios where the table and its indexes persist + while only the data pages are reclaimed. + """ + used_pages = [] + storage_info = conn.execute( + f'call storage_info("{table}") where table_type = "NODE" return start_page_idx, num_pages' + ) + while storage_info.has_next(): + cur_tuple = storage_info.get_next() + if cur_tuple[1] > 0: + used_pages.append(cur_tuple) + return used_pages + + def get_total_used_pages(conn): return conn.execute("call file_info() return num_pages").get_next()[0] @@ -181,7 +199,7 @@ def test_fsm_reclaim_node_table_recopy(fsm_node_table_setup) -> None: def test_fsm_reclaim_node_table_delete(fsm_node_table_setup) -> None: _, conn = fsm_node_table_setup - used_pages = get_used_page_ranges(conn, "person") + used_pages = get_used_node_data_page_ranges(conn, "person") conn.execute("match (p:person) delete p") prevent_data_file_truncation(conn) conn.execute("checkpoint") @@ -201,7 +219,7 @@ def test_fsm_reclaim_rel_table(fsm_rel_table_setup) -> None: def test_fsm_reclaim_rel_table_delete(fsm_node_table_setup) -> None: _, conn = fsm_node_table_setup - used_pages = get_used_page_ranges(conn, "person") + used_pages = get_used_node_data_page_ranges(conn, "person") conn.execute("match (p:person) delete p") prevent_data_file_truncation(conn) conn.execute("checkpoint") diff --git a/test/test_pybind_implicit_prepare_cache.py b/test/test_pybind_implicit_prepare_cache.py new file mode 100644 index 0000000..e0466ca --- /dev/null +++ b/test/test_pybind_implicit_prepare_cache.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import ladybug as lb +import ladybug.connection as lb_connection +import pytest + + +class _FakeResult: + def isSuccess(self) -> bool: + return True + + def hasNextQueryResult(self) -> bool: + return False + + def close(self) -> None: + return + + +class _FakePreparedStatement: + def __init__(self, query: str, parameters: dict[str, object]): + self.query = query + self.parameters = dict(parameters) + + +class _FakePybindConnection: + def __init__(self) -> None: + self.prepare_calls: list[tuple[str, dict[str, object]]] = [] + self.execute_calls: list[tuple[_FakePreparedStatement, dict[str, object]]] = [] + self.query_calls: list[str] = [] + self.closed = False + + def prepare( + self, query: str, parameters: dict[str, object] + ) -> _FakePreparedStatement: + self.prepare_calls.append((query, dict(parameters))) + return _FakePreparedStatement(query, parameters) + + def execute( + self, prepared: _FakePreparedStatement, parameters: dict[str, object] + ) -> _FakeResult: + self.execute_calls.append((prepared, dict(parameters))) + return _FakeResult() + + def query(self, query: str) -> _FakeResult: + self.query_calls.append(query) + return _FakeResult() + + def close(self) -> None: + self.closed = True + + +class _FakeBackendConnection: + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +class _FakeDatabase: + def __init__(self) -> None: + self._use_pybind_backend = True + self._database = object() + self.is_closed = False + self.registered_connections: list[lb.Connection] = [] + + def _register_connection(self, connection: lb.Connection) -> None: + self.registered_connections.append(connection) + + def _unregister_connection(self, connection: lb.Connection) -> None: + self.registered_connections.remove(connection) + + def init_database(self) -> None: + return None + + +@pytest.fixture +def fake_pybind_connection(monkeypatch: pytest.MonkeyPatch) -> _FakePybindConnection: + fake_pybind = _FakePybindConnection() + + monkeypatch.setattr(lb_connection, "get_pybind_module", lambda: SimpleNamespace()) + monkeypatch.setattr( + lb.Connection, + "init_connection", + lambda self: setattr(self, "_connection", _FakeBackendConnection()), + ) + + def _get_pybind_connection(self: lb.Connection) -> _FakePybindConnection: + if self._py_connection is None: + self._py_connection = fake_pybind + return self._py_connection + + monkeypatch.setattr(lb.Connection, "_get_pybind_connection", _get_pybind_connection) + return fake_pybind + + +def test_pybind_implicit_prepare_reuses_same_query( + fake_pybind_connection: _FakePybindConnection, +) -> None: + conn = lb.Connection(_FakeDatabase()) + + conn.execute("RETURN $value", {"value": 1}) + conn.execute("RETURN $value", {"value": 2}) + + assert fake_pybind_connection.query_calls == [] + assert len(fake_pybind_connection.prepare_calls) == 1 + assert [call[0].query for call in fake_pybind_connection.execute_calls] == [ + "RETURN $value", + "RETURN $value", + ] + assert [call[1] for call in fake_pybind_connection.execute_calls] == [ + {"value": 1}, + {"value": 2}, + ] + + +def test_pybind_implicit_prepare_does_not_share_different_queries( + fake_pybind_connection: _FakePybindConnection, +) -> None: + conn = lb.Connection(_FakeDatabase()) + + conn.execute("RETURN $value", {"value": 1}) + conn.execute("RETURN $other", {"other": 1}) + + assert [call[0] for call in fake_pybind_connection.prepare_calls] == [ + "RETURN $value", + "RETURN $other", + ] + + +def test_pybind_implicit_prepare_segregates_by_integer_width( + fake_pybind_connection: _FakePybindConnection, +) -> None: + conn = lb.Connection(_FakeDatabase()) + + conn.execute("RETURN $value", {"value": 1}) + conn.execute("RETURN $value", {"value": 1000}) + conn.execute("RETURN $value", {"value": 2}) + + assert len(fake_pybind_connection.prepare_calls) == 2 + assert [call[1] for call in fake_pybind_connection.execute_calls] == [ + {"value": 1}, + {"value": 1000}, + {"value": 2}, + ] + + +def test_pybind_no_parameter_query_skips_prepare_cache( + fake_pybind_connection: _FakePybindConnection, +) -> None: + conn = lb.Connection(_FakeDatabase()) + + conn.execute("RETURN 1") + + assert fake_pybind_connection.prepare_calls == [] + assert fake_pybind_connection.query_calls == ["RETURN 1"] + + +def test_pybind_close_clears_implicit_prepare_cache( + fake_pybind_connection: _FakePybindConnection, +) -> None: + conn = lb.Connection(_FakeDatabase()) + + conn.execute("RETURN $value", {"value": 1}) + + # Cache key is now (query, parameter-type signature) so that a + # single query with different parameter shapes (e.g. MAP vs STRUCT) + # does not share a prepared statement. + assert len(conn._pybind_implicit_prepared_cache) == 1 + cached_key = next(iter(conn._pybind_implicit_prepared_cache)) + assert cached_key[0] == "RETURN $value" + + conn.close() + + assert conn._pybind_implicit_prepared_cache == {} + assert fake_pybind_connection.closed is True + + +def test_pybind_implicit_prepare_segregates_by_parameter_shape( + fake_pybind_connection: _FakePybindConnection, +) -> None: + """The same query with different parameter shapes must not share a plan.""" + conn = lb.Connection(_FakeDatabase()) + + # Same keys ``key``/``value`` with matching list lengths -> MAP. + conn.execute("RETURN $st", {"st": {"key": [1, 2, 3], "value": [3, 7, 98]}}) + # First key is ``key1`` -> STRUCT. + conn.execute("RETURN $st", {"st": {"key1": [1, 2, 3], "value": [3, 7, 98]}}) + # ``key``/``value`` but mismatched list lengths -> STRUCT (different + # signature from the first call AND from the previous STRUCT because + # the field names differ). + conn.execute("RETURN $st", {"st": {"key": [1, 2], "value": [3, 7, 98, 4]}}) + + # 3 distinct structural shapes -> 3 separate prepared statements. + # Sharing any of them would either misbind the parameter or reroute + # a new shape through a stale plan. + assert len(fake_pybind_connection.prepare_calls) == 3 + assert all(call[0] == "RETURN $st" for call in fake_pybind_connection.prepare_calls) + assert [call[1] for call in fake_pybind_connection.execute_calls] == [ + {"st": {"key": [1, 2, 3], "value": [3, 7, 98]}}, + {"st": {"key1": [1, 2, 3], "value": [3, 7, 98]}}, + {"st": {"key": [1, 2], "value": [3, 7, 98, 4]}}, + ] + + +def test_pybind_implicit_prepare_reuses_same_struct_shape( + fake_pybind_connection: _FakePybindConnection, +) -> None: + """Two calls that produce the same structural signature reuse the cache.""" + conn = lb.Connection(_FakeDatabase()) + + # Same fields, different values -> identical signature. + conn.execute("RETURN $st", {"st": {"key1": [1, 2, 3], "value": [3, 7, 98]}}) + conn.execute("RETURN $st", {"st": {"key1": [4, 5, 6], "value": [7, 8, 9]}}) + + assert len(fake_pybind_connection.prepare_calls) == 1 + assert [call[1] for call in fake_pybind_connection.execute_calls] == [ + {"st": {"key1": [1, 2, 3], "value": [3, 7, 98]}}, + {"st": {"key1": [4, 5, 6], "value": [7, 8, 9]}}, + ] diff --git a/test/test_pybind_implicit_prepare_cache_threading.py b/test/test_pybind_implicit_prepare_cache_threading.py new file mode 100644 index 0000000..bd4ef96 --- /dev/null +++ b/test/test_pybind_implicit_prepare_cache_threading.py @@ -0,0 +1,70 @@ +""" +Thread-safety regression tests for the implicit prepared-statement cache. + +The cache lives on ``Connection._pybind_implicit_prepared_cache`` and is +shared by the pybind and C-API execute paths. When two threads execute +the same parameterized query on the same connection, the cached +prepared statement is shared; if its mutable bound state is torn by +concurrent bind calls, the executor reads garbage and returns the wrong +value. These tests reproduce that scenario and would catch a +regression where the cache is reintroduced without proper +serialization. +""" + +from __future__ import annotations + +import threading + +import ladybug as lb +import pytest + + +@pytest.mark.parametrize( + ("num_threads", "iters"), + [ + # Light load - already covered by the async tests in spirit, but + # we re-test on a synchronous shared connection to make sure the + # per-connection lock covers the multi-threaded case. + (2, 5), + # Moderate load - the smallest case where the race used to + # surface (1 thread never reproduces it). + (8, 5), + # Heavy load - the original failure mode. + (32, 5), + ], +) +def test_shared_connection_concurrent_same_query(num_threads: int, iters: int) -> None: + """Many threads on a single connection must each see its own bound value.""" + # Use an explicit max_db_size (1 GB) so the C-API backend does not default + # to the library's 8 TB mmap region, which fails on CI runners with tight + # virtual-address limits. + db = lb.Database(":memory:", buffer_pool_size=2**28, max_db_size=2**30) + conn = lb.Connection(db) + + errors: list[tuple[int, list]] = [] + errors_lock = threading.Lock() + + def worker(base: int) -> None: + for i in range(iters): + v = base + i + result = conn.execute("RETURN $n", {"n": v}) + try: + got = result.get_next() + finally: + result.close() + if got != [v]: + with errors_lock: + errors.append((v, got)) + + threads = [ + threading.Thread(target=worker, args=(i * 1000,)) for i in range(num_threads) + ] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], ( + f"{len(errors)}/{num_threads * iters} executions returned the " + f"wrong value. First few: {errors[:5]}" + ) diff --git a/test/test_scan_pandas_pyarrow.py b/test/test_scan_pandas_pyarrow.py index 7cfbab4..539f356 100644 --- a/test/test_scan_pandas_pyarrow.py +++ b/test/test_scan_pandas_pyarrow.py @@ -24,10 +24,16 @@ def generate_primitive(dtype): if dtype.startswith("uint64"): return random.randrange(0, 18446744073709551615) if dtype.startswith("float32"): - random_bits = random.getrandbits(32) - random_bytes = struct.pack("