feat: add pipelined stack execution

.gitignore

@@ -19,6 +19,7 @@ target/
 .vscode/
 .cursor/
 .zed/
+.cache
 .coverage*
 # files
 **/*.so

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
       - id: mixed-line-ending
       - id: trailing-whitespace
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: "v0.14.4"
+    rev: "v0.14.5"
     hooks:
       - id: ruff
         args: ["--fix"]
@@ -43,6 +43,7 @@ repos:
     rev: "v1.0.1"
     hooks:
       - id: sphinx-lint
+        args: ["--jobs", "1"]
   - repo: local
     hooks:
       - id: pypi-readme

AGENTS.md

@@ -185,6 +185,25 @@ SQLSpec is a type-safe SQL query mapper designed for minimal abstraction between
 - **Single-Pass Processing**: Parse once → transform once → validate once - SQL object is single source of truth
 - **Abstract Methods with Concrete Implementations**: Protocol defines abstract methods, base classes provide concrete sync/async implementations
 
+### Query Stack Implementation Guidelines
+
+- **Builder Discipline**
+    - `StatementStack` and `StackOperation` are immutable (`__slots__`, tuple storage). Every push helper returns a new stack; never mutate `_operations` in place.
+    - Validate inputs at push time (non-empty SQL, execute_many payloads, reject nested stacks) so drivers can assume well-formed operations.
+- **Adapter Responsibilities**
+    - Add a single capability gate per adapter (e.g., Oracle pipeline version check, `psycopg.capabilities.has_pipeline()`), return `super().execute_stack()` immediately when unsupported.
+    - Preserve `StackResult.result` by building SQL/Arrow results via `create_sql_result()` / `create_arrow_result()` instead of copying row data.
+    - Honor manual toggles via `driver_features={"stack_native_disabled": True}` and document the behavior in the adapter guide.
+- **Telemetry + Tracing**
+    - Always wrap adapter overrides with `StackExecutionObserver(self, stack, continue_on_error, native_pipeline=bool)`.
+    - Do **not** emit duplicate metrics; the observer already increments `stack.execute.*`, logs `stack.execute.start/complete/failed`, and publishes the `sqlspec.stack.execute` span.
+- **Error Handling**
+    - Wrap driver exceptions in `StackExecutionError` with `operation_index`, summarized SQL (`describe_stack_statement()`), adapter name, and execution mode.
+    - Continue-on-error stacks append `StackResult.from_error()` and keep executing. Fail-fast stacks roll back (if they started the transaction) before re-raising the wrapped error.
+- **Testing Expectations**
+    - Add integration tests under `tests/integration/test_adapters/<adapter>/test_driver.py::test_*statement_stack*` that cover native path, sequential fallback, and continue-on-error.
+    - Guard base behavior (empty stacks, large stacks, transaction boundaries) via `tests/integration/test_stack_edge_cases.py`.
+
 ### Driver Parameter Profile Registry
 
 - All adapter parameter defaults live in `DriverParameterProfile` entries inside `sqlspec/core/parameters.py`.

docs/changelog.rst

@@ -10,6 +10,14 @@ SQLSpec Changelog
 Recent Updates
 ==============
 
+Query Stack Documentation Suite
+--------------------------------
+
+- Expanded the :doc:`/reference/query-stack` API reference (``StatementStack``, ``StackResult``, driver hooks, and ``StackExecutionError``) with the high-level workflow, execution modes, telemetry, and troubleshooting tips.
+- Added :doc:`/examples/patterns/stacks/query_stack_example` that runs the same stack against SQLite and AioSQLite.
+- Captured the detailed architecture and performance guidance inside the internal specs workspace for future agent runs.
+- Updated every adapter reference with a **Query Stack Support** section so behavior is documented per database.
+
 Migration Convenience Methods on Config Classes
 ------------------------------------------------
 

docs/examples/README.md

@@ -6,6 +6,7 @@ This directory now mirrors the way developers explore SQLSpec:
 - `frameworks/` groups runnable apps (Litestar for now) that rely on lightweight backends (aiosqlite, duckdb).
 - `adapters/` holds connection-focused snippets for production drivers such as asyncpg, psycopg, and oracledb.
 - `patterns/` demonstrates SQL builder usage, migrations, and multi-tenant routing.
+- `arrow/` collects Arrow integration demos so advanced exports stay discoverable without bloating other folders.
 - `loaders/` shows how to hydrate SQL from files for quick demos.
 - `extensions/` keeps integration-specific samples (Adapter Development Kit in this pass).
 

docs/examples/arrow/__init__.py

@@ -0,0 +1,3 @@
+"""Arrow integration examples for SQLSpec."""
+
+__all__ = ()

docs/examples/arrow/arrow_basic_usage.rst

@@ -0,0 +1,16 @@
+Arrow: Basic Usage
+==================
+
+Demonstrate the ``select_to_arrow()`` helper across multiple adapters and
+conversion targets (native Arrow, pandas, polars, and Parquet exports).
+
+.. code-block:: console
+
+   uv run python docs/examples/arrow/arrow_basic_usage.py
+
+Source
+------
+
+.. literalinclude:: arrow_basic_usage.py
+   :language: python
+   :linenos:

docs/examples/index.rst

@@ -91,6 +91,19 @@ Patterns
      - Routing requests to dedicated SQLite configs per tenant slug.
    * - ``patterns/configs/multi_adapter_registry.py``
      - Register multiple adapters on a single SQLSpec registry.
+   * - ``patterns/stacks/query_stack_example.py``
+     - Immutable StatementStack workflow executed against SQLite and AioSQLite drivers.
+
+Arrow
+-----
+
+.. list-table:: Arrow-powered exports
+   :header-rows: 1
+
+   * - File
+     - Scenario
+   * - ``arrow/arrow_basic_usage.py``
+     - ``select_to_arrow()`` walkthrough covering native Arrow, pandas, polars, and Parquet exports.
 
 Loaders
 -------
@@ -142,4 +155,6 @@ Shared Utilities
    frameworks/starlette/aiosqlite_app
    frameworks/flask/sqlite_app
    patterns/configs/multi_adapter_registry
+   patterns/stacks/query_stack_example
+   arrow/arrow_basic_usage
    README

docs/examples/patterns/stacks/__init__.py

@@ -0,0 +1,3 @@
+"""Statement stack examples for SQLSpec documentation."""
+
+__all__ = ()

docs/examples/patterns/stacks/query_stack_example.py

@@ -0,0 +1,105 @@
+"""Demonstrate StatementStack usage across sync and async SQLite adapters."""
+
+import asyncio
+from typing import Any
+
+from sqlspec import SQLSpec, StatementStack
+from sqlspec.adapters.aiosqlite import AiosqliteConfig
+from sqlspec.adapters.sqlite import SqliteConfig
+
+__all__ = ("build_stack", "main", "run_async_example", "run_sync_example")
+
+SCHEMA_SCRIPT = """
+CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, last_action TEXT);
+CREATE TABLE IF NOT EXISTS audit_log (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    user_id INTEGER NOT NULL,
+    action TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS user_roles (
+    user_id INTEGER NOT NULL,
+    role TEXT NOT NULL
+);
+"""
+
+
+def build_stack(user_id: int, action: str) -> "StatementStack":
+    """Add audit, update, and select operations to the stack."""
+    return (
+        StatementStack()
+        .push_execute(
+            "INSERT INTO audit_log (user_id, action) VALUES (:user_id, :action)", {"user_id": user_id, "action": action}
+        )
+        .push_execute(
+            "UPDATE users SET last_action = :action WHERE id = :user_id", {"action": action, "user_id": user_id}
+        )
+        .push_execute("SELECT role FROM user_roles WHERE user_id = :user_id ORDER BY role", {"user_id": user_id})
+    )
+
+
+def _seed_sync_tables(session: "Any", user_id: int, roles: "tuple[str, ...]") -> None:
+    """Create tables and seed sync demo data."""
+    session.execute_script(SCHEMA_SCRIPT)
+    session.execute(
+        "INSERT INTO users (id, last_action) VALUES (:user_id, :action)", {"user_id": user_id, "action": "start"}
+    )
+    session.execute_many(
+        "INSERT INTO user_roles (user_id, role) VALUES (:user_id, :role)",
+        [{"user_id": user_id, "role": role} for role in roles],
+    )
+
+
+async def _seed_async_tables(session: "Any", user_id: int, roles: "tuple[str, ...]") -> None:
+    """Create tables and seed async demo data."""
+    await session.execute_script(SCHEMA_SCRIPT)
+    await session.execute(
+        "INSERT INTO users (id, last_action) VALUES (:user_id, :action)", {"user_id": user_id, "action": "start"}
+    )
+    await session.execute_many(
+        "INSERT INTO user_roles (user_id, role) VALUES (:user_id, :role)",
+        [{"user_id": user_id, "role": role} for role in roles],
+    )
+
+
+def run_sync_example() -> None:
+    """Execute the stack with the synchronous SQLite adapter."""
+    registry = SQLSpec()
+    config = registry.add_config(SqliteConfig(pool_config={"database": ":memory:"}))
+    with registry.provide_session(config) as session:
+        _seed_sync_tables(session, 1, ("admin", "editor"))
+        results = session.execute_stack(build_stack(user_id=1, action="sync-login"))
+        audit_insert, user_update, role_select = results
+        print("[sync] rows inserted:", audit_insert.rows_affected)
+        print("[sync] rows updated:", user_update.rows_affected)
+        if role_select.result is not None:
+            roles = [row["role"] for row in role_select.result.data]
+            print("[sync] roles:", roles)
+
+
+def run_async_example() -> None:
+    """Execute the stack with the asynchronous AioSQLite adapter."""
+
+    async def _inner() -> None:
+        registry = SQLSpec()
+        config = registry.add_config(AiosqliteConfig(pool_config={"database": ":memory:"}))
+        async with registry.provide_session(config) as session:
+            await _seed_async_tables(session, 2, ("viewer",))
+            results = await session.execute_stack(build_stack(user_id=2, action="async-login"))
+            audit_insert, user_update, role_select = results
+            print("[async] rows inserted:", audit_insert.rows_affected)
+            print("[async] rows updated:", user_update.rows_affected)
+            if role_select.result is not None:
+                roles = [row["role"] for row in role_select.result.data]
+                print("[async] roles:", roles)
+
+    asyncio.run(_inner())
+
+
+def main() -> None:
+    """Run both sync and async StatementStack demonstrations."""
+    run_sync_example()
+    run_async_example()
+
+
+if __name__ == "__main__":
+    main()

docs/examples/patterns/stacks/query_stack_example.rst

@@ -0,0 +1,22 @@
+====================
+Query Stack Example
+====================
+
+This example builds an immutable ``StatementStack`` and executes it against both the synchronous SQLite adapter and the asynchronous AioSQLite adapter. Each stack:
+
+1. Inserts an audit log row
+2. Updates the user's last action
+3. Fetches the user's roles
+
+.. literalinclude:: query_stack_example.py
+   :language: python
+   :caption: ``docs/examples/patterns/stacks/query_stack_example.py``
+   :linenos:
+
+Run the script:
+
+.. code-block:: console
+
+   uv run python docs/examples/patterns/stacks/query_stack_example.py
+
+Expected output shows inserted/updated row counts plus the projected role list for each adapter.

docs/guides/README.md

@@ -36,6 +36,11 @@ Optimization guides for SQLSpec:
 
 - [**SQLglot Guide**](performance/sqlglot.md) - SQL parsing, transformation, and optimization with SQLglot
 - [**MyPyC Guide**](performance/mypyc.md) - Compilation strategies for high-performance Python code
+- [**Batch Execution**](performance/batch-execution.md) - Guidance for Query Stack vs. ``execute_many`` across adapters
+
+## Features
+
+- [**Query Stack Guide**](features/query-stack.md) - Multi-statement execution, execution modes, telemetry, and troubleshooting
 
 ## Migrations
 
@@ -55,6 +60,7 @@ Core architecture and design patterns:
 
 - [**Architecture Guide**](architecture/architecture.md) - SQLSpec architecture overview
 - [**Data Flow Guide**](architecture/data-flow.md) - How data flows through SQLSpec
+- [**Architecture Patterns**](architecture/patterns.md) - Immutable stack builder, native vs. sequential branching, and telemetry requirements
 
 ## Extensions
 

docs/guides/adapters/adbc.md

@@ -17,6 +17,12 @@ This guide provides specific instructions for the `adbc` adapter.
 - **JSON Strategy:** `helper` (shared serializers wrap dict/list/tuple values)
 - **Extras:** `type_coercion_overrides` ensure Arrow arrays map to Python lists; PostgreSQL dialects attach a NULL-handling AST transformer
 
+## Query Stack Support
+
+- Each ADBC backend falls back to SQLSpec's sequential stack executor. There is no driver-agnostic pipeline API today, so stacks simply reuse the same cursor management that individual `execute()` calls use, wrapped in a transaction when the backend supports it (e.g., PostgreSQL) and as independent statements when it does not (e.g., SQLite, DuckDB).
+- Continue-on-error mode is supported on every backend. Successful statements commit as they finish, while failures populate `StackResult.error` for downstream inspection.
+- Telemetry spans (`sqlspec.stack.execute`) and `StackExecutionMetrics` counters emit for all stacks, enabling observability parity with adapters that do have native optimizations.
+
 ## Best Practices
 
 - **Arrow-Native:** The primary benefit of ADBC is its direct integration with Apache Arrow. Use it when you need to move large amounts of data efficiently between the database and data science tools like Pandas or Polars.

docs/guides/adapters/aiosqlite.md

@@ -17,6 +17,11 @@ This guide provides specific instructions for the `aiosqlite` adapter.
 - **JSON Strategy:** `helper` (shared serializer handles dict/list/tuple inputs)
 - **Extras:** None (profile applies bool→int and ISO datetime coercions automatically)
 
+## Query Stack Support
+
+- `StatementStack` executions always use the sequential fallback – SQLite has no notion of pipelined requests – so each operation runs one after another on the same connection. When `continue_on_error=False`, SQLSpec opens a transaction (if one is not already in progress) so the entire stack commits or rolls back together. With `continue_on_error=True`, statements are committed individually after each success.
+- Because pooled in-memory connections share state, prefer per-test temporary database files when running stacks under pytest-xdist (see `tests/integration/test_adapters/test_aiosqlite/test_driver.py::test_aiosqlite_statement_stack_*` for the reference pattern).
+
 ## Best Practices
 
 - **Async Only:** This is an asynchronous driver for SQLite. Use it in `asyncio` applications.

docs/guides/adapters/asyncmy.md

@@ -26,3 +26,12 @@ This guide covers `asyncmy`.
 
 -   **`PyMySQL.err.OperationalError: (1366, ...)`**: Incorrect string value for a column. This is often due to character set issues. Ensure your connection and tables are using `utf8mb4`.
 -   **Authentication Errors:** MySQL 8.0 and later use a different default authentication plugin (`caching_sha2_password`). If you have trouble connecting, you may need to configure the user account to use the older `mysql_native_password` plugin, though this is less secure.
+
+## Query Stack Support
+
+The MySQL wire protocol doesn't offer a pipeline/batch mode like Oracle or PostgreSQL, so `StatementStack` executions use the base sequential implementation:
+
+- All operations run one-by-one within the usual transaction rules (fail-fast stacks open a transaction, continue-on-error stacks stay in autocommit mode).
+- Telemetry spans/metrics/logs are still emitted so you can trace stack executions in production.
+
+If you need reduced round-trips for MySQL/MariaDB, consider consolidating statements into stored procedures or batching logic within application-side transactions.

docs/guides/adapters/asyncpg.md

@@ -134,6 +134,31 @@ pip install cloud-alloydb-python-connector
 
 For comprehensive configuration options and troubleshooting, see the [Google Cloud Connectors Guide](/guides/cloud/google-connectors.md).
 
+## Query Stack Support
+
+`StatementStack` calls execute in a single transaction when `continue_on_error=False`, leveraging asyncpg's fast extended-query protocol to minimize round-trips. When you need partial success handling (`continue_on_error=True`), the adapter automatically disables the shared transaction and reports individual failures via `StackResult.error`.
+
+- Telemetry spans (`sqlspec.stack.execute`), metrics (`stack.execute.*`), and hashed operation logging are emitted for every stack, so production monitoring captures adoption automatically.
+- The pipeline path preserves `StackResult.result` for SELECT statements, so downstream helpers continue to operate on the original `SQLResult` objects.
+- To force the sequential fallback (for incident response or regression tests), pass `driver_features={"stack_native_disabled": True}` to the config.
+
+Example usage:
+
+```python
+from sqlspec import StatementStack
+
+stack = (
+    StatementStack()
+    .push_execute("INSERT INTO audit_log (message) VALUES ($1)", ("login",))
+    .push_execute("UPDATE users SET last_login = NOW() WHERE id = $1", (user_id,))
+    .push_execute("SELECT permissions FROM user_permissions WHERE user_id = $1", (user_id,))
+)
+
+results = await asyncpg_session.execute_stack(stack)
+```
+
+If you enable `continue_on_error=True`, the adapter returns three `StackResult` objects, each recording its own `error`/`warning` state without rolling the entire stack back.
+
 ## MERGE Operations (PostgreSQL 15+)
 
 AsyncPG supports high-performance MERGE operations for bulk upserts using PostgreSQL's native MERGE statement with `jsonb_to_recordset()`.

docs/guides/adapters/bigquery.md

@@ -17,6 +17,12 @@ This guide provides specific instructions for the `bigquery` adapter.
 -   **JSON Strategy:** `helper` with `json_tuple_strategy="tuple"`
 -   **Extras:** `type_coercion_overrides` keep list values intact while converting tuples to lists during binding
 
+## Query Stack Support
+
+-   BigQuery does **not** expose a native pipeline API, so `StatementStack` calls execute sequentially through the core driver. Because BigQuery does not offer transactional semantics, the `begin()`/`commit()` hooks are no-ops—the stack still runs each statement in order and surfaces failures via `StackResult.error`.
+-   Continue-on-error mode is supported. Each failing operation records its own `StackExecutionError` while later statements continue to run, which is particularly helpful for long-running analytical batches.
+-   Telemetry spans (`sqlspec.stack.execute`) and `StackExecutionMetrics` counters are emitted for every stack execution, making it easy to monitor adoption even though the adapter falls back to the sequential path.
+
 ## Best Practices
 
 -   **Authentication:** BigQuery requires authentication with Google Cloud. For local development, the easiest way is to use the Google Cloud CLI and run `gcloud auth application-default login`.