# Streaming Large Result Sets

# Streaming Large Result Sets

<details>
<summary>Relevant source files</summary>

The following files were used as context for generating this wiki page:

- [src/dal/dal.ts](src/dal/dal.ts)
- [src/query/streaming.ts](src/query/streaming.ts)

</details>



The `@primebrick/dal-pg` library provides native support for streaming large datasets using PostgreSQL cursors via the `pg-query-stream` integration. This allows applications to process millions of rows with constant memory overhead by avoiding the default behavior of buffering the entire result set into a JavaScript array.

## Implementation Overview

Streaming is primarily exposed through the `findAll` method in the `Repository` class when the `stream: true` option is provided [src/dal/dal.ts:29-30](). Internally, the library uses the `createStream` utility to wrap a `QueryStream` into an `AsyncIterable` [src/query/streaming.ts:4-6]().

### Data Flow: Query to AsyncIterable

When a stream is requested, the system bypasses the standard `pool.query()` execution path. Instead, it initializes a `QueryStream` which manages a server-side cursor.

1.  **SQL Generation**: The `Repository` builds the SQL text and parameters using the standard Query DSL.
2.  **Stream Creation**: `createStream()` is invoked with the `Queryable` (Pool or PoolClient), the SQL, and the parameters [src/query/streaming.ts:35-39]().
3.  **Lifecycle Management**: If a `Pool` is provided, the wrapper lazily acquires a `PoolClient` when iteration begins and ensures it is released back to the pool once the stream is exhausted or an error occurs [src/query/streaming.ts:45-58]().
4.  **Iteration**: The consumer uses a `for await...of` loop to process rows as they arrive from the database.

### Streaming Lifecycle and Entity Space

The following diagram illustrates how the `streaming.ts` module bridges the gap between the high-level `Dal` gateway and the low-level `pg-query-stream` cursor.

**Diagram: Streaming Execution Pipeline**
```mermaid
sequenceDiagram
    participant App as "Application Code"
    participant Dal as "Dal Gateway (src/dal/dal.ts)"
    participant Repo as "Repository (src/repository/repository.ts)"
    participant Streamer as "createStream (src/query/streaming.ts)"
    participant PQS as "pg-query-stream (QueryStream)"
    participant PG as "PostgreSQL Cursor"

    App->>Dal: findAll(Entity, { stream: true })
    Dal->>Repo: findAll(Entity, { stream: true })
    Repo->>Streamer: createStream(pool, sql, values)
    Streamer-->>App: Return AsyncIterable
    
    Note over App, PG: Iteration begins
    App->>Streamer: [Symbol.asyncIterator].next()
    Streamer->>Dal: pool.connect()
    Dal-->>Streamer: client (PoolClient)
    Streamer->>PQS: client.query(QueryStream)
    PQS->>PG: FETCH 100 (batch)
    PG-->>PQS: Row Data
    PQS-->>App: yield Row
    
    Note over App, PG: Iteration ends or errors
    Streamer->>Dal: client.release()
```
Sources: [src/query/streaming.ts:35-65](), [src/dal/dal.ts:29-31]().

## Connection Lifecycle Management

A critical distinction in the streaming implementation is how database connections are handled based on the input `Queryable` type [src/query/streaming.ts:17]().

| Input Type | Lifecycle Owner | Implementation Detail |
| :--- | :--- | :--- |
| `pg.Pool` | `createStream` | Acquires a client via `pool.connect()` on the first iteration and calls `client.release()` in a `finally` block [src/query/streaming.ts:45-58](). |
| `pg.PoolClient` | Caller | Uses the provided client directly. The caller must ensure the client is released or the transaction is committed/rolled back [src/query/streaming.ts:61-64](). |

## Timeouts and Batching

The streaming mechanism interacts specifically with PostgreSQL's `statement_timeout` setting. Unlike standard queries where the timeout covers the entire result transmission, streaming uses cursors.

### FETCH Batches
`pg-query-stream` fetches rows in batches (defaulting to 100 rows). Because each batch is a distinct network operation, the `statement_timeout` configured in the `Dal` instance (default 30s) applies to the individual `FETCH` commands rather than the total time the stream is open [src/dal/dal.ts:29-30](). This makes streaming naturally resilient to long-running processing logic in the application layer, as the database only "works" during the batch retrieval phase.

### Anti-Throttling Integration
The `Dal` gateway ensures that every connection retrieved from the pool has a `statement_timeout` set via the `onConnect` handler [src/dal/dal.ts:147-163](). This prevents a stalled stream (e.g., a consumer that stops pulling rows but doesn't close the stream) from holding a database backend process indefinitely, although the connection will remain "In Use" in the pool until the `AsyncIterable` is closed.

## Technical Entity Mapping

The following diagram maps the logical streaming components to the specific classes and functions in the codebase.

**Diagram: Code Entity Map for Streaming**
```mermaid
classDiagram
    class Dal {
        +pool: Pool
        +findAll(options: FindOptions)
    }
    class Repository {
        +findAll(options: FindOptions)
    }
    class StreamingModule {
        <<module>>
        +createStream(db: Queryable, text: string, values: any[])
    }
    class QueryStream {
        <<external>>
        +cursor: Cursor
    }

    Dal "1" *-- "1" Repository : delegates to
    Repository ..> StreamingModule : calls if stream=true
    StreamingModule ..> QueryStream : wraps
    StreamingModule ..> pg_PoolClient : manages lifecycle
```
Sources: [src/dal/dal.ts:101-170](), [src/query/streaming.ts:1-65]().

## Usage Example

The `findAll` method returns an `AsyncIterable` when `stream: true` is set, allowing for memory-efficient processing [src/query/streaming.ts:27-33]().

```typescript
// Example informed by src/query/streaming.ts and Repository signatures
const dal = getDal();
const userStream = await dal.findAll(UserEntity, { 
  stream: true,
  where: Filter.fieldValue("active", "=", true)
});

for await (const user of userStream) {
  // Process one user at a time
  // Memory usage remains constant regardless of table size
  await processUser(user);
}
```
Sources: [src/query/streaming.ts:27-33](), [src/dal/dal.ts:29-31]().

---