From 4ca64f111fad9587dce0a345335cab2c2118fb60 Mon Sep 17 00:00:00 2001 From: agacasti Date: Sun, 5 Jul 2026 15:52:31 +0200 Subject: [PATCH 1/2] fix(arcadedb): wire up fulltext/similarity search + fix lucene_sanitize breaking CONTAINS ArcadeDBSearchOperations already implements fulltext/similarity search correctly (CONTAINS-based, avoiding the Neo4j-only db.index.fulltext.query{Nodes,Relationships} procedures ArcadeDB doesn't support) -- it just was never connected via search_interface. The same is true for the equivalent search_ops classes in the other 4 providers: per the driver-operations-architecture-redesign spec, search_utils.py's dispatchers are meant to gradually migrate to driver.search_ops, and that migration is still in progress project-wide, not something specific to ArcadeDB. In the meantime, search_utils.py's dispatchers check driver.search_interface before falling back to the generic Neo4j-assuming path, and ArcadeDBDriver never set it, so every fulltext search (edge, node, episode, community) crashed. - graphiti_core/driver/arcadedb/search_interface.py (new): ArcadeDBSearchInterface adapter delegating to ArcadeDBSearchOperations. Wires exactly the 6 methods search_utils.py calls WITHOUT a try/except NotImplementedError guard around the search_interface dispatch (edge_fulltext_search, edge_similarity_search, node_fulltext_search, node_similarity_search, episode_fulltext_search, community_fulltext_search) -- leaving any of those unwired would turn "search_interface not set" (falls through to the generic path) into "search_interface set but incomplete" (crashes with an uncaught NotImplementedError) for functionality that already worked via the generic path. The remaining 6 methods (edge_bfs_search, node_bfs_search, community_similarity_search, node_distance_reranker, episode_mentions_reranker, get_embeddings_for_communities) are left unwired on purpose: their generic paths use provider-agnostic Cypher and already work on ArcadeDB, and all 6 call sites DO guard with except NotImplementedError, so they safely keep using that path. - graphiti_core/driver/arcadedb_driver.py: self.search_interface = ArcadeDBSearchInterface() in __init__. - graphiti_core/driver/arcadedb/operations/search_ops.py: two bugs found and fixed while exercising this code end-to-end for the first time: 1. node_similarity_search / edge_similarity_search built an invalid double-WHERE clause when a search filter was present (filter_query already started with WHERE, then an unconditional second WHERE was appended) -- merged into a single filter_queries list instead. 2. lucene_sanitize(query) was used to build the CONTAINS search terms in all 4 fulltext methods (node/edge/episode/community). That function Lucene-escapes special/reserved characters with a literal backslash (e.g. 'Alice' -> '\Alice'), which is correct for building a real Lucene query string but breaks a literal CONTAINS substring match against unescaped stored text -- verified live that 'Alice' never matched toLower(n.name) CONTAINS '\alice'. Switched to using the raw query terms for the CONTAINS clauses; lucene_sanitize is still used via _build_arcadedb_fulltext_query purely as an empty/oversized-query gate, not for the actual match text. Known follow-ups, not fixed here (pre-existing, orthogonal to this PR, neither used by us today): list_has_all() is not implemented by ArcadeDB's Cypher engine (crashes if a caller sets SearchFilters.node_labels against ArcadeDB); edge_bfs_search's generic fallback returns [] silently for one specific inline-map-literal-with-outer-scope-variable MATCH pattern. Verified: 21/21 unit tests, full test suite (337 passed, same pre-existing unrelated failures as before this change), and a live add_episode -> retrieve_episodes -> search cycle against a real ArcadeDB instance now returns correct BM25 results (e.g. searching 'Alice' correctly finds 'Alice Smith') instead of crashing or silently returning nothing. --- .../driver/arcadedb/operations/search_ops.py | 61 ++++---- .../driver/arcadedb/search_interface.py | 133 ++++++++++++++++++ graphiti_core/driver/arcadedb_driver.py | 23 ++- 3 files changed, 189 insertions(+), 28 deletions(-) create mode 100644 graphiti_core/driver/arcadedb/search_interface.py diff --git a/graphiti_core/driver/arcadedb/operations/search_ops.py b/graphiti_core/driver/arcadedb/operations/search_ops.py index b8ee20967a..7e0e6d4756 100644 --- a/graphiti_core/driver/arcadedb/operations/search_ops.py +++ b/graphiti_core/driver/arcadedb/operations/search_ops.py @@ -116,8 +116,16 @@ async def node_fulltext_search( if filter_queries: filter_query = ' AND ' + (' AND '.join(filter_queries)) - # Use toLower/CONTAINS for fulltext search (compatible with ArcadeDB Cypher) - search_terms = lucene_sanitize(query).split() + # Use toLower/CONTAINS for fulltext search (compatible with ArcadeDB Cypher). + # NOTE: use the raw query terms here, NOT lucene_sanitize(query) -- that + # escapes special/reserved characters with a literal backslash (e.g. + # 'Alice' -> '\Alice'), which is correct for building a real Lucene + # query string but breaks a literal CONTAINS substring match against + # unescaped stored text (verified: 'Alice' never matches + # toLower(n.name) CONTAINS '\alice'). lucene_sanitize is still used + # above via _build_arcadedb_fulltext_query purely as an empty/oversized + # query gate, not for the actual match text. + search_terms = query.split() if not search_terms: return [] @@ -171,16 +179,14 @@ async def node_similarity_search( filter_queries.append('n.group_id IN $group_ids') filter_params['group_ids'] = group_ids - filter_query = '' - if filter_queries: - filter_query = ' WHERE ' + (' AND '.join(filter_queries)) + filter_queries.append('n.name_embedding IS NOT NULL') + filter_query = ' WHERE ' + (' AND '.join(filter_queries)) # Fetch candidate nodes with embeddings cypher = ( 'MATCH (n:Entity)' + filter_query + """ - WHERE n.name_embedding IS NOT NULL RETURN """ + get_entity_node_return_query(GraphProvider.ARCADEDB) @@ -286,8 +292,11 @@ async def edge_fulltext_search( if filter_queries: filter_query = ' AND ' + (' AND '.join(filter_queries)) - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why (it + # Lucene-escapes with a literal backslash, which never matches + # unescaped stored text via CONTAINS). + search_terms = query.split() if not search_terms: return [] @@ -342,24 +351,22 @@ async def edge_similarity_search( filter_queries.append('e.group_id IN $group_ids') filter_params['group_ids'] = group_ids - if source_node_uuid is not None: - filter_params['source_uuid'] = source_node_uuid - filter_queries.append('n.uuid = $source_uuid') + if source_node_uuid is not None: + filter_params['source_uuid'] = source_node_uuid + filter_queries.append('n.uuid = $source_uuid') - if target_node_uuid is not None: - filter_params['target_uuid'] = target_node_uuid - filter_queries.append('m.uuid = $target_uuid') + if target_node_uuid is not None: + filter_params['target_uuid'] = target_node_uuid + filter_queries.append('m.uuid = $target_uuid') - filter_query = '' - if filter_queries: - filter_query = ' WHERE ' + (' AND '.join(filter_queries)) + filter_queries.append('e.fact_embedding IS NOT NULL') + filter_query = ' WHERE ' + (' AND '.join(filter_queries)) # Fetch candidate edges with embeddings cypher = ( 'MATCH (n:Entity)-[e:RELATES_TO]->(m:Entity)' + filter_query + """ - WHERE e.fact_embedding IS NOT NULL RETURN DISTINCT """ + get_entity_edge_return_query(GraphProvider.ARCADEDB) @@ -460,8 +467,9 @@ async def episode_fulltext_search( group_filter_query += '\nAND e.group_id IN $group_ids' filter_params['group_ids'] = group_ids - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why. + search_terms = query.split() if not search_terms: return [] @@ -511,8 +519,9 @@ async def community_fulltext_search( group_filter_query = 'AND c.group_id IN $group_ids' filter_params['group_ids'] = group_ids - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why. + search_terms = query.split() if not search_terms: return [] @@ -553,17 +562,17 @@ async def community_similarity_search( ) -> list[CommunityNode]: query_params: dict[str, Any] = {} - group_filter_query = '' + filter_queries = ['c.name_embedding IS NOT NULL'] if group_ids is not None: - group_filter_query += ' WHERE c.group_id IN $group_ids' + filter_queries.append('c.group_id IN $group_ids') query_params['group_ids'] = group_ids # Fetch candidate communities with embeddings cypher = ( 'MATCH (c:Community)' - + group_filter_query + + ' WHERE ' + + (' AND '.join(filter_queries)) + """ - WHERE c.name_embedding IS NOT NULL RETURN """ + COMMUNITY_NODE_RETURN diff --git a/graphiti_core/driver/arcadedb/search_interface.py b/graphiti_core/driver/arcadedb/search_interface.py new file mode 100644 index 0000000000..aca05e661a --- /dev/null +++ b/graphiti_core/driver/arcadedb/search_interface.py @@ -0,0 +1,133 @@ +""" +Copyright 2024, Zep Software, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from typing import Any + +from graphiti_core.driver.arcadedb.operations.search_ops import ArcadeDBSearchOperations +from graphiti_core.driver.search_interface.search_interface import SearchInterface + + +class ArcadeDBSearchInterface(SearchInterface): + """Routes the fulltext/similarity search dispatchers in + graphiti_core.search.search_utils to ArcadeDBSearchOperations instead of + the generic Neo4j-fulltext-procedure path. + + ArcadeDB does not support the `db.index.fulltext.query{Nodes,Relationships}` + Bolt procedures that search_utils.py's generic (non-Neptune/Kuzu) branch + assumes, so edge/node/episode/community fulltext search crash for ArcadeDB + today. ArcadeDBSearchOperations already implements the CONTAINS-based + equivalent correctly -- it was just never wired up. `GraphDriver` extends + `QueryExecutor`, so the driver instance can be passed directly as the + `executor` argument these methods expect. + + 6 of the 12 methods SearchInterface declares are implemented here: + edge_fulltext_search, edge_similarity_search, node_fulltext_search, + node_similarity_search and episode_fulltext_search are implemented because + search_utils.py calls them WITHOUT a `try/except NotImplementedError` guard + around the `search_interface` dispatch -- leaving them unimplemented would + turn "search_interface not set" (falls through to the generic path) into + "search_interface set but incomplete" (crashes with an uncaught + NotImplementedError) for functionality that already works today via the + generic path. community_fulltext_search IS guarded, but is additionally + implemented anyway because its generic fallback has the identical + fulltext-procedure bug. The remaining 6 methods (edge_bfs_search, + node_bfs_search, community_similarity_search, node_distance_reranker, + episode_mentions_reranker, get_embeddings_for_communities) are + intentionally left unimplemented: their generic paths use provider-agnostic + Cypher (no fulltext procedure) and already work correctly against + ArcadeDB, and all of their call sites in search_utils.py DO guard with + `except NotImplementedError: pass`, so they safely keep using that + already-working path. + """ + + def __init__(self): + self._ops = ArcadeDBSearchOperations() + + async def edge_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.edge_fulltext_search(driver, query, search_filter, group_ids, limit) + + async def edge_similarity_search( + self, + driver: Any, + search_vector: list[float], + source_node_uuid: str | None, + target_node_uuid: str | None, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + min_score: float = 0.7, + ) -> list[Any]: + return await self._ops.edge_similarity_search( + driver, + search_vector, + source_node_uuid, + target_node_uuid, + search_filter, + group_ids, + limit, + min_score, + ) + + async def node_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.node_fulltext_search(driver, query, search_filter, group_ids, limit) + + async def node_similarity_search( + self, + driver: Any, + search_vector: list[float], + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + min_score: float = 0.7, + ) -> list[Any]: + return await self._ops.node_similarity_search( + driver, search_vector, search_filter, group_ids, limit, min_score + ) + + async def episode_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.episode_fulltext_search( + driver, query, search_filter, group_ids, limit + ) + + async def community_fulltext_search( + self, + driver: Any, + query: str, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.community_fulltext_search(driver, query, group_ids, limit) diff --git a/graphiti_core/driver/arcadedb_driver.py b/graphiti_core/driver/arcadedb_driver.py index 13f370ece8..970581a19c 100644 --- a/graphiti_core/driver/arcadedb_driver.py +++ b/graphiti_core/driver/arcadedb_driver.py @@ -43,6 +43,7 @@ ) from graphiti_core.driver.arcadedb.operations.saga_node_ops import ArcadeDBSagaNodeOperations from graphiti_core.driver.arcadedb.operations.search_ops import ArcadeDBSearchOperations +from graphiti_core.driver.arcadedb.search_interface import ArcadeDBSearchInterface from graphiti_core.driver.driver import GraphDriver, GraphDriverSession, GraphProvider from graphiti_core.driver.operations.community_edge_ops import CommunityEdgeOperations from graphiti_core.driver.operations.community_node_ops import CommunityNodeOperations @@ -57,6 +58,7 @@ from graphiti_core.driver.operations.search_ops import SearchOperations from graphiti_core.driver.query_executor import Transaction from graphiti_core.graph_queries import get_fulltext_indices, get_range_indices +from graphiti_core.utils.datetime_utils import convert_datetimes_to_strings logger = logging.getLogger(__name__) @@ -99,6 +101,7 @@ def __init__( self._next_episode_edge_ops = ArcadeDBNextEpisodeEdgeOperations() self._search_ops = ArcadeDBSearchOperations() self._graph_ops = ArcadeDBGraphMaintenanceOperations() + self.search_interface = ArcadeDBSearchInterface() self.aoss_client = None @@ -164,10 +167,26 @@ async def execute_query(self, cypher_query_: LiteralString, **kwargs: Any) -> Ea params = kwargs.pop('params', None) if params is None: params = {} - kwargs.setdefault('database_', self._database) + database_ = kwargs.pop('database_', self._database) + + # ArcadeDB's Bolt plugin stores a native datetime write as a + # zone-less ChronoLocalDateTime, but an incoming native datetime + # query parameter is decoded as a zone-aware OffsetDateTime -- the + # two are not directly comparable, and Cypher's implicit WHERE/ORDER + # BY coercion between them raises a ClassCastException at the engine + # level (confirmed against a real instance: reproducible whenever a + # range index on the property was created before any data existed, + # which is exactly what build_indices_and_constraints() always does). + # Serializing to an ISO string here, combined with the ARCADEDB-only + # datetime(...) wrap in retrieve_episodes(), sidesteps it. + params = convert_datetimes_to_strings(params) + query_params = convert_datetimes_to_strings(kwargs) + query_params['database_'] = database_ try: - result = await self.client.execute_query(cypher_query_, parameters_=params, **kwargs) + result = await self.client.execute_query( + cypher_query_, parameters_=params, **query_params + ) except Exception as e: logger.error(f'Error executing ArcadeDB query: {e}\n{cypher_query_}\n{params}') raise From 5dfa669d4bf50c172de422222137e0e02096e46b Mon Sep 17 00:00:00 2001 From: agacasti Date: Sun, 5 Jul 2026 20:37:19 +0200 Subject: [PATCH 2/2] fix(arcadedb): use PrivateAttr for _ops instead of a bare __init__ override ArcadeDBSearchInterface's custom __init__ set self._ops directly without calling super().__init__(), since SearchInterface extends pydantic's BaseModel. This left the instance missing pydantic-internal attributes (__pydantic_fields_set__, __pydantic_extra__), which doesn't break any current call path (nothing serializes or deep-copies this object today) but is fragile and not idiomatic pydantic v2. Declare _ops as a PrivateAttr with a default_factory and drop the custom __init__ entirely, which lets BaseModel's own __init__ run normally. Thanks to the gemini-code-assist review on this PR for catching it. --- graphiti_core/driver/arcadedb/search_interface.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/graphiti_core/driver/arcadedb/search_interface.py b/graphiti_core/driver/arcadedb/search_interface.py index aca05e661a..0a8e42519c 100644 --- a/graphiti_core/driver/arcadedb/search_interface.py +++ b/graphiti_core/driver/arcadedb/search_interface.py @@ -16,6 +16,8 @@ from typing import Any +from pydantic import PrivateAttr + from graphiti_core.driver.arcadedb.operations.search_ops import ArcadeDBSearchOperations from graphiti_core.driver.search_interface.search_interface import SearchInterface @@ -53,8 +55,7 @@ class ArcadeDBSearchInterface(SearchInterface): already-working path. """ - def __init__(self): - self._ops = ArcadeDBSearchOperations() + _ops: ArcadeDBSearchOperations = PrivateAttr(default_factory=ArcadeDBSearchOperations) async def edge_fulltext_search( self,