feat: add adapter type-system release groundwork

docs/changelog.rst

@@ -21,6 +21,11 @@ v0.54.0 - SQL processing correctness and cleanup
   statement config plus normalized driver-feature dictionary across backends.
 * MySQL-family adapter config, driver, and pool modules now resolve runtime
   vendor symbols through adapter-local typing modules.
+* Oracle LOB fetches now default to direct string/byte materialization where
+  python-oracledb supports it. Pass ``fetch_lobs=True`` when application code
+  needs native Oracle LOB locators. Unconstrained LOB contents are no longer
+  parsed with content heuristics; native ``JSON``, ``IS JSON`` CLOB/BLOB, and
+  OSON-capable values still decode through Oracle JSON metadata.
 * Driver statement-object caches are now bounded by the configured statement
   cache size, and cached named-parameter rebinding reuses driver-owned
   processing state.

docs/reference/adapters/oracledb.rst

@@ -6,6 +6,32 @@ Sync and async Oracle adapter using `python-oracledb <https://python-oracledb.re
 Features native pipeline mode for multi-statement batching, BLOB support, and
 LOB coercion with byte-length thresholds.
 
+LOB And JSON Fetching
+=====================
+
+Oracle configurations default ``fetch_lobs`` to ``False``. With modern
+``python-oracledb`` this returns supported LOB values under Oracle's 1 GB
+direct-fetch ceiling directly as ``str`` or ``bytes`` for normal SELECTs,
+streaming reads, and Arrow exports. SQLSpec still materializes readable locators
+when Oracle returns one, so buffered results and schema hydration do not expose
+driver handles by default.
+
+Pass ``fetch_lobs=True`` on a query when application code needs native Oracle
+LOB locators, for example in a streaming workflow that wants to control when a
+large value is read.
+
+JSON fetch conversion is metadata-driven:
+
+* native ``JSON`` columns are returned by ``python-oracledb``;
+* ``IS JSON`` CLOB/BLOB/VARCHAR2 columns are decoded through Oracle fetch
+  metadata;
+* OSON BLOB values are decoded through Oracle's OSON support when the server and
+  driver expose it.
+
+Unconstrained CLOB or BLOB columns are returned as text or bytes even when their
+contents look like JSON. Add an Oracle JSON type or ``IS JSON`` constraint when
+you want automatic JSON decoding.
+
 Sync Configuration
 ==================
 

docs/reference/core/type-converters.rst

@@ -10,14 +10,6 @@ drivers to convert between Python types and database-native representations.
 Base Classes
 ============
 
-.. autoclass:: BaseTypeConverter
-   :members:
-   :show-inheritance:
-
-.. autoclass:: CachedOutputConverter
-   :members:
-   :show-inheritance:
-
 .. autoclass:: BaseInputConverter
    :members:
    :show-inheritance:

sqlspec/_typing.py

@@ -61,6 +61,7 @@
     "FailFastStub",
     "Gauge",
     "Histogram",
+    "MsgspecValidationError",
     "NumpyArray",
     "NumpyArrayStub",
     "PandasDataFrame",
@@ -256,15 +257,18 @@ class UnsetTypeStub(enum.Enum):
     from msgspec import UNSET as _REAL_UNSET
     from msgspec import Struct as _RealStruct
     from msgspec import UnsetType as _RealUnsetType
+    from msgspec import ValidationError as _RealMsgspecValidationError
     from msgspec import convert as _real_convert
     from msgspec.structs import fields as _real_msgspec_fields
 
+    MsgspecValidationError: type[Exception] = _RealMsgspecValidationError
     Struct = _RealStruct
     UnsetType = _RealUnsetType
     UNSET = _REAL_UNSET
     convert = _real_convert
     msgspec_fields = _real_msgspec_fields
 except ImportError:
+    MsgspecValidationError = ValueError
     Struct = StructStub  # type: ignore[assignment,misc]
     UnsetType = UnsetTypeStub  # type: ignore[assignment,misc]
     UNSET = UNSET_STUB  # type: ignore[assignment] # pyright: ignore[reportConstantRedefinition]

sqlspec/adapters/adbc/config.py

@@ -185,6 +185,7 @@ class AdbcConfig(NoPoolSyncConfig[AdbcConnection, AdbcDriver]):
     supports_native_arrow_export: "ClassVar[bool]" = True
     supports_native_arrow_import: "ClassVar[bool]" = True
     supports_arrow_streaming: "ClassVar[bool]" = True
+    supports_native_row_streaming: "ClassVar[bool]" = True
     supports_native_parquet_export: "ClassVar[bool]" = True
     supports_native_parquet_import: "ClassVar[bool]" = True
     storage_partition_strategies: "ClassVar[tuple[str, ...]]" = ("fixed", "rows_per_chunk")

sqlspec/adapters/adbc/driver.py

@@ -4,6 +4,7 @@
 database dialects, parameter style conversion, and transaction management.
 """
 
+import contextlib
 from typing import TYPE_CHECKING, Any, Literal, cast
 
 from typing_extensions import final
@@ -36,7 +37,7 @@
     get_cache_config,
     register_driver_profile,
 )
-from sqlspec.driver import BaseSyncExceptionHandler, SyncDriverAdapterBase
+from sqlspec.driver import BaseSyncExceptionHandler, SyncDriverAdapterBase, SyncRowStream
 from sqlspec.exceptions import DatabaseConnectionError, SQLSpecError
 from sqlspec.utils.arrow_helpers import arrow_reader_with_deferred_close
 from sqlspec.utils.logging import get_logger
@@ -82,6 +83,70 @@ def _handle_exception(self, exc_type: "type[BaseException] | None", exc_val: "Ba
         return True
 
 
+class AdbcSelectStreamSource:
+    """Native ADBC chunk source backed by a ``RecordBatchReader``."""
+
+    __slots__ = ("_cursor_manager", "_driver", "_parameters", "_reader", "_sql")
+
+    def __init__(self, driver: "AdbcDriver", sql: str, parameters: Any) -> None:
+        self._driver = driver
+        self._sql = sql
+        self._parameters = parameters
+        self._cursor_manager: AdbcCursor | None = None
+        self._reader: Any = None
+
+    def start(self) -> None:
+        cursor_manager = self._driver.with_cursor(self._driver.connection)
+        reader: Any = None
+        try:
+            cursor = cursor_manager.__enter__()
+            handler = self._driver.handle_database_exceptions()
+            with handler:
+                execute_parameters = normalize_postgres_empty_parameters(self._driver._dialect_name, self._parameters)
+                cursor.execute(self._sql, parameters=execute_parameters)
+                reader = _fetch_record_batch(cursor)
+            self._driver._check_pending_exception(handler)
+        except BaseException:
+            with contextlib.suppress(Exception):
+                cursor_manager.__exit__(None, None, None)
+            raise
+
+        if reader is None:
+            msg = "ADBC did not return a record batch reader."
+            raise SQLSpecError(msg)
+        self._cursor_manager = cursor_manager
+        self._reader = iter(reader)
+
+    def fetch_chunk(self) -> "list[dict[str, Any]]":
+        reader = self._reader
+        if reader is None:
+            return []
+        while True:
+            try:
+                batch = next(reader)
+            except StopIteration:
+                return []
+            rows = cast("list[dict[str, Any]]", batch.to_pylist())
+            if rows:
+                return rows
+
+    def close(self) -> None:
+        self._reader = None
+        cursor_manager = self._cursor_manager
+        self._cursor_manager = None
+        if cursor_manager is not None:
+            with contextlib.suppress(Exception):
+                cursor_manager.__exit__(None, None, None)
+
+
+def _fetch_record_batch(cursor: Any) -> Any:
+    fetch_record_batch = getattr(cursor, "fetch_record_batch", None)
+    if fetch_record_batch is None:
+        msg = "ADBC cursor does not expose fetch_record_batch() for native row streaming."
+        raise SQLSpecError(msg)
+    return fetch_record_batch()
+
+
 @final
 class AdbcDriver(SyncDriverAdapterBase):
     """ADBC driver for Arrow Database Connectivity.
@@ -151,19 +216,16 @@ def dispatch_execute(self, cursor: "AdbcRawCursor", statement: SQL) -> "Executio
         is_select_like = statement.returns_rows() or self._should_force_select(statement, cursor)
 
         if is_select_like:
-            fetched_data = cursor.fetchall()
-            column_names = self._resolve_column_names(cursor.description)
-            data, column_names = collect_rows(
-                cast("list[Any] | None", fetched_data), cursor.description, column_names=column_names
-            )
-            row_format = "dict" if data and isinstance(data[0], dict) else "tuple"
+            arrow_table = cursor.fetch_arrow_table()
+            data = arrow_table.to_pylist()
+            column_names = list(arrow_table.column_names)
             return self.create_execution_result(
                 cursor,
                 selected_data=data,
                 column_names=column_names,
                 data_row_count=len(data),
                 is_select_result=True,
-                row_format=row_format,
+                row_format="dict",
             )
 
         row_count = self._resolve_count_result_rowcount(cursor, fallback=resolve_rowcount(cursor))
@@ -265,6 +327,14 @@ def dispatch_execute_script(self, cursor: "AdbcRawCursor", statement: "SQL") ->
             is_script_result=True,
         )
 
+    def dispatch_select_stream(self, statement: "SQL", chunk_size: int) -> "SyncRowStream[dict[str, Any]] | None":
+        """Return a native ADBC stream backed by ``fetch_record_batch()``."""
+        _ = chunk_size
+        if not statement.returns_rows():
+            return None
+        sql, prepared_parameters = self._compiled_sql(statement, self.statement_config)
+        return SyncRowStream(AdbcSelectStreamSource(self, sql, prepared_parameters))
+
     # ─────────────────────────────────────────────────────────────────────────────
     # TRANSACTION MANAGEMENT
     # ─────────────────────────────────────────────────────────────────────────────

sqlspec/adapters/adbc/type_converter.py

@@ -7,63 +7,24 @@
 
 from typing import Any
 
-from sqlspec.core.type_converter import CachedOutputConverter
 from sqlspec.utils.serializers import to_json
 
-__all__ = ("ADBC_SPECIAL_CHARS", "ADBCOutputConverter", "get_adbc_type_converter")
+__all__ = ("ADBCOutputConverter", "get_adbc_type_converter")
 
-ADBC_SPECIAL_CHARS: "frozenset[str]" = frozenset({"{", "[", "-", ":", "T", "."})
 
-# Native type support by dialect
-_NATIVE_SUPPORT: "dict[str, list[str]]" = {
-    "postgres": ["uuid", "json", "interval", "pg_array"],
-    "postgresql": ["uuid", "json", "interval", "pg_array"],
-    "pgvector": ["uuid", "json", "interval", "pg_array"],
-    "paradedb": ["uuid", "json", "interval", "pg_array"],
-    "duckdb": ["uuid", "json"],
-    "bigquery": ["json"],
-    "sqlite": [],
-    "mysql": ["json"],
-    "snowflake": ["json"],
-}
-
-
-class ADBCOutputConverter(CachedOutputConverter):
-    """ADBC-specific output conversion with dialect awareness.
-
-    Extends CachedOutputConverter with ADBC multi-backend functionality
-    including dialect-specific type handling for different database systems.
-    """
+class ADBCOutputConverter:
+    """ADBC-specific parameter conversion with dialect awareness."""
 
     __slots__ = ("dialect",)
 
-    def __init__(self, dialect: str, cache_size: int = 5000) -> None:
+    def __init__(self, dialect: str) -> None:
         """Initialize with dialect-specific configuration.
 
         Args:
             dialect: Target database dialect (postgres, sqlite, duckdb, etc.)
-            cache_size: Maximum number of string values to cache (default: 5000)
         """
-        super().__init__(special_chars=ADBC_SPECIAL_CHARS, cache_size=cache_size)
         self.dialect = dialect.lower()
 
-    def _convert_detected(self, value: str, detected_type: str) -> Any:
-        """Convert value with dialect-specific handling.
-
-        Args:
-            value: String value to convert.
-            detected_type: Detected type name.
-
-        Returns:
-            Converted value according to dialect requirements.
-        """
-        try:
-            if self.dialect == "sqlite" and detected_type == "uuid":
-                return str(value)
-            return self.convert_value(value, detected_type)
-        except Exception:
-            return value
-
     def convert_dict(self, value: "dict[str, Any]") -> Any:
         """Convert dictionary values with dialect-specific handling.
 
@@ -77,42 +38,14 @@ def convert_dict(self, value: "dict[str, Any]") -> Any:
             return to_json(value)
         return value
 
-    def supports_native_type(self, type_name: str) -> bool:
-        """Check if dialect supports native handling of a type.
-
-        Args:
-            type_name: Type name to check
-
-        Returns:
-            True if dialect supports native handling, False otherwise.
-        """
-        return type_name in _NATIVE_SUPPORT.get(self.dialect, [])
-
-    def get_dialect_specific_converter(self, value: Any, target_type: str) -> Any:
-        """Apply dialect-specific conversion logic.
-
-        Args:
-            value: Value to convert.
-            target_type: Target type for conversion.
-
-        Returns:
-            Converted value according to dialect requirements.
-        """
-        if self.dialect == "sqlite" and target_type == "uuid":
-            return str(value)
-        if self.dialect == "bigquery" and target_type == "uuid":
-            return str(self.convert_value(value, target_type))
-        return self.convert_value(value, target_type)
-
 
-def get_adbc_type_converter(dialect: str, cache_size: int = 5000) -> ADBCOutputConverter:
+def get_adbc_type_converter(dialect: str) -> ADBCOutputConverter:
     """Factory function to create dialect-specific ADBC type converter.
 
     Args:
         dialect: Database dialect name.
-        cache_size: Maximum number of string values to cache (default: 5000)
 
     Returns:
         Configured ADBCOutputConverter instance.
     """
-    return ADBCOutputConverter(dialect, cache_size)
+    return ADBCOutputConverter(dialect)

sqlspec/adapters/aiomysql/config.py

@@ -62,7 +62,6 @@ class AiomysqlConnectionParams(TypedDict):
     allow_local_infile: NotRequired[bool]
     echo: NotRequired[bool]
     local_infile: NotRequired[bool]
-    enable_local_infile: NotRequired[bool]
     ssl: NotRequired["SSLContext"]
     sql_mode: NotRequired[str]
     init_command: NotRequired[str]
@@ -86,20 +85,19 @@ class AiomysqlPoolParams(AiomysqlConnectionParams):
 
 
 def _normalize_local_infile(connection_config: "Mapping[str, Any]", *, strip_consent_gate: bool) -> "dict[str, Any]":
-    """Normalize aiomysql local-infile aliases and SQLSpec's consent gate."""
+    """Normalize aiomysql local-infile settings and SQLSpec's consent gate."""
     config = dict(connection_config)
 
-    enable_local_infile = bool(config.get("enable_local_infile", False))
+    config.pop("enable_local_infile", None)
     allow_local_infile = bool(config.get(_AIOMYSQL_LOCAL_INFILE_GATE, False))
-    local_infile = bool(config.get("local_infile", False) or enable_local_infile)
+    local_infile = bool(config.get("local_infile", False))
     if local_infile and not allow_local_infile:
         msg = (
             "Aiomysql local_infile=True requires allow_local_infile=True because "
             "LOAD DATA LOCAL INFILE can read client files."
         )
         raise ImproperConfigurationError(msg)
     config["local_infile"] = bool(local_infile and allow_local_infile)
-    config.pop("enable_local_infile", None)
     if strip_consent_gate:
         config.pop(_AIOMYSQL_LOCAL_INFILE_GATE, None)
     return config