feat: add .pipe(), toOperator(), fromOperator() for composable pipelines

DataFrame.pipe(fn) injects custom operators mid-chain without breaking
the fluent API. toOperator() escapes to raw operators, fromOperator()
re-enters. PipeStage type and all operators now exported from index.

- pipe stages applied after filter/computed, before agg/sort in buildEdgePipeline
- 6 tests covering pipe basic, chaining, filter-after, immutability, fromOperator
- new docs/composability.mdx page: Unix pipes framing, three levels of control
- updated operators.mdx with pipe and escape hatch sections

Author: teamchong

docs/src/content/docs/composability.mdx

@@ -0,0 +1,121 @@
+---
+title: Composability
+description: Operators are building blocks. Your code assembles the pipeline.
+---
+
+## Operators are Unix pipes
+
+Every operator in QueryMode implements one interface:
+
+```typescript
+interface Operator {
+  next(): Promise<RowBatch | null>
+  close(): Promise<void>
+}
+```
+
+Each operator reads from upstream and writes downstream. Wrap one operator around another — that's composition. Same idea as `cat | grep | sort | uniq`, except the data is columnar row batches instead of text lines.
+
+```
+ScanOperator → FilterOperator → YourCustomOperator → AggregateOperator → ProjectOperator
+```
+
+## Three levels of control
+
+### 1. DataFrame chain
+
+The DataFrame API handles the common case. Filter, aggregate, sort, join — all without thinking about operators:
+
+```typescript
+const result = await qm.table("events")
+  .filter("status", "eq", "active")
+  .groupBy("region")
+  .aggregate("sum", "amount", "total")
+  .sort("total", "desc")
+  .limit(10)
+  .collect()
+```
+
+Under the hood, this builds a pipeline of operators. You don't see them, but they're there — doing page-level skip, SIMD decode, partial aggregation.
+
+### 2. `.pipe()` — inject custom operators mid-chain
+
+When you need something the built-in methods don't cover, `.pipe()` lets you inject any operator without leaving the chain:
+
+```typescript
+const result = await qm.table("events")
+  .filter("created_at", "gte", "2024-01-01")
+  .pipe(upstream => new ComputedColumnOperator(upstream, [
+    { alias: "risk_score", fn: row => riskModel.predict(row) },
+  ]))
+  .pipe(upstream => new FilterOperator(upstream, [
+    { column: "risk_score", op: "gt", value: 0.8 },
+  ]))
+  .sort("risk_score", "desc")
+  .limit(50)
+  .collect()
+```
+
+ML scoring inside the query pipeline. No round-trip to a separate service, no serialization. The model runs on the same row batch that the filter just produced.
+
+### 3. `toOperator()` / `fromOperator()` — full escape hatch
+
+For maximum control, break out of the DataFrame entirely:
+
+```typescript
+// Escape: DataFrame → raw Operator
+const op = await qm.table("events")
+  .filter("status", "eq", "active")
+  .toOperator()
+
+// Your code: arbitrary operator composition
+const enriched = new ComputedColumnOperator(op, [
+  { alias: "score", fn: row => score(row) },
+])
+const deduped = new DistinctOperator(enriched, ["user_id"])
+
+// Re-enter: raw Operator → DataFrame
+const result = await DataFrame.fromOperator(deduped, executor)
+  .sort("score", "desc")
+  .limit(10)
+  .collect()
+```
+
+You left the DataFrame, did whatever you wanted with raw operators, and came back. The DataFrame doesn't know or care what happened in between.
+
+## Write your own operators
+
+Any object that implements `next()` and `close()` is an operator:
+
+```typescript
+class SamplingOperator implements Operator {
+  constructor(
+    private upstream: Operator,
+    private rate: number,
+  ) {}
+
+  async next(): Promise<RowBatch | null> {
+    const batch = await this.upstream.next()
+    if (!batch) return null
+    return batch.filter(() => Math.random() < this.rate)
+  }
+
+  async close() {
+    await this.upstream.close()
+  }
+}
+
+// Use it
+const result = await qm.table("events")
+  .filter("status", "eq", "active")
+  .pipe(upstream => new SamplingOperator(upstream, 0.1))
+  .collect()
+```
+
+No framework, no plugin system, no registration. It's a two-method interface.
+
+## Why this matters
+
+Traditional query engines are black boxes. You send a SQL string, get rows back. You can't put a rate limiter between the scan and the filter. You can't run ML scoring before the aggregation. You can't swap the sort algorithm based on the data size.
+
+With QueryMode, the pipeline is your code. The operators do real query engine work — page-level skip, SIMD decode, memory-bounded spill — but you control how they're assembled. That's the difference between using a query engine and building with one.

docs/src/content/docs/operators.mdx

@@ -119,3 +119,65 @@ const scored = new ComputedColumnOperator(filtered, [
 ])
 const top10 = new TopKOperator(scored, "ml_score", true, 10)
 ```
+
+## Pipe: inject custom stages into the DataFrame
+
+The DataFrame API's `.pipe()` method lets you inject any custom operator into the pipeline without leaving the fluent chain:
+
+```typescript
+import { QueryMode, ComputedColumnOperator } from "querymode"
+
+const qm = QueryMode.local()
+const result = await qm.table("events")
+  .filter("created_at", "gte", "2024-01-01")
+  .pipe(upstream => new ComputedColumnOperator(upstream, [
+    { alias: "score", fn: row => scoreModel.predict(row) },
+  ]))
+  .sort("score", "desc")
+  .limit(10)
+  .collect()
+```
+
+Chain multiple `.pipe()` calls — each one wraps the previous stage:
+
+```typescript
+const result = await qm.table("events")
+  .filter("status", "eq", "active")
+  .pipe(upstream => new ComputedColumnOperator(upstream, [
+    { alias: "risk", fn: row => computeRisk(row) },
+  ]))
+  .pipe(upstream => new FilterOperator(upstream, [
+    { column: "risk", op: "gt", value: 0.8 },
+  ]))
+  .collect()
+```
+
+## Escape hatch: toOperator / fromOperator
+
+For full control, break out of the DataFrame into raw operators, then re-enter:
+
+```typescript
+// Escape: get the raw Operator pipeline
+const op = await qm.table("events")
+  .filter("status", "eq", "active")
+  .toOperator()
+
+// Your code: wrap with custom operators
+const scored = new ComputedColumnOperator(op, [
+  { alias: "score", fn: row => scoreModel.predict(row) },
+])
+const filtered = new FilterOperator(scored, [
+  { column: "score", op: "gt", value: 0.9 },
+])
+
+// Re-enter: wrap back into a DataFrame
+const result = await DataFrame.fromOperator(filtered, executor)
+  .sort("score", "desc")
+  .limit(10)
+  .collect()
+```
+
+Three levels of control:
+1. **DataFrame chain** — convenience methods for common operations
+2. **`.pipe()`** — inject custom operators without leaving the chain
+3. **`toOperator()` / `fromOperator()`** — full escape hatch for arbitrary composition

src/client.ts

@@ -14,6 +14,7 @@ import type {
   VectorSearchParams,
   WindowSpec,
 } from "./types.js";
+import type { Operator, RowBatch } from "./operators.js";
 
 // ---------------------------------------------------------------------------
 // Progress callback for collect()
@@ -99,6 +100,7 @@ export class DataFrame<T extends Row = Row> {
   private _version?: number;
   private _vectorEncoder?: (text: string) => Promise<Float32Array>;
   private _vectorQueryText?: string;
+  private _pipeStages: PipeStage[];
   private _executor: QueryExecutor;
 
   constructor(table: string, executor: QueryExecutor, init?: Partial<DataFrameInit>) {
@@ -124,6 +126,7 @@ export class DataFrame<T extends Row = Row> {
     this._version = init?.version;
     this._vectorEncoder = init?.vectorEncoder;
     this._vectorQueryText = init?.vectorQueryText;
+    this._pipeStages = init?.pipeStages ?? [];
   }
 
   /** Create a shallow clone with overrides — returns a new immutable DataFrame. */
@@ -149,6 +152,7 @@ export class DataFrame<T extends Row = Row> {
       version: this._version,
       vectorEncoder: this._vectorEncoder,
       vectorQueryText: this._vectorQueryText,
+      pipeStages: this._pipeStages,
       ...overrides,
     });
   }
@@ -332,6 +336,25 @@ export class DataFrame<T extends Row = Row> {
     return this.derive({ windows: [...this._windows, spec] }) as DataFrame;
   }
 
+  /**
+   * Inject a custom operator into the pipeline. The function receives the
+   * upstream Operator and must return a new Operator. Runs at collect() time.
+   *
+   *   const result = await qm.table("events")
+   *     .filter("created_at", "gte", startDate)
+   *     .pipe(upstream => new ComputedColumnOperator(upstream, [
+   *       { alias: "score", fn: row => model.predict(row) },
+   *     ]))
+   *     .sort("score", "desc")
+   *     .limit(10)
+   *     .collect()
+   */
+  pipe(fn: (upstream: Operator) => Operator): DataFrame {
+    return this.derive({
+      pipeStages: [...(this._pipeStages ?? []), fn],
+    }) as DataFrame;
+  }
+
   /** Deduplicate rows. If no columns specified, dedup on all columns. */
   distinct(...columns: string[]): DataFrame<T> {
     return this.derive({ distinct: columns.length > 0 ? columns : [] });
@@ -584,6 +607,47 @@ export class DataFrame<T extends Row = Row> {
     return this._executor.executeStream(this.toDescriptor());
   }
 
+  /**
+   * Escape hatch: get the raw Operator for this query. Resolves deferred
+   * subqueries and vector encoders, then returns an Operator you can wrap
+   * with your own operators before re-entering via DataFrame.fromOperator().
+   *
+   * Requires an executor that supports toOperator (local-executor, query-do).
+   */
+  async toOperator(): Promise<Operator> {
+    if (!this._executor.toOperator) {
+      throw new Error("toOperator() requires an executor with operator pipeline support");
+    }
+    const desc = await this.resolveDeferred();
+    return this._executor.toOperator(desc);
+  }
+
+  /**
+   * Re-entry: wrap a raw Operator back into a DataFrame for further chaining.
+   * The operator is drained at collect() time.
+   */
+  static fromOperator<R extends Row = Row>(op: Operator, executor: QueryExecutor): DataFrame<R> {
+    const drainExecutor: QueryExecutor = {
+      async execute(_query: QueryDescriptor): Promise<QueryResult> {
+        const rows: Row[] = [];
+        let batch: RowBatch | null;
+        while ((batch = await op.next()) !== null) {
+          rows.push(...batch);
+        }
+        await op.close();
+        return {
+          rows,
+          rowCount: rows.length,
+          columns: rows.length > 0 ? Object.keys(rows[0]) : [],
+          bytesRead: 0,
+          pagesSkipped: 0,
+          durationMs: 0,
+        };
+      },
+    };
+    return new DataFrame<R>("__from_operator__", drainExecutor);
+  }
+
   toDescriptor(): QueryDescriptor {
     return {
       table: this._table,
@@ -606,6 +670,7 @@ export class DataFrame<T extends Row = Row> {
         ? this._subqueryIn.map(sq => ({ column: sq.column, valueSet: sq.valueSet }))
         : undefined,
       version: this._version,
+      pipeStages: this._pipeStages.length > 0 ? this._pipeStages : undefined,
     };
   }
 }
@@ -736,6 +801,9 @@ export class MaterializedExecutor implements QueryExecutor {
 // DataFrameInit — internal state for derive()
 // ---------------------------------------------------------------------------
 
+/** A function that wraps an upstream Operator, returning a new Operator. */
+export type PipeStage = (upstream: Operator) => Operator;
+
 interface DataFrameInit {
   filters: FilterOp[];
   projections: string[];
@@ -757,6 +825,7 @@ interface DataFrameInit {
   version?: number;
   vectorEncoder?: (text: string) => Promise<Float32Array>;
   vectorQueryText?: string;
+  pipeStages: PipeStage[];
 }
 
 // ---------------------------------------------------------------------------
@@ -784,6 +853,7 @@ export interface QueryDescriptor {
   setOperation?: { mode: "union" | "union_all" | "intersect" | "except"; right: QueryDescriptor };
   subqueryIn?: { column: string; valueSet: Set<string> }[];
   version?: number;
+  pipeStages?: PipeStage[];
 }
 
 /** Interface for query execution backends (local, DO, browser) */
@@ -805,4 +875,6 @@ export interface QueryExecutor {
   explain?(query: QueryDescriptor): Promise<ExplainResult>;
   /** Optional: lazy batch iteration */
   cursor?(query: QueryDescriptor, batchSize: number): AsyncIterable<Row[]>;
+  /** Optional: return the raw Operator pipeline for escape-hatch composition */
+  toOperator?(query: QueryDescriptor): Promise<Operator>;
 }

src/index.ts

@@ -15,6 +15,14 @@ export { ReaderRegistry, FileDataSource, UrlDataSource } from "./reader.js";
 export type { FormatReader, DataSource } from "./reader.js";
 export { DataFrame, TableQuery, MaterializedExecutor } from "./client.js";
 export { LazyResultHandle } from "./client.js";
+export {
+  FilterOperator, AggregateOperator, TopKOperator, ProjectOperator,
+  ComputedColumnOperator, HashJoinOperator, ExternalSortOperator,
+  InMemorySortOperator, WindowOperator, DistinctOperator, SetOperator,
+  LimitOperator, SubqueryInOperator,
+  drainPipeline, buildEdgePipeline,
+} from "./operators.js";
+export type { Operator, RowBatch } from "./operators.js";
 export { QueryModeError } from "./errors.js";
 export type { ErrorCode } from "./errors.js";
 export { LocalExecutor } from "./local-executor.js";
@@ -29,7 +37,7 @@ export { HnswIndex, cosineDistance, l2DistanceSq, dotDistance } from "./hnsw.js"
 export type { HnswOptions } from "./hnsw.js";
 export { MaterializationCache, queryHashKey } from "./lazy.js";
 export type { MaterializationCacheOptions } from "./lazy.js";
-export type { QueryExecutor, QueryDescriptor, ProgressInfo, CollectOptions } from "./client.js";
+export type { QueryExecutor, QueryDescriptor, ProgressInfo, CollectOptions, PipeStage } from "./client.js";
 export type {
   Env,
   Footer,

src/operators.ts

@@ -1977,6 +1977,13 @@ export function buildEdgePipeline(
     pipeline = new ComputedColumnOperator(pipeline, query.computedColumns);
   }
 
+  // User-injected pipe stages (inserted after filter/computed, before agg/sort)
+  if (query.pipeStages) {
+    for (const stage of query.pipeStages) {
+      pipeline = stage(pipeline);
+    }
+  }
+
   // Window functions
   if (query.windows && query.windows.length > 0) {
     pipeline = new WindowOperator(pipeline, query.windows);