diff --git a/.github/workflows/dialect-live-cloud.yml b/.github/workflows/dialect-live-cloud.yml
index 9703f0855..106a7ca7b 100644
--- a/.github/workflows/dialect-live-cloud.yml
+++ b/.github/workflows/dialect-live-cloud.yml
@@ -37,56 +37,13 @@ on:
   workflow_dispatch:
 
 jobs:
-  # Build the dialect_live test binary once, exactly as dialect-live.yml
-  # does, so the per-backend legs download it instead of each recompiling
-  # the same crate against the same feature set.
+  # Build the dialect_live test binary once (reusable workflow shared with
+  # dialect-live.yml), so the per-backend legs download it instead of each
+  # recompiling the same crate against the same feature set. save-cache
+  # stays false: a nightly run must not fight over the shared cache key
+  # owned by build.yaml/main.
   compile:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out repository
-        uses: actions/checkout@v4
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v4
-        with:
-          node-version: "22"
-
-      - name: Install tree-sitter-cli
-        # parser.c is gitignored and generated from grammar.js by the
-        # tree-sitter-ggsql build script on every fresh checkout.
-        run: npm install -g tree-sitter-cli
-
-      - name: Install Rust
-        uses: dtolnay/rust-toolchain@stable
-
-      - name: Caching
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Same shared-key as build.yaml and dialect-live.yml: the prefix
-          # fallback restores their target/ and this job only rebuilds the
-          # feature deltas.
-          shared-key: ${{ runner.os }}-build
-          cache-on-failure: true
-
-      - name: Compile the dialect_live test binary
-        # Same feature set as dialect-live.yml so the restored crate
-        # fingerprints hit; see the long comment there. --no-run plus
-        # --message-format=json yields the binary path; the binary is
-        # self-contained (deps statically linked, ADBC drivers dlopened at
-        # runtime), so the legs can run it directly.
-        run: |
-          bin=$(cargo test -p ggsql --features "adbc sqlite" --test dialect_live \
-                --no-run --message-format=json-render-diagnostics \
-                | jq -rs '[.[] | select(.target.name == "dialect_live" and .executable != null) | .executable][-1]')
-          test -n "$bin" -a -f "$bin"
-          cp "$bin" dialect-live-bin
-
-      - name: Upload the test binary
-        uses: actions/upload-artifact@v4
-        with:
-          name: dialect-live-bin
-          path: dialect-live-bin
-          retention-days: 1
+    uses: ./.github/workflows/dialect-live-compile.yml
 
   live:
     needs: compile
diff --git a/.github/workflows/dialect-live-compile.yml b/.github/workflows/dialect-live-compile.yml
new file mode 100644
index 000000000..93181b0a2
--- /dev/null
+++ b/.github/workflows/dialect-live-compile.yml
@@ -0,0 +1,70 @@
+name: dialect-live compile (reusable)
+
+# Shared compile job for the dialect-live workflows: builds the
+# dialect_live test binary once and hands it to every backend leg as an
+# artifact, so the legs compile nothing. Callers:
+# - dialect-live.yml (PRs + main): passes save-cache on main only.
+# - dialect-live-cloud.yml (nightly): never saves — a scheduled run must
+#   not fight over the shared cache key owned by build.yaml/main.
+
+on:
+  workflow_call:
+    inputs:
+      save-cache:
+        description: Whether to save the rust-cache back under the shared key.
+        type: boolean
+        default: false
+
+jobs:
+  compile:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: "22"
+
+      - name: Install tree-sitter-cli
+        # parser.c is gitignored and generated from grammar.js by the
+        # tree-sitter-ggsql build script on every fresh checkout.
+        run: npm install -g tree-sitter-cli
+
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Caching
+        uses: Swatinem/rust-cache@v2
+        with:
+          # Deliberately the same shared-key as build.yaml: rust-cache's
+          # prefix fallback restores the build job's target/, so this job
+          # only rebuilds the feature deltas instead of every dependency,
+          # and on main it saves the warmed variant back for build.yaml
+          # and PR runs to reuse.
+          shared-key: ${{ runner.os }}-build
+          cache-on-failure: true
+          save-if: ${{ inputs.save-cache }}
+
+      - name: Compile the dialect_live test binary
+        # --no-run plus --message-format=json yields the binary path; the
+        # binary is self-contained (deps statically linked, ODBC/ADBC
+        # drivers dlopened at runtime), so the legs can run it directly.
+        # Feature set is load-bearing: `adbc` compiles in the ADBC reader
+        # path and `sqlite` provides the cache backend Druid wraps in. The
+        # datafusion leg uses the Foundry driver through the driver manager,
+        # so the DataFusion graph never enters the test build.
+        run: |
+          bin=$(cargo test -p ggsql --features "adbc sqlite" --test dialect_live \
+                --no-run --message-format=json-render-diagnostics \
+                | jq -rs '[.[] | select(.target.name == "dialect_live" and .executable != null) | .executable][-1]')
+          test -n "$bin" -a -f "$bin"
+          cp "$bin" dialect-live-bin
+
+      - name: Upload the test binary
+        uses: actions/upload-artifact@v4
+        with:
+          name: dialect-live-bin
+          path: dialect-live-bin
+          retention-days: 1
diff --git a/.github/workflows/dialect-live.yml b/.github/workflows/dialect-live.yml
index 80a693d4c..f67edcad8 100644
--- a/.github/workflows/dialect-live.yml
+++ b/.github/workflows/dialect-live.yml
@@ -30,62 +30,15 @@ on:
   workflow_dispatch:
 
 jobs:
-  # Build the dialect_live test binary once. rust-cache (same shared-key as
-  # build.yaml) restores build.yaml's warmed target/, so this compiles only
-  # the workspace crates and the feature deltas; the binary is then handed
-  # to every backend leg as an artifact, so the legs compile nothing.
+  # Build the dialect_live test binary once (reusable workflow shared with
+  # dialect-live-cloud.yml). rust-cache restores build.yaml's warmed
+  # target/ via the same shared-key; on main the warmed variant is saved
+  # back for build.yaml and PR runs to reuse. The binary is handed to
+  # every backend leg as an artifact, so the legs compile nothing.
   compile:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Check out repository
-        uses: actions/checkout@v4
-
-      - name: Setup Node.js
-        uses: actions/setup-node@v4
-        with:
-          node-version: "22"
-
-      - name: Install tree-sitter-cli
-        # parser.c is gitignored and generated from grammar.js by the
-        # tree-sitter-ggsql build script on every fresh checkout.
-        run: npm install -g tree-sitter-cli
-
-      - name: Install Rust
-        uses: dtolnay/rust-toolchain@stable
-
-      - name: Caching
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Deliberately the same shared-key as build.yaml: rust-cache's
-          # prefix fallback restores the build job's target/, so this job
-          # only rebuilds the feature deltas instead of every dependency,
-          # and on main it saves the warmed variant back for build.yaml
-          # and PR runs to reuse.
-          shared-key: ${{ runner.os }}-build
-          cache-on-failure: true
-          save-if: ${{ github.ref == 'refs/heads/main' }}
-
-      - name: Compile the dialect_live test binary
-        # --no-run plus --message-format=json yields the binary path; the
-        # binary is self-contained (deps statically linked, ODBC/ADBC
-        # drivers dlopened at runtime), so the legs can run it directly.
-        # Feature set is load-bearing: `adbc` compiles in the ADBC reader
-        # path and `sqlite` provides the cache backend Druid wraps in. The
-        # datafusion leg uses the Foundry driver through the driver manager,
-        # so the DataFusion graph never enters the test build.
-        run: |
-          bin=$(cargo test -p ggsql --features "adbc sqlite" --test dialect_live \
-                --no-run --message-format=json-render-diagnostics \
-                | jq -rs '[.[] | select(.target.name == "dialect_live" and .executable != null) | .executable][-1]')
-          test -n "$bin" -a -f "$bin"
-          cp "$bin" dialect-live-bin
-
-      - name: Upload the test binary
-        uses: actions/upload-artifact@v4
-        with:
-          name: dialect-live-bin
-          path: dialect-live-bin
-          retention-days: 1
+    uses: ./.github/workflows/dialect-live-compile.yml
+    with:
+      save-cache: ${{ github.ref == 'refs/heads/main' }}
 
   live:
     needs: compile
diff --git a/src/execute/golden.rs b/src/execute/golden.rs
index b756e25f5..04b332010 100644
--- a/src/execute/golden.rs
+++ b/src/execute/golden.rs
@@ -27,8 +27,7 @@ mod battery;
 
 use crate::reader::registry;
 use crate::reader::test_support::StubReader;
-use crate::reader::{AnsiDialect, Reader, SqlDialect};
-use crate::DataFrame;
+use crate::reader::{AnsiDialect, Reader};
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
@@ -37,64 +36,15 @@ use std::sync::{Arc, Mutex};
 /// predates and outlives any single scheme); everything else resolves
 /// through the registry, so a dialect constructor can only live in one
 /// place. The `goldens_cover_registry` test keeps the file set in step.
-fn dialect_for(name: &str) -> Box<dyn SqlDialect + Send> {
+fn dialect_for(name: &str) -> crate::reader::DialectRef {
     if name == "ansi" {
-        return Box::new(AnsiDialect);
+        return &AnsiDialect;
     }
     registry::by_scheme(name)
         .unwrap_or_else(|| panic!("no registry entry for golden dialect '{name}'"))
         .dialect()
 }
 
-/// The shared fixture as a DataFrame, matching the live table's schema:
-/// id/val/grp/day(a real date)/"mixed Case". `TextDay` swaps the date for
-/// ISO text to force the text→temporal cast path.
-fn fixture_df(fixture: &battery::Fixture) -> DataFrame {
-    use arrow::array::{ArrayRef, Date32Array, Float64Array, Int32Array, StringArray};
-    let day: ArrayRef = match fixture {
-        battery::Fixture::Shared => Arc::new(Date32Array::from(
-            battery::ROWS
-                .iter()
-                .map(|r| r.day_epoch)
-                .collect::<Vec<_>>(),
-        )),
-        battery::Fixture::TextDay => Arc::new(StringArray::from(
-            battery::ROWS
-                .iter()
-                .map(|r| battery::day_iso(r.day_epoch))
-                .collect::<Vec<_>>(),
-        )),
-    };
-    DataFrame::new(vec![
-        (
-            "id",
-            Arc::new(Int32Array::from(
-                battery::ROWS.iter().map(|r| r.id).collect::<Vec<_>>(),
-            )) as ArrayRef,
-        ),
-        (
-            "val",
-            Arc::new(Float64Array::from(
-                battery::ROWS.iter().map(|r| r.val).collect::<Vec<_>>(),
-            )) as ArrayRef,
-        ),
-        (
-            "grp",
-            Arc::new(StringArray::from(
-                battery::ROWS.iter().map(|r| r.grp).collect::<Vec<_>>(),
-            )) as ArrayRef,
-        ),
-        ("day", day),
-        (
-            "mixed Case",
-            Arc::new(Float64Array::from(
-                battery::ROWS.iter().map(|r| r.mixed).collect::<Vec<_>>(),
-            )) as ArrayRef,
-        ),
-    ])
-    .unwrap()
-}
-
 /// Collapse all whitespace runs so goldens don't churn on formatting, and
 /// replace the random per-session hex in generated temp-table names
 /// (`__ggsql_global_<32 hex>__`) with a stable placeholder.
@@ -130,7 +80,7 @@ fn golden_dir() -> PathBuf {
 
 /// Render every battery case's SQL for one dialect. Re-registers the
 /// fixture only when its variant changes.
-fn render_dialect(dialect: Box<dyn SqlDialect + Send>) -> String {
+fn render_dialect(dialect: crate::reader::DialectRef) -> String {
     let (reader, log) = StubReader::new(dialect);
     let mut out = String::new();
     let mut registered: Option<u8> = None;
@@ -145,7 +95,7 @@ fn render_dialect(dialect: Box<dyn SqlDialect + Send>) -> String {
                 reader.unregister(battery::TABLE).unwrap();
             }
             reader
-                .register(battery::TABLE, fixture_df(&case.fixture), true)
+                .register(battery::TABLE, battery::fixture_df(&case.fixture), true)
                 .unwrap();
             registered = Some(variant);
         }
diff --git a/src/reader/adbc.rs b/src/reader/adbc.rs
index 68c4aef7d..dc918ece6 100644
--- a/src/reader/adbc.rs
+++ b/src/reader/adbc.rs
@@ -26,7 +26,7 @@ pub struct AdbcReader<D: Driver> {
     // new_statement / set_sql_query / execute all take &mut, but Reader::execute_sql
     // takes &self.
     connection: RefCell<<D::DatabaseType as Database>::ConnectionType>,
-    dialect: Box<dyn SqlDialect + Send>,
+    dialect: crate::reader::DialectRef,
     /// Registry-declared driver quirks (error-text conventions), when the
     /// reader was built for a registered backend. `None` for ad-hoc drivers.
     adbc_info: Option<crate::reader::registry::AdbcInfo>,
@@ -40,7 +40,7 @@ impl<D: Driver> AdbcReader<D> {
     /// Construct an `AdbcReader` with an explicit `SqlDialect`. Use this to
     /// plug in backend-specific dialects (e.g. a TrinoDialect, SnowflakeDialect)
     /// when the reader is pointed at that backend.
-    pub fn with_dialect(driver: D, dialect: Box<dyn SqlDialect + Send>) -> Result<Self> {
+    pub fn with_dialect(driver: D, dialect: crate::reader::DialectRef) -> Result<Self> {
         Self::new(driver, dialect)
     }
 
@@ -50,7 +50,7 @@ impl<D: Driver> AdbcReader<D> {
     /// options. For convenience, use `from_driver` for the common case
     /// with the ANSI dialect, or pass a custom `SqlDialect`
     /// (e.g. a Trino / Snowflake dialect) here directly.
-    pub fn new(mut driver: D, dialect: Box<dyn SqlDialect + Send>) -> Result<Self> {
+    pub fn new(mut driver: D, dialect: crate::reader::DialectRef) -> Result<Self> {
         let database = driver
             .new_database()
             .map_err(|e| GgsqlError::ReaderError(format!("ADBC new_database failed: {}", e)))?;
@@ -74,7 +74,7 @@ impl<D: Driver> AdbcReader<D> {
     /// or other auth-required backends).
     pub fn new_with_database_opts(
         mut driver: D,
-        dialect: Box<dyn SqlDialect + Send>,
+        dialect: crate::reader::DialectRef,
         opts: impl IntoIterator<
             Item = (
                 adbc_core::options::OptionDatabase,
@@ -156,7 +156,7 @@ impl<D: Driver> AdbcReader<D> {
     /// standards-compliant backends; use `new` directly to plug in a
     /// backend-specific dialect.
     pub fn from_driver(driver: D) -> Result<Self> {
-        Self::new(driver, Box::new(AnsiDialect))
+        Self::new(driver, &AnsiDialect)
     }
 }
 
@@ -493,6 +493,12 @@ fn bigquery_driver_uri(body: &str, query: &str) -> (String, String) {
 
 /// Probe whether an ADBC driver for `scheme` can be loaded, without opening
 /// a connection. Used by reader dispatch to decide between ADBC and ODBC.
+///
+/// Note this loads (dlopens) the driver library and drops it;
+/// `from_connection_string` loads it again when the probe succeeds. That
+/// double-load is accepted connect-time cost: threading the preloaded
+/// driver through dispatch would complicate the probe-then-build flow for
+/// no per-query benefit.
 pub fn adbc_driver_available(scheme: &str) -> bool {
     load_driver_for_scheme(scheme).is_ok()
 }
@@ -629,7 +635,7 @@ where
                 .map_err(|e| GgsqlError::ReaderError(format!("ADBC execute_update DROP: {}", e)))?;
         }
 
-        let create_sql = crate::reader::create_table_sql(name, &schema, &*self.dialect)?;
+        let create_sql = crate::reader::create_table_sql(name, &schema, self.dialect)?;
         self.new_query_statement(&mut conn, &create_sql)?
             .execute_update()
             .map_err(|e| GgsqlError::ReaderError(format!("ADBC execute_update CREATE: {}", e)))?;
@@ -740,7 +746,7 @@ where
     }
 
     fn dialect(&self) -> &dyn SqlDialect {
-        &*self.dialect
+        self.dialect
     }
 }
 
@@ -1018,10 +1024,9 @@ mod equivalence_tests {
             None,
         )
         .expect("`dbc install sqlite` first; see module docs");
-        let dialect: Box<dyn crate::reader::SqlDialect + Send> = Box::new(SqliteDialect);
         AdbcReader::new_with_database_opts(
             driver,
-            dialect,
+            &SqliteDialect,
             std::iter::once((
                 OptionDatabase::Uri,
                 OptionValue::String(format!("file:{}", db_path)),
diff --git a/src/reader/cache.rs b/src/reader/cache.rs
index c2b57d76a..6e7c465c4 100644
--- a/src/reader/cache.rs
+++ b/src/reader/cache.rs
@@ -1424,8 +1424,8 @@ mod behavior_tests {
         use crate::reader::dialects::{DuckDbDialect, MySqlDialect};
         use crate::reader::test_support::StubReader;
 
-        let (primary, primary_sql) = StubReader::new(Box::new(MySqlDialect));
-        let (cache, _cache_sql) = StubReader::new(Box::new(DuckDbDialect));
+        let (primary, primary_sql) = StubReader::new(&MySqlDialect);
+        let (cache, _cache_sql) = StubReader::new(&DuckDbDialect);
         // Memoization disabled: the stub fabricates Float64 results that the
         // memo metadata path cannot store, which is beside the point here.
         let reader = CachingReader::with_config(
diff --git a/src/reader/cache_equivalence.rs b/src/reader/cache_equivalence.rs
index e4c2aac9e..373bfbcaf 100644
--- a/src/reader/cache_equivalence.rs
+++ b/src/reader/cache_equivalence.rs
@@ -291,10 +291,9 @@ mod adbc_mode {
             None,
         )
         .expect("`dbc install sqlite` first; see adbc.rs::equivalence_tests docs");
-        let dialect: Box<dyn SqlDialect + Send> = Box::new(SqliteDialect);
         AdbcReader::new_with_database_opts(
             driver,
-            dialect,
+            &SqliteDialect,
             std::iter::once((
                 OptionDatabase::Uri,
                 OptionValue::String(format!("file:{}", db_path)),
diff --git a/src/reader/connection.rs b/src/reader/connection.rs
index fa3c98b21..594787a06 100644
--- a/src/reader/connection.rs
+++ b/src/reader/connection.rs
@@ -39,13 +39,12 @@ pub fn split_cache_uri(uri: &str) -> Option<(String, String)> {
 /// URI parsing or option map — drivers reject unknown keys.
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct GgsqlParams {
-    /// Whether the connection opts out of the *automatic* caching layer
-    /// (`cache=off`): no `CachingReader` wrap when the backend cannot host
-    /// ggsql's internal tables. An explicit `<cache>+<primary>://` URI is
-    /// unaffected.
+    /// `cache=off`: opt out of the automatic caching layer entirely —
+    /// different from `cache_disabled`, which keeps the caching reader but
+    /// turns its result memo off.
     pub cache_off: bool,
-    /// Forced reader kind (`reader=native|adbc|odbc`); `None` lets dispatch
-    /// pick in its default preference order.
+    /// `reader=native|adbc|odbc`: force one reader kind over the default
+    /// preference order (native → ADBC → ODBC).
     pub reader: Option<String>,
     /// Explicit SQL-dialect pin (`dialect=<scheme|ansi>`), bypassing backend
     /// detection — the escape hatch for backends ggsql doesn't recognise.
@@ -160,7 +159,10 @@ impl ConnUri {
     }
 
     /// True when the URI opts out of the automatic caching layer
-    /// (`cache=off`).
+    /// (`cache=off`). This is one of two distinct "disable cache" knobs:
+    /// `cache=off` never wraps the reader at all, while `cache_disabled=1`
+    /// keeps the caching reader but disables its result memo
+    /// ([`GgsqlParams::cache_disabled`]).
     pub fn cache_disabled_off(&self) -> bool {
         self.ggsql.cache_off
     }
@@ -353,13 +355,12 @@ fn build_backend_reader(
             let dialect = if let Some(name) = &conn.ggsql.dialect {
                 Some(
                     crate::reader::registry::dialect_override(name)
-                        .map(|d| d as Box<dyn crate::reader::SqlDialect>)
                         .ok_or_else(|| crate::reader::registry::unknown_dialect_error(name))?,
                 )
             } else if entry.scheme == "flightsql" {
                 None
             } else {
-                Some(entry.dialect() as Box<dyn crate::reader::SqlDialect>)
+                Some(entry.dialect())
             };
             let reader = crate::reader::OdbcReader::from_odbc_conn_str(&conn_str, dialect)?;
             return Ok(Box::new(reader));
@@ -578,12 +579,12 @@ fn auto_cache_if_needed(
     {
         use crate::reader::cache::CacheConfig;
 
-        let (cache_uri, cache_scheme) = if cfg!(feature = "duckdb") {
-            ("duckdb://memory", "duckdb")
+        let cache_scheme = if cfg!(feature = "duckdb") {
+            "duckdb"
         } else {
-            ("sqlite://:memory:", "sqlite")
+            "sqlite"
         };
-        let cache = build_reader(cache_uri)?;
+        let cache = build_reader(cache_uri(cache_scheme)?)?;
         Ok(Box::new(crate::reader::CachingReader::with_config(
             reader,
             cache,
diff --git a/src/reader/dialects/bigquery.rs b/src/reader/dialects/bigquery.rs
index 518470f2f..04c6069ab 100644
--- a/src/reader/dialects/bigquery.rs
+++ b/src/reader/dialects/bigquery.rs
@@ -27,7 +27,7 @@ impl SqlDialect for BigQueryDialect {
     }
 
     fn quote_ident(&self, name: &str) -> String {
-        format!("`{}`", name.replace('`', "``"))
+        super::backtick_quote_ident(name)
     }
 
     fn sql_cast(&self, expr: &str, type_name: &str) -> String {
diff --git a/src/reader/dialects/clickhouse.rs b/src/reader/dialects/clickhouse.rs
index c1f7e5400..da9b47f40 100644
--- a/src/reader/dialects/clickhouse.rs
+++ b/src/reader/dialects/clickhouse.rs
@@ -43,7 +43,7 @@ impl SqlDialect for ClickHouseDialect {
     }
 
     fn quote_ident(&self, name: &str) -> String {
-        format!("`{}`", name.replace('`', "``"))
+        super::backtick_quote_ident(name)
     }
 
     fn sql_greatest(&self, exprs: &[&str]) -> String {
diff --git a/src/reader/dialects/databricks.rs b/src/reader/dialects/databricks.rs
index ab4a4d7ee..6ca86e6e4 100644
--- a/src/reader/dialects/databricks.rs
+++ b/src/reader/dialects/databricks.rs
@@ -21,7 +21,7 @@ impl SqlDialect for DatabricksDialect {
     }
 
     fn quote_ident(&self, name: &str) -> String {
-        format!("`{}`", name.replace('`', "``"))
+        super::backtick_quote_ident(name)
     }
 
     // Spark treats double quotes as string literals, and SQL warehouses
diff --git a/src/reader/dialects/datafusion.rs b/src/reader/dialects/datafusion.rs
index 10e1f9216..5bd45f1df 100644
--- a/src/reader/dialects/datafusion.rs
+++ b/src/reader/dialects/datafusion.rs
@@ -67,14 +67,8 @@ impl SqlDialect for DataFusionDialect {
         kind: crate::plot::types::CastTargetType,
     ) -> String {
         // DataFusion rejects temporal -> numeric casts ("Unsupported CAST
-        // from Date32 to Float64"); EXTRACT EPOCH yields seconds for both
-        // dates and timestamps, so scale to ggsql's epoch units.
-        use crate::plot::types::CastTargetType as C;
-        match kind {
-            C::Date => format!("(EXTRACT(EPOCH FROM {expr}) / 86400)"),
-            C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
-            _ => format!("CAST({expr} AS DOUBLE)"),
-        }
+        // from Date32 to Float64").
+        super::epoch_via_extract_seconds(self, expr, kind)
     }
 }
 
diff --git a/src/reader/dialects/drill.rs b/src/reader/dialects/drill.rs
index b16663f9b..a8194b181 100644
--- a/src/reader/dialects/drill.rs
+++ b/src/reader/dialects/drill.rs
@@ -35,7 +35,7 @@ impl SqlDialect for DrillDialect {
     }
 
     fn quote_ident(&self, name: &str) -> String {
-        format!("`{}`", name.replace('`', "``"))
+        super::backtick_quote_ident(name)
     }
 }
 
diff --git a/src/reader/dialects/mod.rs b/src/reader/dialects/mod.rs
index f90aaa14f..7bd7b91a7 100644
--- a/src/reader/dialects/mod.rs
+++ b/src/reader/dialects/mod.rs
@@ -50,6 +50,58 @@ pub struct AnsiDialect;
 
 impl SqlDialect for AnsiDialect {}
 
+/// Backtick identifier quoting (MySQL/MariaDB, ClickHouse, BigQuery,
+/// Databricks, Drill); embedded backticks are doubled.
+pub(crate) fn backtick_quote_ident(name: &str) -> String {
+    format!("`{}`", name.replace('`', "``"))
+}
+
+/// `1`/`0` boolean literal for backends without a real BOOLEAN literal
+/// (SQL Server, Oracle, SQLite).
+pub(crate) fn one_zero_boolean_literal(value: bool) -> String {
+    if value { "1" } else { "0" }.to_string()
+}
+
+/// Postgres-style epoch conversion for backends that reject temporal →
+/// numeric casts: date subtraction yields integer days, EXTRACT EPOCH
+/// yields seconds scaled to microseconds. Shared by the Postgres and
+/// Redshift dialects.
+pub(crate) fn epoch_via_subtract_extract<D: SqlDialect + ?Sized>(
+    dialect: &D,
+    expr: &str,
+    kind: crate::plot::types::CastTargetType,
+) -> String {
+    use crate::plot::types::CastTargetType as C;
+    match kind {
+        C::Date => format!("({expr} - DATE '1970-01-01')"),
+        C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
+        _ => {
+            let ty = dialect.number_type_name().unwrap_or("DOUBLE PRECISION");
+            dialect.sql_cast(expr, ty)
+        }
+    }
+}
+
+/// Epoch conversion for backends where EXTRACT EPOCH yields seconds for
+/// both dates and timestamps (DataFusion, MonetDB): dates scale to days,
+/// datetimes to microseconds; anything else falls through to a numeric
+/// cast in the dialect's number type.
+pub(crate) fn epoch_via_extract_seconds<D: SqlDialect + ?Sized>(
+    dialect: &D,
+    expr: &str,
+    kind: crate::plot::types::CastTargetType,
+) -> String {
+    use crate::plot::types::CastTargetType as C;
+    match kind {
+        C::Date => format!("(EXTRACT(EPOCH FROM {expr}) / 86400)"),
+        C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
+        _ => {
+            let ty = dialect.number_type_name().unwrap_or("DOUBLE");
+            dialect.sql_cast(expr, ty)
+        }
+    }
+}
+
 /// CASE-based scalar greatest/least for backends without `GREATEST`/`LEAST`
 /// (SQL Server, SQLite, Druid, Drill, MonetDB). Builds a left-folded chain of
 /// two-way comparisons.
@@ -83,7 +135,7 @@ mod tests {
     use super::*;
     use crate::reader::registry::{by_scheme, detect_or_err as detect_dialect};
 
-    fn dialect_for_scheme(scheme: &str) -> Option<Box<dyn SqlDialect + Send>> {
+    fn dialect_for_scheme(scheme: &str) -> Option<crate::reader::DialectRef> {
         by_scheme(scheme).map(|e| e.dialect())
     }
 
@@ -94,31 +146,28 @@ mod tests {
     #[test]
     fn detects_from_dbms_name() {
         assert_type_name(
-            &*detect_dialect(Some("PostgreSQL"), None).unwrap(),
+            detect_dialect(Some("PostgreSQL"), None).unwrap(),
             Some("DOUBLE PRECISION"),
         );
+        assert_type_name(detect_dialect(Some("MySQL"), None).unwrap(), Some("DOUBLE"));
         assert_type_name(
-            &*detect_dialect(Some("MySQL"), None).unwrap(),
-            Some("DOUBLE"),
-        );
-        assert_type_name(
-            &*detect_dialect(Some("Microsoft SQL Server"), None).unwrap(),
+            detect_dialect(Some("Microsoft SQL Server"), None).unwrap(),
             Some("FLOAT"),
         );
         assert_type_name(
-            &*detect_dialect(Some("Snowflake"), None).unwrap(),
+            detect_dialect(Some("Snowflake"), None).unwrap(),
             Some("DOUBLE"),
         );
         assert_type_name(
-            &*detect_dialect(Some("Oracle"), None).unwrap(),
+            detect_dialect(Some("Oracle"), None).unwrap(),
             Some("BINARY_DOUBLE"),
         );
         assert_type_name(
-            &*detect_dialect(Some("ClickHouse"), None).unwrap(),
+            detect_dialect(Some("ClickHouse"), None).unwrap(),
             Some("Nullable(Float64)"),
         );
         assert_type_name(
-            &*detect_dialect(Some("Amazon Redshift"), None).unwrap(),
+            detect_dialect(Some("Amazon Redshift"), None).unwrap(),
             Some("DOUBLE PRECISION"),
         );
     }
@@ -126,11 +175,11 @@ mod tests {
     #[test]
     fn falls_through_to_driver_hint() {
         assert_type_name(
-            &*detect_dialect(Some("Unknown DBMS"), Some("PostgreSQL Unicode")).unwrap(),
+            detect_dialect(Some("Unknown DBMS"), Some("PostgreSQL Unicode")).unwrap(),
             Some("DOUBLE PRECISION"),
         );
         assert_type_name(
-            &*detect_dialect(None, Some("msodbcsql18")).unwrap(),
+            detect_dialect(None, Some("msodbcsql18")).unwrap(),
             Some("FLOAT"),
         );
     }
@@ -183,42 +232,24 @@ mod tests {
         );
     }
 
-    /// All schemes the registry knows, for conformance sweeps.
-    const ALL_SCHEMES: &[&str] = &[
-        "postgres",
-        "redshift",
-        "mysql",
-        "mariadb",
-        "snowflake",
-        "mssql",
-        "bigquery",
-        "databricks",
-        "clickhouse",
-        "oracle",
-        "trino",
-        "exasol",
-        "monetdb",
-        "druid",
-        "drill",
-        "datafusion",
-        "duckdb",
-        "sqlite",
-    ];
-
     /// Contract for the quantile hook: callers pass the raw (unquoted)
     /// column name and the dialect quotes it. A name needing quoting must
     /// appear quoted — interpolating it raw breaks on any real column whose
     /// name is not a bare lowercase identifier.
     #[test]
     fn quantile_hook_quotes_raw_column_names() {
-        for scheme in ALL_SCHEMES {
-            let d = dialect_for_scheme(scheme).unwrap();
-            let quoted = d.quote_ident("mixed Case");
-            let sql = d.sql_quantile("mixed Case", 0.5, crate::sql::FromItem::Table("t"), &[]);
-            assert!(
-                sql.contains(&quoted),
-                "{scheme}: sql_quantile does not quote its column: {sql}"
-            );
+        // Sweeps the registry directly so new backends are covered
+        // automatically; canonical schemes and aliases alike.
+        for entry in crate::reader::registry::REGISTRY {
+            for scheme in entry.schemes() {
+                let d = dialect_for_scheme(scheme).unwrap();
+                let quoted = d.quote_ident("mixed Case");
+                let sql = d.sql_quantile("mixed Case", 0.5, crate::sql::FromItem::Table("t"), &[]);
+                assert!(
+                    sql.contains(&quoted),
+                    "{scheme}: sql_quantile does not quote its column: {sql}"
+                );
+            }
         }
     }
 
diff --git a/src/reader/dialects/monetdb.rs b/src/reader/dialects/monetdb.rs
index df7efe991..4cabef8ff 100644
--- a/src/reader/dialects/monetdb.rs
+++ b/src/reader/dialects/monetdb.rs
@@ -61,13 +61,8 @@ impl SqlDialect for MonetDbDialect {
         kind: crate::plot::types::CastTargetType,
     ) -> String {
         // MonetDB has no direct temporal -> double cast ("types date and
-        // double are not equal"); EXTRACT EPOCH yields seconds.
-        use crate::plot::types::CastTargetType as C;
-        match kind {
-            C::Date => format!("(EXTRACT(EPOCH FROM {expr}) / 86400)"),
-            C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
-            _ => format!("CAST({expr} AS DOUBLE)"),
-        }
+        // double are not equal").
+        super::epoch_via_extract_seconds(self, expr, kind)
     }
 }
 
diff --git a/src/reader/dialects/mssql.rs b/src/reader/dialects/mssql.rs
index d35853a6a..4615be99f 100644
--- a/src/reader/dialects/mssql.rs
+++ b/src/reader/dialects/mssql.rs
@@ -35,11 +35,7 @@ impl SqlDialect for MssqlDialect {
     }
 
     fn sql_boolean_literal(&self, value: bool) -> String {
-        if value {
-            "1".to_string()
-        } else {
-            "0".to_string()
-        }
+        super::one_zero_boolean_literal(value)
     }
 
     fn sql_ceil(&self, expr: &str) -> String {
diff --git a/src/reader/dialects/mysql.rs b/src/reader/dialects/mysql.rs
index 381d89fc9..15003dfa3 100644
--- a/src/reader/dialects/mysql.rs
+++ b/src/reader/dialects/mysql.rs
@@ -20,7 +20,7 @@ impl SqlDialect for MySqlDialect {
     }
 
     fn quote_ident(&self, name: &str) -> String {
-        format!("`{}`", name.replace('`', "``"))
+        super::backtick_quote_ident(name)
     }
 
     fn sql_null_safe_eq(&self, left: &str, right: &str) -> String {
diff --git a/src/reader/dialects/oracle.rs b/src/reader/dialects/oracle.rs
index 625f2b943..b2e8aef52 100644
--- a/src/reader/dialects/oracle.rs
+++ b/src/reader/dialects/oracle.rs
@@ -27,11 +27,7 @@ impl SqlDialect for OracleDialect {
     }
 
     fn sql_boolean_literal(&self, value: bool) -> String {
-        if value {
-            "1".to_string()
-        } else {
-            "0".to_string()
-        }
+        super::one_zero_boolean_literal(value)
     }
 
     fn sql_limit(&self, query: &str, n: usize) -> String {
diff --git a/src/reader/dialects/postgres.rs b/src/reader/dialects/postgres.rs
index f974010d2..54e550401 100644
--- a/src/reader/dialects/postgres.rs
+++ b/src/reader/dialects/postgres.rs
@@ -30,14 +30,8 @@ impl SqlDialect for PostgresDialect {
         expr: &str,
         kind: crate::plot::types::CastTargetType,
     ) -> String {
-        // Postgres rejects temporal -> numeric casts; date subtraction
-        // yields integer days, EXTRACT EPOCH covers datetimes.
-        use crate::plot::types::CastTargetType as C;
-        match kind {
-            C::Date => format!("({expr} - DATE '1970-01-01')"),
-            C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
-            _ => format!("CAST({expr} AS DOUBLE PRECISION)"),
-        }
+        // Postgres rejects temporal -> numeric casts.
+        super::epoch_via_subtract_extract(self, expr, kind)
     }
 
     fn sql_quantile(
diff --git a/src/reader/dialects/redshift.rs b/src/reader/dialects/redshift.rs
index e579b3a66..71e86d5c5 100644
--- a/src/reader/dialects/redshift.rs
+++ b/src/reader/dialects/redshift.rs
@@ -22,11 +22,6 @@ impl SqlDialect for RedshiftDialect {
         expr: &str,
         kind: crate::plot::types::CastTargetType,
     ) -> String {
-        use crate::plot::types::CastTargetType as C;
-        match kind {
-            C::Date => format!("({expr} - DATE '1970-01-01')"),
-            C::DateTime => format!("(EXTRACT(EPOCH FROM {expr}) * 1000000)"),
-            _ => format!("CAST({expr} AS DOUBLE PRECISION)"),
-        }
+        super::epoch_via_subtract_extract(self, expr, kind)
     }
 }
diff --git a/src/reader/dialects/sqlite.rs b/src/reader/dialects/sqlite.rs
index c4c0e8ebe..10485637e 100644
--- a/src/reader/dialects/sqlite.rs
+++ b/src/reader/dialects/sqlite.rs
@@ -68,11 +68,7 @@ impl SqlDialect for SqliteDialect {
     }
 
     fn sql_boolean_literal(&self, value: bool) -> String {
-        if value {
-            "1".to_string()
-        } else {
-            "0".to_string()
-        }
+        super::one_zero_boolean_literal(value)
     }
 
     fn sql_spatial_setup(&self) -> Vec<String> {
diff --git a/src/reader/mod.rs b/src/reader/mod.rs
index f4fb21f38..6a8fcfa16 100644
--- a/src/reader/mod.rs
+++ b/src/reader/mod.rs
@@ -945,6 +945,11 @@ pub fn default_sql_aggregate(
 
 pub use dialects::AnsiDialect;
 
+/// A shared dialect instance. Dialects are stateless unit structs, so
+/// registry lookups hand out `'static` references rather than boxing a
+/// fresh trait object per call.
+pub type DialectRef = &'static (dyn SqlDialect + Send + Sync);
+
 /// Fail fast when a spatial feature is used on a backend whose dialect
 /// reports `supports_spatial() == false`, rather than emitting spatial SQL
 /// the backend cannot run.
diff --git a/src/reader/odbc/mod.rs b/src/reader/odbc/mod.rs
index 5933744a8..8cb3cbdfe 100644
--- a/src/reader/odbc/mod.rs
+++ b/src/reader/odbc/mod.rs
@@ -22,10 +22,9 @@ use wrapper::{Connection, Statement};
 ///
 /// Delegates to the shared matcher in [`crate::reader::dialects`]; the
 /// `Driver=` value from the ODBC connection string serves as the driver hint.
-fn detect_dialect(dbms_name: Option<&str>, conn_str: &str) -> Result<Box<dyn super::SqlDialect>> {
+fn detect_dialect(dbms_name: Option<&str>, conn_str: &str) -> Result<super::DialectRef> {
     let driver = super::connection::extract_odbc_value(conn_str, "driver");
     super::registry::detect_or_err(dbms_name, driver.as_deref())
-        .map(|d| d as Box<dyn super::SqlDialect>)
 }
 
 /// Pull ggsql-owned keys (`Dialect=<name>`) out of an ODBC connection
@@ -39,9 +38,10 @@ fn take_dialect_override(conn_str: &str) -> (String, Option<String>) {
 /// Generic ODBC reader implementing the `Reader` trait.
 pub struct OdbcReader {
     connection: Connection,
-    dialect: Box<dyn super::SqlDialect>,
+    dialect: super::DialectRef,
     registered_tables: crate::reader::RegisteredTables,
     batch_size: usize,
+    numeric_as_double: bool,
 }
 
 // Safety: ODBC connections are safe to use from one thread at a time.
@@ -63,10 +63,7 @@ impl OdbcReader {
     /// by the connection, falling back to the `Driver=` value in the
     /// connection string, then ANSI. Pass `Some(...)` to pin the dialect
     /// (e.g. from a backend-specific ggsql URI scheme).
-    pub fn from_odbc_conn_str(
-        conn_str: &str,
-        dialect: Option<Box<dyn super::SqlDialect>>,
-    ) -> Result<Self> {
+    pub fn from_odbc_conn_str(conn_str: &str, dialect: Option<super::DialectRef>) -> Result<Self> {
         ffi::try_load()
             .map_err(|e| GgsqlError::ReaderError(format!("ODBC is not available: {}", e)))?;
 
@@ -80,7 +77,6 @@ impl OdbcReader {
             (d @ Some(_), _) => d,
             (None, Some(name)) => Some(
                 crate::reader::registry::dialect_override(&name)
-                    .map(|d| d as Box<dyn super::SqlDialect>)
                     .ok_or_else(|| crate::reader::registry::unknown_dialect_error(&name))?,
             ),
             (None, None) => None,
@@ -120,22 +116,25 @@ impl OdbcReader {
             None => detect_dialect(dbms_name.as_deref(), &conn_str)?,
         };
 
-        // Fetch batch size is a per-backend capability (registry
-        // `odbc_row_array_size`): Oracle ODBC rejects block cursors
-        // (SQL_ATTR_ROW_ARRAY_SIZE > 1) with HY090 at SQLFetch time, and the
-        // failed fetch leaves the cursor unusable.
-        let batch_size = super::registry::detect(
+        // Fetch behavior is a per-backend capability: Oracle ODBC rejects
+        // block cursors (`odbc_row_array_size`; HY090 at SQLFetch time, and
+        // the failed fetch leaves the cursor unusable) and cannot convert
+        // DECIMAL to integer C types (`odbc_numeric_as_double`).
+        let entry = super::registry::detect(
             dbms_name.as_deref(),
             super::connection::extract_odbc_value(&conn_str, "driver").as_deref(),
-        )
-        .and_then(|e| e.odbc_row_array_size)
-        .unwrap_or(BATCH_SIZE);
+        );
+        let batch_size = entry
+            .and_then(|e| e.odbc_row_array_size)
+            .unwrap_or(BATCH_SIZE);
+        let numeric_as_double = entry.is_some_and(|e| e.odbc_numeric_as_double);
 
         Ok(Self {
             connection,
             dialect,
             registered_tables: crate::reader::RegisteredTables::new(),
             batch_size,
+            numeric_as_double,
         })
     }
 }
@@ -148,7 +147,7 @@ impl Reader for OdbcReader {
             return Ok(DataFrame::empty());
         };
 
-        cursor_to_dataframe(cursor, self.batch_size).map_err(|e| {
+        cursor_to_dataframe(cursor, self.batch_size, self.numeric_as_double).map_err(|e| {
             let snippet: String = sql.chars().take(200).collect();
             GgsqlError::ReaderError(format!("{e} [statement: {snippet}]"))
         })
@@ -274,13 +273,13 @@ impl Reader for OdbcReader {
     }
 
     fn dialect(&self) -> &dyn super::SqlDialect {
-        &*self.dialect
+        self.dialect
     }
 
     fn list_catalogs(&self) -> Result<Vec<String>> {
         // ODBC spec: CatalogName="%", SchemaName="", TableName=""
         let stmt = wrapper::sql_tables(&self.connection, Some("%"), Some(""), Some(""), None)?;
-        let df = cursor_to_dataframe(stmt, self.batch_size)?;
+        let df = cursor_to_dataframe(stmt, self.batch_size, self.numeric_as_double)?;
         let mut catalogs = extract_string_column_ci(&df, "TABLE_CAT")?;
         catalogs.sort();
         catalogs.dedup();
@@ -290,7 +289,7 @@ impl Reader for OdbcReader {
     fn list_schemas(&self, _catalog: &str) -> Result<Vec<String>> {
         // ODBC spec: CatalogName="", SchemaName="%", TableName=""
         let stmt = wrapper::sql_tables(&self.connection, Some(""), Some("%"), Some(""), None)?;
-        let df = cursor_to_dataframe(stmt, self.batch_size)?;
+        let df = cursor_to_dataframe(stmt, self.batch_size, self.numeric_as_double)?;
         let mut schemas = extract_string_column_ci(&df, "TABLE_SCHEM")?;
         schemas.sort();
         schemas.dedup();
@@ -309,7 +308,7 @@ impl Reader for OdbcReader {
             Some(schema)
         };
         let stmt = wrapper::sql_tables(&self.connection, cat, sch, Some("%"), Some("TABLE,VIEW"))?;
-        let df = cursor_to_dataframe(stmt, self.batch_size)?;
+        let df = cursor_to_dataframe(stmt, self.batch_size, self.numeric_as_double)?;
         extract_table_infos_ci(&df)
     }
 
@@ -330,7 +329,7 @@ impl Reader for OdbcReader {
             Some(schema)
         };
         let stmt = wrapper::sql_columns(&self.connection, cat, sch, Some(table), None)?;
-        let df = cursor_to_dataframe(stmt, self.batch_size)?;
+        let df = cursor_to_dataframe(stmt, self.batch_size, self.numeric_as_double)?;
         extract_column_infos_ci(&df)
     }
 }
@@ -419,8 +418,9 @@ enum ColumnBuilder {
 impl ColumnBuilder {
     fn from_sql_type(
         sql_type: SqlSmallInt,
-        _col_size: SqlULen,
-        _decimal_digits: SqlSmallInt,
+        col_size: SqlULen,
+        decimal_digits: SqlSmallInt,
+        numeric_as_double: bool,
     ) -> Self {
         match sql_type {
             SQL_TINYINT => Self::Int8(Vec::new()),
@@ -429,10 +429,16 @@ impl ColumnBuilder {
             SQL_BIGINT => Self::Int64(Vec::new()),
             SQL_REAL => Self::Float32(Vec::new()),
             SQL_DOUBLE | SQL_FLOAT => Self::Float64(Vec::new()),
-            // Bind as double rather than a sized integer: Oracle ODBC fails
-            // with HY090 when converting SQL_DECIMAL to SQL_C_SLONG/SBIGINT,
-            // and numeric→double conversion is supported by every driver.
-            // Values are bound for plot coordinates (f64) anyway.
+            // Integer-scale numerics keep an integer type (and their
+            // precision past 2^53) — except where the driver cannot convert
+            // DECIMAL to an integer C type (Oracle fails with HY090; see the
+            // registry's `odbc_numeric_as_double`). >15 digits cannot be
+            // trusted to Int64, so those bind as double too.
+            SQL_NUMERIC | SQL_DECIMAL
+                if !numeric_as_double && decimal_digits == 0 && (1..=15).contains(&col_size) =>
+            {
+                Self::Int64(Vec::new())
+            }
             SQL_NUMERIC | SQL_DECIMAL => Self::Float64(Vec::new()),
             SQL_BIT => Self::Boolean(Vec::new()),
             SQL_TYPE_DATE => Self::Date(Vec::new()),
@@ -536,6 +542,10 @@ const BATCH_SIZE: usize = 1000;
 /// silently truncated; 16 KiB keeps a 1000-row batch of one text column at
 /// 16 MiB.
 const DEFAULT_TEXT_BUF_SIZE: usize = 16384;
+/// Cap on the per-cell text buffer sized from the driver-reported column
+/// size, bounding batch memory (a 1000-row batch of one text column at the
+/// cap is ~1 GiB; values beyond the buffer size are an explicit error).
+const MAX_TEXT_BUF_SIZE: usize = 1024 * 1024;
 
 struct ColumnBuffer {
     data: Vec<u8>,
@@ -543,7 +553,11 @@ struct ColumnBuffer {
     text_buf_size: usize,
 }
 
-fn cursor_to_dataframe(stmt: Statement, batch_size: usize) -> Result<DataFrame> {
+fn cursor_to_dataframe(
+    stmt: Statement,
+    batch_size: usize,
+    numeric_as_double: bool,
+) -> Result<DataFrame> {
     let col_count = stmt.num_result_cols()?;
     if col_count == 0 {
         return Ok(DataFrame::empty());
@@ -551,15 +565,18 @@ fn cursor_to_dataframe(stmt: Statement, batch_size: usize) -> Result<DataFrame>
 
     // Describe all columns
     let mut col_names = Vec::with_capacity(col_count);
+    let mut col_sizes = Vec::with_capacity(col_count);
     let mut builders = Vec::with_capacity(col_count);
 
     for i in 1..=col_count as u16 {
         let (name, data_type, col_size, decimal_digits, _nullable) = stmt.describe_col(i)?;
         col_names.push(name);
+        col_sizes.push(col_size);
         builders.push(ColumnBuilder::from_sql_type(
             data_type,
             col_size,
             decimal_digits,
+            numeric_as_double,
         ));
     }
 
@@ -574,9 +591,22 @@ fn cursor_to_dataframe(stmt: Statement, batch_size: usize) -> Result<DataFrame>
         .enumerate()
         .map(|(i, builder)| {
             let (elem_size, text_buf_size) = if matches!(builder, ColumnBuilder::Text(_)) {
-                // +2: null terminator, and keep the bind length even — some
-                // Unicode drivers (Oracle) reject odd buffer lengths with HY090
-                (DEFAULT_TEXT_BUF_SIZE + 2, DEFAULT_TEXT_BUF_SIZE)
+                // Size the per-cell buffer from the driver-reported column
+                // size when available (×4 for UTF-8 expansion), so a
+                // VARCHAR(4000) doesn't pay the default and long values in
+                // sized columns fit. Unsized columns (LONG/CLOB) get the
+                // default; anything longer is an explicit error, never a
+                // silent truncation. +2: null terminator, and keep the bind
+                // length even — some Unicode drivers (Oracle) reject odd
+                // buffer lengths with HY090.
+                let size = if col_sizes[i] > 0 {
+                    col_sizes[i]
+                        .saturating_mul(4)
+                        .clamp(DEFAULT_TEXT_BUF_SIZE, MAX_TEXT_BUF_SIZE)
+                } else {
+                    DEFAULT_TEXT_BUF_SIZE
+                };
+                (size + 2, size)
             } else {
                 (builder.element_size(), 0)
             };
@@ -838,78 +868,92 @@ mod tests {
     #[test]
     fn test_column_builder_from_sql_type() {
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_TINYINT, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_TINYINT, 0, 0, false),
             ColumnBuilder::Int8(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_SMALLINT, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_SMALLINT, 0, 0, false),
             ColumnBuilder::Int16(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_INTEGER, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_INTEGER, 0, 0, false),
             ColumnBuilder::Int32(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_BIGINT, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_BIGINT, 0, 0, false),
             ColumnBuilder::Int64(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_REAL, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_REAL, 0, 0, false),
             ColumnBuilder::Float32(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_DOUBLE, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_DOUBLE, 0, 0, false),
             ColumnBuilder::Float64(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_FLOAT, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_FLOAT, 0, 0, false),
             ColumnBuilder::Float64(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_BIT, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_BIT, 0, 0, false),
             ColumnBuilder::Boolean(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_TYPE_DATE, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_TYPE_DATE, 0, 0, false),
             ColumnBuilder::Date(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_TYPE_TIME, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_TYPE_TIME, 0, 0, false),
             ColumnBuilder::Time(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_TYPE_TIMESTAMP, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_TYPE_TIMESTAMP, 0, 0, false),
             ColumnBuilder::Timestamp(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_VARCHAR, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_VARCHAR, 0, 0, false),
             ColumnBuilder::Text(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_WVARCHAR, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_WVARCHAR, 0, 0, false),
             ColumnBuilder::Text(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_LONGVARCHAR, 0, 0),
+            ColumnBuilder::from_sql_type(SQL_LONGVARCHAR, 0, 0, false),
             ColumnBuilder::Text(_)
         ));
-        // Decimals bind as double regardless of precision/scale: Oracle ODBC
-        // rejects SQL_DECIMAL → SQL_C_SBIGINT conversion with HY090
+        // Integer-scale numerics bind as Int64 where the driver supports
+        // it; past 15 digits, with a fractional scale, or with
+        // numeric_as_double (Oracle rejects SQL_DECIMAL → SQL_C_SBIGINT
+        // with HY090), they bind as double.
+        assert!(matches!(
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 5, 0, false),
+            ColumnBuilder::Int64(_)
+        ));
+        assert!(matches!(
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 15, 0, false),
+            ColumnBuilder::Int64(_)
+        ));
+        assert!(matches!(
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 20, 0, false),
+            ColumnBuilder::Float64(_)
+        ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_NUMERIC, 5, 0),
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 10, 2, false),
             ColumnBuilder::Float64(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_NUMERIC, 15, 0),
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 5, 0, true),
             ColumnBuilder::Float64(_)
         ));
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_NUMERIC, 25, 0),
+            ColumnBuilder::from_sql_type(SQL_NUMERIC, 25, 0, false),
             ColumnBuilder::Float64(_)
         ));
         // Decimal with scale>0 maps to Float64
         assert!(matches!(
-            ColumnBuilder::from_sql_type(SQL_DECIMAL, 10, 2),
+            ColumnBuilder::from_sql_type(SQL_DECIMAL, 10, 2, false),
             ColumnBuilder::Float64(_)
         ));
     }
diff --git a/src/reader/registry.rs b/src/reader/registry.rs
index 996df0f41..0b079acf7 100644
--- a/src/reader/registry.rs
+++ b/src/reader/registry.rs
@@ -6,9 +6,29 @@
 //! patterns, dialect constructor, ADBC driver details, and ODBC synthesis
 //! hints. Dispatch (`connection.rs`), the ADBC reader, and the Jupyter
 //! kernel all consult this table rather than keeping their own lookups.
+//!
+//! # Adding a dialect
+//!
+//! 1. Create `src/reader/dialects/<scheme>.rs` with a unit struct
+//!    implementing [`SqlDialect`](super::SqlDialect), and register the
+//!    module in `dialects/mod.rs`.
+//! 2. Add a [`DatabaseEntry`] below (schemes, detection patterns, ADBC
+//!    info, ODBC quirks). Detection tests live in this file.
+//! 3. Add the golden file: `GGSQL_BLESS=1 cargo test -p ggsql --lib
+//!    golden::<scheme>` (see `src/execute/golden.rs`); the
+//!    `goldens_cover_registry` test enforces this.
+//! 4. Live tests (`src/tests/dialect_live.rs`): a `live_<scheme>` test, a
+//!    `GGSQL_TEST_URI_<SCHEME>` doc entry, DDL branches in
+//!    `create_table_sql`/`date_literal`/`ddl_quote`/`insert_sql`, and a
+//!    startup script under `.github/scripts/live/` plus a leg in
+//!    `.github/workflows/dialect-live.yml`. Cases that cannot run live get
+//!    a `live_skip` with a reason in `src/tests/battery/mod.rs`.
+//! 5. If the backend cannot host ggsql's internal tables, set
+//!    `requires_cache`/`temp_table_style` on the dialect instead of
+//!    special-casing readers.
 
 use super::dialects::*;
-use super::{AnsiDialect, SqlDialect};
+use super::AnsiDialect;
 
 /// How the URI handed to an ADBC driver is derived from the ggsql URI.
 ///
@@ -132,8 +152,10 @@ pub struct DatabaseEntry {
     /// order, most specific first (SQL Server before anything containing
     /// "sql", Redshift before Postgres).
     pub detect: &'static [DetectPattern],
-    /// Construct the backend's SQL dialect.
-    dialect: fn() -> Box<dyn SqlDialect + Send>,
+    /// The backend's SQL dialect. Dialects are stateless unit structs, so
+    /// the registry shares one static instance rather than boxing per
+    /// lookup.
+    dialect: crate::reader::DialectRef,
     /// ADBC driver details; `None` when no usable dedicated driver exists
     /// (Drill, MonetDB) — dispatch falls through to ODBC.
     pub adbc: Option<AdbcInfo>,
@@ -146,6 +168,11 @@ pub struct DatabaseEntry {
     /// (SQL_ATTR_ROW_ARRAY_SIZE > 1) with HY090 at SQLFetch time, and the
     /// failed fetch leaves the cursor unusable — fetch row-by-row (1).
     pub odbc_row_array_size: Option<usize>,
+    /// Bind NUMERIC/DECIMAL as double even when scale is 0 and the value
+    /// fits an integer: Oracle ODBC fails with HY090 converting SQL_DECIMAL
+    /// to SQL_C_SLONG/SBIGINT. Elsewhere integer-scale numerics keep their
+    /// Int64 type (and precision past 2^53).
+    pub odbc_numeric_as_double: bool,
     /// In-process reader to prefer when its cargo feature is compiled in
     /// (duckdb, sqlite). `None` for backends reached only through external
     /// ADBC/ODBC drivers.
@@ -153,9 +180,9 @@ pub struct DatabaseEntry {
 }
 
 impl DatabaseEntry {
-    /// The backend's SQL dialect, freshly boxed.
-    pub fn dialect(&self) -> Box<dyn SqlDialect + Send> {
-        (self.dialect)()
+    /// The backend's SQL dialect (shared static instance).
+    pub fn dialect(&self) -> crate::reader::DialectRef {
+        self.dialect
     }
 
     /// All URI schemes naming this backend, canonical first.
@@ -175,6 +202,7 @@ macro_rules! entry {
             adbc: $adbc,
             odbc_dbq_style: false,
             odbc_row_array_size: None,
+            odbc_numeric_as_double: false,
             native_reader: None,
         }
     };
@@ -221,7 +249,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
             Has("sqlserver"),
             Has("mssql"),
         ],
-        || Box::new(MssqlDialect),
+        &MssqlDialect,
         adbc!("adbc_driver_mssql", "mssql", Passthrough, false)
     ),
     entry!(
@@ -229,7 +257,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Redshift",
         &[Has("redshift")],
-        || Box::new(RedshiftDialect),
+        &RedshiftDialect,
         adbc!(
             "adbc_driver_postgresql",
             "postgresql",
@@ -242,7 +270,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &["postgresql"],
         "PostgreSQL",
         &[Has("postgres"), Has("psql")],
-        || Box::new(PostgresDialect),
+        &PostgresDialect,
         adbc!("adbc_driver_postgresql", "postgresql", Passthrough, true)
     ),
     entry!(
@@ -250,7 +278,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &["mariadb"],
         "MySQL",
         &[Has("mariadb"), Has("mysql")],
-        || Box::new(MySqlDialect),
+        &MySqlDialect,
         adbc!("adbc_driver_mysql", "mysql", MySqlGoDsn, true)
     ),
     entry!(
@@ -258,7 +286,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Snowflake",
         &[Has("snowflake")],
-        || Box::new(SnowflakeDialect),
+        &SnowflakeDialect,
         adbc!("adbc_driver_snowflake", "snowflake", Passthrough, true)
     ),
     entry!(
@@ -266,7 +294,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "BigQuery",
         &[Has("bigquery")],
-        || Box::new(BigQueryDialect),
+        &BigQueryDialect,
         adbc!(
             "adbc_driver_bigquery",
             "bigquery",
@@ -280,7 +308,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &["spark"],
         "Databricks",
         &[Has("databricks"), Has("spark")],
-        || Box::new(DatabricksDialect),
+        &DatabricksDialect,
         adbc!(
             "adbc_driver_databricks",
             "databricks",
@@ -294,18 +322,19 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "ClickHouse",
         &[Has("clickhouse")],
-        || Box::new(ClickHouseDialect),
+        &ClickHouseDialect,
         adbc!("adbc_driver_clickhouse", "clickhouse", ClickHouseHttp, true)
     ),
     DatabaseEntry {
         odbc_dbq_style: true,
         odbc_row_array_size: Some(1),
+        odbc_numeric_as_double: true,
         ..entry!(
             "oracle",
             &[],
             "Oracle",
             &[Has("oracle"), DetectPattern::All(&["ora", "driver"])],
-            || Box::new(OracleDialect),
+            &OracleDialect,
             adbc!("adbc_driver_oracle", "oracle", Passthrough, true)
         )
     },
@@ -314,7 +343,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Trino",
         &[Has("trino")],
-        || Box::new(TrinoDialect),
+        &TrinoDialect,
         adbc!("adbc_driver_trino", "trino", Passthrough, true)
     ),
     entry!(
@@ -322,7 +351,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Exasol",
         &[Has("exasol"), Has("exaodbc")],
-        || Box::new(ExasolDialect),
+        &ExasolDialect,
         adbc!("adbc_driver_exasol", "exasol", Passthrough, true)
     ),
     entry!(
@@ -330,7 +359,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "MonetDB",
         &[Has("monet")],
-        || Box::new(MonetDbDialect),
+        &MonetDbDialect,
         None
     ),
     entry!(
@@ -338,7 +367,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Apache Druid",
         &[Has("druid")],
-        || Box::new(DruidDialect),
+        &DruidDialect,
         adbc!("adbc_driver_druid", "druid", Passthrough, false)
     ),
     entry!(
@@ -346,7 +375,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Apache Drill",
         &[Has("drill")],
-        || Box::new(DrillDialect),
+        &DrillDialect,
         None
     ),
     entry!(
@@ -354,7 +383,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "DataFusion",
         &[Has("datafusion")],
-        || Box::new(DataFusionDialect),
+        &DataFusionDialect,
         adbc!(
             "adbc_driver_datafusion",
             "datafusion",
@@ -370,7 +399,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
             &[],
             "DuckDB",
             &[Has("duckdb")],
-            || Box::new(DuckDbDialect),
+            &DuckDbDialect,
             adbc!("adbc_driver_duckdb", "duckdb", Passthrough, true)
         )
     },
@@ -381,7 +410,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
             &[],
             "SQLite",
             &[Has("sqlite")],
-            || Box::new(SqliteDialect),
+            &SqliteDialect,
             adbc!("adbc_driver_sqlite", "sqlite", Passthrough, true)
         )
     },
@@ -393,7 +422,7 @@ pub static REGISTRY: &[DatabaseEntry] = &[
         &[],
         "Flight SQL",
         &[Has("flightsql"), DetectPattern::All(&["flight", "sql"])],
-        || Box::new(AnsiDialect),
+        &AnsiDialect,
         adbc!("adbc_driver_flightsql", "flightsql", Passthrough, true)
     ),
 ];
@@ -463,9 +492,9 @@ pub fn odbc_driver_env_var(scheme: &str) -> String {
 /// Resolve a `dialect=` override value (`ansi` or any registry scheme) to a
 /// dialect. This is the explicit escape hatch for backends ggsql doesn't
 /// know: unknown backends are an error unless the user pins a dialect.
-pub fn dialect_override(name: &str) -> Option<Box<dyn SqlDialect + Send>> {
+pub fn dialect_override(name: &str) -> Option<crate::reader::DialectRef> {
     if name.eq_ignore_ascii_case("ansi") {
-        return Some(Box::new(AnsiDialect));
+        return Some(&AnsiDialect);
     }
     by_scheme(name).map(|e| e.dialect())
 }
@@ -486,7 +515,7 @@ pub fn unknown_dialect_error(name: &str) -> crate::GgsqlError {
 /// ANSI fallback.
 pub fn resolve_dialect(
     conn: &crate::reader::connection::ConnUri,
-) -> crate::Result<Box<dyn SqlDialect + Send>> {
+) -> crate::Result<crate::reader::DialectRef> {
     if let Some(name) = &conn.ggsql.dialect {
         return dialect_override(name).ok_or_else(|| unknown_dialect_error(name));
     }
@@ -514,7 +543,7 @@ pub fn resolve_dialect(
 pub fn detect_or_err(
     dbms_name: Option<&str>,
     driver_hint: Option<&str>,
-) -> crate::Result<Box<dyn SqlDialect + Send>> {
+) -> crate::Result<crate::reader::DialectRef> {
     detect(dbms_name, driver_hint)
         .map(|e| e.dialect())
         .ok_or_else(|| {
diff --git a/src/reader/test_support.rs b/src/reader/test_support.rs
index 34f7bf44c..408f27f70 100644
--- a/src/reader/test_support.rs
+++ b/src/reader/test_support.rs
@@ -149,7 +149,7 @@ impl Reader for ReadOnlyReader {
 /// `Float64`, which is the right shape for the derived columns
 /// (bins, densities, quantiles) the pipeline produces.
 pub(crate) struct StubReader {
-    dialect: Box<dyn SqlDialect + Send>,
+    dialect: crate::reader::DialectRef,
     tables: Mutex<HashMap<String, Arc<Schema>>>,
     log: Arc<Mutex<Vec<String>>>,
 }
@@ -157,7 +157,7 @@ pub(crate) struct StubReader {
 impl StubReader {
     /// Create a stub for `dialect`, returning it together with a
     /// handle to the shared SQL log.
-    pub(crate) fn new(dialect: Box<dyn SqlDialect + Send>) -> (Self, Arc<Mutex<Vec<String>>>) {
+    pub(crate) fn new(dialect: crate::reader::DialectRef) -> (Self, Arc<Mutex<Vec<String>>>) {
         let log = Arc::new(Mutex::new(Vec::new()));
         (
             Self {
@@ -419,7 +419,7 @@ impl Reader for StubReader {
         execute_with_reader(self, query)
     }
     fn dialect(&self) -> &dyn SqlDialect {
-        &*self.dialect
+        self.dialect
     }
 }
 
@@ -739,3 +739,66 @@ pub(crate) fn assert_dataframes_equal(a: &DataFrame, b: &DataFrame, ctx: &str) {
         );
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn select_list_span_basic() {
+        let (list, table) = select_list_span("SELECT a, b AS c FROM t WHERE x").unwrap();
+        assert_eq!(list, "a, b AS c");
+        assert_eq!(table.as_deref(), Some("t"));
+    }
+
+    #[test]
+    fn select_list_span_skips_nested_select_and_keywords_in_literals() {
+        // The inner SELECT/FROM (parenthesized) and the 'from' inside the
+        // string literal must not be seen as top-level keywords.
+        let (list, table) =
+            select_list_span("SELECT x, 'from nowhere' AS s FROM (SELECT y FROM inner_t) AS sub")
+                .unwrap();
+        assert_eq!(list, "x, 'from nowhere' AS s");
+        assert_eq!(table.as_deref(), None);
+    }
+
+    #[test]
+    fn select_list_span_strips_distinct_and_top() {
+        let (list, _) = select_list_span("SELECT DISTINCT a FROM t").unwrap();
+        assert_eq!(list, "a");
+    }
+
+    #[test]
+    fn split_commas_respects_parens_and_quotes() {
+        // Items are not trimmed; callers trim.
+        assert_eq!(
+            split_top_level_commas("a, f(1, (2)), 'x,y', \"z,w\""),
+            vec!["a", " f(1, (2))", " 'x,y'", " \"z,w\""]
+        );
+    }
+
+    #[test]
+    fn output_name_alias_and_plain_identifiers() {
+        assert_eq!(output_name("expr + 1 AS total"), Some("total".into()));
+        assert_eq!(
+            output_name("COUNT(*) AS \"my Count\""),
+            Some("my Count".into())
+        );
+        assert_eq!(output_name("t.col"), Some("col".into()));
+        assert_eq!(output_name("`bt`"), Some("bt".into()));
+        // Unaliased expressions have no name.
+        assert_eq!(output_name("a + b"), None);
+    }
+
+    #[test]
+    fn stub_fabricates_columns_for_generated_sql() {
+        let (stub, _log) = StubReader::new(&crate::reader::AnsiDialect);
+        let df = stub
+            .execute_sql("SELECT a AS one, COUNT(*) AS n FROM anything")
+            .unwrap();
+        let schema = df.schema();
+        assert_eq!(schema.fields().len(), 2);
+        assert_eq!(schema.field(0).name(), "one");
+        assert_eq!(schema.field(1).name(), "n");
+    }
+}
diff --git a/src/tests/battery/mod.rs b/src/tests/battery/mod.rs
index 17dd9e7cb..d44b0eeb5 100644
--- a/src/tests/battery/mod.rs
+++ b/src/tests/battery/mod.rs
@@ -4,8 +4,9 @@
 //! the case list and dataset exist exactly once; it lives in a
 //! subdirectory so cargo does not treat it as its own test target.
 //!
-//! Pure data and std-only helpers — no crate imports — so the file compiles
-//! in both including contexts (lib unit tests and the integration test).
+//! Data plus helpers over `arrow`/`crate::DataFrame` (`fixture_df`); both
+//! including contexts are targets of this crate, so those deps resolve in
+//! either one.
 
 // Each including tier reads only its own subset: the golden tests ignore
 // the live assertions (Expect, live_skip, live, runs_live_on), and the
@@ -142,6 +143,55 @@ pub enum Fixture {
     TextDay,
 }
 
+/// The fixture as a DataFrame, matching the live table's schema:
+/// id/val/grp/day(a real date)/"mixed Case". `TextDay` swaps the date for
+/// ISO text. Used by the golden tier's stub registrations and by live
+/// backends whose setup goes through `register()` rather than DDL
+/// (DataFusion). BigQuery is the exception: its ADBC register path needs
+/// Int64 ids, so `dialect_live.rs` builds that batch by hand.
+pub fn fixture_df(fixture: &Fixture) -> crate::DataFrame {
+    use arrow::array::{ArrayRef, Date32Array, Float64Array, Int32Array, StringArray};
+    use std::sync::Arc;
+    let day: ArrayRef = match fixture {
+        Fixture::Shared => Arc::new(Date32Array::from(
+            ROWS.iter().map(|r| r.day_epoch).collect::<Vec<_>>(),
+        )),
+        Fixture::TextDay => Arc::new(StringArray::from(
+            ROWS.iter()
+                .map(|r| day_iso(r.day_epoch))
+                .collect::<Vec<_>>(),
+        )),
+    };
+    crate::DataFrame::new(vec![
+        (
+            "id",
+            Arc::new(Int32Array::from(
+                ROWS.iter().map(|r| r.id).collect::<Vec<_>>(),
+            )) as ArrayRef,
+        ),
+        (
+            "val",
+            Arc::new(Float64Array::from(
+                ROWS.iter().map(|r| r.val).collect::<Vec<_>>(),
+            )) as ArrayRef,
+        ),
+        (
+            "grp",
+            Arc::new(StringArray::from(
+                ROWS.iter().map(|r| r.grp).collect::<Vec<_>>(),
+            )) as ArrayRef,
+        ),
+        ("day", day),
+        (
+            "mixed Case",
+            Arc::new(Float64Array::from(
+                ROWS.iter().map(|r| r.mixed).collect::<Vec<_>>(),
+            )) as ArrayRef,
+        ),
+    ])
+    .unwrap()
+}
+
 pub struct Case {
     pub name: &'static str,
     /// Query template; `{table}` is substituted by the harness.
diff --git a/src/tests/dialect_live.rs b/src/tests/dialect_live.rs
index 5fd064455..72e30b22d 100644
--- a/src/tests/dialect_live.rs
+++ b/src/tests/dialect_live.rs
@@ -59,6 +59,9 @@
 //! The DataFusion case runs in-process via the Foundry ADBC driver and
 //! needs no container, so one non-DuckDB engine always runs in CI.
 //!
+//! Drill has no live leg at all: it is covered by golden SQL-shape tests
+//! only (there is no GGSQL_TEST_URI_DRILL).
+//!
 //! The battery itself — case list, per-case assertions, and the dataset —
 //! lives in `src/tests/battery/mod.rs`, shared with the Tier 1 golden
 //! tests (`src/execute/golden.rs`) so the two tiers cannot drift. This
@@ -459,6 +462,7 @@ fn live_backend(scheme: &str) {
                 )),
             ],
         )
+        // Not battery::fixture_df: BigQuery's ADBC register path needs Int64 ids.
         .expect("build bigquery test batch");
         reader
             .register(table, ggsql::DataFrame::from_record_batch(batch), true)
@@ -629,46 +633,10 @@ fn live_datafusion() {
         return;
     }
     use ggsql::reader::adbc::AdbcReader;
-    use std::sync::Arc;
 
     let reader = AdbcReader::from_connection_string("datafusion://").expect("datafusion init");
 
-    let df = ggsql::DataFrame::new(vec![
-        (
-            "id",
-            Arc::new(arrow::array::Int32Array::from(
-                battery::ROWS.iter().map(|r| r.id).collect::<Vec<_>>(),
-            )) as arrow::array::ArrayRef,
-        ),
-        (
-            "val",
-            Arc::new(arrow::array::Float64Array::from(
-                battery::ROWS.iter().map(|r| r.val).collect::<Vec<_>>(),
-            )) as arrow::array::ArrayRef,
-        ),
-        (
-            "grp",
-            Arc::new(arrow::array::StringArray::from(
-                battery::ROWS.iter().map(|r| r.grp).collect::<Vec<_>>(),
-            )) as arrow::array::ArrayRef,
-        ),
-        (
-            "day",
-            Arc::new(arrow::array::Date32Array::from(
-                battery::ROWS
-                    .iter()
-                    .map(|r| r.day_epoch)
-                    .collect::<Vec<_>>(),
-            )) as arrow::array::ArrayRef,
-        ),
-        (
-            "mixed Case",
-            Arc::new(arrow::array::Float64Array::from(
-                battery::ROWS.iter().map(|r| r.mixed).collect::<Vec<_>>(),
-            )) as arrow::array::ArrayRef,
-        ),
-    ])
-    .expect("test dataframe");
+    let df = battery::fixture_df(&battery::Fixture::Shared);
     reader.register(TABLE, df, true).expect("register table");
 
     run_battery(&reader, "datafusion", TABLE);