Skip to content

Fix: PostgreSQL Data Migration Performance Problem #2259

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danielaskdd merged 1 commit into from
Oct 25, 2025
+24 −11
Merged

Fix: PostgreSQL Data Migration Performance Problem #2259

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 24 additions & 11 deletions lightrag/kg/postgres_impl.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4613,16 +4613,19 @@ async def get_all_nodes(self) -> list[dict]:
Returns:
A list of all nodes, where each node is a dictionary of its properties
"""
query = f"""SELECT * FROM cypher('{self.graph_name}', $$
MATCH (n:base)
RETURN n
$$) AS (n agtype)"""
# Use native SQL to avoid Cypher wrapper overhead
# Original: SELECT * FROM cypher(...) with MATCH (n:base)
# Optimized: Direct table access for better performance
query = f"""
SELECT properties
FROM {self.graph_name}.base
"""

results = await self._query(query)
nodes = []
for result in results:
if result["n"]:
node_dict = result["n"]["properties"]
if result.get("properties"):
node_dict = result["properties"]

# Process string result, parse it to JSON dictionary
if isinstance(node_dict, str):
Expand All @@ -4632,6 +4635,7 @@ async def get_all_nodes(self) -> list[dict]:
logger.warning(
f"[{self.workspace}] Failed to parse node string: {node_dict}"
)
continue

# Add node id (entity_id) to the dictionary for easier access
node_dict["id"] = node_dict.get("entity_id")
Expand All @@ -4643,12 +4647,21 @@ async def get_all_edges(self) -> list[dict]:

Returns:
A list of all edges, where each edge is a dictionary of its properties
(The edge is bidirectional; deduplication must be handled by the caller)
(If 2 directional edges exist between the same pair of nodes, deduplication must be handled by the caller)
"""
# Use native SQL to avoid Cartesian product (N×N) in Cypher MATCH
# Original Cypher: MATCH (a:base)-[r]-(b:base) creates ~50 billion row combinations
# Optimized: Start from edges table, join to nodes only to get entity_id
# Performance: O(E) instead of O(N²), ~50,000x faster for large graphs
query = f"""
SELECT DISTINCT
(ag_catalog.agtype_access_operator(VARIADIC ARRAY[a.properties, '"entity_id"'::agtype]))::text AS source,
(ag_catalog.agtype_access_operator(VARIADIC ARRAY[b.properties, '"entity_id"'::agtype]))::text AS target,
r.properties
FROM {self.graph_name}."DIRECTED" r
JOIN {self.graph_name}.base a ON r.start_id = a.id
JOIN {self.graph_name}.base b ON r.end_id = b.id
"""
query = f"""SELECT * FROM cypher('{self.graph_name}', $$
MATCH (a:base)-[r]-(b:base)
RETURN DISTINCT a.entity_id AS source, b.entity_id AS target, properties(r) AS properties
$$) AS (source text, target text, properties agtype)"""

results = await self._query(query)
edges = []
Expand Down