From 2108da3e3d0696990a4adac444896c9b599c1db8 Mon Sep 17 00:00:00 2001 From: Jordan Matelsky Date: Sat, 25 Jul 2026 14:25:36 -0400 Subject: [PATCH 1/4] fix: preserve data during SQL ingestion --- grand/backends/_sqlbackend.py | 36 +----------- grand/backends/test_sql_transactions.py | 74 +++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 34 deletions(-) diff --git a/grand/backends/_sqlbackend.py b/grand/backends/_sqlbackend.py index af20e23..689d322 100644 --- a/grand/backends/_sqlbackend.py +++ b/grand/backends/_sqlbackend.py @@ -774,22 +774,9 @@ def ingest_from_edgelist_dataframe( else [{} for _ in range(len(edgelist))] ) - edge_rows = [ - { - self._edge_source_key: source, - self._edge_target_key: target, - self._primary_key: f"__{source}__{target}", - "_metadata": metadata, - } - for source, target, metadata in zip(sources, targets, edge_metadata) - ] - - with self._mutation(): - if edge_rows: - self._connection.execute(self._edge_table.insert(), edge_rows) - + with self.transaction(): + self.add_edges_from(zip(sources, targets, edge_metadata)) edge_toc = time.time() - edge_tic - node_tic = time.time() nodes = pd.unique( pd.concat( @@ -798,25 +785,6 @@ def ingest_from_edgelist_dataframe( ) ) - node_rows = [ - { - self._primary_key: str(node), - "_metadata": {}, - } - for node in nodes - ] - - if node_rows: - node_insert = self._node_table.insert() - if self._engine.dialect.name == "sqlite": - node_insert = node_insert.prefix_with("OR IGNORE") - self._connection.execute(node_insert, node_rows) - elif self._engine.dialect.name in {"mysql", "mariadb"}: - node_insert = node_insert.prefix_with("IGNORE") - self._connection.execute(node_insert, node_rows) - else: - for node in nodes: - self._insert_empty_node_if_missing(node) return { "node_count": len(nodes), diff --git a/grand/backends/test_sql_transactions.py b/grand/backends/test_sql_transactions.py index 1ffad3c..7406829 100644 --- a/grand/backends/test_sql_transactions.py +++ b/grand/backends/test_sql_transactions.py @@ -1,6 +1,7 @@ import pytest from concurrent.futures import ThreadPoolExecutor from unittest.mock import Mock +import pandas as pd sqlalchemy = pytest.importorskip("sqlalchemy") @@ -174,3 +175,76 @@ def test_concurrent_mutations_do_not_share_connection_simultaneously(tmp_path): assert backend.get_node_count() == 20 backend.close() + + +def test_ingest_preserves_existing_nodes_metadata_and_edges(tmp_path): + backend = SQLBackend( + db_url=f"sqlite:///{tmp_path / 'graph.db'}", + directed=True, + ) + backend.add_node("existing", {"kind": "preserved"}) + backend.add_edge("existing", "A", {"weight": 1, "label": "old"}) + edgelist = pd.DataFrame( + { + "source": ["existing", "A"], + "target": ["A", "B"], + "weight": [2, 3], + } + ) + + backend.ingest_from_edgelist_dataframe(edgelist, "source", "target") + + assert backend.get_node_by_id("existing") == {"kind": "preserved"} + assert backend.get_edge_by_id("existing", "A") == { + "weight": 2, + "label": "old", + } + assert backend.get_edge_by_id("A", "B") == {"weight": 3} + assert set(backend.all_nodes_as_iterable()) == {"existing", "A", "B"} + backend.close() + + +def test_ingest_preserves_primary_key_schema(tmp_path): + backend = SQLBackend(db_url=f"sqlite:///{tmp_path / 'graph.db'}") + + backend.ingest_from_edgelist_dataframe( + pd.DataFrame({"source": ["A"], "target": ["B"]}), + "source", + "target", + ) + + assert backend._node_table.primary_key.columns.keys() == ["ID"] + with pytest.raises(sqlalchemy.exc.IntegrityError): + backend._connection.execute( + backend._node_table.insert(), + [ + {"ID": "duplicate", "_metadata": {}}, + {"ID": "duplicate", "_metadata": {}}, + ], + ) + backend._connection.rollback() + backend.close() + + +def test_ingest_rolls_back_nodes_and_edges_on_failure(tmp_path): + backend = SQLBackend(db_url=f"sqlite:///{tmp_path / 'graph.db'}") + original_execute = backend._connection.execute + + def fail_edge_insert(statement, *args, **kwargs): + if getattr(statement, "table", None) is backend._edge_table: + raise RuntimeError("ingest failed") + return original_execute(statement, *args, **kwargs) + + backend._connection.execute = fail_edge_insert + + with pytest.raises(RuntimeError, match="ingest failed"): + backend.ingest_from_edgelist_dataframe( + pd.DataFrame({"source": ["A"], "target": ["B"]}), + "source", + "target", + ) + + backend._connection.execute = original_execute + assert backend.get_node_count() == 0 + assert backend.get_edge_count() == 0 + backend.close() From 31ad275ddd0e9acbaeef7e4aae423761de539080 Mon Sep 17 00:00:00 2001 From: Jordan Matelsky Date: Sat, 25 Jul 2026 14:37:06 -0400 Subject: [PATCH 2/4] perf: batch SQL ingestion lookups --- grand/backends/_sqlbackend.py | 38 ++++++++++++++++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/grand/backends/_sqlbackend.py b/grand/backends/_sqlbackend.py index 689d322..d86a939 100644 --- a/grand/backends/_sqlbackend.py +++ b/grand/backends/_sqlbackend.py @@ -774,8 +774,42 @@ def ingest_from_edgelist_dataframe( else [{} for _ in range(len(edgelist))] ) + edge_rows = [ + { + self._primary_key: f"__{source}__{target}", + self._edge_source_key: source, + self._edge_target_key: target, + "_metadata": metadata, + } + for source, target, metadata in zip(sources, targets, edge_metadata) + ] + edge_ids = [row[self._primary_key] for row in edge_rows] + with self.transaction(): - self.add_edges_from(zip(sources, targets, edge_metadata)) + existing = { + row[self._primary_key]: row["_metadata"] + for row in self._connection.execute( + select( + self._edge_table.c[self._primary_key], + self._edge_table.c["_metadata"], + ).where(self._edge_table.c[self._primary_key].in_(edge_ids)) + ).mappings() + } + new_edges = [ + row for row in edge_rows if row[self._primary_key] not in existing + ] + if new_edges: + self._connection.execute(self._edge_table.insert(), new_edges) + for row in edge_rows: + edge_id = row[self._primary_key] + if edge_id in existing: + metadata = {**existing[edge_id], **row["_metadata"]} + self._connection.execute( + self._edge_table.update().where( + self._edge_table.c[self._primary_key] == edge_id + ), + parameters={"_metadata": metadata}, + ) edge_toc = time.time() - edge_tic node_tic = time.time() nodes = pd.unique( @@ -784,6 +818,8 @@ def ingest_from_edgelist_dataframe( ignore_index=True, ) ) + for node in nodes: + self._insert_empty_node_if_missing(node) return { From 9d236764fb694eac4f53c822fd05eaea2c05d107 Mon Sep 17 00:00:00 2001 From: Jordan Matelsky Date: Sat, 25 Jul 2026 14:47:39 -0400 Subject: [PATCH 3/4] perf: bulk insert SQL ingest nodes --- grand/backends/_sqlbackend.py | 28 ++++++++++++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/grand/backends/_sqlbackend.py b/grand/backends/_sqlbackend.py index d86a939..7473ec2 100644 --- a/grand/backends/_sqlbackend.py +++ b/grand/backends/_sqlbackend.py @@ -818,8 +818,32 @@ def ingest_from_edgelist_dataframe( ignore_index=True, ) ) - for node in nodes: - self._insert_empty_node_if_missing(node) + node_rows = [ + {self._primary_key: str(node), "_metadata": {}} for node in nodes + ] + if node_rows: + insert = self._node_table.insert() + if self._engine.dialect.name == "sqlite": + insert = insert.prefix_with("OR IGNORE") + elif self._engine.dialect.name in {"mysql", "mariadb"}: + insert = insert.prefix_with("IGNORE") + else: + existing_nodes = set( + self._connection.execute( + select(self._node_table.c[self._primary_key]).where( + self._node_table.c[self._primary_key].in_( + [row[self._primary_key] for row in node_rows] + ) + ) + ).scalars() + ) + node_rows = [ + row + for row in node_rows + if row[self._primary_key] not in existing_nodes + ] + if node_rows: + self._connection.execute(insert, node_rows) return { From 30be5b7b3e245e54635e027de65a725c66552e04 Mon Sep 17 00:00:00 2001 From: Jordan Matelsky Date: Sat, 25 Jul 2026 14:55:08 -0400 Subject: [PATCH 4/4] perf: defer SQL ingest conflict lookups --- grand/backends/_sqlbackend.py | 53 +++++++++++++++++++---------------- 1 file changed, 29 insertions(+), 24 deletions(-) diff --git a/grand/backends/_sqlbackend.py b/grand/backends/_sqlbackend.py index 7473ec2..67a65bf 100644 --- a/grand/backends/_sqlbackend.py +++ b/grand/backends/_sqlbackend.py @@ -786,30 +786,35 @@ def ingest_from_edgelist_dataframe( edge_ids = [row[self._primary_key] for row in edge_rows] with self.transaction(): - existing = { - row[self._primary_key]: row["_metadata"] - for row in self._connection.execute( - select( - self._edge_table.c[self._primary_key], - self._edge_table.c["_metadata"], - ).where(self._edge_table.c[self._primary_key].in_(edge_ids)) - ).mappings() - } - new_edges = [ - row for row in edge_rows if row[self._primary_key] not in existing - ] - if new_edges: - self._connection.execute(self._edge_table.insert(), new_edges) - for row in edge_rows: - edge_id = row[self._primary_key] - if edge_id in existing: - metadata = {**existing[edge_id], **row["_metadata"]} - self._connection.execute( - self._edge_table.update().where( - self._edge_table.c[self._primary_key] == edge_id - ), - parameters={"_metadata": metadata}, - ) + try: + with self._connection.begin_nested(): + if edge_rows: + self._connection.execute(self._edge_table.insert(), edge_rows) + except sqlalchemy.exc.IntegrityError: + existing = { + row[self._primary_key]: row["_metadata"] + for row in self._connection.execute( + select( + self._edge_table.c[self._primary_key], + self._edge_table.c["_metadata"], + ).where(self._edge_table.c[self._primary_key].in_(edge_ids)) + ).mappings() + } + new_edges = [ + row for row in edge_rows if row[self._primary_key] not in existing + ] + if new_edges: + self._connection.execute(self._edge_table.insert(), new_edges) + for row in edge_rows: + edge_id = row[self._primary_key] + if edge_id in existing: + metadata = {**existing[edge_id], **row["_metadata"]} + self._connection.execute( + self._edge_table.update().where( + self._edge_table.c[self._primary_key] == edge_id + ), + parameters={"_metadata": metadata}, + ) edge_toc = time.time() - edge_tic node_tic = time.time() nodes = pd.unique(