arrow-odbc

Sync Arrow-over-ODBC adapter built on arrow-odbc. Streams pyarrow.RecordBatchReader results from any ODBC-compliant driver, making it a good fit for read-heavy analytical transfer between SQL Server, PostgreSQL, MySQL, or other ODBC sources and the Arrow ecosystem.

SQL Server coverage is exercised in CI against SQL Server 2022 through pytest-databases and Microsoft ODBC Driver 18. The shared contract matrix verifies native Arrow reads, Arrow reader/batch output, and Arrow bulk ingest for this adapter. Use load_from_arrow() or bulk_insert_arrow() for native Arrow bulk writes, or execute_many() for row-oriented batches.

The adapter exports a table-backed events queue store, a Litestar session store, and Google ADK session/event and memory stores. They support SQL Server connections through Microsoft ODBC Driver 18 and IBM Db2 LUW connections through the IBM CLI/ODBC driver (see IBM Db2).

IBM Db2

arrow-odbc reads Db2 LUW 11.5 and later through the IBM CLI/ODBC driver. Db2 for z/OS and Db2 for IBM i are not supported. For row-oriented work, pooling, and async access use the Db2 adapter.

Dialect detection. The adapter picks the SQL dialect from the ODBC driver name only: the Driver value of the connection string, or its DSN value when there is no Driver. Names containing db2, IBM Data Server Driver, clidriver, or libdb2o select db2. Database, host, and user names are never used. When the driver is reached through a DSN whose name does not say Db2, set driver_features={"dbms_name": "DB2"}.

Connection keywords. For Db2, host (or server) and port render as the IBM CLI keywords Hostname and Port, and Protocol=TCPIP is added when a host is set. SQL Server options such as encrypt, trust_server_certificate, and trusted_connection raise ImproperConfigurationError.

The ibm_db package (installed by sqlspec[db2]) bundles the CLI driver, so no separate ODBC driver installation is needed; unixODBC must be present on Linux. Point Driver at the bundled libdb2o.so:

from pathlib import Path

import ibm_db
from arrow_odbc import TextEncoding

from sqlspec.adapters.arrow_odbc import ArrowOdbcConfig

package_dir = Path(ibm_db.__file__).resolve().parent
driver = next(package_dir.rglob("clidriver/lib/libdb2o.so"), package_dir / "clidriver/lib/libdb2o.so")

config = ArrowOdbcConfig(
    connection_config={
        "connection_string": f"Driver={driver};LongDataCompat=1;",
        "host": "db2.example.com",
        "port": 50000,
        "database": "SAMPLE",
        "uid": "db2inst1",
        "pwd": "secret",
        "autocommit": False,
    },
    driver_features={"payload_text_encoding": TextEncoding.UTF16},
)

print(config.statement_config.dialect)
# db2

Transactions. Db2 has no BEGIN statement. begin() and transaction() require a connection opened with autocommit off, so set connection_config={"autocommit": False} for transactional work; on an autocommit connection they raise ImproperConfigurationError. Savepoints use Db2 syntax.

Large objects. Add LongDataCompat=1 to the connection string so BLOB columns arrive as binary rather than hexadecimal text and CLOB columns as text.

Text encoding. For non-ASCII text, either set the DB2CODEPAGE=1208 environment variable before connecting or set driver_features={"payload_text_encoding": TextEncoding.UTF16} (TextEncoding comes from the arrow_odbc package). Text parameters are always bound as text.

Results. Implicitly uppercase column names are lowercased (disable with driver_features={"enable_lowercase_column_names": False}); quoted mixed-case names are kept. Timezone-aware datetime parameters are bound as UTC. Db2 TIMESTAMP(12) values are truncated to nanoseconds, and DECFLOAT, XML, and BOOLEAN columns arrive as UTF-8 text.

Configuration

class sqlspec.adapters.arrow_odbc.ArrowOdbcConfig

Bases: NoPoolSyncConfig[Connection, ArrowOdbcDriver]

Configuration for synchronous arrow-odbc connections.

driver_type

alias of ArrowOdbcDriver

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, observability_config=None, **kwargs)

Initialize arrow-odbc configuration.

create_connection()

Create and return a new arrow-odbc connection.

Return type:

Connection

provide_connection(*args, **kwargs)

Provide a connection context manager.

Return type:

ArrowOdbcConnectionContext

provide_session(*_args, statement_config=None, **_kwargs)

Provide a driver session context manager.

Return type:

ArrowOdbcSessionContext

get_signature_namespace()

Get the signature namespace for ArrowOdbcConfig types.

Return type:

dict[str, typing.Any]

get_event_runtime_hints()

Return polling defaults suitable for generic ODBC sources.

Return type:

EventRuntimeHints

Connection Parameters

class sqlspec.adapters.arrow_odbc.ArrowOdbcConnectionParams

Bases: TypedDict

arrow-odbc connection parameters.

Driver Features

class sqlspec.adapters.arrow_odbc.ArrowOdbcDriverFeatures

Bases: TypedDict

arrow-odbc driver feature flags.

connection_autocommit is the autocommit mode of connections created by the config; the config sets it from connection_config["autocommit"]. enable_lowercase_column_names lowercases result column names the database folded to uppercase (quoted mixed-case names are kept); it defaults to on for Db2 and off for every other dialect.

Driver

class sqlspec.adapters.arrow_odbc.ArrowOdbcDriver

Bases: SyncDriverAdapterBase

Sync driver for generic ODBC connections with Arrow-native transfer.

__init__(connection, statement_config=None, driver_features=None)

Initialize driver adapter with connection and configuration.

Parameters:

• connection (Connection) -- Database connection instance

• statement_config (StatementConfig | None) -- Statement configuration for the driver

• driver_features (dict[str, typing.Any] | None) -- Driver-specific features like extensions, secrets, and connection callbacks

• observability -- Optional runtime handling lifecycle hooks, observers, and spans

property data_dictionary: ArrowOdbcDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

dispatch_execute(cursor, statement)

Execute a single SQL statement.

Must be implemented by each driver for database-specific execution logic.

Parameters:

• cursor (Connection) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data

dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets (executemany).

Must be implemented by each driver for database-specific executemany logic.

Parameters:

• cursor (Connection) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data for the many operation

dispatch_execute_script(cursor, statement)

Execute a SQL script containing multiple statements.

Default implementation splits the script and executes statements individually. Drivers can override for database-specific script execution methods.

Parameters:

• cursor (Connection) -- Database cursor/connection object

• statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution data including statement counts

dispatch_select_stream(statement, chunk_size)

Return a native Arrow ODBC row stream backed by record batches.

Return type:

Optional[SyncRowStream[dict[str, typing.Any]]]

collect_rows(cursor, fetched)

Collect rows from cursor after fetchall for the direct execution path.

Adapters should override this method to provide optimized row collection that bypasses full dispatch_execute overhead.

Parameters:

• cursor (Connection) -- Database cursor with description metadata.

• fetched (list[typing.Any]) -- Rows returned from cursor.fetchall().

Return type:

tuple[list[typing.Any], list[str], int]

Returns:

Tuple of (data, column_names, row_count).

Raises:

NotImplementedError -- If the adapter does not implement this method.

resolve_rowcount(cursor)

Resolve the number of affected rows from cursor for the direct execution path.

Adapters should override this method to provide optimized rowcount resolution that bypasses full dispatch_execute overhead.

Parameters:

cursor (Connection) -- Database cursor with rowcount metadata.

Return type:

int

Returns:

Number of affected rows, or 0 when unknown.

Raises:

NotImplementedError -- If the adapter does not implement this method.

begin()

Begin an explicit transaction.

SQL Server starts one with BEGIN TRANSACTION. Db2 has no begin statement: a connection opened with autocommit off is always inside a unit of work, so only the boundary is recorded, and an autocommit connection is refused because each statement would commit on its own.

Raises:

• ImproperConfigurationError -- If the Db2 connection was opened with autocommit on.

• SQLSpecError -- If the begin statement fails.

Return type:

None

commit()

Commit the current transaction on the current connection.

Return type:

None

rollback()

Rollback the current transaction on the current connection.

Return type:

None

with_cursor(connection)

Create and return a context manager for cursor acquisition and cleanup.

Returns a context manager that yields a cursor for database operations. Concrete implementations handle database-specific cursor creation and cleanup.

Return type:

ArrowOdbcCursor

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

ArrowOdbcExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility. The handler stores mapped exceptions in pending_exception rather than raising from __exit__ to avoid ABI boundary violations.

create_savepoint(name)

Create a savepoint within the current transaction.

Return type:

None

release_savepoint(name)

Release a previously created savepoint.

Return type:

None

rollback_to_savepoint(name)

Roll back the current transaction to a previously created savepoint.

Return type:

None

select_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)

Execute a query and return native Arrow results.

bulk_insert_arrow(target_table, source, *, chunk_size=None)

Insert an Arrow table or reader into a database table.

Return type:

None

load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)

Load Arrow data into a table via arrow-odbc bulk insert.

Return type:

StorageBridgeJob

load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)

Load staged artifacts from storage into a table via arrow-odbc bulk insert.

Return type:

StorageBridgeJob

Data Dictionary

class sqlspec.adapters.arrow_odbc.data_dictionary.ArrowOdbcDataDictionary

Bases: SyncDataDictionaryBase

Runtime-dialect data dictionary for generic ODBC connections.

dialect: ClassVar[str] = 'sqlite'

Dialect identifier. Must be defined by subclasses as a class attribute.

__init__(dialect='sqlite')

get_dialect_config()

Return the runtime dialect configuration for this data dictionary.

Return type:

DialectConfig

get_query(domain, operation, *, mode=None)

Return an exact domain query for the runtime dialect.

Return type:

SQL

get_query_text(domain, operation, *, mode=None)

Return raw SQL text for an exact runtime dialect query.

Return type:

str

get_metadata_capabilities(driver, domains=None)

Report Arrow ODBC metadata capabilities without claiming raw catalog support.

Return type:

MetadataCapabilityProfile

resolve_schema(schema)

Return a schema name using runtime dialect defaults when missing.

Return type:

str | None

resolve_identifier(identifier)

Return a runtime-dialect-normalized identifier.

Return type:

str

get_version(driver)

Get database version information when the runtime dialect provides a query.

Return type:

VersionInfo | None

get_feature_flag(driver, feature)

Check whether the runtime dialect supports a feature.

Return type:

bool

get_optimal_type(driver, type_category)

Get the optimal runtime dialect type for a category.

Return type:

str

get_tables(driver, schema=None)

Get table metadata for dialects with bundled catalog queries.

Return type:

list[TableMetadata]

get_columns(driver, table=None, schema=None)

Get column metadata for dialects with bundled catalog queries.

Return type:

list[ColumnMetadata]

get_indexes(driver, table=None, schema=None)

Get index metadata for dialects with bundled catalog queries.

Return type:

list[IndexMetadata]

get_foreign_keys(driver, table=None, schema=None)

Get foreign-key metadata for dialects with bundled catalog queries.

Return type:

list[ForeignKeyMetadata]

get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)

Fail closed because arrow-odbc does not expose lossless DDL metadata.

Return type:

DDLResult

Extensions

class sqlspec.adapters.arrow_odbc.events.ArrowOdbcEventQueueStore

Bases: BaseEventQueueStore[ArrowOdbcConfig]

Event queue DDL for arrow-odbc configs.

SQL Server DDL with OBJECT_ID guards is used by default. A config whose dialect resolves to db2 gets plain Db2 CREATE TABLE/CREATE INDEX statements; the events migration checks the catalog for the table and the index before running them.

__init__(config)

class sqlspec.adapters.arrow_odbc.litestar.ArrowOdbcStore

Bases: BaseSQLSpecStore[ArrowOdbcConfig]

Session store using arrow-odbc sessions.

SQL Server statements are used by default; a config whose dialect resolves to db2 uses Db2 statements with bound UTC times and catalog-probed DDL.

__init__(config)

Initialize the session store.

Parameters:

config (ArrowOdbcConfig) -- SQLSpec database configuration.

async create_table()

Create the session table if it doesn't exist.

Return type:

None

async get(key, renew_for=None)

Get a session value by key.

Return type:

bytes | None

async set(key, value, expires_in=None)

Store a session value.

Return type:

None

async delete(key)

Delete a session by key.

Return type:

None

async delete_all()

Delete all sessions from the store.

Return type:

None

async exists(key)

Check if a session key exists and is not expired.

Return type:

bool

async expires_in(key)

Get the time in seconds until the session expires.

Return type:

int | None

async delete_expired()

Delete all expired sessions.

Return type:

int

class sqlspec.adapters.arrow_odbc.adk.ArrowOdbcADKStore

Bases: BaseSyncADKStore[ArrowOdbcConfig]

Synchronous ADK session/event store using arrow-odbc.

SQL Server statements are used by default; a config whose dialect resolves to db2 uses Db2 statements with bound UTC times and catalog-probed DDL.

__init__(config)

Initialize the ADK store.

Parameters:

config (ArrowOdbcConfig) -- SQLSpec database configuration.

Notes

Reads configuration from config.extension_config["adk"]: - session_table: Sessions table name (default: "adk_session") - events_table: Events table name (default: "adk_event") - app_state_table: App-scoped state table name (default: "adk_app_state") - user_state_table: User-scoped state table name (default: "adk_user_state") - metadata_table: Internal metadata table name (default: "adk_internal_metadata") - owner_id_column: Optional owner FK column DDL (default: None)

create_tables()

Create the ADK tables and indexes the catalog reports as missing.

Return type:

None

create_session(session_id, app_name, user_id, state, owner_id=None)

Create a new ADK session.

Return type:

StoredSession

get_session(app_name, user_id, session_id, *, renew_for=None)

Return a scoped session or None if absent.

Return type:

StoredSession | None

update_session_state(app_name, user_id, session_id, state)

Replace a session's durable state.

Return type:

None

list_sessions(app_name, user_id=None, *, order_by='update_time', descending=True, limit=None, offset=None)

List ADK sessions for an application, optionally scoped to a user.

Return type:

list[StoredSession]

delete_session(app_name, user_id, session_id)

Delete a session. Event rows cascade through the FK.

Return type:

None

append_event(event_record)

Append an event to a session.

Return type:

None

append_event_and_update_state(event_record, app_name, user_id, session_id, state, *, app_state=None, user_state=None)

Atomically append an event and update durable session/scoped state.

Return type:

StoredSession

get_events(app_name, user_id, session_id, after_timestamp=None, limit=None)

Return events for a scoped session ordered by event timestamp.

Return type:

list[StoredEvent]

delete_expired_events(before, app_name=None)

Delete events older than before.

Return type:

int

delete_idle_sessions(updated_before, app_name=None)

Delete sessions whose update_time is older than updated_before.

Return type:

int

delete_idle_user_states(updated_before, app_name=None)

Delete user state rows whose update_time is older than updated_before.

Return type:

int

get_app_state(app_name)

Return app-scoped state.

Return type:

dict[str, typing.Any] | None

get_user_state(app_name, user_id)

Return user-scoped state.

Return type:

dict[str, typing.Any] | None

upsert_app_state(app_name, state)

Insert or replace app-scoped state.

Return type:

None

upsert_user_state(app_name, user_id, state)

Insert or replace user-scoped state.

Return type:

None

get_metadata(key)

Return an ADK metadata value.

Return type:

str | None

set_metadata(key, value)

Set an ADK metadata value.

Return type:

None

class sqlspec.adapters.arrow_odbc.adk.ArrowOdbcADKMemoryStore

Bases: BaseSyncADKMemoryStore[ArrowOdbcConfig]

ADK memory store using arrow-odbc.

SQL Server statements are used by default; a config whose dialect resolves to db2 uses Db2 statements with bound UTC times and catalog-probed DDL.

__init__(config)

Initialize the ADK memory store.

Parameters:

config (ArrowOdbcConfig) -- SQLSpec database configuration.

create_tables()

Create the memory table and indexes the catalog reports as missing.

Return type:

None

insert_memory_entries(entries, owner_id=None)

Insert memory entries, skipping duplicates by event_id.

Return type:

int

search_entries(query, app_name, user_id, limit=None, scope_filter='all', embedding=None)

Search memory entries with SQL Server LIKE matching.

Return type:

list[StoredMemory]

delete_entries_by_session(session_id)

Delete all memory entries for a specific session.

Return type:

int

delete_entries_older_than(days, app_name=None, scope=None)

Delete memory entries older than days days.

Return type:

int

Schema Discovery

ArrowOdbcDataDictionary.get_columns first uses bundled dialect catalog queries. When no query exists for the detected dialect (or it returns no rows) and a table name is given, the driver issues a zero-row probe (SELECT * FROM "schema"."table" WHERE 1=0) and derives column names, ordering, nullability, and SQL type names from the Arrow reader schema. Arrow-derived type names are approximations (for example VARCHAR for any string column); mssql_python and other ODBC adapters without native metadata APIs remain SQL-only.

Extension Settings

Use these types inside extension_config["adk"].

class sqlspec.adapters.arrow_odbc.adk.ArrowOdbcADKConfig

Bases: ADKConfig

arrow-odbc ADK extension settings.

native_json: NotRequired[bool]

Accepted for parity with SQL Server adapters; arrow-odbc uses NVARCHAR(MAX).

Row-oriented batch execution

execute_many() executes each parameter set through the native execute API without rewriting SQL. Nonempty batches report an unknown affected-row count (-1); empty batches report zero. Use bulk_insert_arrow() for native Arrow bulk ingestion.