Copy page

DAL Library

Streaming Large Result Sets

Streaming Large Result Sets

Relevant source files

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

• src/dal/dal.ts
• src/query/streaming.ts

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

Code
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

Code
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.

Code
// 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.

---