Use native prepared batches through query sources

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
 }
 

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)

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)

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)

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)

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)

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

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[]
 

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;

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

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

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.