From f03f224af067db1d6904475e00090038e2a54d8e Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Tue, 27 Jan 2026 00:07:01 +0000 Subject: [PATCH] Cassandra indexes --- .../messaging/translators/triples.py | 10 +- .../trustgraph/schema/services/query.py | 1 + .../trustgraph/direct/cassandra_kg.py | 525 ++++++++++++------ .../query/triples/cassandra/service.py | 116 ++-- .../storage/triples/cassandra/write.py | 30 +- 5 files changed, 452 insertions(+), 230 deletions(-) diff --git a/trustgraph-base/trustgraph/messaging/translators/triples.py b/trustgraph-base/trustgraph/messaging/translators/triples.py index 1c08625b..2b01b1bc 100644 --- a/trustgraph-base/trustgraph/messaging/translators/triples.py +++ b/trustgraph-base/trustgraph/messaging/translators/triples.py @@ -14,11 +14,13 @@ class TriplesQueryRequestTranslator(MessageTranslator): s = self.value_translator.to_pulsar(data["s"]) if "s" in data else None p = self.value_translator.to_pulsar(data["p"]) if "p" in data else None o = self.value_translator.to_pulsar(data["o"]) if "o" in data else None - + g = data.get("g") # None=default graph, "*"=all graphs + return TriplesQueryRequest( s=s, p=p, o=o, + g=g, limit=int(data.get("limit", 10000)), user=data.get("user", "trustgraph"), collection=data.get("collection", "default") @@ -30,14 +32,16 @@ class TriplesQueryRequestTranslator(MessageTranslator): "user": obj.user, "collection": obj.collection } - + if obj.s: result["s"] = self.value_translator.from_pulsar(obj.s) if obj.p: result["p"] = self.value_translator.from_pulsar(obj.p) if obj.o: result["o"] = self.value_translator.from_pulsar(obj.o) - + if obj.g is not None: + result["g"] = obj.g + return result diff --git a/trustgraph-base/trustgraph/schema/services/query.py b/trustgraph-base/trustgraph/schema/services/query.py index 991e8729..dc33febe 100644 --- a/trustgraph-base/trustgraph/schema/services/query.py +++ b/trustgraph-base/trustgraph/schema/services/query.py @@ -30,6 +30,7 @@ class TriplesQueryRequest: s: Term | None = None p: Term | None = None o: Term | None = None + g: str | None = None # Graph IRI. None=default graph, "*"=all graphs limit: int = 0 @dataclass diff --git a/trustgraph-flow/trustgraph/direct/cassandra_kg.py b/trustgraph-flow/trustgraph/direct/cassandra_kg.py index 116abe02..f8a20041 100644 --- a/trustgraph-flow/trustgraph/direct/cassandra_kg.py +++ b/trustgraph-flow/trustgraph/direct/cassandra_kg.py @@ -11,7 +11,24 @@ _active_clusters = [] logger = logging.getLogger(__name__) +# Sentinel value for wildcard graph queries +GRAPH_WILDCARD = "*" + +# Default graph stored as empty string +DEFAULT_GRAPH = "" + + class KnowledgeGraph: + """ + Cassandra-backed knowledge graph supporting quads (s, p, o, g). + + Uses 7 tables to support all 16 query patterns efficiently: + - Family A (g-wildcard): SPOG, POSG, OSPG + - Family B (g-specified): GSPO, GPOS, GOSP + - Collection table: COLL (for iteration/deletion) + + Plus a metadata table for tracking collections. + """ def __init__( self, hosts=None, @@ -24,12 +41,22 @@ class KnowledgeGraph: self.keyspace = keyspace self.username = username - # Optimized multi-table schema with collection deletion support - self.subject_table = "triples_s" - self.po_table = "triples_p" - self.object_table = "triples_o" - self.collection_table = "triples_collection" # For SPO queries and deletion - self.collection_metadata_table = "collection_metadata" # For tracking which collections exist + # 7-table schema for quads with full query pattern support + # Family A: g-wildcard queries (g in clustering columns) + self.spog_table = "quads_spog" # partition (collection, s), cluster (p, o, g) + self.posg_table = "quads_posg" # partition (collection, p), cluster (o, s, g) + self.ospg_table = "quads_ospg" # partition (collection, o), cluster (s, p, g) + + # Family B: g-specified queries (g in partition key) + self.gspo_table = "quads_gspo" # partition (collection, g, s), cluster (p, o) + self.gpos_table = "quads_gpos" # partition (collection, g, p), cluster (o, s) + self.gosp_table = "quads_gosp" # partition (collection, g, o), cluster (s, p) + + # Collection table for iteration and bulk deletion + self.coll_table = "quads_coll" # partition (collection), cluster (g, s, p, o) + + # Collection metadata tracking + self.collection_metadata_table = "collection_metadata" if username and password: ssl_context = SSLContext(PROTOCOL_TLSv1_2) @@ -46,237 +73,376 @@ class KnowledgeGraph: self.prepare_statements() def clear(self): - self.session.execute(f""" drop keyspace if exists {self.keyspace}; - """); - + """) self.init() def init(self): - self.session.execute(f""" create keyspace if not exists {self.keyspace} with replication = {{ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }}; - """); + """) self.session.set_keyspace(self.keyspace) - self.init_optimized_schema() + self.init_quad_schema() + def init_quad_schema(self): + """Initialize 7-table schema for quads with full query pattern support""" - def init_optimized_schema(self): - """Initialize optimized multi-table schema for performance""" - # Table 1: Subject-centric queries (get_s, get_sp, get_os) - # Compound partition key for optimal data distribution + # Family A: g-wildcard queries (g in clustering columns) + + # SPOG: partition (collection, s), cluster (p, o, g) + # Supports: (?, s, ?, ?), (?, s, p, ?), (?, s, p, o) self.session.execute(f""" - CREATE TABLE IF NOT EXISTS {self.subject_table} ( + CREATE TABLE IF NOT EXISTS {self.spog_table} ( collection text, s text, p text, o text, - PRIMARY KEY ((collection, s), p, o) + g text, + PRIMARY KEY ((collection, s), p, o, g) ); - """); + """) - # Table 2: Predicate-Object queries (get_p, get_po) - eliminates ALLOW FILTERING! - # Compound partition key for optimal data distribution + # POSG: partition (collection, p), cluster (o, s, g) + # Supports: (?, ?, p, ?), (?, ?, p, o) self.session.execute(f""" - CREATE TABLE IF NOT EXISTS {self.po_table} ( + CREATE TABLE IF NOT EXISTS {self.posg_table} ( collection text, p text, o text, s text, - PRIMARY KEY ((collection, p), o, s) + g text, + PRIMARY KEY ((collection, p), o, s, g) ); - """); + """) - # Table 3: Object-centric queries (get_o) - # Compound partition key for optimal data distribution + # OSPG: partition (collection, o), cluster (s, p, g) + # Supports: (?, ?, ?, o), (?, s, ?, o) self.session.execute(f""" - CREATE TABLE IF NOT EXISTS {self.object_table} ( + CREATE TABLE IF NOT EXISTS {self.ospg_table} ( collection text, o text, s text, p text, - PRIMARY KEY ((collection, o), s, p) + g text, + PRIMARY KEY ((collection, o), s, p, g) ); - """); + """) - # Table 4: Collection management and SPO queries (get_spo) - # Simple partition key enables efficient collection deletion + # Family B: g-specified queries (g in partition key) + + # GSPO: partition (collection, g, s), cluster (p, o) + # Supports: (g, s, ?, ?), (g, s, p, ?), (g, s, p, o) self.session.execute(f""" - CREATE TABLE IF NOT EXISTS {self.collection_table} ( + CREATE TABLE IF NOT EXISTS {self.gspo_table} ( collection text, + g text, s text, p text, o text, - PRIMARY KEY (collection, s, p, o) + PRIMARY KEY ((collection, g, s), p, o) ); - """); + """) - # Table 5: Collection metadata tracking - # Tracks which collections exist without polluting triple data + # GPOS: partition (collection, g, p), cluster (o, s) + # Supports: (g, ?, p, ?), (g, ?, p, o) + self.session.execute(f""" + CREATE TABLE IF NOT EXISTS {self.gpos_table} ( + collection text, + g text, + p text, + o text, + s text, + PRIMARY KEY ((collection, g, p), o, s) + ); + """) + + # GOSP: partition (collection, g, o), cluster (s, p) + # Supports: (g, ?, ?, o), (g, s, ?, o) + self.session.execute(f""" + CREATE TABLE IF NOT EXISTS {self.gosp_table} ( + collection text, + g text, + o text, + s text, + p text, + PRIMARY KEY ((collection, g, o), s, p) + ); + """) + + # Collection table for iteration and bulk deletion + # COLL: partition (collection), cluster (g, s, p, o) + self.session.execute(f""" + CREATE TABLE IF NOT EXISTS {self.coll_table} ( + collection text, + g text, + s text, + p text, + o text, + PRIMARY KEY (collection, g, s, p, o) + ); + """) + + # Collection metadata tracking self.session.execute(f""" CREATE TABLE IF NOT EXISTS {self.collection_metadata_table} ( collection text, created_at timestamp, PRIMARY KEY (collection) ); - """); + """) - logger.info("Optimized multi-table schema initialized (5 tables)") + logger.info("Quad schema initialized (7 tables + metadata)") def prepare_statements(self): - """Prepare statements for optimal performance""" - # Insert statements for batch operations - self.insert_subject_stmt = self.session.prepare( - f"INSERT INTO {self.subject_table} (collection, s, p, o) VALUES (?, ?, ?, ?)" + """Prepare statements for all 7 tables""" + + # Insert statements + self.insert_spog_stmt = self.session.prepare( + f"INSERT INTO {self.spog_table} (collection, s, p, o, g) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_posg_stmt = self.session.prepare( + f"INSERT INTO {self.posg_table} (collection, p, o, s, g) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_ospg_stmt = self.session.prepare( + f"INSERT INTO {self.ospg_table} (collection, o, s, p, g) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_gspo_stmt = self.session.prepare( + f"INSERT INTO {self.gspo_table} (collection, g, s, p, o) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_gpos_stmt = self.session.prepare( + f"INSERT INTO {self.gpos_table} (collection, g, p, o, s) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_gosp_stmt = self.session.prepare( + f"INSERT INTO {self.gosp_table} (collection, g, o, s, p) VALUES (?, ?, ?, ?, ?)" + ) + self.insert_coll_stmt = self.session.prepare( + f"INSERT INTO {self.coll_table} (collection, g, s, p, o) VALUES (?, ?, ?, ?, ?)" ) - self.insert_po_stmt = self.session.prepare( - f"INSERT INTO {self.po_table} (collection, p, o, s) VALUES (?, ?, ?, ?)" + # Delete statements (for single quad deletion) + self.delete_spog_stmt = self.session.prepare( + f"DELETE FROM {self.spog_table} WHERE collection = ? AND s = ? AND p = ? AND o = ? AND g = ?" + ) + self.delete_posg_stmt = self.session.prepare( + f"DELETE FROM {self.posg_table} WHERE collection = ? AND p = ? AND o = ? AND s = ? AND g = ?" + ) + self.delete_ospg_stmt = self.session.prepare( + f"DELETE FROM {self.ospg_table} WHERE collection = ? AND o = ? AND s = ? AND p = ? AND g = ?" + ) + self.delete_gspo_stmt = self.session.prepare( + f"DELETE FROM {self.gspo_table} WHERE collection = ? AND g = ? AND s = ? AND p = ? AND o = ?" + ) + self.delete_gpos_stmt = self.session.prepare( + f"DELETE FROM {self.gpos_table} WHERE collection = ? AND g = ? AND p = ? AND o = ? AND s = ?" + ) + self.delete_gosp_stmt = self.session.prepare( + f"DELETE FROM {self.gosp_table} WHERE collection = ? AND g = ? AND o = ? AND s = ? AND p = ?" + ) + self.delete_coll_stmt = self.session.prepare( + f"DELETE FROM {self.coll_table} WHERE collection = ? AND g = ? AND s = ? AND p = ? AND o = ?" ) - self.insert_object_stmt = self.session.prepare( - f"INSERT INTO {self.object_table} (collection, o, s, p) VALUES (?, ?, ?, ?)" + # Query statements - Family A (g-wildcard, g in clustering) + + # SPOG table queries + self.get_s_wildcard_stmt = self.session.prepare( + f"SELECT p, o, g FROM {self.spog_table} WHERE collection = ? AND s = ? LIMIT ?" + ) + self.get_sp_wildcard_stmt = self.session.prepare( + f"SELECT o, g FROM {self.spog_table} WHERE collection = ? AND s = ? AND p = ? LIMIT ?" + ) + self.get_spo_wildcard_stmt = self.session.prepare( + f"SELECT g FROM {self.spog_table} WHERE collection = ? AND s = ? AND p = ? AND o = ? LIMIT ?" ) - self.insert_collection_stmt = self.session.prepare( - f"INSERT INTO {self.collection_table} (collection, s, p, o) VALUES (?, ?, ?, ?)" + # POSG table queries + self.get_p_wildcard_stmt = self.session.prepare( + f"SELECT o, s, g FROM {self.posg_table} WHERE collection = ? AND p = ? LIMIT ?" + ) + self.get_po_wildcard_stmt = self.session.prepare( + f"SELECT s, g FROM {self.posg_table} WHERE collection = ? AND p = ? AND o = ? LIMIT ?" ) - # Query statements for optimized access + # OSPG table queries + self.get_o_wildcard_stmt = self.session.prepare( + f"SELECT s, p, g FROM {self.ospg_table} WHERE collection = ? AND o = ? LIMIT ?" + ) + self.get_os_wildcard_stmt = self.session.prepare( + f"SELECT p, g FROM {self.ospg_table} WHERE collection = ? AND o = ? AND s = ? LIMIT ?" + ) + + # Query statements - Family B (g-specified, g in partition) + + # GSPO table queries + self.get_gs_stmt = self.session.prepare( + f"SELECT p, o FROM {self.gspo_table} WHERE collection = ? AND g = ? AND s = ? LIMIT ?" + ) + self.get_gsp_stmt = self.session.prepare( + f"SELECT o FROM {self.gspo_table} WHERE collection = ? AND g = ? AND s = ? AND p = ? LIMIT ?" + ) + self.get_gspo_stmt = self.session.prepare( + f"SELECT s FROM {self.gspo_table} WHERE collection = ? AND g = ? AND s = ? AND p = ? AND o = ? LIMIT ?" + ) + + # GPOS table queries + self.get_gp_stmt = self.session.prepare( + f"SELECT o, s FROM {self.gpos_table} WHERE collection = ? AND g = ? AND p = ? LIMIT ?" + ) + self.get_gpo_stmt = self.session.prepare( + f"SELECT s FROM {self.gpos_table} WHERE collection = ? AND g = ? AND p = ? AND o = ? LIMIT ?" + ) + + # GOSP table queries + self.get_go_stmt = self.session.prepare( + f"SELECT s, p FROM {self.gosp_table} WHERE collection = ? AND g = ? AND o = ? LIMIT ?" + ) + self.get_gos_stmt = self.session.prepare( + f"SELECT p FROM {self.gosp_table} WHERE collection = ? AND g = ? AND o = ? AND s = ? LIMIT ?" + ) + + # Collection table query (for get_all and iteration) self.get_all_stmt = self.session.prepare( - f"SELECT s, p, o FROM {self.subject_table} WHERE collection = ? LIMIT ? ALLOW FILTERING" + f"SELECT g, s, p, o FROM {self.coll_table} WHERE collection = ? LIMIT ?" + ) + self.get_g_stmt = self.session.prepare( + f"SELECT s, p, o FROM {self.coll_table} WHERE collection = ? AND g = ? LIMIT ?" ) - self.get_s_stmt = self.session.prepare( - f"SELECT p, o FROM {self.subject_table} WHERE collection = ? AND s = ? LIMIT ?" - ) + logger.info("Prepared statements initialized for quad schema (7 tables)") - self.get_p_stmt = self.session.prepare( - f"SELECT s, o FROM {self.po_table} WHERE collection = ? AND p = ? LIMIT ?" - ) + def insert(self, collection, s, p, o, g=None): + """Insert a quad into all 7 tables""" + # Default graph stored as empty string + if g is None: + g = DEFAULT_GRAPH - self.get_o_stmt = self.session.prepare( - f"SELECT s, p FROM {self.object_table} WHERE collection = ? AND o = ? LIMIT ?" - ) - - self.get_sp_stmt = self.session.prepare( - f"SELECT o FROM {self.subject_table} WHERE collection = ? AND s = ? AND p = ? LIMIT ?" - ) - - # The critical optimization: get_po without ALLOW FILTERING! - self.get_po_stmt = self.session.prepare( - f"SELECT s FROM {self.po_table} WHERE collection = ? AND p = ? AND o = ? LIMIT ?" - ) - - self.get_os_stmt = self.session.prepare( - f"SELECT p FROM {self.object_table} WHERE collection = ? AND o = ? AND s = ? LIMIT ?" - ) - - self.get_spo_stmt = self.session.prepare( - f"SELECT s as x FROM {self.collection_table} WHERE collection = ? AND s = ? AND p = ? AND o = ? LIMIT ?" - ) - - # Delete statements for collection deletion - self.delete_subject_stmt = self.session.prepare( - f"DELETE FROM {self.subject_table} WHERE collection = ? AND s = ? AND p = ? AND o = ?" - ) - - self.delete_po_stmt = self.session.prepare( - f"DELETE FROM {self.po_table} WHERE collection = ? AND p = ? AND o = ? AND s = ?" - ) - - self.delete_object_stmt = self.session.prepare( - f"DELETE FROM {self.object_table} WHERE collection = ? AND o = ? AND s = ? AND p = ?" - ) - - self.delete_collection_stmt = self.session.prepare( - f"DELETE FROM {self.collection_table} WHERE collection = ? AND s = ? AND p = ? AND o = ?" - ) - - logger.info("Prepared statements initialized for optimal performance (4 tables)") - - def insert(self, collection, s, p, o): - # Batch write to all four tables for consistency batch = BatchStatement() - # Insert into subject table - batch.add(self.insert_subject_stmt, (collection, s, p, o)) + # Family A tables + batch.add(self.insert_spog_stmt, (collection, s, p, o, g)) + batch.add(self.insert_posg_stmt, (collection, p, o, s, g)) + batch.add(self.insert_ospg_stmt, (collection, o, s, p, g)) - # Insert into predicate-object table (column order: collection, p, o, s) - batch.add(self.insert_po_stmt, (collection, p, o, s)) + # Family B tables + batch.add(self.insert_gspo_stmt, (collection, g, s, p, o)) + batch.add(self.insert_gpos_stmt, (collection, g, p, o, s)) + batch.add(self.insert_gosp_stmt, (collection, g, o, s, p)) - # Insert into object table (column order: collection, o, s, p) - batch.add(self.insert_object_stmt, (collection, o, s, p)) - - # Insert into collection table for SPO queries and deletion tracking - batch.add(self.insert_collection_stmt, (collection, s, p, o)) + # Collection table + batch.add(self.insert_coll_stmt, (collection, g, s, p, o)) self.session.execute(batch) + def delete_quad(self, collection, s, p, o, g=None): + """Delete a single quad from all 7 tables""" + if g is None: + g = DEFAULT_GRAPH + + batch = BatchStatement() + + batch.add(self.delete_spog_stmt, (collection, s, p, o, g)) + batch.add(self.delete_posg_stmt, (collection, p, o, s, g)) + batch.add(self.delete_ospg_stmt, (collection, o, s, p, g)) + batch.add(self.delete_gspo_stmt, (collection, g, s, p, o)) + batch.add(self.delete_gpos_stmt, (collection, g, p, o, s)) + batch.add(self.delete_gosp_stmt, (collection, g, o, s, p)) + batch.add(self.delete_coll_stmt, (collection, g, s, p, o)) + + self.session.execute(batch) + + # ======================================================================== + # Query methods + # g=None means default graph, g="*" means all graphs + # ======================================================================== + def get_all(self, collection, limit=50): - # Use subject table for get_all queries - return self.session.execute( - self.get_all_stmt, - (collection, limit) - ) + """Get all quads in collection""" + return self.session.execute(self.get_all_stmt, (collection, limit)) - def get_s(self, collection, s, limit=10): - # Optimized: Direct partition access with (collection, s) - return self.session.execute( - self.get_s_stmt, - (collection, s, limit) - ) + def get_s(self, collection, s, g=None, limit=10): + """Query by subject. g=None: default graph, g='*': all graphs""" + if g is None or g == DEFAULT_GRAPH: + # Default graph - use GSPO table + return self.session.execute(self.get_gs_stmt, (collection, DEFAULT_GRAPH, s, limit)) + elif g == GRAPH_WILDCARD: + # All graphs - use SPOG table + return self.session.execute(self.get_s_wildcard_stmt, (collection, s, limit)) + else: + # Specific graph - use GSPO table + return self.session.execute(self.get_gs_stmt, (collection, g, s, limit)) - def get_p(self, collection, p, limit=10): - # Optimized: Use po_table for direct partition access - return self.session.execute( - self.get_p_stmt, - (collection, p, limit) - ) + def get_p(self, collection, p, g=None, limit=10): + """Query by predicate""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_gp_stmt, (collection, DEFAULT_GRAPH, p, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_p_wildcard_stmt, (collection, p, limit)) + else: + return self.session.execute(self.get_gp_stmt, (collection, g, p, limit)) - def get_o(self, collection, o, limit=10): - # Optimized: Use object_table for direct partition access - return self.session.execute( - self.get_o_stmt, - (collection, o, limit) - ) + def get_o(self, collection, o, g=None, limit=10): + """Query by object""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_go_stmt, (collection, DEFAULT_GRAPH, o, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_o_wildcard_stmt, (collection, o, limit)) + else: + return self.session.execute(self.get_go_stmt, (collection, g, o, limit)) - def get_sp(self, collection, s, p, limit=10): - # Optimized: Use subject_table with clustering key access - return self.session.execute( - self.get_sp_stmt, - (collection, s, p, limit) - ) + def get_sp(self, collection, s, p, g=None, limit=10): + """Query by subject and predicate""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_gsp_stmt, (collection, DEFAULT_GRAPH, s, p, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_sp_wildcard_stmt, (collection, s, p, limit)) + else: + return self.session.execute(self.get_gsp_stmt, (collection, g, s, p, limit)) - def get_po(self, collection, p, o, limit=10): - # CRITICAL OPTIMIZATION: Use po_table - NO MORE ALLOW FILTERING! - return self.session.execute( - self.get_po_stmt, - (collection, p, o, limit) - ) + def get_po(self, collection, p, o, g=None, limit=10): + """Query by predicate and object""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_gpo_stmt, (collection, DEFAULT_GRAPH, p, o, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_po_wildcard_stmt, (collection, p, o, limit)) + else: + return self.session.execute(self.get_gpo_stmt, (collection, g, p, o, limit)) - def get_os(self, collection, o, s, limit=10): - # Optimized: Use subject_table with clustering access (no more ALLOW FILTERING) - return self.session.execute( - self.get_os_stmt, - (collection, s, o, limit) - ) + def get_os(self, collection, o, s, g=None, limit=10): + """Query by object and subject""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_gos_stmt, (collection, DEFAULT_GRAPH, o, s, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_os_wildcard_stmt, (collection, o, s, limit)) + else: + return self.session.execute(self.get_gos_stmt, (collection, g, o, s, limit)) - def get_spo(self, collection, s, p, o, limit=10): - # Optimized: Use collection_table for exact key lookup - return self.session.execute( - self.get_spo_stmt, - (collection, s, p, o, limit) - ) + def get_spo(self, collection, s, p, o, g=None, limit=10): + """Query by subject, predicate, object (find which graphs)""" + if g is None or g == DEFAULT_GRAPH: + return self.session.execute(self.get_gspo_stmt, (collection, DEFAULT_GRAPH, s, p, o, limit)) + elif g == GRAPH_WILDCARD: + return self.session.execute(self.get_spo_wildcard_stmt, (collection, s, p, o, limit)) + else: + return self.session.execute(self.get_gspo_stmt, (collection, g, s, p, o, limit)) + + def get_g(self, collection, g, limit=50): + """Get all quads in a specific graph""" + if g is None: + g = DEFAULT_GRAPH + return self.session.execute(self.get_g_stmt, (collection, g, limit)) + + # ======================================================================== + # Collection management + # ======================================================================== def collection_exists(self, collection): - """Check if collection exists by querying collection_metadata table""" + """Check if collection exists""" try: result = self.session.execute( f"SELECT collection FROM {self.collection_metadata_table} WHERE collection = %s LIMIT 1", @@ -301,63 +467,52 @@ class KnowledgeGraph: raise e def delete_collection(self, collection): - """Delete all triples for a specific collection - - Uses collection_table to enumerate all triples, then deletes from all 4 tables - using full partition keys for optimal performance with compound keys. - """ - # Step 1: Read all triples from collection_table (single partition read) + """Delete all quads for a collection from all 7 tables""" + # Read all quads from collection table rows = self.session.execute( - f"SELECT s, p, o FROM {self.collection_table} WHERE collection = %s", + f"SELECT g, s, p, o FROM {self.coll_table} WHERE collection = %s", (collection,) ) - # Step 2: Delete each triple from all 4 tables using full partition keys - # Batch deletions for efficiency batch = BatchStatement() count = 0 for row in rows: - s, p, o = row.s, row.p, row.o + g, s, p, o = row.g, row.s, row.p, row.o - # Delete from subject table (partition key: collection, s) - batch.add(self.delete_subject_stmt, (collection, s, p, o)) - - # Delete from predicate-object table (partition key: collection, p) - batch.add(self.delete_po_stmt, (collection, p, o, s)) - - # Delete from object table (partition key: collection, o) - batch.add(self.delete_object_stmt, (collection, o, s, p)) - - # Delete from collection table (partition key: collection only) - batch.add(self.delete_collection_stmt, (collection, s, p, o)) + # Delete from all 7 tables + batch.add(self.delete_spog_stmt, (collection, s, p, o, g)) + batch.add(self.delete_posg_stmt, (collection, p, o, s, g)) + batch.add(self.delete_ospg_stmt, (collection, o, s, p, g)) + batch.add(self.delete_gspo_stmt, (collection, g, s, p, o)) + batch.add(self.delete_gpos_stmt, (collection, g, p, o, s)) + batch.add(self.delete_gosp_stmt, (collection, g, o, s, p)) + batch.add(self.delete_coll_stmt, (collection, g, s, p, o)) count += 1 - # Execute batch every 25 triples to avoid oversized batches - # (Each triple adds ~4 statements, so 25 triples = ~100 statements) - if count % 25 == 0: + # Execute batch every 15 quads (7 deletes each = 105 statements) + if count % 15 == 0: self.session.execute(batch) batch = BatchStatement() - # Execute remaining deletions - if count % 25 != 0: + # Execute remaining + if count % 15 != 0: self.session.execute(batch) - # Step 3: Delete collection metadata + # Delete collection metadata self.session.execute( f"DELETE FROM {self.collection_metadata_table} WHERE collection = %s", (collection,) ) - logger.info(f"Deleted {count} triples from collection {collection}") + logger.info(f"Deleted {count} quads from collection {collection}") def close(self): - """Close the Cassandra session and cluster connections properly""" + """Close connections""" if hasattr(self, 'session') and self.session: self.session.shutdown() if hasattr(self, 'cluster') and self.cluster: self.cluster.shutdown() - # Remove from global tracking if self.cluster in _active_clusters: _active_clusters.remove(self.cluster) diff --git a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py index 13726ac3..51bbefa9 100755 --- a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py +++ b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py @@ -1,14 +1,14 @@ """ -Triples query service. Input is a (s, p, o) triple, some values may be -null. Output is a list of triples. +Triples query service. Input is a (s, p, o, g) quad pattern, some values may be +null. Output is a list of quads. """ import logging -from .... direct.cassandra_kg import KnowledgeGraph +from .... direct.cassandra_kg import KnowledgeGraph, GRAPH_WILDCARD, DEFAULT_GRAPH from .... schema import TriplesQueryRequest, TriplesQueryResponse, Error -from .... schema import Value, Triple +from .... schema import Term, Triple, IRI, LITERAL from .... base import TriplesQueryService from .... base.cassandra_config import add_cassandra_args, resolve_cassandra_config @@ -18,6 +18,27 @@ logger = logging.getLogger(__name__) default_ident = "triples-query" +def get_term_value(term): + """Extract the string value from a Term""" + if term is None: + return None + if term.type == IRI: + return term.iri + elif term.type == LITERAL: + return term.value + else: + # For blank nodes or other types, use id or value + return term.id or term.value + + +def create_term(value): + """Create a Term from a string value""" + if value.startswith("http://") or value.startswith("https://"): + return Term(type=IRI, iri=value) + else: + return Term(type=LITERAL, value=value) + + class Processor(TriplesQueryService): def __init__(self, **params): @@ -46,12 +67,6 @@ class Processor(TriplesQueryService): self.cassandra_password = password self.table = None - def create_value(self, ent): - if ent.startswith("http://") or ent.startswith("https://"): - return Value(value=ent, is_uri=True) - else: - return Value(value=ent, is_uri=False) - async def query_triples(self, query): try: @@ -72,77 +87,103 @@ class Processor(TriplesQueryService): ) self.table = user - triples = [] + # Extract values from query + s_val = get_term_value(query.s) + p_val = get_term_value(query.p) + o_val = get_term_value(query.o) + g_val = query.g # Already a string or None - if query.s is not None: - if query.p is not None: - if query.o is not None: + quads = [] + + # Route to appropriate query method based on which fields are specified + if s_val is not None: + if p_val is not None: + if o_val is not None: + # SPO specified - find matching graphs resp = self.tg.get_spo( - query.collection, query.s.value, query.p.value, query.o.value, + query.collection, s_val, p_val, o_val, g=g_val, limit=query.limit ) - triples.append((query.s.value, query.p.value, query.o.value)) + for t in resp: + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((s_val, p_val, o_val, g)) else: + # SP specified resp = self.tg.get_sp( - query.collection, query.s.value, query.p.value, + query.collection, s_val, p_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((query.s.value, query.p.value, t.o)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((s_val, p_val, t.o, g)) else: - if query.o is not None: + if o_val is not None: + # SO specified resp = self.tg.get_os( - query.collection, query.o.value, query.s.value, + query.collection, o_val, s_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((query.s.value, t.p, query.o.value)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((s_val, t.p, o_val, g)) else: + # S only resp = self.tg.get_s( - query.collection, query.s.value, + query.collection, s_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((query.s.value, t.p, t.o)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((s_val, t.p, t.o, g)) else: - if query.p is not None: - if query.o is not None: + if p_val is not None: + if o_val is not None: + # PO specified resp = self.tg.get_po( - query.collection, query.p.value, query.o.value, + query.collection, p_val, o_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((t.s, query.p.value, query.o.value)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((t.s, p_val, o_val, g)) else: + # P only resp = self.tg.get_p( - query.collection, query.p.value, + query.collection, p_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((t.s, query.p.value, t.o)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((t.s, p_val, t.o, g)) else: - if query.o is not None: + if o_val is not None: + # O only resp = self.tg.get_o( - query.collection, query.o.value, + query.collection, o_val, g=g_val, limit=query.limit ) for t in resp: - triples.append((t.s, t.p, query.o.value)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((t.s, t.p, o_val, g)) else: + # Nothing specified - get all resp = self.tg.get_all( query.collection, limit=query.limit ) for t in resp: - triples.append((t.s, t.p, t.o)) + g = t.g if hasattr(t, 'g') else DEFAULT_GRAPH + quads.append((t.s, t.p, t.o, g)) + # Convert to Triple objects (with g field) triples = [ Triple( - s=self.create_value(t[0]), - p=self.create_value(t[1]), - o=self.create_value(t[2]) + s=create_term(q[0]), + p=create_term(q[1]), + o=create_term(q[2]), + g=q[3] if q[3] != DEFAULT_GRAPH else None ) - for t in triples + for q in quads ] return triples @@ -162,4 +203,3 @@ class Processor(TriplesQueryService): def run(): Processor.launch(default_ident, __doc__) - diff --git a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py index b9b42375..6ed93b7d 100755 --- a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py @@ -10,11 +10,12 @@ import argparse import time import logging -from .... direct.cassandra_kg import KnowledgeGraph +from .... direct.cassandra_kg import KnowledgeGraph, DEFAULT_GRAPH from .... base import TriplesStoreService, CollectionConfigHandler from .... base import AsyncProcessor, Consumer, Producer from .... base import ConsumerMetrics, ProducerMetrics from .... base.cassandra_config import add_cassandra_args, resolve_cassandra_config +from .... schema import IRI, LITERAL # Module logger logger = logging.getLogger(__name__) @@ -22,6 +23,19 @@ logger = logging.getLogger(__name__) default_ident = "triples-write" +def get_term_value(term): + """Extract the string value from a Term""" + if term is None: + return None + if term.type == IRI: + return term.iri + elif term.type == LITERAL: + return term.value + else: + # For blank nodes or other types, use id or value + return term.id or term.value + + class Processor(CollectionConfigHandler, TriplesStoreService): def __init__(self, **params): @@ -84,11 +98,19 @@ class Processor(CollectionConfigHandler, TriplesStoreService): self.table = user for t in message.triples: + # Extract values from Term objects + s_val = get_term_value(t.s) + p_val = get_term_value(t.p) + o_val = get_term_value(t.o) + # t.g is None for default graph, or a graph IRI + g_val = t.g if t.g is not None else DEFAULT_GRAPH + self.tg.insert( message.metadata.collection, - t.s.value, - t.p.value, - t.o.value + s_val, + p_val, + o_val, + g=g_val ) async def create_collection(self, user: str, collection: str, metadata: dict):