MaterializeStore — Materialize streaming analytics store

MaterializeStore — Materialize streaming analytics store Overview Materialize is a streaming (incrementally-updated) analytics database: you define SOURCEs over event streams and MATERIALIZED VIEWs whose results are kept continuously up to date as new data arrives, then read them with low-latency SQL. Materialize speaks the PostgreSQL wire protocol, so it is driven with the pg client over a normal Postgres-style connection. Although the wire is Postgres, Materialize's compute model (streaming dataflows, SUBSCRIBE, CREATE SOURCE) is not a general OLTP relational store and does not fit this ORM's SQL Dialect abstraction, so it is modelled as a NoSqlStore (src/nosql/store.ts) exposing a direct query surface plus source/view helpers. Identity: Property | Value | --------- | --------------- | name | 'materialize' | library | 'pg' | Driver: pg (PostgreSQL-wire), lazy-loaded The pg package is an optional peer dependency. It is lazily required (require('pg')) inside connect(), so importing this module never forces it to be installed — it is only needed when the store actually connects and builds a real client. Injected client MaterializeStoreOptions accepts a pre-built client — a pg Client (or a compatible mock) implementing the small PgClientLike shape. When provided, connect() uses it directly and skips require('pg') entirely; either way connect() then calls client.connect(). This is how the test suite injects a mock without the real driver installed. Connection Build a store from connection options and call connect(): Connection options (all optional; map onto pg client config): Option | Type | Purpose | ------------------- | -------------------------- | ------------------------------------------------------------------------------- | host | string | Materialize host. | port | number | SQL port (Materialize default 6875). | user | string | User name. | password | string | Password. | database | string | Database. | connectionString | string | Full postgres://... connection string, an alternative to the discrete fields. | connectionOptions | Record<string, unknown> | Any other pg client option, spread into the config. | client | PgClientLike | Pre-built pg Client (or mock). When set, require('pg') is skipped. | Injected-client form Methods Failures from the driver are wrapped in a DatabaseError (via DatabaseError.from, message prefixed Materialize query failed:, carrying the offending sql). Using query/getClient and the DDL helpers before connect() (or after disconnect()) throws a ConnectionError. Lifecycle Method | Signature | Behavior | ------------- | ----------------------------- | ------------------------------------------------------------------------------------------------- | connect | connect(): Promise<void> | Uses an injected client, otherwise lazy-requires pg and new pg.Client(...); then calls client.connect(). Idempotent. Failures wrapped in ConnectionError. | disconnect | disconnect(): Promise<void> | Calls the client's end() and clears state. | isConnected | isConnected(): boolean | true only when connected and a client is present. | getClient | getClient(): PgClientLike | Returns the underlying pg client. Throws ConnectionError if not connected. | Query + streaming DDL Method | Signature | Behavior | ------------------------ | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------- | query | query<T = Record<string, unknown>>(sql: string, params?: unknown[]): Promise<T[]> | Executes SQL over the Postgres wire with optional positional params, returning pg's result.rows. | createSource | createSource(name: string, definition: string): Promise<any[]> | Convenience helper — runs CREATE SOURCE <name> <definition> via query() and returns the result rows. | createMaterializedView | createMaterializedView(name: string, selectSql: string): Promise<any[]> | Convenience helper — runs CREATE MATERIALIZED VIEW <name> AS <selectSql> (kept incrementally up to date) via query() and returns the result rows. | Example Verification status Unit / mock-verified only. The tests in tests/nosql/materialize.test.ts are fully mock-driven: pg is not installed, so every test injects a mock client (with connect/query/end spies where query() resolves to { rows }) via MaterializeStoreOptions.client, which makes connect() skip require('pg'). There is no network and no live Materialize server. What this proves: connect() calls client.connect() exactly once and is idempotent; a failing client.connect() is wrapped in ConnectionError and leaves the store disconnected. query() forwards sql + params and returns result.rows. createSource() issues CREATE SOURCE kafkasrc FROM KAFKA ... and createMaterializedView() issues CREATE MATERIALIZED VIEW mv AS ... (each with params undefined). Lifecycle: injected-client connect, disconnect() calling client.end(), and ConnectionError from getClient() before connect(). Error handling: a failing query() wrapped in DatabaseError. What this does not prove: live execution against a real Materialize instance over the Postgres wire. Statement/parameter shapes are asserted against pg's documented contract, but end-to-end execution and actual streaming/incremental-view semantics have not been exercised here. Related reading All data stores — the full catalogue, grouped by purpose Database types — where this sits among the 22 categories