feat: add lazy materialization, result caching, and query plan inspection

- Add .count(), .exists(), .first() short-circuit terminals that avoid
  full row materialization (count with no filters is zero I/O)
- Add .explain() for query plan inspection without execution
- Add .cache({ ttl }) for query result caching with TTL and auto-
  invalidation on table updates (VipCache extended with TTL support)
- Add .cursor({ batchSize }) for lazy page-at-a-time batch iteration
  with early termination and backpressure support

Author: teamchong

src/client.ts

@@ -1,6 +1,7 @@
 import type {
   AggregateOp,
   AppendResult,
+  ExplainResult,
   FilterOp,
   QueryResult,
   Row,
@@ -29,6 +30,7 @@ export class TableQuery {
   private _vectorSearch?: VectorSearchParams;
   private _aggregates: AggregateOp[] = [];
   private _groupBy: string[] = [];
+  private _cacheTTL?: number;
   private _executor: QueryExecutor;
 
   constructor(table: string, executor: QueryExecutor) {
@@ -84,6 +86,56 @@ export class TableQuery {
     return this;
   }
 
+  /** Enable query result caching with a TTL in milliseconds. Remote executor only. */
+  cache(opts: { ttl: number }): this {
+    this._cacheTTL = opts.ttl;
+    return this;
+  }
+
+  /** Return the count of matching rows without full materialization. */
+  async count(): Promise<number> {
+    if (this._executor.count) return this._executor.count(this.toDescriptor());
+    const desc = this.toDescriptor();
+    desc.aggregates = [{ fn: "count", column: "*" }];
+    const result = await this._executor.execute(desc);
+    return (result.rows[0]?.["count_*"] as number) ?? 0;
+  }
+
+  /** Return true if at least one row matches. */
+  async exists(): Promise<boolean> {
+    if (this._executor.exists) return this._executor.exists(this.toDescriptor());
+    const desc = this.toDescriptor();
+    desc.limit = 1;
+    if (desc.projections.length === 0) desc.projections = [];
+    const result = await this._executor.execute(desc);
+    return result.rowCount > 0;
+  }
+
+  /** Return the first matching row, or null. */
+  async first(): Promise<Row | null> {
+    if (this._executor.first) return this._executor.first(this.toDescriptor());
+    const desc = this.toDescriptor();
+    desc.limit = 1;
+    const result = await this._executor.execute(desc);
+    return result.rows[0] ?? null;
+  }
+
+  /** Inspect the query plan without executing. No data I/O is performed. */
+  async explain(): Promise<ExplainResult> {
+    if (!this._executor.explain) {
+      throw new Error("explain() requires an executor with plan inspection support");
+    }
+    return this._executor.explain(this.toDescriptor());
+  }
+
+  /** Iterate over results in batches. Processes pages lazily — stops when consumer breaks. */
+  cursor(opts?: { batchSize?: number }): AsyncIterable<Row[]> {
+    if (!this._executor.cursor) {
+      throw new Error("cursor() requires an executor with cursor support");
+    }
+    return this._executor.cursor(this.toDescriptor(), opts?.batchSize ?? 1000);
+  }
+
   /** Append rows to this table. Uses CAS coordination for safe concurrent writes. */
   async append(rows: Record<string, unknown>[]): Promise<AppendResult> {
     if (!this._executor.append) {
@@ -116,6 +168,7 @@ export class TableQuery {
       vectorSearch: this._vectorSearch,
       aggregates: this._aggregates.length > 0 ? this._aggregates : undefined,
       groupBy: this._groupBy.length > 0 ? this._groupBy : undefined,
+      cacheTTL: this._cacheTTL,
     };
   }
 }
@@ -131,6 +184,7 @@ export interface QueryDescriptor {
   vectorSearch?: VectorSearchParams;
   aggregates?: AggregateOp[];
   groupBy?: string[];
+  cacheTTL?: number;
 }
 
 /** Interface for query execution backends (local, DO, browser) */
@@ -140,4 +194,14 @@ export interface QueryExecutor {
   append?(table: string, rows: Record<string, unknown>[]): Promise<AppendResult>;
   /** Optional: streaming execution (available on remote executors) */
   executeStream?(query: QueryDescriptor): Promise<ReadableStream<Row>>;
+  /** Optional: count without full materialization */
+  count?(query: QueryDescriptor): Promise<number>;
+  /** Optional: existence check with limit(1) */
+  exists?(query: QueryDescriptor): Promise<boolean>;
+  /** Optional: return first matching row */
+  first?(query: QueryDescriptor): Promise<Row | null>;
+  /** Optional: plan inspection without execution */
+  explain?(query: QueryDescriptor): Promise<ExplainResult>;
+  /** Optional: lazy batch iteration */
+  cursor?(query: QueryDescriptor, batchSize: number): AsyncIterable<Row[]>;
 }

src/index.ts

@@ -1,10 +1,11 @@
 import { TableQuery } from "./client.js";
 import type { QueryDescriptor, QueryExecutor } from "./client.js";
-import type { AppendResult, ColumnMeta, Env, QueryResult, Row, TableMeta, DatasetMeta } from "./types.js";
+import type { AppendResult, ColumnMeta, DataType, Env, ExplainResult, QueryResult, Row, TableMeta, DatasetMeta } from "./types.js";
 import { parseFooter, parseColumnMetaFromProtobuf, FOOTER_SIZE } from "./footer.js";
 import { parseManifest, type ManifestInfo } from "./manifest.js";
 import { detectFormat, getParquetFooterLength, parseParquetFooter, parquetMetaToTableMeta } from "./parquet.js";
 import { assembleRows, canSkipPage, bigIntReplacer } from "./decode.js";
+import { coalesceRanges } from "./coalesce.js";
 import { instantiateWasm, WasmEngine } from "./wasm-engine.js";
 
 export { MasterDO } from "./master-do.js";
@@ -31,6 +32,7 @@ export type {
   IcebergSchema,
   IcebergDatasetMeta,
   AppendResult,
+  ExplainResult,
   VectorIndexInfo,
 } from "./types.js";
 
@@ -203,6 +205,39 @@ class RemoteExecutor implements QueryExecutor {
       },
     });
   }
+
+  private async postQuery<T>(path: string, query: QueryDescriptor): Promise<T> {
+    const queryDo = this.getQueryDo();
+    const response = await queryDo.fetch(new Request(`http://internal${path}`, {
+      method: "POST",
+      body: JSON.stringify(query, bigIntReplacer),
+      headers: { "content-type": "application/json" },
+    }));
+    if (!response.ok) {
+      const error = await response.text();
+      throw new Error(`QueryMode ${path} failed: ${error}`);
+    }
+    return response.json() as Promise<T>;
+  }
+
+  async count(query: QueryDescriptor): Promise<number> {
+    const body = await this.postQuery<{ count: number }>("/query/count", query);
+    return body.count;
+  }
+
+  async exists(query: QueryDescriptor): Promise<boolean> {
+    const body = await this.postQuery<{ exists: boolean }>("/query/exists", query);
+    return body.exists;
+  }
+
+  async first(query: QueryDescriptor): Promise<Row | null> {
+    const body = await this.postQuery<{ row: Row | null }>("/query/first", query);
+    return body.row;
+  }
+
+  async explain(query: QueryDescriptor): Promise<ExplainResult> {
+    return this.postQuery<ExplainResult>("/query/explain", query);
+  }
 }
 
 /**
@@ -321,6 +356,181 @@ class LocalExecutor implements QueryExecutor {
     };
   }
 
+  /** Count matching rows. No-filter case uses metadata only (zero I/O). */
+  async count(query: QueryDescriptor): Promise<number> {
+    const meta = await this.getOrLoadMeta(query.table);
+    if (query.filters.length === 0) {
+      return meta.columns[0]?.pages.reduce((s, p) => s + p.rowCount, 0) ?? 0;
+    }
+    // With filters: fall through to aggregate path
+    const desc = { ...query, aggregates: [{ fn: "count" as const, column: "*" }] };
+    const result = await this.execute(desc);
+    return (result.rows[0]?.["count_*"] as number) ?? 0;
+  }
+
+  async exists(query: QueryDescriptor): Promise<boolean> {
+    const desc = { ...query, limit: 1 };
+    const result = await this.execute(desc);
+    return result.rowCount > 0;
+  }
+
+  async first(query: QueryDescriptor): Promise<Row | null> {
+    const desc = { ...query, limit: 1 };
+    const result = await this.execute(desc);
+    return result.rows[0] ?? null;
+  }
+
+  async explain(query: QueryDescriptor): Promise<ExplainResult> {
+    const meta = await this.getOrLoadMeta(query.table);
+    const { columns } = meta;
+    const projectedColumns = query.projections.length > 0
+      ? columns.filter(c => query.projections.includes(c.name))
+      : columns;
+
+    let pagesTotal = 0;
+    let pagesSkipped = 0;
+    const ranges: { column: string; offset: number; length: number }[] = [];
+    const colDetails: ExplainResult["columns"] = [];
+
+    for (const col of projectedColumns) {
+      let colBytes = 0;
+      let colPages = 0;
+      for (const page of col.pages) {
+        pagesTotal++;
+        if (!query.vectorSearch && canSkipPage(page, query.filters, col.name)) {
+          pagesSkipped++;
+          continue;
+        }
+        colPages++;
+        colBytes += page.byteLength;
+        ranges.push({ column: col.name, offset: Number(page.byteOffset), length: page.byteLength });
+      }
+      colDetails.push({ name: col.name, dtype: col.dtype as DataType, pages: colPages, bytes: colBytes });
+    }
+
+    const coalesced = coalesceRanges(ranges, 64 * 1024);
+    const estimatedBytes = ranges.reduce((s, r) => s + r.length, 0);
+    const totalRows = columns[0]?.pages.reduce((s, p) => s + p.rowCount, 0) ?? 0;
+
+    return {
+      table: query.table,
+      format: "lance",
+      totalRows,
+      columns: colDetails,
+      pagesTotal,
+      pagesSkipped,
+      pagesScanned: pagesTotal - pagesSkipped,
+      estimatedBytes,
+      estimatedR2Reads: coalesced.length,
+      fragments: 1,
+      filters: query.filters.map(f => ({
+        column: f.column,
+        op: f.op,
+        pushable: f.op !== "in" && f.op !== "neq",
+      })),
+      metaCached: this.metaCache.has(query.table),
+    };
+  }
+
+  async *cursor(query: QueryDescriptor, batchSize: number): AsyncIterable<Row[]> {
+    const meta = await this.getOrLoadMeta(query.table);
+    const { columns } = meta;
+    const projectedColumns = query.projections.length > 0
+      ? columns.filter(c => query.projections.includes(c.name))
+      : columns;
+
+    const isUrl = query.table.startsWith("http://") || query.table.startsWith("https://");
+    const firstCol = projectedColumns[0];
+    if (!firstCol) return;
+
+    // If sorted, must buffer all rows then chunk
+    if (query.sortColumn) {
+      const result = await this.execute(query);
+      for (let i = 0; i < result.rows.length; i += batchSize) {
+        yield result.rows.slice(i, i + batchSize);
+      }
+      return;
+    }
+
+    const totalPages = firstCol.pages.length;
+    let pageIdx = 0;
+    let totalYielded = 0;
+
+    const fs = isUrl ? null : await import("node:fs/promises");
+    const handle = isUrl ? null : await fs!.open(query.table, "r");
+    const wasm = await this.getWasm();
+
+    try {
+      while (pageIdx < totalPages) {
+        let batchRows = 0;
+        const batchStartPage = pageIdx;
+        while (pageIdx < totalPages && batchRows < batchSize) {
+          const page = firstCol.pages[pageIdx];
+          if (!query.vectorSearch && canSkipPage(page, query.filters, firstCol.name)) {
+            pageIdx++;
+            continue;
+          }
+          batchRows += page.rowCount;
+          pageIdx++;
+        }
+
+        if (batchRows === 0) continue;
+
+        const columnData = new Map<string, ArrayBuffer[]>();
+        for (const col of projectedColumns) {
+          for (let pi = batchStartPage; pi < pageIdx; pi++) {
+            const page = col.pages[pi];
+            if (!page) continue;
+            if (!query.vectorSearch && canSkipPage(page, query.filters, col.name)) continue;
+
+            let ab: ArrayBuffer;
+            if (isUrl) {
+              const start = Number(page.byteOffset);
+              const end = start + page.byteLength - 1;
+              const resp = await fetch(query.table, { headers: { Range: `bytes=${start}-${end}` } });
+              ab = await resp.arrayBuffer();
+            } else {
+              const buf = Buffer.alloc(page.byteLength);
+              await handle!.read(buf, 0, page.byteLength, Number(page.byteOffset));
+              ab = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
+            }
+
+            const arr = columnData.get(col.name) ?? [];
+            arr.push(ab);
+            columnData.set(col.name, arr);
+          }
+        }
+
+        const rows = assembleRows(columnData, projectedColumns, query, wasm);
+
+        if (rows.length > 0) {
+          if (query.limit && totalYielded + rows.length > query.limit) {
+            yield rows.slice(0, query.limit - totalYielded);
+            return;
+          }
+          yield rows;
+          totalYielded += rows.length;
+        }
+      }
+    } finally {
+      if (handle) await handle.close();
+    }
+  }
+
+  /** Get or load table metadata (footer + columns). Caches in-memory. */
+  private async getOrLoadMeta(table: string): Promise<{ columns: ColumnMeta[]; fileSize: number }> {
+    let cached = this.metaCache.get(table);
+    if (cached) return cached;
+    const isUrl = table.startsWith("http://") || table.startsWith("https://");
+    cached = isUrl ? await this.loadMetaFromUrl(table) : await this.loadMetaFromFile(table);
+    this.metaCache.set(table, cached);
+    if (this.metaCache.size > 1000) {
+      const firstKey = this.metaCache.keys().next().value;
+      if (firstKey) this.metaCache.delete(firstKey);
+    }
+    return cached;
+  }
+
   async execute(query: QueryDescriptor): Promise<QueryResult> {
     const startTime = Date.now();
     const isUrl = query.table.startsWith("http://") || query.table.startsWith("https://");