From 93cd357776a9833b8ce76b2bf1b41d6cdcd45c4b Mon Sep 17 00:00:00 2001 From: Teun Huijben Date: Thu, 30 Jul 2026 09:10:39 -0700 Subject: [PATCH 1/2] fix np scalars becoming blobs in SQLGraph --- src/tracksdata/graph/_sql_graph.py | 38 +++++--- .../graph/_test/test_graph_backends.py | 95 +++++++++++++++++++ 2 files changed, 121 insertions(+), 12 deletions(-) diff --git a/src/tracksdata/graph/_sql_graph.py b/src/tracksdata/graph/_sql_graph.py index b2f66b46..475b778f 100644 --- a/src/tracksdata/graph/_sql_graph.py +++ b/src/tracksdata/graph/_sql_graph.py @@ -59,13 +59,21 @@ def _data_numpy_to_native(data: dict[str, Any]) -> None: """ Convert numpy scalars to native Python scalars in place. + Database drivers do not know about numpy scalar types. ``sqlite3``, for example, + falls back to the buffer protocol and stores ``np.int64(7)`` as its raw + little-endian byte buffer (a BLOB), silently corrupting a column declared as + ``BIGINT``. Numpy floats and strings happen to survive because they subclass + their Python counterparts, which makes the corruption look selective. + Parameters ---------- data : dict[str, Any] The data to convert. Modified in place. """ for k, v in data.items(): - if np.isscalar(v) and hasattr(v, "item"): + # `np.generic` is the base class of every numpy scalar, and excludes + # (0-dim) arrays, which must be passed through untouched. + if isinstance(v, np.generic): data[k] = v.item() @@ -972,6 +980,11 @@ def _flatten_attrs_for_write( write path (``bulk_add_nodes``, ``bulk_add_edges``, ``update_node_attrs``, ``update_edge_attrs``) since the single-node/edge wrappers in :class:`BaseGraph` now delegate to the bulk variants. + + Numpy scalars are converted to their native Python counterparts here -- + including the ones nested inside a struct value -- so that every write path + hands the driver values it can map onto the column's declared dtype, + see :func:`_data_numpy_to_native`. """ result: dict[str, Any] = {} for key, value in attrs.items(): @@ -980,6 +993,7 @@ def _flatten_attrs_for_write( result.update(flatten_struct_value(key, value, schema.dtype)) else: result[key] = value + _data_numpy_to_native(result) return result def bulk_add_nodes( @@ -1030,7 +1044,12 @@ def bulk_add_nodes( node_ids = [] insert_rows = [] for i, node in enumerate(nodes): - time = node["t"] + # numpy times must be converted before the id arithmetic below, otherwise + # `time * node_id_time_multiplier` silently overflows for narrow dtypes + # (e.g. np.int32) and the resulting id is a numpy scalar itself. + time = node[DEFAULT_ATTR_KEYS.T] + if isinstance(time, np.generic): + time = time.item() if indices is None: default_node_id = (time * self.node_id_time_multiplier) - 1 @@ -1038,10 +1057,10 @@ def bulk_add_nodes( # Update max_id tracking only for auto-generated IDs self._max_id_per_time[time] = node_id else: - node_id = indices[i] + node_id = int(indices[i]) node_ids.append(node_id) - insert_rows.append({**node, DEFAULT_ATTR_KEYS.NODE_ID: node_id}) + insert_rows.append({**node, DEFAULT_ATTR_KEYS.T: time, DEFAULT_ATTR_KEYS.NODE_ID: node_id}) # Flatten struct-typed attrs into their physical leaf columns before write. # Non-struct keys (incl. NODE_ID) pass through unchanged. @@ -1163,9 +1182,6 @@ def bulk_add_edges( return None edge_schemas = self._edge_attr_schemas() - for edge in edges: - _data_numpy_to_native(edge) - edges = [self._flatten_attrs_for_write(edge, edge_schemas) for edge in edges] if return_ids: @@ -1200,8 +1216,8 @@ def add_overlap( The ID of the added overlap. """ overlap = self.Overlap( - source_id=source_id, - target_id=target_id, + source_id=int(source_id), + target_id=int(target_id), ) with Session(self._engine) as session: session.add(overlap) @@ -1228,7 +1244,7 @@ def bulk_add_overlaps( if hasattr(overlaps, "tolist"): overlaps = overlaps.tolist() - overlaps = [{"source_id": source_id, "target_id": target_id} for source_id, target_id in overlaps] + overlaps = [{"source_id": int(source_id), "target_id": int(target_id)} for source_id, target_id in overlaps] self._chunked_sa_write(Session.bulk_insert_mappings, overlaps, self.Overlap) def overlaps( @@ -2034,8 +2050,6 @@ def _update_table( ids = ids.tolist() # Handle array values with bulk_update_mappings - attrs = attrs.copy() - _data_numpy_to_native(attrs) schemas = self._attr_schemas_for_table(table_class) attrs = self._flatten_attrs_for_write(attrs, schemas) diff --git a/src/tracksdata/graph/_test/test_graph_backends.py b/src/tracksdata/graph/_test/test_graph_backends.py index 3c8095c6..19cb66a6 100644 --- a/src/tracksdata/graph/_test/test_graph_backends.py +++ b/src/tracksdata/graph/_test/test_graph_backends.py @@ -109,6 +109,101 @@ def test_add_edge(graph_backend: BaseGraph) -> None: assert df["weight"].to_list() == [0.5, 0.1] +def test_add_node_and_edge_with_numpy_scalars(graph_backend: BaseGraph) -> None: + """Numpy scalars must be stored with the column's declared dtype, not as raw byte buffers. + + Indexing any value out of a numpy array yields a numpy scalar, so this is the + norm for importers (geff/CSV/...) feeding values into the graph. + """ + graph_backend.add_node_attr_key("val", dtype=pl.Int64, default_value=-1) + graph_backend.add_node_attr_key("pos", dtype=pl.Float64, default_value=0.0) + graph_backend.add_node_attr_key("flag", dtype=pl.Boolean, default_value=False) + graph_backend.add_edge_attr_key("weight", dtype=pl.Int64, default_value=0) + + node_1 = graph_backend.add_node( + {"t": np.int64(0), "val": np.int64(7), "pos": np.float32(1.5), "flag": np.bool_(True)} + ) + node_2, node_3 = graph_backend.bulk_add_nodes( + [ + {"t": np.int32(1), "val": np.int32(8), "pos": np.float64(2.5), "flag": np.bool_(False)}, + {"t": 2, "val": 9, "pos": 3.5, "flag": True}, + ] + ) + + nodes_df = graph_backend.node_attrs(attr_keys=["t", "val", "pos", "flag"]).sort("t") + assert nodes_df.schema["val"] == pl.Int64 + assert nodes_df["t"].to_list() == [0, 1, 2] + assert nodes_df["val"].to_list() == [7, 8, 9] + assert nodes_df["pos"].to_list() == [1.5, 2.5, 3.5] + assert nodes_df["flag"].to_list() == [True, False, True] + + graph_backend.add_edge(np.int64(node_1), np.int64(node_2), {"weight": np.int64(3)}) + graph_backend.bulk_add_edges( + [ + { + DEFAULT_ATTR_KEYS.EDGE_SOURCE: np.int64(node_2), + DEFAULT_ATTR_KEYS.EDGE_TARGET: np.int64(node_3), + "weight": np.int32(4), + } + ] + ) + + edges_df = graph_backend.edge_attrs( + attr_keys=[DEFAULT_ATTR_KEYS.EDGE_SOURCE, DEFAULT_ATTR_KEYS.EDGE_TARGET, "weight"] + ).sort("weight") + assert edges_df["weight"].to_list() == [3, 4] + assert edges_df[DEFAULT_ATTR_KEYS.EDGE_SOURCE].to_list() == [node_1, node_2] + assert edges_df[DEFAULT_ATTR_KEYS.EDGE_TARGET].to_list() == [node_2, node_3] + + graph_backend.add_overlap(np.int64(node_1), np.int64(node_2)) + graph_backend.bulk_add_overlaps([[np.int64(node_2), np.int64(node_3)]]) + assert sorted(graph_backend.overlaps()) == sorted([[node_1, node_2], [node_2, node_3]]) + + +def test_sql_node_ids_from_narrow_numpy_time() -> None: + """A narrow numpy `t` must not overflow the `t * node_id_time_multiplier` id arithmetic.""" + graph = SQLGraph( + drivername="sqlite", + database=":memory:", + engine_kwargs={"connect_args": {"check_same_thread": False}}, + ) + # np.int32(3) * 1_000_000_000 wraps around to a negative number in int32 arithmetic + node_ids = graph.bulk_add_nodes([{"t": np.int32(3)}, {"t": np.int32(3)}]) + + assert node_ids == [3 * graph.node_id_time_multiplier, 3 * graph.node_id_time_multiplier + 1] + assert all(isinstance(node_id, int) for node_id in node_ids) + assert graph.node_ids() == node_ids + + +def test_add_node_with_numpy_scalars_in_struct(graph_backend: BaseGraph) -> None: + """Numpy scalars nested inside a struct attribute must also honor the declared dtype.""" + graph_backend.add_node_attr_key("m", dtype=pl.Struct({"a": pl.Int64, "b": pl.Float64})) + + graph_backend.add_node({"t": 0, "m": {"a": np.int64(3), "b": np.float64(0.25)}}) + graph_backend.bulk_add_nodes([{"t": 1, "m": {"a": np.int32(4), "b": np.float32(0.5)}}]) + + nodes_df = graph_backend.node_attrs(attr_keys=["t", "m"]).sort("t") + assert nodes_df["m"].to_list() == [{"a": 3, "b": 0.25}, {"a": 4, "b": 0.5}] + + +def test_update_attrs_with_numpy_scalars(graph_backend: BaseGraph) -> None: + """The update path must coerce numpy scalars just like the insert path.""" + graph_backend.add_node_attr_key("val", dtype=pl.Int64, default_value=-1) + graph_backend.add_edge_attr_key("weight", dtype=pl.Int64, default_value=0) + + node_1 = graph_backend.add_node({"t": 0, "val": 0}) + node_2 = graph_backend.add_node({"t": 1, "val": 0}) + edge_id = graph_backend.add_edge(node_1, node_2, {"weight": 0}) + + graph_backend.update_node_attrs(attrs={"val": np.int64(5)}, node_ids=[node_1]) + graph_backend.update_node_attrs(attrs={"val": [np.int32(6)]}, node_ids=[node_2]) + graph_backend.update_edge_attrs(attrs={"weight": np.int64(7)}, edge_ids=[edge_id]) + + nodes_df = graph_backend.node_attrs(attr_keys=["t", "val"]).sort("t") + assert nodes_df["val"].to_list() == [5, 6] + assert graph_backend.edge_attrs(attr_keys=["weight"])["weight"].to_list() == [7] + + def test_remove_edge_by_id(graph_backend: BaseGraph) -> None: """Test removing an edge by ID across backends using unified API.""" # Setup From 3c7f901caff9bf020a67e40598e598941a931ca4 Mon Sep 17 00:00:00 2001 From: Teun Huijben Date: Thu, 30 Jul 2026 13:29:59 -0700 Subject: [PATCH 2/2] remove unnecessary addition --- src/tracksdata/graph/_sql_graph.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tracksdata/graph/_sql_graph.py b/src/tracksdata/graph/_sql_graph.py index 475b778f..ff0ae6be 100644 --- a/src/tracksdata/graph/_sql_graph.py +++ b/src/tracksdata/graph/_sql_graph.py @@ -1060,7 +1060,7 @@ def bulk_add_nodes( node_id = int(indices[i]) node_ids.append(node_id) - insert_rows.append({**node, DEFAULT_ATTR_KEYS.T: time, DEFAULT_ATTR_KEYS.NODE_ID: node_id}) + insert_rows.append({**node, DEFAULT_ATTR_KEYS.NODE_ID: node_id}) # Flatten struct-typed attrs into their physical leaf columns before write. # Non-struct keys (incl. NODE_ID) pass through unchanged.