diff --git a/docs/tech-specs/collection-management.md b/docs/tech-specs/collection-management.md index 55e50ea1..73b76381 100644 --- a/docs/tech-specs/collection-management.md +++ b/docs/tech-specs/collection-management.md @@ -72,18 +72,18 @@ The collection metadata will be stored in a structured Cassandra table in the li ```sql CREATE TABLE collections ( user text, - collection_id text, + collection text, name text, description text, tags set, created_at timestamp, updated_at timestamp, - PRIMARY KEY (user, collection_id) + PRIMARY KEY (user, collection) ); ``` Table structure: -- **user** + **collection_id**: Composite primary key ensuring user isolation +- **user** + **collection**: Composite primary key ensuring user isolation - **name**: Human-readable collection name - **description**: Detailed description of collection purpose - **tags**: Set of tags for categorization and filtering @@ -175,7 +175,7 @@ All store writers will implement a standardized collection management interface { "operation": "delete-collection", "user": "user123", - "collection_id": "documents-2024", + "collection": "documents-2024", "timestamp": "2024-01-15T10:30:00Z" } ``` @@ -206,15 +206,15 @@ As part of this implementation, the Cassandra triple store will be refactored fr **New Architecture:** - Keyspace per user, single "triples" table for all collections -- Schema: `(collection_id, s, p, o)` with `PRIMARY KEY (collection_id, s, p, o)` -- Collection isolation through collection_id partitioning +- Schema: `(collection, s, p, o)` with `PRIMARY KEY (collection, s, p, o)` +- Collection isolation through collection partitioning **Changes Required:** 1. **TrustGraph Class Refactor** (`trustgraph/direct/cassandra.py`): - Remove `table` parameter from constructor, use fixed "triples" table - - Add `collection_id` parameter to all methods - - Update schema to include collection_id as first column + - Add `collection` parameter to all methods + - Update schema to include collection as first column - **Index Updates**: New indexes will be created to support all 8 query patterns: - Index on `(s)` for subject-based queries - Index on `(p)` for predicate-based queries @@ -222,11 +222,11 @@ As part of this implementation, the Cassandra triple store will be refactored fr - Note: Cassandra doesn't support multi-column secondary indexes, so these are single-column indexes - **Query Pattern Performance**: - - ✅ `get_all()` - partition scan on `collection_id` - - ✅ `get_s(s)` - uses primary key efficiently (`collection_id, s`) - - ✅ `get_p(p)` - uses `idx_p` with `collection_id` filtering - - ✅ `get_o(o)` - uses `idx_o` with `collection_id` filtering - - ✅ `get_sp(s, p)` - uses primary key efficiently (`collection_id, s, p`) + - ✅ `get_all()` - partition scan on `collection` + - ✅ `get_s(s)` - uses primary key efficiently (`collection, s`) + - ✅ `get_p(p)` - uses `idx_p` with `collection` filtering + - ✅ `get_o(o)` - uses `idx_o` with `collection` filtering + - ✅ `get_sp(s, p)` - uses primary key efficiently (`collection, s, p`) - ⚠️ `get_po(p, o)` - requires `ALLOW FILTERING` (uses either `idx_p` or `idx_o` plus filtering) - ✅ `get_os(o, s)` - uses `idx_o` with additional filtering on `s` - ✅ `get_spo(s, p, o)` - uses full primary key efficiently @@ -235,16 +235,16 @@ As part of this implementation, the Cassandra triple store will be refactored fr 2. **Storage Writer Updates** (`trustgraph/storage/triples/cassandra/write.py`): - Maintain single TrustGraph connection per user instead of per (user, collection) - - Pass collection_id to insert operations + - Pass collection to insert operations - Improved resource utilization with fewer connections 3. **Query Service Updates** (`trustgraph/query/triples/cassandra/service.py`): - Single TrustGraph connection per user - - Pass collection_id to all query operations + - Pass collection to all query operations - Maintain same query logic with collection parameter **Benefits:** -- **Simplified Collection Deletion**: Simple `DELETE FROM triples WHERE collection_id = ?` instead of dropping tables +- **Simplified Collection Deletion**: Simple `DELETE FROM triples WHERE collection = ?` instead of dropping tables - **Resource Efficiency**: Fewer database connections and table objects - **Cross-Collection Operations**: Easier to implement operations spanning multiple collections - **Consistent Architecture**: Aligns with unified collection metadata approach @@ -279,7 +279,7 @@ Existing collections will need to be registered in the new Cassandra collections ### Cassandra Triple Store Migration The Cassandra storage refactor requires data migration from table-per-collection to unified table: - **Pre-migration**: Identify all user keyspaces and collection tables -- **Data Transfer**: Copy triples from individual collection tables to unified "triples" table with collection_id +- **Data Transfer**: Copy triples from individual collection tables to unified "triples" table with collection - **Schema Validation**: Ensure new primary key structure maintains query performance - **Cleanup**: Remove old collection tables after successful migration - **Rollback Plan**: Maintain ability to restore table-per-collection structure if needed diff --git a/trustgraph-flow/trustgraph/direct/cassandra.py b/trustgraph-flow/trustgraph/direct/cassandra.py index f7ca7e5e..d6e44327 100644 --- a/trustgraph-flow/trustgraph/direct/cassandra.py +++ b/trustgraph-flow/trustgraph/direct/cassandra.py @@ -10,14 +10,14 @@ class TrustGraph: def __init__( self, hosts=None, - keyspace="trustgraph", table="default", username=None, password=None + keyspace="trustgraph", username=None, password=None ): if hosts is None: hosts = ["localhost"] self.keyspace = keyspace - self.table = table + self.table = "triples" # Fixed table name for unified schema self.username = username if username and password: @@ -55,13 +55,19 @@ class TrustGraph: self.session.execute(f""" create table if not exists {self.table} ( + collection text, s text, p text, o text, - PRIMARY KEY (s, p, o) + PRIMARY KEY (collection, s, p, o) ); """); + self.session.execute(f""" + create index if not exists {self.table}_s + ON {self.table} (s); + """); + self.session.execute(f""" create index if not exists {self.table}_p ON {self.table} (p); @@ -72,58 +78,66 @@ class TrustGraph: ON {self.table} (o); """); - def insert(self, s, p, o): - + def insert(self, collection, s, p, o): + self.session.execute( - f"insert into {self.table} (s, p, o) values (%s, %s, %s)", - (s, p, o) + f"insert into {self.table} (collection, s, p, o) values (%s, %s, %s, %s)", + (collection, s, p, o) ) - def get_all(self, limit=50): + def get_all(self, collection, limit=50): return self.session.execute( - f"select s, p, o from {self.table} limit {limit}" + f"select s, p, o from {self.table} where collection = %s limit {limit}", + (collection,) ) - def get_s(self, s, limit=10): + def get_s(self, collection, s, limit=10): return self.session.execute( - f"select p, o from {self.table} where s = %s limit {limit}", - (s,) + f"select p, o from {self.table} where collection = %s and s = %s limit {limit}", + (collection, s) ) - def get_p(self, p, limit=10): + def get_p(self, collection, p, limit=10): return self.session.execute( - f"select s, o from {self.table} where p = %s limit {limit}", - (p,) + f"select s, o from {self.table} where collection = %s and p = %s limit {limit}", + (collection, p) ) - def get_o(self, o, limit=10): + def get_o(self, collection, o, limit=10): return self.session.execute( - f"select s, p from {self.table} where o = %s limit {limit}", - (o,) + f"select s, p from {self.table} where collection = %s and o = %s limit {limit}", + (collection, o) ) - def get_sp(self, s, p, limit=10): + def get_sp(self, collection, s, p, limit=10): return self.session.execute( - f"select o from {self.table} where s = %s and p = %s limit {limit}", - (s, p) + f"select o from {self.table} where collection = %s and s = %s and p = %s limit {limit}", + (collection, s, p) ) - def get_po(self, p, o, limit=10): + def get_po(self, collection, p, o, limit=10): return self.session.execute( - f"select s from {self.table} where p = %s and o = %s limit {limit} allow filtering", - (p, o) + f"select s from {self.table} where collection = %s and p = %s and o = %s limit {limit} allow filtering", + (collection, p, o) ) - def get_os(self, o, s, limit=10): + def get_os(self, collection, o, s, limit=10): return self.session.execute( - f"select p from {self.table} where o = %s and s = %s limit {limit}", - (o, s) + f"select p from {self.table} where collection = %s and o = %s and s = %s limit {limit} allow filtering", + (collection, o, s) ) - def get_spo(self, s, p, o, limit=10): + def get_spo(self, collection, s, p, o, limit=10): return self.session.execute( - f"""select s as x from {self.table} where s = %s and p = %s and o = %s limit {limit}""", - (s, p, o) + f"""select s as x from {self.table} where collection = %s and s = %s and p = %s and o = %s limit {limit}""", + (collection, s, p, o) + ) + + def delete_collection(self, collection): + """Delete all triples for a specific collection""" + self.session.execute( + f"delete from {self.table} where collection = %s", + (collection,) ) def close(self): diff --git a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py index a0dde295..fae8af7d 100755 --- a/trustgraph-flow/trustgraph/query/triples/cassandra/service.py +++ b/trustgraph-flow/trustgraph/query/triples/cassandra/service.py @@ -56,21 +56,21 @@ class Processor(TriplesQueryService): try: - table = (query.user, query.collection) + user = query.user - if table != self.table: + if user != self.table: if self.cassandra_username and self.cassandra_password: self.tg = TrustGraph( hosts=self.cassandra_host, - keyspace=query.user, table=query.collection, + keyspace=query.user, username=self.cassandra_username, password=self.cassandra_password ) else: self.tg = TrustGraph( hosts=self.cassandra_host, - keyspace=query.user, table=query.collection, + keyspace=query.user, ) - self.table = table + self.table = user triples = [] @@ -78,13 +78,13 @@ class Processor(TriplesQueryService): if query.p is not None: if query.o is not None: resp = self.tg.get_spo( - query.s.value, query.p.value, query.o.value, + query.collection, query.s.value, query.p.value, query.o.value, limit=query.limit ) triples.append((query.s.value, query.p.value, query.o.value)) else: resp = self.tg.get_sp( - query.s.value, query.p.value, + query.collection, query.s.value, query.p.value, limit=query.limit ) for t in resp: @@ -92,14 +92,14 @@ class Processor(TriplesQueryService): else: if query.o is not None: resp = self.tg.get_os( - query.o.value, query.s.value, + query.collection, query.o.value, query.s.value, limit=query.limit ) for t in resp: triples.append((query.s.value, t.p, query.o.value)) else: resp = self.tg.get_s( - query.s.value, + query.collection, query.s.value, limit=query.limit ) for t in resp: @@ -108,14 +108,14 @@ class Processor(TriplesQueryService): if query.p is not None: if query.o is not None: resp = self.tg.get_po( - query.p.value, query.o.value, + query.collection, query.p.value, query.o.value, limit=query.limit ) for t in resp: triples.append((t.s, query.p.value, query.o.value)) else: resp = self.tg.get_p( - query.p.value, + query.collection, query.p.value, limit=query.limit ) for t in resp: @@ -123,13 +123,14 @@ class Processor(TriplesQueryService): else: if query.o is not None: resp = self.tg.get_o( - query.o.value, + query.collection, query.o.value, limit=query.limit ) for t in resp: triples.append((t.s, t.p, query.o.value)) else: resp = self.tg.get_all( + query.collection, limit=query.limit ) for t in resp: diff --git a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py index 06e8f4e0..6c9c2677 100755 --- a/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py +++ b/trustgraph-flow/trustgraph/storage/triples/cassandra/write.py @@ -52,9 +52,9 @@ class Processor(TriplesStoreService): async def store_triples(self, message): - table = (message.metadata.user, message.metadata.collection) + user = message.metadata.user - if self.table is None or self.table != table: + if self.table is None or self.table != user: self.tg = None @@ -63,24 +63,23 @@ class Processor(TriplesStoreService): self.tg = TrustGraph( hosts=self.cassandra_host, keyspace=message.metadata.user, - table=message.metadata.collection, username=self.cassandra_username, password=self.cassandra_password ) else: self.tg = TrustGraph( hosts=self.cassandra_host, keyspace=message.metadata.user, - table=message.metadata.collection, ) except Exception as e: logger.error(f"Exception: {e}", exc_info=True) time.sleep(1) raise e - self.table = table + self.table = user for t in message.triples: self.tg.insert( + message.metadata.collection, t.s.value, t.p.value, t.o.value