Skip to content

feat(oracledb): add enterprise topology, session context hooks, streaming Arrow, HNSW/RRF, and 26ai capabilities #849

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cofin merged 1 commit into from
Oct 22, 2026
Merged

feat(oracledb): add enterprise topology, session context hooks, streaming Arrow, HNSW/RRF, and 26ai capabilities #849

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions sqlspec/adapters/oracledb/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,35 @@
OracleSyncConfig,
)
from sqlspec.adapters.oracledb.core import (
VALID_ORACLE_DRIVER_FEATURE_KEYS,
OracleBlob,
OracleClob,
OracleEndpoint,
OracleJson,
OracleSessionContextConfig,
OracleTopologyConfig,
build_arrow_fetch_kwargs,
build_connection_config,
build_duality_etag_update_sql,
build_session_checkout_plsql,
build_session_context_plsql,
build_session_release_plsql,
build_session_reset_plsql,
build_tns_description,
coerce_oracle_df_to_arrow_batches,
coerce_oracle_df_to_arrow_table,
default_statement_config,
execute_session_checkout_async,
execute_session_checkout_sync,
execute_session_release_async,
execute_session_release_sync,
extract_duality_etag,
extract_duality_view_etag,
extract_oracle_wallet_zip,
inject_duality_view_etag,
oracle_df_batches_to_arrow_reader,
oracle_df_to_arrow_table,
sanitize_oracle_text_query,
)
from sqlspec.adapters.oracledb.data_dictionary import (
OracledbAsyncDataDictionary,
Expand All @@ -28,22 +52,46 @@
from sqlspec.adapters.oracledb.type_converter import OracleOutputConverter

__all__ = (
"VALID_ORACLE_DRIVER_FEATURE_KEYS",
"OracleAsyncConfig",
"OracleAsyncDriver",
"OracleAsyncExceptionHandler",
"OracleBlob",
"OracleClob",
"OracleConnectionParams",
"OracleDriverFeatures",
"OracleEndpoint",
"OracleJson",
"OracleOutputConverter",
"OraclePoolParams",
"OracleSessionContextConfig",
"OracleSyncConfig",
"OracleSyncDriver",
"OracleSyncExceptionHandler",
"OracleTopologyConfig",
"OracleVersionInfo",
"OracledbAsyncDataDictionary",
"OracledbSyncDataDictionary",
"build_arrow_fetch_kwargs",
"build_connection_config",
"build_duality_etag_update_sql",
"build_session_checkout_plsql",
"build_session_context_plsql",
"build_session_release_plsql",
"build_session_reset_plsql",
"build_tns_description",
"coerce_oracle_df_to_arrow_batches",
"coerce_oracle_df_to_arrow_table",
"default_statement_config",
"execute_session_checkout_async",
"execute_session_checkout_sync",
"execute_session_release_async",
"execute_session_release_sync",
"extract_duality_etag",
"extract_duality_view_etag",
"extract_oracle_wallet_zip",
"inject_duality_view_etag",
"oracle_df_batches_to_arrow_reader",
"oracle_df_to_arrow_table",
"sanitize_oracle_text_query",
)
158 changes: 155 additions & 3 deletions sqlspec/adapters/oracledb/adk/store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,13 @@
from sqlspec import SQL
from sqlspec.adapters.oracledb._storage import oracle_table_feature_report, validate_oracle_identifier
from sqlspec.adapters.oracledb._typing import DatabaseError as OracleDatabaseError
from sqlspec.adapters.oracledb.core import coerce_large_parameters_async, coerce_large_parameters_sync
from sqlspec.adapters.oracledb.core import (
coerce_large_parameters_async,
coerce_large_parameters_sync,
sanitize_oracle_text_query,
)
from sqlspec.adapters.oracledb.data_dictionary import JSONStorageType, OracleVersionInfo, storage_type_from_version
from sqlspec.builder._vector_distance import build_oracle_hnsw_subquery, build_oracle_hybrid_rrf_sql
from sqlspec.config import ADKConfig
from sqlspec.extensions.adk import (
BaseAsyncADKStore,
Expand Down Expand Up @@ -2463,6 +2468,10 @@ async def search_entries(
effective_limit = limit if limit is not None else self._max_results

try:
if embedding is not None and self._use_fts:
return await self._search_entries_hybrid_rrf(
query, embedding, app_name, user_id, effective_limit, scope_filter
)
if self._use_fts:
return await self._search_entries_fts(query, app_name, user_id, effective_limit, scope_filter)
return await self._search_entries_simple(query, app_name, user_id, effective_limit, scope_filter)
Expand Down Expand Up @@ -2659,6 +2668,9 @@ async def _execute_insert_entry(self, cursor: Any, sql: str, params: "dict[str,
async def _search_entries_fts(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
sanitized_query = sanitize_oracle_text_query(query)
if not sanitized_query:
return []
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
Expand All @@ -2674,7 +2686,74 @@ async def _search_entries_fts(
)
WHERE ROWNUM <= :limit
"""
params = {**scope_params, "query": query, "limit": limit}
params = {**scope_params, "query": sanitized_query, "limit": limit}
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
rows = await cursor.fetchall()
return await self._rows_to_records(rows)

async def _search_entries_hybrid_rrf(
self,
query: str,
embedding: "Sequence[float] | Any",
app_name: str,
user_id: str,
limit: int,
scope_filter: Literal["all", "user", "app"] = "all",
*,
vector_column: str = "embedding",
rrf_k: int = 60,
metric: str = "COSINE",
) -> "list[StoredMemory]":
sanitized_query = sanitize_oracle_text_query(query)
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
select_cols = (
"id",
"session_id",
"app_name",
"user_id",
"scope",
"event_id",
"author",
"timestamp",
"content_json",
"content_text",
"metadata_json",
"inserted_at",
)
if not sanitized_query:
sql = build_oracle_hnsw_subquery(
table=self._memory_table,
vector_column=vector_column,
select_columns=select_cols,
vector_param="query_vector",
top_k_param="top_k",
metric=metric,
where_clause=where_scope,
)
params: dict[str, Any] = {**scope_params, "query_vector": embedding, "top_k": limit}
else:
sql = build_oracle_hybrid_rrf_sql(
table=self._memory_table,
key_column="id",
vector_column=vector_column,
text_column="content_text",
select_columns=select_cols,
vector_param="query_vector",
text_param="text_query",
top_k_param="top_k",
rrf_k_param="rrf_k",
metric=metric,
where_clause=where_scope,
)
params = {
**scope_params,
"query_vector": embedding,
"text_query": sanitized_query,
"top_k": limit,
"rrf_k": rrf_k,
}
async with self._config.provide_connection() as conn:
with conn.cursor() as cursor:
await cursor.execute(sql, params)
Expand Down Expand Up @@ -2823,6 +2902,10 @@ def search_entries(
effective_limit = limit if limit is not None else self._max_results

try:
if embedding is not None and self._use_fts:
return self._search_entries_hybrid_rrf(
query, embedding, app_name, user_id, effective_limit, scope_filter
)
if self._use_fts:
return self._search_entries_fts(query, app_name, user_id, effective_limit, scope_filter)
return self._search_entries_simple(query, app_name, user_id, effective_limit, scope_filter)
Expand Down Expand Up @@ -3017,6 +3100,9 @@ def _execute_insert_entry(self, cursor: Any, sql: str, params: "dict[str, Any]")
def _search_entries_fts(
self, query: str, app_name: str, user_id: str, limit: int, scope_filter: Literal["all", "user", "app"] = "all"
) -> "list[StoredMemory]":
sanitized_query = sanitize_oracle_text_query(query)
if not sanitized_query:
return []
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
sql = f"""
SELECT id, session_id, app_name, user_id, scope, event_id, author,
Expand All @@ -3032,7 +3118,73 @@ def _search_entries_fts(
)
WHERE ROWNUM <= :limit
"""
params = {**scope_params, "query": query, "limit": limit}
params = {**scope_params, "query": sanitized_query, "limit": limit}
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
return self._rows_to_records(rows)

def _search_entries_hybrid_rrf(
self,
query: str,
embedding: "Sequence[float] | Any",
app_name: str,
user_id: str,
limit: int,
scope_filter: Literal["all", "user", "app"] = "all",
*,
vector_column: str = "embedding",
rrf_k: int = 60,
metric: str = "COSINE",
) -> "list[StoredMemory]":
sanitized_query = sanitize_oracle_text_query(query)
where_scope, scope_params = _build_oracle_scope_where(app_name, user_id, scope_filter)
select_cols = (
"id",
"session_id",
"app_name",
"user_id",
"scope",
"event_id",
"author",
"timestamp",
"content_json",
"content_text",
"metadata_json",
"inserted_at",
)
if not sanitized_query:
sql = build_oracle_hnsw_subquery(
table=self._memory_table,
vector_column=vector_column,
select_columns=select_cols,
vector_param="query_vector",
top_k_param="top_k",
metric=metric,
where_clause=where_scope,
)
params: dict[str, Any] = {**scope_params, "query_vector": embedding, "top_k": limit}
else:
sql = build_oracle_hybrid_rrf_sql(
table=self._memory_table,
key_column="id",
vector_column=vector_column,
text_column="content_text",
select_columns=select_cols,
vector_param="query_vector",
text_param="text_query",
top_k_param="top_k",
rrf_k_param="rrf_k",
metric=metric,
where_clause=where_scope,
)
params = {
**scope_params,
"query_vector": embedding,
"text_query": sanitized_query,
"top_k": limit,
"rrf_k": rrf_k,
}
with self._config.provide_connection() as conn, conn.cursor() as cursor:
cursor.execute(sql, params)
rows = cursor.fetchall()
Expand Down
Loading
Loading