diff --git a/graphiti_core/graphiti.py b/graphiti_core/graphiti.py index 51c3099fa4..dcd86e5735 100644 --- a/graphiti_core/graphiti.py +++ b/graphiti_core/graphiti.py @@ -26,7 +26,7 @@ from graphiti_core.cross_encoder.client import CrossEncoderClient from graphiti_core.cross_encoder.openai_reranker_client import OpenAIRerankerClient from graphiti_core.decorators import handle_multiple_group_ids -from graphiti_core.driver.driver import GraphDriver +from graphiti_core.driver.driver import GraphDriver, GraphProvider from graphiti_core.driver.neo4j_driver import Neo4jDriver from graphiti_core.edges import ( CommunityEdge, @@ -403,12 +403,21 @@ async def _saga_get_previous_episode_uuid( except NotImplementedError: pass + # ArcadeDB-only: wrap valid_at/created_at in datetime(...) to avoid a + # ClassCastException comparing native-datetime representations once a + # range index exists on the property (same class of bug fixed in + # graph_data_operations.py::retrieve_episodes()). + order_by = ( + 'ORDER BY datetime(e.valid_at) DESC, datetime(e.created_at) DESC' + if self.driver.provider == GraphProvider.ARCADEDB + else 'ORDER BY e.valid_at DESC, e.created_at DESC' + ) records, _, _ = await self.driver.execute_query( - """ - MATCH (s:Saga {uuid: $saga_uuid})-[:HAS_EPISODE]->(e:Episodic) + f""" + MATCH (s:Saga {{uuid: $saga_uuid}})-[:HAS_EPISODE]->(e:Episodic) WHERE e.uuid <> $current_episode_uuid RETURN e.uuid AS uuid - ORDER BY e.valid_at DESC, e.created_at DESC + {order_by} LIMIT 1 """, saga_uuid=saga_uuid, @@ -485,13 +494,25 @@ async def summarize_saga(self, saga_id: str) -> SagaNode: saga_id, since=since, limit=max_episodes ) if episodes_data is None: + # ArcadeDB-only: see the note in _saga_get_previous_episode_uuid. + is_arcadedb = self.driver.provider == GraphProvider.ARCADEDB if since is not None: + created_at_where = ( + 'WHERE datetime(e.created_at) > datetime($since)' + if is_arcadedb + else 'WHERE e.created_at > $since' + ) + order_by = ( + 'ORDER BY datetime(e.valid_at) ASC, datetime(e.created_at) ASC' + if is_arcadedb + else 'ORDER BY e.valid_at ASC, e.created_at ASC' + ) records, _, _ = await self.driver.execute_query( - """ - MATCH (s:Saga {uuid: $saga_uuid})-[:HAS_EPISODE]->(e:Episodic) - WHERE e.created_at > $since + f""" + MATCH (s:Saga {{uuid: $saga_uuid}})-[:HAS_EPISODE]->(e:Episodic) + {created_at_where} RETURN e.content AS content, e.valid_at AS valid_at - ORDER BY e.valid_at ASC, e.created_at ASC + {order_by} LIMIT $limit """, saga_uuid=saga_id, @@ -500,11 +521,16 @@ async def summarize_saga(self, saga_id: str) -> SagaNode: routing_='r', ) else: + order_by = ( + 'ORDER BY datetime(e.valid_at) DESC, datetime(e.created_at) DESC' + if is_arcadedb + else 'ORDER BY e.valid_at DESC, e.created_at DESC' + ) records, _, _ = await self.driver.execute_query( - """ - MATCH (s:Saga {uuid: $saga_uuid})-[:HAS_EPISODE]->(e:Episodic) + f""" + MATCH (s:Saga {{uuid: $saga_uuid}})-[:HAS_EPISODE]->(e:Episodic) RETURN e.content AS content, e.valid_at AS valid_at - ORDER BY e.valid_at DESC, e.created_at DESC + {order_by} LIMIT $limit """, saga_uuid=saga_id, diff --git a/graphiti_core/search/search_filters.py b/graphiti_core/search/search_filters.py index c64c5f097c..afdff37d8d 100644 --- a/graphiti_core/search/search_filters.py +++ b/graphiti_core/search/search_filters.py @@ -105,8 +105,25 @@ def node_search_filter_query_constructor( def date_filter_query_constructor( - value_name: str, param_name: str, operator: ComparisonOperator + value_name: str, + param_name: str, + operator: ComparisonOperator, + provider: GraphProvider | None = None, ) -> str: + # ArcadeDB stores a native datetime write as a zone-less ChronoLocalDateTime + # but decodes an incoming native datetime parameter as a zone-aware + # OffsetDateTime -- comparing them directly can raise a ClassCastException + # at the engine level once a range index exists on the property (same class + # of bug as retrieve_episodes(), see graph_data_operations.py). Wrapping + # both sides in datetime(...) sidesteps it; scoped to ARCADEDB only, and + # skipped for IS NULL/IS NOT NULL since there's no value to wrap there. + if provider == GraphProvider.ARCADEDB and operator not in ( + ComparisonOperator.is_null, + ComparisonOperator.is_not_null, + ): + value_name = f'datetime({value_name})' + param_name = f'datetime({param_name})' + query = '(' + value_name + ' ' if operator == ComparisonOperator.is_null or operator == ComparisonOperator.is_not_null: @@ -158,7 +175,7 @@ def edge_search_filter_query_constructor( and_filters = [ date_filter_query_constructor( - 'e.valid_at', f'$valid_at_{j}', date_filter.comparison_operator + 'e.valid_at', f'$valid_at_{j}', date_filter.comparison_operator, provider ) for j, date_filter in enumerate(or_list) ] @@ -189,7 +206,7 @@ def edge_search_filter_query_constructor( and_filters = [ date_filter_query_constructor( - 'e.invalid_at', f'$invalid_at_{j}', date_filter.comparison_operator + 'e.invalid_at', f'$invalid_at_{j}', date_filter.comparison_operator, provider ) for j, date_filter in enumerate(or_list) ] @@ -220,7 +237,7 @@ def edge_search_filter_query_constructor( and_filters = [ date_filter_query_constructor( - 'e.created_at', f'$created_at_{j}', date_filter.comparison_operator + 'e.created_at', f'$created_at_{j}', date_filter.comparison_operator, provider ) for j, date_filter in enumerate(or_list) ] @@ -251,7 +268,7 @@ def edge_search_filter_query_constructor( and_filters = [ date_filter_query_constructor( - 'e.expired_at', f'$expired_at_{j}', date_filter.comparison_operator + 'e.expired_at', f'$expired_at_{j}', date_filter.comparison_operator, provider ) for j, date_filter in enumerate(or_list) ] diff --git a/graphiti_core/utils/maintenance/graph_data_operations.py b/graphiti_core/utils/maintenance/graph_data_operations.py index 8f01132ef2..23dbd20361 100644 --- a/graphiti_core/utils/maintenance/graph_data_operations.py +++ b/graphiti_core/utils/maintenance/graph_data_operations.py @@ -96,6 +96,25 @@ async def retrieve_episodes( except NotImplementedError: pass + # ArcadeDB stores a native datetime write as a zone-less ChronoLocalDateTime + # but decodes an incoming native datetime parameter as a zone-aware + # OffsetDateTime -- comparing them directly raises a ClassCastException at + # the engine level (confirmed: reproducible whenever a range index on the + # property was created before any data existed, which is exactly what + # build_indices_and_constraints() always does). Wrapping both sides in + # datetime(...) sidesteps it; scoped to ARCADEDB only so the other + # providers' generated Cypher is untouched. + valid_at_where = ( + 'WHERE datetime(e.valid_at) <= datetime($reference_time)' + if driver.provider == GraphProvider.ARCADEDB + else 'WHERE e.valid_at <= $reference_time' + ) + valid_at_order = ( + 'ORDER BY datetime(e.valid_at) DESC' + if driver.provider == GraphProvider.ARCADEDB + else 'ORDER BY e.valid_at DESC' + ) + # If saga is provided, retrieve episodes from that saga only if saga is not None: group_id = group_ids[0] if group_ids else None @@ -104,7 +123,7 @@ async def retrieve_episodes( records, _, _ = await driver.execute_query( f""" MATCH (s:Saga {{name: $saga_name, group_id: $group_id}})-[:HAS_EPISODE]->(e:Episodic) - WHERE e.valid_at <= $reference_time + {valid_at_where} {source_filter} RETURN """ @@ -113,8 +132,8 @@ async def retrieve_episodes( if driver.provider == GraphProvider.NEPTUNE else EPISODIC_NODE_RETURN ) - + """ - ORDER BY e.valid_at DESC + + f""" + {valid_at_order} LIMIT $num_episodes """, saga_name=saga, @@ -138,9 +157,9 @@ async def retrieve_episodes( query_params['source'] = source.name query: LiteralString = ( - """ + f""" MATCH (e:Episodic) - WHERE e.valid_at <= $reference_time + {valid_at_where} """ + query_filter + """ @@ -151,8 +170,8 @@ async def retrieve_episodes( if driver.provider == GraphProvider.NEPTUNE else EPISODIC_NODE_RETURN ) - + """ - ORDER BY e.valid_at DESC + + f""" + {valid_at_order} LIMIT $num_episodes """ )