diff --git a/hypaware-core/plugins-workspace/ai-gateway/src/dataset.js b/hypaware-core/plugins-workspace/ai-gateway/src/dataset.js
index df30050a..9ce48c5a 100644
--- a/hypaware-core/plugins-workspace/ai-gateway/src/dataset.js
+++ b/hypaware-core/plugins-workspace/ai-gateway/src/dataset.js
@@ -10,7 +10,7 @@ import { AI_GATEWAY_MESSAGE_COLUMNS, aiGatewayRowsFromProjectedExchange } from '
 import { isPlainObject, stringValue } from 'hypaware/core/util'
 
 /**
- * @import { AiGatewayProjectedExchange, BackfillItem, BackfillMaterializeContext, BackfillMaterializerContribution, CachePartitionMeta, ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, DatasetSettleContext, QueryPartition, QueryStorageService } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { AiGatewayProjectedExchange, BackfillItem, BackfillMaterializeContext, BackfillMaterializerContribution, CachePartitionMeta, ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, DatasetSettleContext, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
  * @import { GatewayState } from './types.js'
  * @import { AsyncDataSource } from 'squirreling'
@@ -120,6 +120,7 @@ export async function refreshPartition() {
  *
  * @param {QueryPartition[]} partitions
  * @param {DatasetDataSourceContext} ctx
+ * @returns {Promise<ScannableDataSource>}
  */
 export async function createDataSource(partitions, ctx) {
   const storage = /** @type {ExtendedQueryStorageService} */ (ctx.storage)
@@ -137,7 +138,7 @@ export async function createDataSource(partitions, ctx) {
     tablePaths.add(p.path)
   }
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
@@ -169,12 +170,12 @@ const SCHEMA_COLUMN_NAMES = AI_GATEWAY_SCHEMA_COLUMNS.map((c) => c.name)
  * (LLP 0015#multi-partition-union).
  *
  * @ref LLP 0032#capture [implements]: additive columns stay queryable over old partitions; no partition-label bump / cache wipe needed
- * @param {AsyncDataSource} source
- * @returns {AsyncDataSource}
+ * @param {ScannableDataSource} source
+ * @returns {ScannableDataSource}
  */
 function withSchemaColumns(source) {
   const columns = Array.from(new Set([...source.columns, ...SCHEMA_COLUMN_NAMES]))
-  /** @type {AsyncDataSource} */
+  /** @type {ScannableDataSource} */
   const wrapped = {
     columns,
     numRows: source.numRows,
@@ -236,6 +237,20 @@ function withSchemaColumns(source) {
       }
     }
   }
+  // Native batches are transparent only when the physical prepared schema
+  // already covers the full declared schema. A drifted source stays on the
+  // row/column paths above, which own its absent-column semantics.
+  // @ref LLP 0294#schema-drift [implements]: prepared batches never invent a value for a declared-but-absent field
+  if (source.schema && source.prepareScan) {
+    const prepareScan = source.prepareScan
+    const fieldsByName = new Map(source.schema.fields.map((field) => [field.name, field]))
+    if (columns.every((column) => fieldsByName.has(column))) {
+      wrapped.schema = {
+        fields: columns.map((column) => /** @type {NonNullable<ReturnType<typeof fieldsByName.get>>} */ (fieldsByName.get(column))),
+      }
+      wrapped.prepareScan = (request) => prepareScan.call(source, request)
+    }
+  }
   return wrapped
 }
 
diff --git a/hypaware-core/plugins-workspace/claude/src/telemetry/events_dataset.js b/hypaware-core/plugins-workspace/claude/src/telemetry/events_dataset.js
index 17a3fb4a..2447d915 100644
--- a/hypaware-core/plugins-workspace/claude/src/telemetry/events_dataset.js
+++ b/hypaware-core/plugins-workspace/claude/src/telemetry/events_dataset.js
@@ -8,10 +8,9 @@ import { BODY_EVENT_NAMES } from './bodies.js'
 import { CONTENT_EVENT_NAMES } from './projection.js'
 
 /**
- * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService } from '../../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../../src/core/cache/types.js'
  * @import { ClaudeTelemetryEvent } from '../types.js'
- * @import { AsyncDataSource } from 'squirreling'
  */
 
 const PLUGIN_NAME = '@hypaware/claude'
@@ -267,7 +266,7 @@ async function createDataSource(partitions, ctx) {
   }
   for (const p of fresh) tablePaths.add(p.path)
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
diff --git a/hypaware-core/plugins-workspace/context-graph-enrich/src/datasets.js b/hypaware-core/plugins-workspace/context-graph-enrich/src/datasets.js
index 2a4566ae..32c88983 100644
--- a/hypaware-core/plugins-workspace/context-graph-enrich/src/datasets.js
+++ b/hypaware-core/plugins-workspace/context-graph-enrich/src/datasets.js
@@ -7,9 +7,8 @@ import { discoverCachePartitions } from '../../../../src/core/cache/partition.js
 import { unionSources, emptySource } from 'hypaware/core/query'
 
 /**
- * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
- * @import { AsyncDataSource } from 'squirreling'
  */
 
 export const PLUGIN_NAME = '@hypaware/context-graph-enrich'
@@ -165,7 +164,7 @@ async function discoverParts(ctx, dataset) {
  * @param {QueryPartition[]} partitions
  * @param {DatasetDataSourceContext} ctx
  * @param {string} dataset
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function createDataSource(partitions, ctx, dataset) {
   const storage = /** @type {ExtendedQueryStorageService} */ (ctx.storage)
@@ -178,7 +177,7 @@ async function createDataSource(partitions, ctx, dataset) {
   }
   for (const p of fresh) tablePaths.add(p.path)
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
diff --git a/hypaware-core/plugins-workspace/context-graph/src/datasets.js b/hypaware-core/plugins-workspace/context-graph/src/datasets.js
index 18a2c7e7..543aa53b 100644
--- a/hypaware-core/plugins-workspace/context-graph/src/datasets.js
+++ b/hypaware-core/plugins-workspace/context-graph/src/datasets.js
@@ -6,9 +6,8 @@ import { discoverCachePartitions } from '../../../../src/core/cache/partition.js
 import { unionSources, emptySource } from 'hypaware/core/query'
 
 /**
- * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
- * @import { AsyncDataSource } from 'squirreling'
  */
 
 export const PLUGIN_NAME = '@hypaware/context-graph'
@@ -147,7 +146,7 @@ async function discoverParts(ctx, dataset) {
  * @param {QueryPartition[]} partitions
  * @param {DatasetDataSourceContext} ctx
  * @param {'node' | 'edge'} dataset
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function createDataSource(partitions, ctx, dataset) {
   const storage = /** @type {ExtendedQueryStorageService} */ (ctx.storage)
@@ -160,7 +159,7 @@ async function createDataSource(partitions, ctx, dataset) {
   }
   for (const p of fresh) tablePaths.add(p.path)
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
diff --git a/hypaware-core/plugins-workspace/gascity/src/dataset.js b/hypaware-core/plugins-workspace/gascity/src/dataset.js
index 465ea198..5957349e 100644
--- a/hypaware-core/plugins-workspace/gascity/src/dataset.js
+++ b/hypaware-core/plugins-workspace/gascity/src/dataset.js
@@ -6,9 +6,8 @@ import { discoverCachePartitions } from '../../../../src/core/cache/partition.js
 import { unionSources, emptySource } from 'hypaware/core/query'
 
 /**
- * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
- * @import { AsyncDataSource } from 'squirreling'
  */
 
 export const DATASET_NAME = 'gascity_messages'
@@ -112,6 +111,7 @@ export async function refreshPartition(_partition) {
  *
  * @param {QueryPartition[]} partitions
  * @param {DatasetDataSourceContext} ctx
+ * @returns {Promise<ScannableDataSource>}
  */
 export async function createDataSource(partitions, ctx) {
   const storage = /** @type {ExtendedQueryStorageService} */ (ctx.storage)
@@ -125,7 +125,7 @@ export async function createDataSource(partitions, ctx) {
   }
   for (const p of fresh) tablePaths.add(p.path)
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
diff --git a/hypaware-core/plugins-workspace/otel/src/datasets.js b/hypaware-core/plugins-workspace/otel/src/datasets.js
index d067e8d2..20822c32 100644
--- a/hypaware-core/plugins-workspace/otel/src/datasets.js
+++ b/hypaware-core/plugins-workspace/otel/src/datasets.js
@@ -6,9 +6,8 @@ import { discoverCachePartitions } from '../../../../src/core/cache/partition.js
 import { unionSources, emptySource } from 'hypaware/core/query'
 
 /**
- * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRefreshResult, DatasetRegistration, QueryPartition, QueryStorageService, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../../src/core/cache/types.js'
- * @import { AsyncDataSource } from 'squirreling'
  */
 
 export const PARTITION_LABEL = 'all'
@@ -205,7 +204,7 @@ async function createDataSource(partitions, ctx, dataset) {
   }
   for (const p of fresh) tablePaths.add(p.path)
 
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const tablePath of tablePaths) {
     const source = await storage.dataSourceForTable(tablePath)
diff --git a/hypaware-core/plugins-workspace/s3/src/query-dataset.js b/hypaware-core/plugins-workspace/s3/src/query-dataset.js
index 9d5a3f30..c18521c8 100644
--- a/hypaware-core/plugins-workspace/s3/src/query-dataset.js
+++ b/hypaware-core/plugins-workspace/s3/src/query-dataset.js
@@ -7,8 +7,7 @@ import { parquetDataSource, unionSources, emptySource } from 'hypaware/core/quer
 
 /**
  * @import { AsyncBuffer } from 'hyparquet'
- * @import { AsyncDataSource } from 'squirreling/src/types.js'
- * @import { BlobStore, ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRegistration, PluginName, QueryPartition } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { BlobStore, ColumnSpec, DatasetDataSourceContext, DatasetDiscoveryContext, DatasetRegistration, PluginName, QueryPartition, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { S3QuerySourceConfig } from './types.js'
  */
 
@@ -35,7 +34,7 @@ export function buildS3QueryDataset({ source, blobStore, plugin }) {
     /**
      * @param {QueryPartition[]} partitions
      * @param {DatasetDataSourceContext} _ctx
-     * @returns {Promise<AsyncDataSource>}
+     * @returns {Promise<ScannableDataSource>}
      */
     createDataSource: (partitions, _ctx) => createDataSource(source, blobStore, partitions),
   }
@@ -74,13 +73,13 @@ async function discoverPartitions(source, blobStore) {
  * @param {S3QuerySourceConfig} source
  * @param {BlobStore} blobStore
  * @param {QueryPartition[]} partitions
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function createDataSource(source, blobStore, partitions) {
   if (source.format === 'iceberg') {
     return createIcebergDataSource(source, blobStore)
   }
-  /** @type {AsyncDataSource[]} */
+  /** @type {ScannableDataSource[]} */
   const sources = []
   for (const partition of partitions) {
     const key = partition.tableUrl
@@ -103,7 +102,7 @@ async function createDataSource(source, blobStore, partitions) {
  *
  * @param {S3QuerySourceConfig} source
  * @param {BlobStore} blobStore
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function createIcebergDataSource(source, blobStore) {
   // Guard against a missing/empty table the way the local cache does
diff --git a/hypaware-plugin-kernel-types.d.ts b/hypaware-plugin-kernel-types.d.ts
index 24d338bb..70d576e1 100644
--- a/hypaware-plugin-kernel-types.d.ts
+++ b/hypaware-plugin-kernel-types.d.ts
@@ -17,6 +17,16 @@ import type { UsagePolicyDrop } from './src/core/usage-policy/types.d.ts'
 
 export type { AsyncDataSource, ScanOptions, ScanResults }
 
+/**
+ * A data source that retains Squirreling's row interface. Hypaware's storage,
+ * union, visibility, and legacy parquet adapters all guarantee this stronger
+ * shape even when they also expose prepared native batches.
+ */
+export type ScannableDataSource = AsyncDataSource & {
+  columns: string[]
+  scan(options: ScanOptions): ScanResults
+}
+
 export type JsonPrimitive = string | number | boolean | null
 export type JsonValue = JsonPrimitive | JsonObject | JsonValue[]
 
diff --git a/llp/0015-query-and-datasets.spec.md b/llp/0015-query-and-datasets.spec.md
index 739ea6d9..e00da0f6 100644
--- a/llp/0015-query-and-datasets.spec.md
+++ b/llp/0015-query-and-datasets.spec.md
@@ -22,6 +22,11 @@
 > OOM the host by buffering an unbounded scan
 > ([hyparam/hypaware-server#9](https://github.com/hyparam/hypaware-server/issues/9)).
 
+> **Extended by [LLP 0294](./0294-native-prepared-batches-through-query-sources.decision.md).**
+> Compatible partition unions now concatenate native prepared batches, remap
+> per-table field ids, and keep LIMIT/OFFSET on the merged stream. Drifted
+> schemas retain the row-padding behavior specified below.
+
 ## Query is intrinsic
 
 Query and Iceberg storage are intrinsic services. Plugins register datasets;
diff --git a/llp/0098-scancolumn-where-pushdown.decision.md b/llp/0098-scancolumn-where-pushdown.decision.md
index a6b68e70..d0fab065 100644
--- a/llp/0098-scancolumn-where-pushdown.decision.md
+++ b/llp/0098-scancolumn-where-pushdown.decision.md
@@ -12,6 +12,10 @@
 > flags, so a filtered `COUNT` keeps the streaming fast path instead of
 > falling back to per-row materialization.
 
+> **Extended by [LLP 0294](./0294-native-prepared-batches-through-query-sources.decision.md).**
+> The same storage, union, schema, and heap wrappers now preserve Squirreling's
+> schema-addressed prepared batches when their semantics are transparent.
+
 ## Context
 
 LLP 0055 lit the engine's streaming-aggregate fast path by implementing
diff --git a/llp/0261-absent-column-pad-value-per-path.decision.md b/llp/0261-absent-column-pad-value-per-path.decision.md
index f38a3c0a..b30c4c29 100644
--- a/llp/0261-absent-column-pad-value-per-path.decision.md
+++ b/llp/0261-absent-column-pad-value-per-path.decision.md
@@ -9,6 +9,8 @@
 **Extends:** LLP 0032 (#capture: the parenthetical "padding absent physical
 columns to null" holds only on the column-stream path; the row path pads with
 a cell that reads `undefined`)
+**Extended by:** LLP 0294 (prepared batches are disabled across schema drift,
+so this document's row and column-stream values remain authoritative)
 
 > Corrects one sentence of [LLP 0032 §Capture](./0032-github-llm-graph-bridge.decision.md#capture)
 > without changing what it decided. The guarantee 0032 needed, a
diff --git a/llp/0294-native-prepared-batches-through-query-sources.decision.md b/llp/0294-native-prepared-batches-through-query-sources.decision.md
new file mode 100644
index 00000000..687844a8
--- /dev/null
+++ b/llp/0294-native-prepared-batches-through-query-sources.decision.md
@@ -0,0 +1,107 @@
+# LLP 0294: Native prepared batches survive the query source stack
+
+**Type:** Decision
+**Status:** Accepted
+**Systems:** Query, Cache, Sources
+**Author:** Kenny / Codex
+**Date:** 2026-08-18
+**Related:** LLP 0015, LLP 0055, LLP 0097, LLP 0098, LLP 0105, LLP 0241, LLP 0261
+
+> Squirreling 0.16.1 and Icebird 0.8.23 add schema-addressed native batch
+> scans. Hypaware forwards that path through semantically transparent source
+> wrappers, concatenates only compatible partition schemas, keeps ranges
+> global to a union, and retains the established row fallback wherever schema
+> drift or visibility filtering needs row-level semantics.
+
+## Context
+
+Icebird now exposes `schema` plus `prepareScan()`. Squirreling uses them in
+preference to `scan()` so parquet vectors and deferred columns can flow through
+execution without allocating one `AsyncRow` and one promise per cell. A source
+must expose both properties for the native path to light.
+
+Hypaware wrapped every Icebird source several times before execution:
+
+- storage removes internal cache fields;
+- multi-partition datasets concatenate sources;
+- ai-gateway advertises declared columns over old physical schemas;
+- the execution budget samples heap growth while rows or column chunks flow;
+- local-only visibility may filter or suppress individual rows.
+
+Those wrappers predated prepared scans and exposed only `scan()` and
+`scanColumn()`. Updating package versions alone therefore produced correct
+queries but could not reach the new path.
+
+## Decision
+
+### Transparent wrappers {#transparent-wrappers}
+
+Storage forwards `schema` and `prepareScan()` after removing internal fields
+from the advertised schema. Squirreling derives field demands from that public
+schema, so an internal field cannot enter a prepared request.
+
+The heap-budget wrapper forwards prepared metadata and samples after native
+batch production and after each deferred column materializes. It also accepts
+a prepared-only third-party source, matching Squirreling's widened
+`AsyncDataSource` contract. Hypaware names the stronger shape used by its own
+storage and parquet layers `ScannableDataSource`: those sources always retain
+`columns` and `scan()` for wrappers whose semantics require rows.
+
+The visibility wrapper does not forward prepared scans. It needs each row's
+`cwd`, may suppress selected content cells, and must apply LIMIT/OFFSET after
+withholding. A governable prepared-only third-party source is refused when a
+non-top caller needs the visibility wrapper, rather than bypassing the privacy
+rule.
+
+### Partition union {#partition-union}
+
+`unionSources` advertises a prepared schema only when every child has a
+prepared scan and the schemas have identical ordered names, data types, and
+nullability. Field ids may differ between separately created Iceberg tables;
+the union uses the first schema as its logical schema and remaps each demand by
+field name to the child's id.
+
+The union strips LIMIT/OFFSET before preparing children. Those hints apply once
+to the concatenated stream, as LLP 0015 already requires for row and column
+scans. A filter is forwarded to each child for file and row-group pruning. The
+native batches are concatenated only when every child reports the same filter
+residual. If residual contracts differ, the prepared result adapts the union's
+established row scan back into batches and leaves the full filter and range as
+residual work.
+
+Prepared `exactRows` and `maxRows` are summed only when every child knows the
+corresponding value. The union's legacy `numRows` follows the same rule: one
+unknown child makes the total unknown instead of silently contributing zero.
+
+### Schema drift {#schema-drift}
+
+A partition union with different logical schemas does not expose
+`prepareScan()`. The row path remains authoritative for absent-field padding,
+including the `undefined` versus `null` split recorded by LLP 0261. Native
+batches do not invent a third representation.
+
+The ai-gateway declared-schema wrapper follows the same gate. It forwards a
+prepared scan only when the physical schema already contains every column the
+wrapper advertises. If an old partition lacks a declared field, the existing
+row and `scanColumn` paths retain control.
+
+Icebird 0.8.23 has a pre-existing row-scan defect when a pushed filter and
+position deletes coexist: filtering loses physical row ordinals before delete
+application, so a live matching row can be dropped. This is tracked in
+[icebird#41](https://github.com/hyparam/icebird/issues/41). The prepared path
+avoids the defect by withholding the filter when deletes exist, but visibility
+wrappers, `scanColumn` aggregates, and schema-drift fallbacks remain exposed
+until Icebird fixes its row path.
+
+## Consequences
+
+- Current, schema-aligned Iceberg partitions reach native batch execution end
+  to end through storage, union, ai-gateway, and heap-budget wrappers.
+- Additive schema drift retains the established absent-column semantics and
+  stays on the older paths until all participating physical schemas align.
+- LIMIT/OFFSET and residual filters remain globally correct over
+  multi-partition datasets.
+- Internal cache fields remain absent from planning and query results.
+- Third-party prepared-only datasets work when no row-level visibility filter
+  is required. A privacy-sensitive prepared-only dataset must also implement
+  `scan()` before a restricted caller can query it.
diff --git a/package.json b/package.json
index 538ef37b..ac5554f4 100644
--- a/package.json
+++ b/package.json
@@ -79,9 +79,9 @@
     "hyparquet": "1.28.2",
     "hyparquet-compressors": "1.1.1",
     "hypgrep": "0.5.1",
-    "icebird": "0.8.22",
+    "icebird": "0.8.23",
     "marked": "18.0.9",
-    "squirreling": "0.15.3"
+    "squirreling": "0.16.1"
   },
   "optionalDependencies": {
     "hyparquet-writer": "0.16.6",
diff --git a/src/core/cache/iceberg/store.js b/src/core/cache/iceberg/store.js
index 5ee9a2e8..a9c3fe1b 100644
--- a/src/core/cache/iceberg/store.js
+++ b/src/core/cache/iceberg/store.js
@@ -40,7 +40,7 @@ import {
 import { INGEST_SEQ_COLUMN } from '../streaming-reader.js'
 
 /**
- * @import { ColumnSpec } from '../../../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, ScannableDataSource } from '../../../../hypaware-plugin-kernel-types.js'
  * @import { AppendOptions } from '../../../../src/core/cache/types.js'
  * @import { Catalog, Lister, Manifest, ManifestEntry, PartitionSpec, Resolver, Schema, TableMetadata } from 'icebird/src/types.js'
  * @import { AsyncDataSource, AsyncRow } from 'squirreling'
@@ -546,7 +546,7 @@ export function seqValue(raw) {
  * an empty table.
  *
  * @param {string} tablePath
- * @returns {Promise<AsyncDataSource | null>}
+ * @returns {Promise<ScannableDataSource | null>}
  */
 export async function dataSourceForTable(tablePath) {
   if (!tableExists(tablePath)) return null
diff --git a/src/core/cache/storage.js b/src/core/cache/storage.js
index bf591b03..51db9c79 100644
--- a/src/core/cache/storage.js
+++ b/src/core/cache/storage.js
@@ -26,7 +26,7 @@ import { createHash } from 'node:crypto'
 import path from 'node:path'
 
 /**
- * @import { SinkContinuation } from '../../../hypaware-plugin-kernel-types.js'
+ * @import { ScannableDataSource, SinkContinuation } from '../../../hypaware-plugin-kernel-types.js'
  * @import { CachePartitioningDeclaration, ExtendedQueryStorageService, SourceWithholdResolver } from '../../../src/core/cache/types.js'
  * @import { UsagePolicyResolver } from '../../../src/core/usage-policy/types.js'
  * @import { AsyncDataSource } from 'squirreling'
@@ -370,7 +370,7 @@ export function createQueryStorageService({ cacheRoot, getDeclaration, getSettle
       const source = await dataSourceForTable(resolveIcebergDir(tablePath))
       if (!source) return null
       const publicColumns = source.columns.filter((c) => !INTERNAL_FIELDS.includes(c))
-      /** @type {AsyncDataSource} */
+      /** @type {ScannableDataSource} */
       const wrapped = {
         numRows: source.numRows,
         columns: publicColumns,
@@ -402,6 +402,18 @@ export function createQueryStorageService({ cacheRoot, getDeclaration, getSettle
       if (typeof source.scanColumn === 'function') {
         wrapped.scanColumn = (options) => /** @type {NonNullable<AsyncDataSource['scanColumn']>} */ (source.scanColumn)(options)
       }
+      // Icebird's prepared scan is safe to expose after applying the same
+      // internal-field projection to its logical schema. Squirreling plans
+      // requests from this schema, so no prepared demand can name an internal
+      // field and the inner source returns only the requested public fields.
+      // @ref LLP 0294#transparent-wrappers [implements]: storage preserves native batches without advertising internal cache fields
+      if (source.schema && source.prepareScan) {
+        const prepareScan = source.prepareScan
+        wrapped.schema = {
+          fields: source.schema.fields.filter((field) => !INTERNAL_FIELDS.includes(field.name)),
+        }
+        wrapped.prepareScan = (request) => prepareScan.call(source, request)
+      }
       return wrapped
     },
 
diff --git a/src/core/cache/types.d.ts b/src/core/cache/types.d.ts
index 5ebe95a2..797d09db 100644
--- a/src/core/cache/types.d.ts
+++ b/src/core/cache/types.d.ts
@@ -1,8 +1,7 @@
-import type { ColumnSpec, QueryScope, QueryStorageService } from '../../../hypaware-plugin-kernel-types.d.ts'
+import type { ColumnSpec, QueryScope, QueryStorageService, ScannableDataSource } from '../../../hypaware-plugin-kernel-types.d.ts'
 import type { ParquetWriter } from 'hyparquet-writer'
 import type { Writer } from 'hyparquet-writer/src/types.js'
 import type { PartitionSpec } from 'icebird/src/types.js'
-import type { AsyncDataSource } from 'squirreling'
 import type { UsagePolicyResolver } from '../usage-policy/types.d.ts'
 // Partitioning declaration promoted to a neutral core home
 // (LLP 0003 / LLP 0022#shared-core-helpers). Re-exported here so existing
@@ -448,7 +447,7 @@ export interface SourceWithholdResolver {
 }
 
 export type ExtendedQueryStorageService = QueryStorageService & {
-  dataSourceForTable(tablePath: string): Promise<AsyncDataSource | null>
+  dataSourceForTable(tablePath: string): Promise<ScannableDataSource | null>
   flushTable(tablePath: string, opts?: { reason?: string; force?: boolean }): Promise<FlushResult>
   flushAll(opts?: { reason?: string; force?: boolean }): Promise<FlushResult>
   pendingInfo(tablePath: string): Promise<PendingInfo>
diff --git a/src/core/query/parquet-source.js b/src/core/query/parquet-source.js
index 2e0a877d..a0af6d2e 100644
--- a/src/core/query/parquet-source.js
+++ b/src/core/query/parquet-source.js
@@ -8,7 +8,8 @@ import { whereToParquetFilter } from './parquet-pushdown.js'
 
 /**
  * @import { AsyncBuffer, FileMetaData } from 'hyparquet'
- * @import { AsyncDataSource, ScanOptions, ScanResults, SqlPrimitive } from 'squirreling/src/types.js'
+ * @import { ScannableDataSource } from '../../../hypaware-plugin-kernel-types.js'
+ * @import { ScanOptions, ScanResults, SqlPrimitive } from 'squirreling'
  */
 
 /**
@@ -35,7 +36,7 @@ import { whereToParquetFilter } from './parquet-pushdown.js'
  *
  * @param {AsyncBuffer} file
  * @param {FileMetaData} metadata
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 export function parquetDataSource(file, metadata) {
   const schema = parquetSchema(metadata)
diff --git a/src/core/query/sql.js b/src/core/query/sql.js
index a1f7d470..bb994087 100644
--- a/src/core/query/sql.js
+++ b/src/core/query/sql.js
@@ -16,11 +16,11 @@ import {
 } from './visibility.js'
 
 /**
- * @import { PluginLogger } from '../../../hypaware-plugin-kernel-types.js'
+ * @import { PluginLogger, ScannableDataSource } from '../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../src/core/cache/types.js'
  * @import { ExecuteSqlOptions, ExecuteSqlResult, LocalOnlyVisibilityReport, RefreshMode } from '../../../src/core/query/types.js'
  * @import { UsagePolicyResolver } from '../../../src/core/usage-policy/types.js'
- * @import { AsyncDataSource } from 'squirreling'
+ * @import { AsyncBatch, AsyncDataSource, PrepareScan } from 'squirreling'
  */
 
 /**
@@ -150,12 +150,26 @@ function resolveForcedGc() {
  * @returns {AsyncDataSource}
  */
 function withHeapBudget(source, guard) {
+  if (!source.scan) {
+    const schema = /** @type {NonNullable<AsyncDataSource['schema']>} */ (source.schema)
+    const prepareScan = /** @type {NonNullable<AsyncDataSource['prepareScan']>} */ (source.prepareScan)
+    /** @type {AsyncDataSource} */
+    const bounded = {
+      numRows: source.numRows,
+      columns: source.columns ?? schema.fields.map((field) => field.name),
+      schema,
+      prepareScan: budgetedPrepareScan((request) => prepareScan.call(source, request), guard),
+    }
+    forwardBudgetedScanColumn(bounded, source, guard)
+    return bounded
+  }
+  const scan = source.scan
   /** @type {AsyncDataSource} */
   const bounded = {
     numRows: source.numRows,
-    columns: source.columns,
+    columns: source.columns ?? source.schema?.fields.map((field) => field.name) ?? [],
     scan(options) {
-      const inner = source.scan(options)
+      const inner = scan.call(source, options)
       return {
         appliedWhere: inner.appliedWhere,
         appliedLimitOffset: inner.appliedLimitOffset,
@@ -172,24 +186,90 @@ function withHeapBudget(source, guard) {
       }
     },
   }
-  if (typeof source.scanColumn === 'function') {
-    const scanColumn = /** @type {NonNullable<AsyncDataSource['scanColumn']>} */ (source.scanColumn)
-    // @ref LLP 0098#wrapper-duties [implements]: the budget decoration must pass appliedWhere/appliedLimitOffset through untouched, or the engine re-slices a filtered stream
-    bounded.scanColumn = (options) => {
-      const inner = normalizeScanColumn(scanColumn(options), options)
+  if (source.schema && source.prepareScan) {
+    const prepareScan = source.prepareScan
+    bounded.schema = source.schema
+    bounded.prepareScan = budgetedPrepareScan((request) => prepareScan.call(source, request), guard)
+  }
+  forwardBudgetedScanColumn(bounded, source, guard)
+  return bounded
+}
+
+/**
+ * Forward a column stream with its negotiation flags intact and sample each
+ * materialized chunk.
+ *
+ * @param {AsyncDataSource} bounded
+ * @param {AsyncDataSource} source
+ * @param {{ check: (site: string) => void }} guard
+ * @returns {void}
+ */
+function forwardBudgetedScanColumn(bounded, source, guard) {
+  if (typeof source.scanColumn !== 'function') return
+  const scanColumn = /** @type {NonNullable<AsyncDataSource['scanColumn']>} */ (source.scanColumn)
+  // @ref LLP 0098#wrapper-duties [implements]: the budget decoration must pass appliedWhere/appliedLimitOffset through untouched, or the engine re-slices a filtered stream
+  bounded.scanColumn = (options) => {
+    const inner = normalizeScanColumn(scanColumn.call(source, options), options)
+    return {
+      appliedWhere: inner.appliedWhere,
+      appliedLimitOffset: inner.appliedLimitOffset,
+      async *chunks() {
+        for await (const chunk of inner.chunks()) {
+          guard.check('column_chunk')
+          yield chunk
+        }
+      },
+    }
+  }
+}
+
+/**
+ * Preserve a prepared source through the heap-budget decoration. Direct
+ * vectors are sampled after batch production; deferred vectors are sampled
+ * immediately after their lazy read resolves.
+ *
+ * @param {PrepareScan} prepareScan
+ * @param {{ check: (site: string) => void }} guard
+ * @returns {PrepareScan}
+ */
+function budgetedPrepareScan(prepareScan, guard) {
+  return function prepareWithBudget(request) {
+    const inner = prepareScan(request)
+    return {
+      schema: inner.schema,
+      residual: inner.residual,
+      properties: inner.properties,
+      async *batches(options = {}) {
+        for await (const batch of inner.batches(options)) {
+          guard.check('native_batch')
+          yield budgetedBatch(batch, guard)
+        }
+      },
+    }
+  }
+}
+
+/**
+ * @param {AsyncBatch} batch
+ * @param {{ check: (site: string) => void }} guard
+ * @returns {AsyncBatch}
+ */
+function budgetedBatch(batch, guard) {
+  return {
+    selection: batch.selection,
+    columns: batch.columns.map((column) => {
+      if (!('read' in column)) return column
+      const read = column.read
       return {
-        appliedWhere: inner.appliedWhere,
-        appliedLimitOffset: inner.appliedLimitOffset,
-        async *chunks() {
-          for await (const chunk of inner.chunks()) {
-            guard.check('column_chunk')
-            yield chunk
-          }
+        ...column,
+        async read(request) {
+          const vector = await read.call(column, request)
+          guard.check('native_batch')
+          return vector
         },
       }
-    }
+    }),
   }
-  return bounded
 }
 
 /**
@@ -319,14 +399,18 @@ export async function executeQuerySql(args) {
           let table = source
           if (!includeLocalOnly) {
             const contentColumns = dataset.localOnlyContentColumns ?? []
-            const governable = source.columns.includes('cwd') ||
-              contentColumns.some((c) => source.columns.includes(c))
+            const sourceColumns = source.columns ?? source.schema?.fields.map((field) => field.name) ?? []
+            const governable = sourceColumns.includes('cwd') ||
+              contentColumns.some((c) => sourceColumns.includes(c))
             if (governable) {
               const vis = getVisibility()
               localOnly.callerClass = vis.callerClass
               if (!callerSeesEverything(vis.callerRank)) {
+                if (!source.scan || !source.columns) {
+                  throw new Error(`Dataset "${name}" must provide columns and scan() to enforce local-only visibility`)
+                }
                 localOnly.filtered = true
-                table = withLocalOnlyVisibility(source, {
+                table = withLocalOnlyVisibility(/** @type {ScannableDataSource} */ (source), {
                   resolver: vis.resolver,
                   callerRank: vis.callerRank,
                   contentColumns,
diff --git a/src/core/query/union-source.js b/src/core/query/union-source.js
index 8cf86692..01a0b7f3 100644
--- a/src/core/query/union-source.js
+++ b/src/core/query/union-source.js
@@ -1,9 +1,14 @@
 // @ts-check
 
+import { isDeepStrictEqual } from 'node:util'
+
+import { rowsToBatches } from 'squirreling'
+
 import { normalizeScanColumn } from './scan-column.js'
 
 /**
- * @import { AsyncCell, AsyncRow, AsyncDataSource, ExprNode } from 'squirreling/src/types.js'
+ * @import { ScannableDataSource } from '../../../hypaware-plugin-kernel-types.js'
+ * @import { AsyncCells, AsyncRow, AsyncDataSource, ExprNode, PreparedScan, RelationSchema, ScanProperties, ScanRequest } from 'squirreling'
  */
 
 /**
@@ -13,7 +18,7 @@ import { normalizeScanColumn } from './scan-column.js'
  * builds a cell per REQUESTED key and resolves it off an object that has no
  * such key), so padding introduces no new value.
  */
-const absentCell = /** @type {AsyncCell} */ (/** @type {unknown} */ (() => Promise.resolve(undefined)))
+const absentCell = /** @type {AsyncCells[string]} */ (/** @type {unknown} */ (() => Promise.resolve(undefined)))
 
 /**
  * Re-key one scanned row onto the exact column list the scan advertises,
@@ -46,7 +51,7 @@ export function alignRowColumns(row, columns) {
     }
     if (same) return row
   }
-  /** @type {Record<string, AsyncCell>} */
+  /** @type {AsyncCells} */
   const cells = {}
   for (const name of columns) cells[name] = row.cells[name] ?? absentCell
   // `resolved` is keyed by name and only ever read by name, so the original
@@ -130,22 +135,25 @@ export async function* alignRows(rows, columns) {
  * already present even though `appliedWhere: false` never asks for them
  * explicitly.
  *
- * @param {AsyncDataSource[]} sources
- * @returns {AsyncDataSource}
+ * @param {ScannableDataSource[]} sources
+ * @returns {ScannableDataSource}
  * @ref LLP 0015#multi-partition-union [constrained-by]: the union must not forward limit/offset or offsets apply twice, nor push a filter a partition can't satisfy
  */
 export function unionSources(sources) {
   /** @type {Set<string>} */
   const allColumns = new Set()
   let totalRows = 0
+  let totalRowsKnown = true
   for (const s of sources) {
     for (const col of s.columns) allColumns.add(col)
-    totalRows += s.numRows ?? 0
+    if (s.numRows === undefined) totalRowsKnown = false
+    else totalRows += s.numRows
   }
-  /** @type {AsyncDataSource} */
+  const columns = Array.from(allColumns)
+  /** @type {ScannableDataSource} */
   const union = {
-    columns: Array.from(allColumns),
-    numRows: totalRows,
+    columns,
+    numRows: totalRowsKnown ? totalRows : undefined,
     scan(options) {
       // Defends against a runtime scan() with no options even though the
       // AsyncDataSource contract types it as required.
@@ -175,6 +183,11 @@ export function unionSources(sources) {
       }
     },
   }
+  const preparedSchema = commonPreparedSchema(sources, columns)
+  if (preparedSchema) {
+    union.schema = preparedSchema
+    union.prepareScan = (request) => prepareUnionScan({ union, sources, schema: preparedSchema, request })
+  }
   // The column-stream hook is offered only when EVERY partition can stream
   // the column; a mixed union stays row-based so the engine's fallback owns
   // correctness.
@@ -274,6 +287,179 @@ export function unionSources(sources) {
   return union
 }
 
+/**
+ * Find one logical schema the union can advertise without changing its
+ * existing column order. Field ids may differ between independent Iceberg
+ * tables, so compatibility is by ordered name, type, and nullability; each
+ * prepared request is remapped to the child table's ids below.
+ *
+ * A drifted union deliberately returns undefined. Its row path owns absent
+ * column padding, whose undefined/null semantics are not part of the native
+ * batch contract (LLP 0261), so native batches must not guess a third answer.
+ *
+ * @param {ScannableDataSource[]} sources
+ * @param {string[]} columns
+ * @returns {RelationSchema | undefined}
+ * @ref LLP 0294#partition-union [implements]: only aligned schemas expose one logical prepared union
+ */
+function commonPreparedSchema(sources, columns) {
+  if (sources.length === 0) return undefined
+  const first = sources[0]
+  if (!first.schema || !first.prepareScan) return undefined
+  if (!schemaMatchesColumns(first.schema, columns)) return undefined
+  for (let i = 1; i < sources.length; i++) {
+    const source = sources[i]
+    if (!source.schema || !source.prepareScan) return undefined
+    if (!schemasAreCompatible(first.schema, source.schema)) return undefined
+  }
+  return first.schema
+}
+
+/**
+ * @param {RelationSchema} schema
+ * @param {string[]} columns
+ * @returns {boolean}
+ */
+function schemaMatchesColumns(schema, columns) {
+  if (schema.fields.length !== columns.length) return false
+  return schema.fields.every((field, index) => field.name === columns[index])
+}
+
+/**
+ * @param {RelationSchema} left
+ * @param {RelationSchema} right
+ * @returns {boolean}
+ */
+function schemasAreCompatible(left, right) {
+  if (left.fields.length !== right.fields.length) return false
+  return left.fields.every((field, index) => {
+    const candidate = right.fields[index]
+    return field.name === candidate.name &&
+      field.nullable === candidate.nullable &&
+      isDeepStrictEqual(field.dataType, candidate.dataType)
+  })
+}
+
+/**
+ * Prepare one native scan over a concatenation. Range hints are never sent to
+ * children because LIMIT/OFFSET are not distributive over partitions. Filter
+ * hints are sent for pruning; native batches are concatenated only when every
+ * child reports the same residual contract. A mixed residual falls back to
+ * the union's established row semantics and adapts those rows to batches.
+ *
+ * @param {object} options
+ * @param {ScannableDataSource} options.union
+ * @param {ScannableDataSource[]} options.sources
+ * @param {RelationSchema} options.schema
+ * @param {ScanRequest} options.request
+ * @returns {PreparedScan}
+ * @ref LLP 0294#partition-union [implements]: remap field ids per child and keep range hints on the concatenated stream
+ */
+function prepareUnionScan({ union, sources, schema, request }) {
+  const fieldsById = new Map(schema.fields.map((field) => [field.id, field]))
+  const requestedFields = request.columns.map((demand) => {
+    const field = fieldsById.get(demand.field)
+    if (!field) throw new Error(`Prepared union requested unknown field id ${demand.field}`)
+    return field
+  })
+  const requestedNames = requestedFields.map((field) => field.name)
+  const childScans = sources.map((source) => {
+    const fieldsByName = new Map(/** @type {RelationSchema} */ (source.schema).fields.map((field) => [field.name, field]))
+    const columns = request.columns.map((demand, index) => ({
+      ...demand,
+      field: /** @type {NonNullable<ReturnType<typeof fieldsByName.get>>} */ (fieldsByName.get(requestedNames[index])).id,
+    }))
+    return /** @type {NonNullable<AsyncDataSource['prepareScan']>} */ (source.prepareScan)({
+      ...request,
+      columns,
+      limit: undefined,
+      offset: undefined,
+    })
+  })
+  const nativeCompatible = childScans.every((scan) => {
+    if (scan.schema.fields.length !== requestedNames.length) return false
+    return scan.schema.fields.every((field, index) => field.name === requestedNames[index])
+  }) && childScans.every((scan) => scan.residual.filter === childScans[0].residual.filter) &&
+    (childScans[0].residual.filter === undefined || childScans[0].residual.filter === request.filter)
+
+  if (!nativeCompatible) {
+    return rowFallbackPreparedScan({ union, schema: { fields: requestedFields }, request })
+  }
+
+  /** @type {ScanProperties} */
+  const properties = {}
+  const exactRows = sumPreparedProperty(childScans, 'exactRows')
+  const maxRows = sumPreparedProperty(childScans, 'maxRows')
+  if (exactRows !== undefined) properties.exactRows = exactRows
+  if (maxRows !== undefined) properties.maxRows = maxRows
+  return {
+    schema: { fields: requestedFields },
+    residual: {
+      filter: childScans[0].residual.filter,
+      limit: request.limit,
+      offset: request.offset,
+    },
+    properties,
+    async *batches(options = {}) {
+      for (const scan of childScans) {
+        options.signal?.throwIfAborted()
+        yield* scan.batches(options)
+      }
+    },
+  }
+}
+
+/**
+ * @param {PreparedScan[]} scans
+ * @param {'exactRows' | 'maxRows'} property
+ * @returns {number | undefined}
+ */
+function sumPreparedProperty(scans, property) {
+  let total = 0
+  for (const scan of scans) {
+    const value = scan.properties[property]
+    if (value === undefined) return undefined
+    total += value
+  }
+  return total
+}
+
+/**
+ * Preserve correctness when otherwise-compatible prepared children disagree
+ * about residual work. The engine still gets a PreparedScan, but its batches
+ * come from the union's established row implementation and the whole request
+ * remains residual.
+ *
+ * @param {object} options
+ * @param {ScannableDataSource} options.union
+ * @param {RelationSchema} options.schema
+ * @param {ScanRequest} options.request
+ * @returns {PreparedScan}
+ */
+function rowFallbackPreparedScan({ union, schema, request }) {
+  const names = schema.fields.map((field) => field.name)
+  return {
+    schema,
+    residual: {
+      filter: request.filter,
+      limit: request.limit,
+      offset: request.offset,
+    },
+    properties: {
+      ...(request.filter === undefined && union.numRows !== undefined ? { exactRows: union.numRows } : {}),
+      ...(union.numRows !== undefined ? { maxRows: union.numRows } : {}),
+    },
+    async *batches({ signal } = {}) {
+      const scan = /** @type {NonNullable<AsyncDataSource['scan']>} */ (union.scan)({
+        columns: names,
+        where: request.filter,
+        signal,
+      })
+      yield* rowsToBatches(scan.rows(), names, { signal })
+    },
+  }
+}
+
 /**
  * Whether `where` can be pushed to `source`: only when the predicate's column
  * set is fully enumerable and every column it names is present on the source.
@@ -368,7 +554,7 @@ export function whereColumns(where) {
  * rather than throwing `ColumnNotFoundError`.
  *
  * @param {string[]} columns
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 export function emptySource(columns) {
   return {
diff --git a/src/core/query/visibility.js b/src/core/query/visibility.js
index 41057111..a208bfa4 100644
--- a/src/core/query/visibility.js
+++ b/src/core/query/visibility.js
@@ -7,6 +7,7 @@ import { localOnlyListPath } from '../usage-policy/local_only.js'
 
 /**
  * @import { AsyncDataSource, AsyncRow } from 'squirreling'
+ * @import { ScannableDataSource } from '../../../hypaware-plugin-kernel-types.js'
  * @import { ExtendedQueryStorageService } from '../../../src/core/cache/types.js'
  * @import { UsageClass, UsagePolicyResolver } from '../../../src/core/usage-policy/types.js'
  * @import { LocalOnlyVisibilityReport } from '../../../src/core/query/types.js'
@@ -111,21 +112,21 @@ export function callerSeesEverything(callerRank) {
  *
  * @ref LLP 0105 [implements]: the one shared filter at the query read path; caller class >= row class on the lattice, never per-command
  * @ref LLP 0105#graph-provenance [implements]: rows lacking per-row cwd provenance get their declared content-bearing columns suppressed, never surfaced
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {{
  *   resolver: UsagePolicyResolver,
  *   callerRank: number,
  *   contentColumns: string[],
  *   report: LocalOnlyVisibilityReport,
  * }} opts
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 export function withLocalOnlyVisibility(source, opts) {
   const { resolver, callerRank, report } = opts
   const hasCwd = source.columns.includes('cwd')
   const declaredContent = opts.contentColumns.filter((c) => source.columns.includes(c))
 
-  /** @type {AsyncDataSource} */
+  /** @type {ScannableDataSource} */
   const guarded = {
     // numRows intentionally absent (see fast-path discipline above).
     columns: source.columns,
diff --git a/test/core/ai-gateway-dataset.test.js b/test/core/ai-gateway-dataset.test.js
index ca39718c..c4eae5db 100644
--- a/test/core/ai-gateway-dataset.test.js
+++ b/test/core/ai-gateway-dataset.test.js
@@ -10,6 +10,7 @@ import { appendRowsToPartition, appendRowsToSourceTable } from '../../src/core/c
 import { createQueryStorageService } from '../../src/core/cache/storage.js'
 import { createQueryRegistry } from '../../src/core/registry/datasets.js'
 import {
+  AI_GATEWAY_SCHEMA_COLUMNS,
   aiGatewayDatasetRegistration,
   createDataSource,
   DATASET_NAME,
@@ -17,8 +18,9 @@ import {
 } from '../../hypaware-core/plugins-workspace/ai-gateway/src/dataset.js'
 
 /**
- * @import { ColumnSpec, QueryScope } from '../../hypaware-plugin-kernel-types.js'
- * @import { AsyncDataSource, ExprNode, SqlPrimitive } from 'squirreling/src/types.js'
+ * @import { ColumnSpec, QueryScope, ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
+ * @import { ExtendedQueryStorageService } from '../../src/core/cache/types.js'
+ * @import { AsyncDataSource, ExprNode, RelationSchema, SqlPrimitive } from 'squirreling/src/types.js'
  */
 
 /** @param {string} prefix */
@@ -271,6 +273,7 @@ test('ai-gateway createDataSource advertises declared schema columns absent from
     const scope = { limit: 1000 }
     const partitions = await discoverParts({ cacheDir: cacheRoot, scope, config: { version: 2 } })
     const source = await createDataSource(partitions, { scope, storage })
+    assert.equal(source.prepareScan, undefined, 'schema drift stays on the padding-aware row and column paths')
 
     // The declared v7 columns are advertised even though the partition lacks them.
     for (const col of ['git_remote', 'head_sha', 'repo_root']) {
@@ -293,6 +296,60 @@ test('ai-gateway createDataSource advertises declared schema columns absent from
   }
 })
 
+// @ref LLP 0294#schema-drift [tests]: a complete declared schema forwards native batches, while the drifted test above does not
+test('ai-gateway createDataSource preserves prepared scans when the physical schema is complete', async () => {
+  const cacheRoot = await makeTmpDir('prepared-schema')
+  try {
+    const columns = AI_GATEWAY_SCHEMA_COLUMNS.map((column) => column.name)
+    let preparedCalls = 0
+    /** @type {RelationSchema} */
+    const schema = {
+      fields: columns.map((name, index) => ({
+        id: index + 1,
+        name,
+        dataType: { type: 'unknown' },
+        nullable: true,
+      })),
+    }
+    /** @type {ScannableDataSource} */
+    const physical = {
+      columns,
+      numRows: 1,
+      schema,
+      scan() {
+        return { appliedWhere: false, appliedLimitOffset: false, async *rows() {} }
+      },
+      prepareScan(request) {
+        preparedCalls++
+        const requested = request.columns.map((demand) => schema.fields.find((field) => field.id === demand.field))
+        assert.ok(requested.every(Boolean))
+        return {
+          schema: { fields: /** @type {NonNullable<(typeof requested)[number]>[]} */ (requested) },
+          residual: {},
+          properties: { exactRows: 0, maxRows: 0 },
+          async *batches() {},
+        }
+      },
+    }
+    const storage = /** @type {ExtendedQueryStorageService} */ (/** @type {unknown} */ ({
+      cacheRoot,
+      dataSourceForTable: async () => physical,
+    }))
+    /** @type {QueryScope} */
+    const scope = { limit: 1000 }
+    const source = await createDataSource([
+      { dataset: DATASET_NAME, partition: {}, tablePath: path.join(cacheRoot, 'complete') },
+    ], { scope, storage })
+
+    assert.equal(typeof source.prepareScan, 'function')
+    assert.deepEqual(source.schema?.fields.map((field) => field.name), columns)
+    source.prepareScan?.({ columns: [{ field: 1, phase: 1, purpose: 'output', mode: 'deferred' }] })
+    assert.equal(preparedCalls, 1)
+  } finally {
+    await fs.rm(cacheRoot, { recursive: true, force: true })
+  }
+})
+
 test('ai-gateway createDataSource streams scanColumn with nulls for a physically absent column', async () => {
   // The column-stream analog of the schema-padding row test above: the
   // engine's streaming-aggregate fast path consumes scanColumn, and a
diff --git a/test/core/cache-storage.test.js b/test/core/cache-storage.test.js
index 9309bea7..67b7aa49 100644
--- a/test/core/cache-storage.test.js
+++ b/test/core/cache-storage.test.js
@@ -7,6 +7,8 @@ import fsSync from 'node:fs'
 import os from 'node:os'
 import path from 'node:path'
 
+import { collect, executeSql } from 'squirreling'
+
 import { readCursorSync } from '../../src/core/cache/partition.js'
 import { createQueryStorageService } from '../../src/core/cache/storage.js'
 import { DEFAULT_SPOOL_BYTES_THRESHOLD, SPOOL_DIR } from '../../src/core/cache/spool.js'
@@ -114,6 +116,18 @@ test('storage.dataSourceForTable keeps columns and cells aligned after internal-
 
     const source = await storage.dataSourceForTable(storage.cacheTablePath('dataset', ['all']))
     assert.ok(source)
+    assert.ok(source.schema, 'storage forwards the public prepared schema')
+    assert.equal(typeof source.prepareScan, 'function', 'storage forwards native batches')
+    assert.deepEqual(source.schema.fields.map((field) => field.name), ['id', 'value'])
+
+    const rowScan = source.scan
+    source.scan = () => { throw new Error('legacy row scan should not run') }
+    const preparedRows = await collect(executeSql({
+      tables: { t: source },
+      query: 'SELECT id, value FROM t',
+    }))
+    assert.deepEqual(preparedRows, [{ id: 7, value: 'kept' }], 'prepared scan returns only public fields')
+    source.scan = rowScan
 
     const scan = source.scan({})
     for await (const row of scan.rows()) {
diff --git a/test/core/iceberg-source-parity.test.js b/test/core/iceberg-source-parity.test.js
index 6754f739..e258b2b9 100644
--- a/test/core/iceberg-source-parity.test.js
+++ b/test/core/iceberg-source-parity.test.js
@@ -34,6 +34,7 @@ import { rowsToColumnSources } from '../../hypaware-core/plugins-workspace/forma
 
 /**
  * @import { AsyncBuffer } from 'hyparquet'
+ * @import { ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
  * @import { AsyncDataSource, ExprNode, ScanColumnResults, SelectStatement, SqlPrimitive } from 'squirreling/src/types.js'
  * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js'
  */
@@ -171,7 +172,7 @@ function asyncBufferFromBytes(bytes) {
  * The parquet-file backend over the fixture, at the small row-group size the
  * differential harness uses so multi-row-group iteration is exercised.
  *
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeParquetSource() {
   const columnData = rowsToColumnSources(NULLABLE_COLUMNS, NULLABLE_ROWS)
@@ -185,7 +186,7 @@ async function makeParquetSource() {
  * `dataSourceForTable` seam `hyp query sql` reaches.
  *
  * @param {string} tablePath
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeIcebergSource(tablePath) {
   await appendRowsToTable(tablePath, NULLABLE_COLUMNS, NULLABLE_ROWS)
@@ -195,7 +196,7 @@ async function makeIcebergSource(tablePath) {
 }
 
 /**
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {string} predicate
  * @returns {Promise<number[]>}
  */
@@ -510,7 +511,7 @@ function whereOf(sql) {
 }
 
 /**
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {string} query
  * @returns {Promise<number[]>}
  */
diff --git a/test/core/parquet-source.test.js b/test/core/parquet-source.test.js
index bcce85f0..d3eeec56 100644
--- a/test/core/parquet-source.test.js
+++ b/test/core/parquet-source.test.js
@@ -14,7 +14,7 @@ import { asyncBufferFromBytes, parquetSourceFromRows } from '../helpers/parquet_
 
 /**
  * @import { AsyncDataSource, ExprNode, SelectStatement } from 'squirreling/src/types.js'
- * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js'
+ * @import { ColumnSpec, ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
  */
 
 /** @type {ColumnSpec[]} */
@@ -58,7 +58,7 @@ const NULLABLE_ROWS = [
  * Build an in-memory parquet file from ROWS with a small row-group size
  * so the scan exercises multi-row-group iteration (2 + 2 + 1).
  *
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeSource() {
   return parquetSourceFromRows(COLUMNS, ROWS, { rowGroupSize: 2 })
@@ -86,7 +86,7 @@ const TIMESTAMP_ROWS = [
 ]
 
 /**
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeTimestampSource() {
   const columnData = rowsToColumnSources(TIMESTAMP_COLUMNS, TIMESTAMP_ROWS)
@@ -99,7 +99,7 @@ async function makeTimestampSource() {
 /**
  * Same, over `NULLABLE_ROWS`.
  *
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeNullableSource() {
   return parquetSourceFromRows(NULLABLE_COLUMNS, NULLABLE_ROWS, { rowGroupSize: 2 })
@@ -115,7 +115,7 @@ function whereOf(sql) {
 }
 
 /**
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {string} query
  */
 async function run(source, query) {
diff --git a/test/core/query-sql-budget.test.js b/test/core/query-sql-budget.test.js
index 834ac89a..efa097c9 100644
--- a/test/core/query-sql-budget.test.js
+++ b/test/core/query-sql-budget.test.js
@@ -50,11 +50,15 @@ function memorySource(rows, opts = {}) {
   return source
 }
 
-/** @param {AsyncDataSource} source */
-function registryFor(source) {
+/**
+ * @param {AsyncDataSource} source
+ * @param {{ localOnlyContentColumns?: string[] }} [extras]
+ */
+function registryFor(source, extras = {}) {
   const dataset = {
     discoverPartitions: async () => [],
     createDataSource: async () => source,
+    ...extras,
   }
   return /** @type {any} */ ({ getDataset: () => dataset, listDatasets: () => [] })
 }
@@ -167,6 +171,166 @@ test('the streaming-aggregate scanColumn fast path stays lit through the budget
   assert.deepEqual(scanColumnCalls, ['a'], 'the engine consumed the column stream, not buffered rows')
 })
 
+test('the budget decoration preserves the scan receiver', async () => {
+  const source = memorySource([{ a: 1 }])
+  const scan = /** @type {NonNullable<AsyncDataSource['scan']>} */ (source.scan)
+  source.scan = function scanWithReceiver(options) {
+    assert.equal(this, source)
+    return scan.call(source, options)
+  }
+  const result = await executeQuerySql({
+    query: 'SELECT a FROM t',
+    registry: registryFor(source),
+    storage,
+  })
+  assert.deepEqual(result.rows, [{ a: 1 }])
+})
+
+// @ref LLP 0294#transparent-wrappers [tests]: the heap decoration samples native batches without forcing the source back through scan()
+test('the prepared native-batch path stays lit through the budget decoration', async () => {
+  let preparedCalls = 0
+  const schema = {
+    fields: [{ id: 7, name: 'a', dataType: /** @type {const} */ ({ type: 'number' }), nullable: false }],
+  }
+  /** @type {AsyncDataSource} */
+  const source = {
+    columns: ['a'],
+    numRows: 4,
+    schema,
+    scan() {
+      throw new Error('legacy row scan should not run')
+    },
+    prepareScan(request) {
+      preparedCalls++
+      assert.deepEqual(request.columns.map((demand) => demand.field), [7])
+      return {
+        schema,
+        residual: {},
+        properties: { exactRows: 4, maxRows: 4 },
+        async *batches() {
+          yield {
+            selection: { type: 'all', length: 4 },
+            columns: [{ type: 'typed', values: new Float64Array([4, 3, 2, 1]), length: 4 }],
+          }
+        },
+      }
+    },
+  }
+  const result = await executeQuerySql({
+    query: 'SELECT MIN(a) AS n FROM t',
+    registry: registryFor(source),
+    storage,
+  })
+  assert.equal(result.rows[0].n, 1)
+  assert.equal(preparedCalls, 1)
+})
+
+test('a prepared-only source keeps its scanColumn fast path through the budget decoration', async () => {
+  const schema = {
+    fields: [{ id: 7, name: 'a', dataType: /** @type {const} */ ({ type: 'string' }), nullable: false }],
+  }
+  /** @type {string[]} */
+  const scanColumnCalls = []
+  /** @type {AsyncDataSource} */
+  const source = {
+    columns: ['a'],
+    schema,
+    prepareScan() {
+      throw new Error('prepared scan should not replace the column fast path')
+    },
+    scanColumn({ column }) {
+      scanColumnCalls.push(column)
+      return {
+        appliedWhere: true,
+        appliedLimitOffset: true,
+        async *chunks() {
+          yield ['x', 'y', 'x']
+        },
+      }
+    },
+  }
+  const result = await executeQuerySql({
+    query: 'SELECT COUNT(DISTINCT a) AS n FROM t',
+    registry: registryFor(source),
+    storage,
+  })
+  assert.equal(result.rows[0].n, 2)
+  assert.deepEqual(scanColumnCalls, ['a'])
+})
+
+test('the native budget guard samples a lazy column immediately after materialization', async () => {
+  const rowCount = 200_000
+  const schema = {
+    fields: [{ id: 7, name: 'a', dataType: /** @type {const} */ ({ type: 'string' }), nullable: false }],
+  }
+  let readCalls = 0
+  /** @type {AsyncDataSource} */
+  const source = {
+    columns: ['a'],
+    schema,
+    prepareScan() {
+      return {
+        schema,
+        residual: {},
+        properties: { exactRows: rowCount, maxRows: rowCount },
+        async *batches() {
+          yield {
+            selection: { type: 'all', length: rowCount },
+            columns: [{
+              async read() {
+                readCalls++
+                return {
+                  type: 'values',
+                  values: Array.from({ length: rowCount }, (_, i) => `retained-${i}-${'x'.repeat(100)}`),
+                  length: rowCount,
+                }
+              },
+            }],
+          }
+        },
+      }
+    },
+  }
+  await assert.rejects(
+    executeQuerySql({
+      query: 'SELECT MIN(a) AS n FROM t',
+      registry: registryFor(source),
+      storage,
+      maxHeapBytes: 8 * 1024 * 1024,
+    }),
+    (err) => {
+      assert.ok(err instanceof QueryExecutionBudgetError)
+      assert.equal(err.diagnostics?.site, 'native_batch')
+      return true
+    }
+  )
+  assert.equal(readCalls, 1)
+})
+
+test('a governable prepared-only source refuses restricted visibility without a row scan', async () => {
+  const schema = {
+    fields: [{ id: 7, name: 'content', dataType: /** @type {const} */ ({ type: 'string' }), nullable: true }],
+  }
+  /** @type {AsyncDataSource} */
+  const source = {
+    columns: ['content'],
+    schema,
+    prepareScan() {
+      throw new Error('privacy refusal must happen before the prepared scan')
+    },
+  }
+  await assert.rejects(
+    executeQuerySql({
+      query: 'SELECT content FROM t',
+      registry: registryFor(source, { localOnlyContentColumns: ['content'] }),
+      storage,
+    }),
+    {
+      message: 'Dataset "t" must provide columns and scan() to enforce local-only visibility',
+    }
+  )
+})
+
 test('transient scan garbage does not trip the budget; only retained growth refuses', async () => {
   // The guard confirms a crossing with a forced GC before refusing (LLP
   // 0097#confirm-with-gc). Each chunk allocates ~25MB, holds it long
diff --git a/test/core/star-expansion-drifted-union.test.js b/test/core/star-expansion-drifted-union.test.js
index faf1bedc..51f061dd 100644
--- a/test/core/star-expansion-drifted-union.test.js
+++ b/test/core/star-expansion-drifted-union.test.js
@@ -31,6 +31,7 @@ import {
 
 /**
  * @import { ColumnSpec, QueryScope } from '../../hypaware-plugin-kernel-types.js'
+ * @import { ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
  * @import { AsyncDataSource, AsyncRow, SqlPrimitive } from 'squirreling/src/types.js'
  */
 
@@ -58,7 +59,7 @@ const REMOTE = 'git@example.com:acme/app.git'
  * - `drifted`: TWO partitions, one with `git_remote` and one without.
  *
  * @param {'lone' | 'drifted'} shape
- * @returns {Promise<{ cacheRoot: string, source: AsyncDataSource }>}
+ * @returns {Promise<{ cacheRoot: string, source: ScannableDataSource }>}
  */
 async function stageFixture(shape) {
   const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), `hyp-star-${shape}-`))
@@ -82,7 +83,7 @@ async function stageFixture(shape) {
 
 /**
  * @param {'lone' | 'drifted'} shape
- * @param {(source: AsyncDataSource) => Promise<void>} body
+ * @param {(source: ScannableDataSource) => Promise<void>} body
  */
 async function withFixture(shape, body) {
   const { cacheRoot, source } = await stageFixture(shape)
@@ -97,7 +98,7 @@ async function withFixture(shape, body) {
  * Run a SELECT the way `hyp query sql` does, ordered by `id` so partition
  * scan order cannot make an assertion flap.
  *
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {string} query
  * @returns {Promise<Record<string, unknown>[]>}
  */
@@ -208,7 +209,7 @@ test('a clause above the scan reads a column some partition lacks without throwi
  *
  * @param {string[]} columns
  * @param {Record<string, SqlPrimitive>[]} objects
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 function narrowSource(columns, objects) {
   return {
diff --git a/test/core/union-source.test.js b/test/core/union-source.test.js
index 2e0d09e0..d2615339 100644
--- a/test/core/union-source.test.js
+++ b/test/core/union-source.test.js
@@ -9,7 +9,9 @@ import { normalizeScanColumn } from '../../src/core/query/scan-column.js'
 import { parquetSourceFromRows } from '../helpers/parquet_source_fixture.js'
 
 /**
- * @import { AsyncCells, AsyncDataSource, ExprNode, IdentifierNode, ScanColumnResults, ScanOptions, SqlPrimitive } from 'squirreling/src/types.js'
+ * @import { ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
+ * @import { AsyncCells, AsyncDataSource, ExprNode, Field, ScanOptions, ScanRequest, SqlPrimitive } from 'squirreling'
+ * @import { IdentifierNode, ScanColumnResults } from 'squirreling/src/types.js'
  * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js'
  */
 
@@ -20,7 +22,7 @@ import { parquetSourceFromRows } from '../helpers/parquet_source_fixture.js'
  *
  * @param {Record<string, SqlPrimitive>[]} rows
  * @param {ScanOptions[]} seenOptions
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 function fakeSource(rows, seenOptions) {
   const columns = Object.keys(rows[0] ?? {})
@@ -54,6 +56,13 @@ test('unionSources unions columns and sums numRows', () => {
   assert.equal(union.numRows, 3)
 })
 
+test('unionSources leaves numRows unknown when any partition count is unknown', () => {
+  const known = fakeSource([{ a: 1 }], [])
+  const unknown = fakeSource([{ a: 2 }], [])
+  unknown.numRows = undefined
+  assert.equal(unionSources([known, unknown]).numRows, undefined)
+})
+
 test('unionSources does not forward limit/offset to sub-sources', async () => {
   /** @type {ScanOptions[]} */
   const seen = []
@@ -84,6 +93,182 @@ test('unionSources does not forward limit/offset to sub-sources', async () => {
   }
 })
 
+/**
+ * A native-batch source whose legacy row scan throws, so a successful query
+ * proves every wrapper stayed on prepareScan. Each instance may use different
+ * field ids, matching separately-created Iceberg tables.
+ *
+ * @param {Field[]} fields
+ * @param {Record<string, SqlPrimitive>[]} rows
+ * @param {ScanRequest[]} seen
+ * @returns {ScannableDataSource}
+ */
+function preparedSource(fields, rows, seen) {
+  const columns = fields.map((field) => field.name)
+  return {
+    columns,
+    numRows: rows.length,
+    schema: { fields },
+    scan() {
+      throw new Error('legacy row scan should not run')
+    },
+    prepareScan(request) {
+      seen.push(request)
+      const requestedFields = request.columns.map((demand) => {
+        const field = fields.find((candidate) => candidate.id === demand.field)
+        if (!field) throw new Error(`unknown test field ${demand.field}`)
+        return field
+      })
+      return {
+        schema: { fields: requestedFields },
+        residual: { filter: request.filter },
+        properties: { maxRows: rows.length },
+        async *batches() {
+          yield {
+            selection: { type: 'all', length: rows.length },
+            columns: requestedFields.map((field) => ({
+              type: 'values',
+              values: rows.map((row) => row[field.name]),
+              length: rows.length,
+            })),
+          }
+        },
+      }
+    },
+  }
+}
+
+/**
+ * A prepared source with a working row scan and configurable negotiation.
+ * Its native batches throw so a successful query proves the union selected
+ * the row fallback.
+ *
+ * @param {Field[]} fields
+ * @param {Record<string, SqlPrimitive>[]} rows
+ * @param {{ appliesFilter?: boolean, mismatchesEmptySchema?: boolean }} [options]
+ * @returns {ScannableDataSource}
+ */
+function fallbackPreparedSource(fields, rows, options = {}) {
+  const source = fakeSource(rows, [])
+  source.schema = { fields }
+  source.prepareScan = (request) => {
+    const requestedFields = request.columns.map((demand) => {
+      const field = fields.find((candidate) => candidate.id === demand.field)
+      if (!field) throw new Error(`unknown test field ${demand.field}`)
+      return field
+    })
+    return {
+      schema: { fields: options.mismatchesEmptySchema && request.columns.length === 0 ? fields : requestedFields },
+      residual: { filter: options.appliesFilter ? undefined : request.filter },
+      properties: {},
+      async *batches() {
+        throw new Error('native batches should not run after incompatible negotiation')
+      },
+    }
+  }
+  return source
+}
+
+// @ref LLP 0294#partition-union [tests]: field ids are local to each table, filters may prune each child, and ranges belong to the concatenated stream
+test('unionSources concatenates prepared batches, remaps field ids, and keeps range hints global', async () => {
+  /** @type {ScanRequest[]} */
+  const seenA = []
+  /** @type {ScanRequest[]} */
+  const seenB = []
+  /** @type {Field[]} */
+  const fieldsA = [
+    { id: 1, name: 'k', dataType: { type: 'string' }, nullable: false },
+    { id: 2, name: 'v', dataType: { type: 'number' }, nullable: false },
+  ]
+  /** @type {Field[]} */
+  const fieldsB = [
+    { id: 101, name: 'k', dataType: { type: 'string' }, nullable: false },
+    { id: 102, name: 'v', dataType: { type: 'number' }, nullable: false },
+  ]
+  const union = unionSources([
+    preparedSource(fieldsA, [{ k: 'x', v: 1 }, { k: 'y', v: 2 }], seenA),
+    preparedSource(fieldsB, [{ k: 'x', v: 3 }, { k: 'x', v: 4 }], seenB),
+  ])
+
+  assert.ok(union.schema)
+  assert.equal(typeof union.prepareScan, 'function')
+  const rows = await collect(executeSql({
+    tables: { t: union },
+    query: "SELECT v FROM t WHERE k = 'x' LIMIT 2 OFFSET 1",
+  }))
+  assert.deepEqual(rows, [{ v: 3 }, { v: 4 }])
+  for (const request of [...seenA, ...seenB]) {
+    assert.ok(request.filter, 'filter is forwarded for per-table pruning')
+    assert.equal(request.limit, undefined, 'limit remains global')
+    assert.equal(request.offset, undefined, 'offset remains global')
+  }
+  assert.deepEqual(seenA[0].columns.map((demand) => demand.field), [2, 1])
+  assert.deepEqual(seenB[0].columns.map((demand) => demand.field), [102, 101])
+})
+
+test('unionSources declines prepared batches when partition schemas drift', () => {
+  const seen = []
+  const older = preparedSource([
+    { id: 1, name: 'id', dataType: { type: 'number' }, nullable: false },
+  ], [{ id: 1 }], seen)
+  const newer = preparedSource([
+    { id: 1, name: 'id', dataType: { type: 'number' }, nullable: false },
+    { id: 2, name: 'extra', dataType: { type: 'string' }, nullable: true },
+  ], [{ id: 2, extra: 'x' }], seen)
+  const union = unionSources([older, newer])
+  assert.equal(union.schema, undefined)
+  assert.equal(union.prepareScan, undefined, 'row padding remains authoritative for drifted schemas')
+})
+
+test('unionSources falls back to rows when prepared children report different filter residuals', async () => {
+  /** @type {Field[]} */
+  const fields = [
+    { id: 1, name: 'k', dataType: { type: 'string' }, nullable: false },
+    { id: 2, name: 'v', dataType: { type: 'number' }, nullable: false },
+  ]
+  const union = unionSources([
+    fallbackPreparedSource(fields, [{ k: 'x', v: 1 }, { k: 'y', v: 2 }]),
+    fallbackPreparedSource(fields, [{ k: 'x', v: 3 }], { appliesFilter: true }),
+  ])
+  const rows = await collect(executeSql({
+    tables: { t: union },
+    query: "SELECT v FROM t WHERE k = 'x'",
+  }))
+  assert.deepEqual(rows, [{ v: 1 }, { v: 3 }])
+})
+
+test('unionSources row fallback preserves counts with an empty prepared projection', async () => {
+  /** @type {Field[]} */
+  const fields = [{ id: 1, name: 'v', dataType: { type: 'number' }, nullable: false }]
+  const first = fallbackPreparedSource(fields, [{ v: 1 }, { v: 2 }])
+  const second = fallbackPreparedSource(fields, [{ v: 3 }], { mismatchesEmptySchema: true })
+  first.numRows = undefined
+  const rows = await collect(executeSql({ tables: { t: unionSources([first, second]) }, query: 'SELECT COUNT(*) AS n FROM t' }))
+  assert.deepEqual(rows, [{ n: 3 }])
+})
+
+test('unionSources treats equivalent data types as compatible regardless of object key order', () => {
+  /** @type {Field[]} */
+  const fieldsA = [{
+    id: 1,
+    name: 'values',
+    dataType: { type: 'array', items: { type: 'string' } },
+    nullable: false,
+  }]
+  /** @type {Field[]} */
+  const fieldsB = [{
+    id: 2,
+    name: 'values',
+    dataType: { items: { type: 'string' }, type: 'array' },
+    nullable: false,
+  }]
+  const union = unionSources([
+    preparedSource(fieldsA, [{ values: ['a'] }], []),
+    preparedSource(fieldsB, [{ values: ['b'] }], []),
+  ])
+  assert.equal(typeof union.prepareScan, 'function')
+})
+
 /**
  * A `col = value` predicate as a squirreling ExprNode.
  *
@@ -135,7 +320,7 @@ const PARQUET_PARTITION_COLUMNS = [
  * test below exercises actual hyparquet reads and pushdown, not a fake source.
  *
  * @param {Record<string, SqlPrimitive>[]} rows
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function makeParquetPartition(rows) {
   return parquetSourceFromRows(PARQUET_PARTITION_COLUMNS, rows, { rowGroupSize: 2 })
@@ -235,10 +420,10 @@ test('unionSources tolerates a scan with no options', async () => {
  * flags) to a fake source, honoring its own limit/offset. Exercises the
  * union's normalization shim for pre-0.15 plugin sources.
  *
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {Record<string, SqlPrimitive>[]} rows
  * @param {{ column: string, where?: ExprNode, limit?: number, offset?: number }[]} seenColumnScans
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 function withFakeScanColumn(source, rows, seenColumnScans) {
   source.scanColumn = ({ column, where, limit, offset }) => ({
@@ -257,10 +442,10 @@ function withFakeScanColumn(source, rows, seenColumnScans) {
  * applies an equality `where` like the icebird source does, reporting
  * `appliedWhere` honestly.
  *
- * @param {AsyncDataSource} source
+ * @param {ScannableDataSource} source
  * @param {Record<string, SqlPrimitive>[]} rows
  * @param {{ column: string, where?: ExprNode, limit?: number, offset?: number }[]} seenColumnScans
- * @returns {AsyncDataSource}
+ * @returns {ScannableDataSource}
  */
 function withFlaggedScanColumn(source, rows, seenColumnScans) {
   source.scanColumn = ({ column, where, limit, offset }) => {
@@ -452,7 +637,7 @@ const DRIFT_BASE_COLUMNS = [
  * Two real parquet partitions with additive drift: `extra` exists only in the
  * newer one, the shape a cache takes on the day a dataset gains a column.
  *
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 async function driftedUnion() {
   const older = await parquetSourceFromRows(DRIFT_BASE_COLUMNS, [
@@ -547,7 +732,7 @@ test('a partition whose rows carry no resolved map reads the same, because the u
    *
    * @param {string[]} columns
    * @param {Record<string, SqlPrimitive>[]} rows
-   * @returns {AsyncDataSource}
+   * @returns {ScannableDataSource}
    */
   function unresolvedSource(columns, rows) {
     return {
diff --git a/test/helpers/parquet_source_fixture.js b/test/helpers/parquet_source_fixture.js
index ff8281c6..1885327f 100644
--- a/test/helpers/parquet_source_fixture.js
+++ b/test/helpers/parquet_source_fixture.js
@@ -17,7 +17,8 @@ import { rowsToColumnSources } from '../../hypaware-core/plugins-workspace/forma
 
 /**
  * @import { AsyncBuffer } from 'hyparquet'
- * @import { AsyncDataSource, SqlPrimitive } from 'squirreling/src/types.js'
+ * @import { ScannableDataSource } from '../../hypaware-plugin-kernel-types.js'
+ * @import { SqlPrimitive } from 'squirreling'
  * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js'
  */
 
@@ -44,7 +45,7 @@ export function asyncBufferFromBytes(bytes) {
  * @param {ColumnSpec[]} columns
  * @param {Record<string, SqlPrimitive>[]} rows
  * @param {{ rowGroupSize?: number }} [options] `rowGroupSize` forces multi-row-group iteration
- * @returns {Promise<AsyncDataSource>}
+ * @returns {Promise<ScannableDataSource>}
  */
 export async function parquetSourceFromRows(columns, rows, options = {}) {
   const columnData = rowsToColumnSources(columns, rows)
diff --git a/test/plugins/claude-telemetry-events-dataset.test.js b/test/plugins/claude-telemetry-events-dataset.test.js
index 3030a8f2..0b0fd008 100644
--- a/test/plugins/claude-telemetry-events-dataset.test.js
+++ b/test/plugins/claude-telemetry-events-dataset.test.js
@@ -305,9 +305,11 @@ test('rows written through storage flush, discover, and read back through the re
     assert.equal(partitions[0].tablePath, path.join(cacheRoot, 'datasets', TELEMETRY_EVENTS_DATASET, 'all'))
 
     const source = await registration.createDataSource(partitions, /** @type {any} */ ({ scope: {}, storage }))
+    const scan = source.scan
+    assert.ok(scan, 'cache-backed registration retains the row scan')
     /** @type {Record<string, unknown>[]} */
     const seen = []
-    for await (const row of source.scan({}).rows()) {
+    for await (const row of scan.call(source, {}).rows()) {
       if (/** @type {any} */ (row).resolved) seen.push(/** @type {any} */ (row).resolved)
     }
     assert.equal(seen.length, 2)