diff --git a/trustgraph-flow/trustgraph/query/triples/neo4j/service.py b/trustgraph-flow/trustgraph/query/triples/neo4j/service.py index 69e10d62..0e84d733 100755 --- a/trustgraph-flow/trustgraph/query/triples/neo4j/service.py +++ b/trustgraph-flow/trustgraph/query/triples/neo4j/service.py @@ -55,6 +55,10 @@ class Processor(TriplesQueryService): try: + # Extract user and collection, use defaults if not provided + user = query.user if query.user else "default" + collection = query.collection if query.collection else "default" + triples = [] if query.s is not None: @@ -64,9 +68,12 @@ class Processor(TriplesQueryService): # SPO records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal {value: $value}) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {uri: $rel, user: $user, collection: $collection}]->" + "(dest:Literal {value: $value, user: $user, collection: $collection}) " "RETURN $src as src", src=query.s.value, rel=query.p.value, value=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -74,9 +81,12 @@ class Processor(TriplesQueryService): triples.append((query.s.value, query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node {uri: $uri}) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {uri: $rel, user: $user, collection: $collection}]->" + "(dest:Node {uri: $uri, user: $user, collection: $collection}) " "RETURN $src as src", src=query.s.value, rel=query.p.value, uri=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -88,9 +98,12 @@ class Processor(TriplesQueryService): # SP records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Literal) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {uri: $rel, user: $user, collection: $collection}]->" + "(dest:Literal {user: $user, collection: $collection}) " "RETURN dest.value as dest", src=query.s.value, rel=query.p.value, + user=user, collection=collection, database_=self.db, ) @@ -99,9 +112,12 @@ class Processor(TriplesQueryService): triples.append((query.s.value, query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel {uri: $rel}]->(dest:Node) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {uri: $rel, user: $user, collection: $collection}]->" + "(dest:Node {user: $user, collection: $collection}) " "RETURN dest.uri as dest", src=query.s.value, rel=query.p.value, + user=user, collection=collection, database_=self.db, ) @@ -116,9 +132,12 @@ class Processor(TriplesQueryService): # SO records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal {value: $value}) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Literal {value: $value, user: $user, collection: $collection}) " "RETURN rel.uri as rel", src=query.s.value, value=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -127,9 +146,12 @@ class Processor(TriplesQueryService): triples.append((query.s.value, data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node {uri: $uri}) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Node {uri: $uri, user: $user, collection: $collection}) " "RETURN rel.uri as rel", src=query.s.value, uri=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -142,9 +164,12 @@ class Processor(TriplesQueryService): # S records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Literal) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Literal {user: $user, collection: $collection}) " "RETURN rel.uri as rel, dest.value as dest", src=query.s.value, + user=user, collection=collection, database_=self.db, ) @@ -153,9 +178,12 @@ class Processor(TriplesQueryService): triples.append((query.s.value, data["rel"], data["dest"])) records, summary, keys = self.io.execute_query( - "MATCH (src:Node {uri: $src})-[rel:Rel]->(dest:Node) " + "MATCH (src:Node {uri: $src, user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Node {user: $user, collection: $collection}) " "RETURN rel.uri as rel, dest.uri as dest", src=query.s.value, + user=user, collection=collection, database_=self.db, ) @@ -173,9 +201,12 @@ class Processor(TriplesQueryService): # PO records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal {value: $value}) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {uri: $uri, user: $user, collection: $collection}]->" + "(dest:Literal {value: $value, user: $user, collection: $collection}) " "RETURN src.uri as src", uri=query.p.value, value=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -184,9 +215,12 @@ class Processor(TriplesQueryService): triples.append((data["src"], query.p.value, query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node {uri: $dest}) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {uri: $uri, user: $user, collection: $collection}]->" + "(dest:Node {uri: $dest, user: $user, collection: $collection}) " "RETURN src.uri as src", uri=query.p.value, dest=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -199,9 +233,12 @@ class Processor(TriplesQueryService): # P records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Literal) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {uri: $uri, user: $user, collection: $collection}]->" + "(dest:Literal {user: $user, collection: $collection}) " "RETURN src.uri as src, dest.value as dest", uri=query.p.value, + user=user, collection=collection, database_=self.db, ) @@ -210,9 +247,12 @@ class Processor(TriplesQueryService): triples.append((data["src"], query.p.value, data["dest"])) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel {uri: $uri}]->(dest:Node) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {uri: $uri, user: $user, collection: $collection}]->" + "(dest:Node {user: $user, collection: $collection}) " "RETURN src.uri as src, dest.uri as dest", uri=query.p.value, + user=user, collection=collection, database_=self.db, ) @@ -227,9 +267,12 @@ class Processor(TriplesQueryService): # O records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel]->(dest:Literal {value: $value}) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Literal {value: $value, user: $user, collection: $collection}) " "RETURN src.uri as src, rel.uri as rel", value=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -238,9 +281,12 @@ class Processor(TriplesQueryService): triples.append((data["src"], data["rel"], query.o.value)) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel]->(dest:Node {uri: $uri}) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Node {uri: $uri, user: $user, collection: $collection}) " "RETURN src.uri as src, rel.uri as rel", uri=query.o.value, + user=user, collection=collection, database_=self.db, ) @@ -253,8 +299,11 @@ class Processor(TriplesQueryService): # * records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel]->(dest:Literal) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Literal {user: $user, collection: $collection}) " "RETURN src.uri as src, rel.uri as rel, dest.value as dest", + user=user, collection=collection, database_=self.db, ) @@ -263,8 +312,11 @@ class Processor(TriplesQueryService): triples.append((data["src"], data["rel"], data["dest"])) records, summary, keys = self.io.execute_query( - "MATCH (src:Node)-[rel:Rel]->(dest:Node) " + "MATCH (src:Node {user: $user, collection: $collection})-" + "[rel:Rel {user: $user, collection: $collection}]->" + "(dest:Node {user: $user, collection: $collection}) " "RETURN src.uri as src, rel.uri as rel, dest.uri as dest", + user=user, collection=collection, database_=self.db, ) diff --git a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py index e1913c14..c33478eb 100755 --- a/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/neo4j/write.py @@ -61,6 +61,7 @@ class Processor(TriplesStoreService): logger.info("Create indexes...") + # Legacy indexes for backwards compatibility try: session.run( "CREATE INDEX Node_uri FOR (n:Node) ON (n.uri)", @@ -88,15 +89,50 @@ class Processor(TriplesStoreService): # Maybe index already exists logger.warning("Index create failure ignored") + # New compound indexes for user/collection filtering + try: + session.run( + "CREATE INDEX node_user_collection_uri FOR (n:Node) ON (n.user, n.collection, n.uri)", + ) + except Exception as e: + logger.warning(f"Compound index create failure: {e}") + logger.warning("Index create failure ignored") + + try: + session.run( + "CREATE INDEX literal_user_collection_value FOR (n:Literal) ON (n.user, n.collection, n.value)", + ) + except Exception as e: + logger.warning(f"Compound index create failure: {e}") + logger.warning("Index create failure ignored") + + # Note: Neo4j doesn't support compound indexes on relationships in all versions + # Try to create individual indexes on relationship properties + try: + session.run( + "CREATE INDEX rel_user FOR ()-[r:Rel]-() ON (r.user)", + ) + except Exception as e: + logger.warning(f"Relationship index create failure: {e}") + logger.warning("Index create failure ignored") + + try: + session.run( + "CREATE INDEX rel_collection FOR ()-[r:Rel]-() ON (r.collection)", + ) + except Exception as e: + logger.warning(f"Relationship index create failure: {e}") + logger.warning("Index create failure ignored") + logger.info("Index creation done") - def create_node(self, uri): + def create_node(self, uri, user, collection): - logger.debug(f"Create node {uri}") + logger.debug(f"Create node {uri} for user={user}, collection={collection}") summary = self.io.execute_query( - "MERGE (n:Node {uri: $uri})", - uri=uri, + "MERGE (n:Node {uri: $uri, user: $user, collection: $collection})", + uri=uri, user=user, collection=collection, database_=self.db, ).summary @@ -105,13 +141,13 @@ class Processor(TriplesStoreService): time=summary.result_available_after )) - def create_literal(self, value): + def create_literal(self, value, user, collection): - logger.debug(f"Create literal {value}") + logger.debug(f"Create literal {value} for user={user}, collection={collection}") summary = self.io.execute_query( - "MERGE (n:Literal {value: $value})", - value=value, + "MERGE (n:Literal {value: $value, user: $user, collection: $collection})", + value=value, user=user, collection=collection, database_=self.db, ).summary @@ -120,15 +156,15 @@ class Processor(TriplesStoreService): time=summary.result_available_after )) - def relate_node(self, src, uri, dest): + def relate_node(self, src, uri, dest, user, collection): - logger.debug(f"Create node rel {src} {uri} {dest}") + logger.debug(f"Create node rel {src} {uri} {dest} for user={user}, collection={collection}") summary = self.io.execute_query( - "MATCH (src:Node {uri: $src}) " - "MATCH (dest:Node {uri: $dest}) " - "MERGE (src)-[:Rel {uri: $uri}]->(dest)", - src=src, dest=dest, uri=uri, + "MATCH (src:Node {uri: $src, user: $user, collection: $collection}) " + "MATCH (dest:Node {uri: $dest, user: $user, collection: $collection}) " + "MERGE (src)-[:Rel {uri: $uri, user: $user, collection: $collection}]->(dest)", + src=src, dest=dest, uri=uri, user=user, collection=collection, database_=self.db, ).summary @@ -137,15 +173,15 @@ class Processor(TriplesStoreService): time=summary.result_available_after )) - def relate_literal(self, src, uri, dest): + def relate_literal(self, src, uri, dest, user, collection): - logger.debug(f"Create literal rel {src} {uri} {dest}") + logger.debug(f"Create literal rel {src} {uri} {dest} for user={user}, collection={collection}") summary = self.io.execute_query( - "MATCH (src:Node {uri: $src}) " - "MATCH (dest:Literal {value: $dest}) " - "MERGE (src)-[:Rel {uri: $uri}]->(dest)", - src=src, dest=dest, uri=uri, + "MATCH (src:Node {uri: $src, user: $user, collection: $collection}) " + "MATCH (dest:Literal {value: $dest, user: $user, collection: $collection}) " + "MERGE (src)-[:Rel {uri: $uri, user: $user, collection: $collection}]->(dest)", + src=src, dest=dest, uri=uri, user=user, collection=collection, database_=self.db, ).summary @@ -156,16 +192,20 @@ class Processor(TriplesStoreService): async def store_triples(self, message): + # Extract user and collection from metadata + user = message.metadata.user if message.metadata.user else "default" + collection = message.metadata.collection if message.metadata.collection else "default" + for t in message.triples: - self.create_node(t.s.value) + self.create_node(t.s.value, user, collection) if t.o.is_uri: - self.create_node(t.o.value) - self.relate_node(t.s.value, t.p.value, t.o.value) + self.create_node(t.o.value, user, collection) + self.relate_node(t.s.value, t.p.value, t.o.value, user, collection) else: - self.create_literal(t.o.value) - self.relate_literal(t.s.value, t.p.value, t.o.value) + self.create_literal(t.o.value, user, collection) + self.relate_literal(t.s.value, t.p.value, t.o.value, user, collection) @staticmethod def add_args(parser):