mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-22 11:41:02 +02:00
Cassandra indexes
This commit is contained in:
Cyber MacGeddon
parent
c03d2eed30
commit
f03f224af0
5 changed files with 452 additions and 230 deletions
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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__)
|
||||
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue