# Serve results Serving results from Materialize In Materialize, indexed views and materialized views maintain up-to-date query results. This allows Materialize to serve fresh query results with low latency. To serve results, you can: - [Query using `SELECT` and `SUBSCRIBE` statements](/serve-results/query-results/) - [Use BI/data collaboration tools](/serve-results/bi-tools/) - [Sink results to to external systems](/export-data/) - [Use Foreign Data Wrapper (FDW)](/serve-results/fdw/)
SELECT/SUBSCRIBE statements
External BI tools
Sink results
--- ## `SELECT` and `SUBSCRIBE` You can query results from Materialize using `SELECT` and `SUBSCRIBE` SQL statements. Because Materialize uses the PostgreSQL wire protocol, it works out-of-the-box with a wide range of SQL clients and tools that support PostgreSQL. ## SELECT You can query data in Materialize using the [`SELECT` statement](/sql/select/). For example: ```mzsql SELECT region.id, sum(purchase.total) FROM mysql_simple_purchase AS purchase JOIN mysql_simple_user AS user ON purchase.user_id = user.id JOIN mysql_simple_region AS region ON user.region_id = region.id GROUP BY region.id; ``` Performing a `SELECT` on an indexed view or an indexed materialized view is Materialize's ideal operation. When Materialize receives such a `SELECT` query, it quickly returns the maintained results from memory. Materialize also quickly returns results for queries that only filter, project, transform with scalar functions, and re-order data that is maintained by an index. Queries that can't simply read out from an index will create an ephemeral dataflow to compute the results. These dataflows are bound to the active [cluster](/fundamentals/concepts/clusters/), which you can change using: ```mzsql SET cluster = ; ``` Materialize will remove the dataflow as soon as it has returned the query results to you. For more information, see [`SELECT`](/sql/select/) reference page. See also the following client library guides: ## SUBSCRIBE You can use [`SUBSCRIBE`](/sql/subscribe/) to stream query results. For example: ```mzsql BEGIN; DECLARE c CURSOR FOR SUBSCRIBE (SELECT * FROM mv_counter_sum); FETCH 10 c WITH (timeout='1s'); FETCH 20 c WITH (timeout='1s'); COMMIT; ``` The [`SUBSCRIBE`](/sql/subscribe/) statement is a more general form of a `SELECT` statement. While a `SELECT` statement computes a relation at a moment in time, a `SUBSCRIBE` operation computes how a relation changes over time. You can use `SUBSCRIBE` to: - Power event processors that react to every change to a relation or an arbitrary `SELECT` statement. - Replicate the complete history of a relation while `SUBSCRIBE` is active. > **Tip:** Use materialized view (instead of an indexed view) with `SUBSCRIBE`. For more information, see [`SUBSCRIBE`](/sql/subscribe/) reference page. See also the following client library guides: --- ## ADBC (Arrow Database Connectivity) [ADBC (Arrow Database Connectivity)](https://arrow.apache.org/adbc/) is a standard, columnar database API from Apache Arrow. Because Materialize is **wire-compatible** with PostgreSQL, the community [ADBC PostgreSQL driver](https://arrow.apache.org/adbc/current/driver/postgresql.html) (`adbc_driver_postgresql`) can connect directly to Materialize; i.e., no Materialize-specific driver is required. The benefit over a standard PostgreSQL client is the result format. ADBC returns query results as [Apache Arrow](https://arrow.apache.org/) tables, which you can hand directly to Arrow-native tools such as [DuckDB](https://duckdb.org/), [pandas](https://pandas.pydata.org/), and [Polars](https://pola.rs/) without an intermediate object-store or file hop. ## Prerequisites Requires **Materialize v26.40 or later**. When it opens a connection, the driver reads the binary send and receive functions of every type from `pg_catalog.pg_type`. Earlier versions of Materialize do not expose `pg_type.typsend`, so `connect()` fails before it can run a query. Install the driver manager, the PostgreSQL driver, and PyArrow. The examples on this page also use DuckDB: ```bash pip install adbc-driver-manager adbc-driver-postgresql pyarrow duckdb ``` The examples were tested with `adbc-driver-manager` and `adbc-driver-postgresql` 1.12.0, `pyarrow` 25.0.1, and `duckdb` 1.5.5. ## Connect Connect using `adbc_driver_postgresql.dbapi.connect` with a PostgreSQL connection URI. Use the connection as a context manager so that it closes when the block exits: ```python import adbc_driver_postgresql.dbapi from urllib.parse import quote_plus username = quote_plus("MATERIALIZE_USERNAME") password = quote_plus("APP_SPECIFIC_PASSWORD") uri = ( f"postgresql://{username}:{password}" "@MATERIALIZE_HOST:6875/materialize?sslmode=require" ) with adbc_driver_postgresql.dbapi.connect(uri) as conn: # Run queries here. ... ``` > **Tip:** Percent-encode the username and password before putting them in the URI, as in > the example above. Materialize usernames are email addresses, so they contain an > `@`, which delimits the host in a URI. App passwords can likewise contain > characters that a URI reserves. > **Note:** `sslmode=require` applies to Materialize Cloud, which accepts only TLS > connections. A self-managed deployment accepts TLS only if you have configured > it to serve TLS, so set `sslmode` to match your deployment. For where to find your host name, database, and app password, see [SQL clients](/serve-results/sql-clients/). ## Query and fetch as Arrow Execute a query with a cursor, then call `fetch_arrow_table` to get a [`pyarrow.Table`](https://arrow.apache.org/docs/python/generated/pyarrow.Table.html): ```python with adbc_driver_postgresql.dbapi.connect(uri) as conn: with conn.cursor() as cur: cur.execute("SELECT n, n * 2 AS doubled FROM generate_series(1, 100) AS g(n)") table = cur.fetch_arrow_table() print(table.schema) print(table.num_rows) ``` ```nofmt n: int32 doubled: int32 100 ``` The Arrow table holds the full result in memory, so it stays usable after the connection closes. ## Hand off to DuckDB Because the result is already an Arrow table, an Arrow-native engine can read it in place. DuckDB queries the `table` from the previous section directly, with no S3 or file hop: ```python import duckdb con = duckdb.connect() con.register("t", table) print(con.execute("SELECT sum(doubled) FROM t WHERE n > 50").fetchall()) ``` ```nofmt [(7550,)] ``` The same pattern works for any Arrow consumer. For example, `table.to_pandas()` produces a pandas DataFrame, and `polars.from_arrow(table)` produces a Polars DataFrame. ## Type mapping The driver maps the types Materialize shares with PostgreSQL to Arrow types as follows: Materialize / PostgreSQL type | Arrow type ----------------------------- | ---------- `bool` | `bool` `int2` (`smallint`) | `int16` `int4` (`integer`) | `int32` `int8` (`bigint`) | `int64` `float4` (`real`) | `float32` `float8` (`double precision`) | `float64` `text`, `varchar` | `string` `bytea` | `binary` `date` | `date32` `time` | `time64` (microseconds) `timestamp` | `timestamp` (microseconds) `timestamptz` | `timestamp` (microseconds, UTC) `interval` | `month_day_nano_interval` `jsonb` | `json` (over `string`) `numeric` | `opaque` extension (over `string`) `uuid` | `opaque` extension (over `binary`) arrays (for example `int4[]`) | `list` of the element type > **Note:** `numeric` and `uuid` have no native Arrow type, so the driver wraps them in an > [Arrow `opaque` extension > type](https://arrow.apache.org/docs/format/CanonicalExtensions.html#opaque) > over `string` and `binary` storage respectively. The driver picks the Arrow type for a column from the name of that type's PostgreSQL binary receive function. Materialize-specific types ([`list`](/sql/types/list/), [`map`](/sql/types/map/), [`uint2`/`uint4`/`uint8`](/sql/types/uint/), `mz_timestamp`, and `mz_aclitem`) have no PostgreSQL receive function. Rather than erroring, the driver falls back to an `opaque` extension type over binary for them, which holds the raw binary `COPY` encoding of the value. To get a usable Arrow value for one of these types, cast it in the query, for example `my_list::text`. The driver retrieves results using PostgreSQL's binary `COPY` protocol. ## Learn more - [SQL clients](/serve-results/sql-clients/) for connection parameters and app passwords. - [Client libraries](/serve-results/client-libraries/) for other language clients. - [Apache Arrow ADBC documentation](https://arrow.apache.org/adbc/) for the ADBC API and driver details. --- ## Client libraries Applications can use various common language-specific PostgreSQL client libraries to interact with Materialize and **create relations**, **execute queries** and **stream out results**. > **Note:** Client libraries tend to run complex introspection queries that may use configuration settings, system tables or features not yet implemented in Materialize. This means that even if PostgreSQL is supported, it's **not guaranteed** that the same integration will work out-of-the-box. | Language | Tested drivers | Notes | | -------- | --------------------------------------------------------------- | ----------------------------------------------------- | | Go | [`pgx`](https://github.com/jackc/pgx) | See the [Go cheatsheet](/serve-results/client-libraries/golang/). | | Java | [PostgreSQL JDBC driver](https://jdbc.postgresql.org/) | See the [Java cheatsheet](/serve-results/client-libraries/java-jdbc/). | | Node.js | [`node-postgres`](https://node-postgres.com/) | See the [Node.js cheatsheet](/serve-results/client-libraries/node-js/). | | PHP | [`pdo_pgsql`](https://www.php.net/manual/en/ref.pgsql.php) | See the [PHP cheatsheet](/serve-results/client-libraries/php/). | | Python | [`psycopg2`](https://pypi.org/project/psycopg2/) | See the [Python cheatsheet](/serve-results/client-libraries/python/) and its [SQLAlchemy section](/serve-results/client-libraries/python/#sqlalchemy). | | Ruby | [`pg` gem](https://rubygems.org/gems/pg/) | See the [Ruby cheatsheet](/serve-results/client-libraries/ruby/). | | Rust | [`postgres-openssl`](https://crates.io/crates/postgres-openssl) | See the [Rust cheatsheet](/serve-results/client-libraries/rust/). | 👋 _Is there another client library you'd like to use with Materialize? Submit a [feature request](https://github.com/MaterializeInc/materialize/discussions/new?category=feature-requests&labels=A-integration)._ --- ## Connect to Materialize via HTTP You can access Materialize through its "session-less" HTTP API endpoint: ```bash https:///api/sql ``` ## Details ### General semantics The API: - Requires username/password authentication, just as connecting via a SQL client (e.g. `psql`). Materialize provides you the username and password upon setting up your account. - Requires that you provide the entirety of your request. The API does not provide session-like semantics, so there is no way to e.g. interactively use transactions. - Ceases to process requests upon encountering the first error. - Does not support statements whose semantics rely on sessions or whose state is indeterminate at the end of processing, including: - `CLOSE` - `COPY` - `DECLARE` - `FETCH` - `SUBSCRIBE` - Supports specifying run-time [configuration parameters](/sql/set) via URL query parameters. ### Transactional semantics The HTTP API provides two modes with slightly different transactional semantics from one another: - **Simple**, which mirrors PostgreSQL's [Simple Query][simple-query] protocol. - Supports a single query, but the single query string may contain multiple statements, e.g. `SELECT 1; SELECT 2;` - Treats all statements as in an implicit transaction unless other transaction control is invoked. - **Extended**, which mirrors PostgreSQL's [Extended Query][extended-query] protocol. - Supports multiple queries, but only one statement per query string. - Supports parameters. - Eagerly commits DDL (e.g. `CREATE TABLE`) in implicit transactions, but not DML (e.g. `INSERT`). ### OpenAPI spec Download our [OpenAPI](https://swagger.io/specification/) v3 spec for this interface: [environmentd-openapi.yml](/materialize-openapi.yml). ## Usage ### Endpoint ``` https:///api/sql ``` Accessing the endpoint requires [basic authentication](https://developer.mozilla.org/en-US/docs/Web/HTTP/Authentication#basic_authentication_scheme). Reuse the same credentials as with a SQL client (e.g. `psql`): * **User ID:** Your email to access Materialize. * **Password:** Your app password. #### Query parameters You can optionally specify configuration parameters for each request, as a URL-encoded JSON object, with the `options` query parameter: ``` https:///api/sql?options= ``` For example, this is how you could specify the `application_name` configuration parameter with JavaScript: ```javascript // Create and encode our parameters object. const options = { application_name: "my_app" }; const encoded = encodeURIComponent(JSON.stringify(options)); // Add the object to our URL as the "options" query parameter. const url = new URL(`https://${mzHostAddress}/api/sql`); url.searchParams.append("options", encoded); ``` ### Input format #### Simple The request body is a JSON object containing a key, `query`, which specifies the SQL string to execute. `query` may contain multiple SQL statements separated by semicolons. ```json { "query": "select * from a; select * from b;" } ``` #### Extended The request body is a JSON object containing a key `queries`, whose value is array of objects, whose structure is: Key | Value ----|------ `query` | A SQL string containing one statement to execute `params` | An optional array of text values to be used as the parameters to `query`. _null_ values are converted to _null_ values in Materialize. Note that all parameter values' elements must be text or _null_; the API will not accept JSON numbers. ```json { "queries": [ { "query": "select * from a;" }, { "query": "select a + $1 from a;", "params": ["100"] } { "query": "select a + $1 from a;", "params": [null] } ] } ``` ### Output format The output format is a JSON object with one key, `results`, whose value is an array of the following: Result | JSON value ---------------------|------------ Rows | `{"rows": <2D array of JSON-ified results>, "desc": , "notices": }` Error | `{"error": , "notices": }` Ok | `{"ok": , "notices": }` Each committed statement returns exactly one of these values; e.g. in the case of "complex responses", such as `INSERT INTO...RETURNING`, the presence of a `"rows"` object implies `"ok"`. The `"notices"` array is present in all types of results and contains any diagnostic messages that were generated during execution of the query. It has the following structure: ``` {"severity": <"warning"|"notice"|"debug"|"info"|"log">, "message": } ``` Note that the returned values include the results of statements which were ultimately rolled back because of an error in a later part of the transaction. You must parse the results to understand which statements ultimately reflect the resultant state. Numeric results are converted to strings to avoid possible JavaScript number inaccuracy. Column descriptions contain the name, oid, data type size and type modifier of a returned column. #### TypeScript definition You can model these with the following TypeScript definitions: ```typescript interface Simple { query: string; } interface ExtendedRequest { query: string; params?: (string | null)[]; } interface Extended { queries: ExtendedRequest[]; } type SqlRequest = Simple | Extended; interface Notice { message: string; severity: string; detail?: string; hint?: string; } interface Error { message: string; code: string; detail?: string; hint?: string; } interface Column { name: string; type_oid: number; // u32 type_len: number; // i16 type_mod: number; // i32 } interface Description { columns: Column[]; } type SqlResult = | { tag: string; rows: any[][]; desc: Description; notices: Notice[]; } | { ok: string; notices: Notice[]; } | { error: Error; notices: Notice[]; }; ``` ## Examples ### Run a transaction Use the [extended input format](#extended) to run a transaction: ```bash curl 'https:///api/sql' \ --header 'Content-Type: application/json' \ --user ':' \ --data '{ "queries": [ { "query": "CREATE TABLE IF NOT EXISTS t (a int);" }, { "query": "CREATE TABLE IF NOT EXISTS s (a int);" }, { "query": "BEGIN;" }, { "query": "INSERT INTO t VALUES ($1), ($2)", "params": ["100", "200"] }, { "query": "COMMIT;" }, { "query": "BEGIN;" }, { "query": "INSERT INTO s VALUES ($1), ($2)", "params": ["9", null] }, { "query": "COMMIT;" } ] }' ``` Response: ```json { "results": [ {"ok": "CREATE TABLE", "notices": []}, {"ok": "CREATE TABLE", "notices": []}, {"ok": "BEGIN", "notices": []}, {"ok": "INSERT 0 2", "notices": []}, {"ok": "COMMIT", "notices": []}, {"ok": "BEGIN", "notices": []}, {"ok": "INSERT 0 2", "notices": []}, {"ok": "COMMIT", "notices": []} ] } ``` ### Run a query Use the [simple input format](#simple) to run a query: ```bash curl 'https:///api/sql' \ --header 'Content-Type: application/json' \ --user ':' \ --data '{ "query": "SELECT t.a + s.a AS cross_add FROM t CROSS JOIN s; SELECT a FROM t WHERE a IS NOT NULL;" }' ``` Response: ```json { "results": [ { "desc": { "columns": [ { "name": "cross_add", "type_len": 4, "type_mod": -1, "type_oid": 23 } ] }, "notices": [], "rows": [], "tag": "SELECT 0" }, { "desc": { "columns": [ { "name": "a", "type_len": 4, "type_mod": -1, "type_oid": 23 } ] }, "notices": [], "rows": [], "tag": "SELECT 0" } ] } ``` ## See also - [SQL Clients](../sql-clients) [simple-query]: https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.4 [extended-query]: https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY --- ## Connect to Materialize via WebSocket You can access Materialize through its interactive WebSocket API endpoint: ```bash wss:///api/experimental/sql ``` ## Details ### General semantics The API: - Requires username/password authentication, just as connecting via a SQL client (e.g. `psql`). Materialize provides you the username and password upon setting up your account. - Maintains an interactive session. - Does not support some statements: - `CLOSE` - `COPY` - `DECLARE` - `FETCH` - Supports specifying run-time configuration parameters ([session variables](https://www.postgresql.org/docs/current/sql-set.html)) via the initial authentication message. ### Transactional semantics The WebSocket API provides two modes with slightly different transactional semantics from one another: - **Simple**, which mirrors PostgreSQL's [Simple Query][simple-query] protocol. - Supports a single query, but the single query string may contain multiple statements, e.g. `SELECT 1; SELECT 2;` - Treats all statements as in an implicit transaction unless other transaction control is invoked. - **Extended**, which mirrors PostgreSQL's [Extended Query][extended-query] protocol. - Supports multiple queries, but only one statement per query string. - Supports parameters. - Eagerly commits DDL (e.g. `CREATE TABLE`) in implicit transactions, but not DML (e.g. `INSERT`). ## Usage ### Endpoint ``` wss:///api/experimental/sql ``` To authenticate using a username and password, send an initial text or binary message containing a JSON object: ``` { "user": "", "password": "", "options": { } } ``` To authenticate using a token, send an initial text or binary message containing a JSON object: ``` { "token": "", "options": { } } ``` Successful authentication will result in: - Some `ParameterStatus` messages indicating the values of some initial session settings. - One `BackendKeyData` message that can be used for cancellation. - One `ReadyForQuery`message, at which point the server is ready to receive requests. An error during authentication is indicated by a websocket Close message. HTTP `Authorization` headers are ignored. ### Messages WebSocket Text or Binary messages can be sent. The payload is described below in the [Input format](#input-format) section. Each request will respond with some number of response messages, followed by a `ReadyForQuery` message. There is exactly one `ReadyForQuery` message for each request, regardless of how many queries the request contains. ### Input format #### Simple The message payload is a JSON object containing a key, `query`, which specifies the SQL string to execute. `query` may contain multiple SQL statements separated by semicolons. ```json { "query": "select * from a; select * from b;" } ``` #### Extended The message payload is a JSON object containing a key `queries`, whose value is array of objects, whose structure is: Key | Value ----|------ `query` | A SQL string containing one statement to execute `params` | An optional array of text values to be used as the parameters to `query`. _null_ values are converted to _null_ values in Materialize. Note that all parameter values' elements must be text or _null_; the API will not accept JSON numbers. ```json { "queries": [ { "query": "select * from a;" }, { "query": "select a + $1 from a;", "params": ["100"] } { "query": "select a + $1 from a;", "params": [null] } ] } ``` ### Output format The response messages are WebSocket Text messages containing a JSON object that contains keys `type` and `payload`. `type` value | Description ---------------------|------------ `ReadyForQuery` | Sent at the end of each response batch `Notice` | An informational notice. `CommandStarting` | A command has executed and response data will be returned. `CommandComplete` | Executing a statement succeeded. `Error` | Executing a statement resulted in an error. `Rows` | A rows-returning statement is executing, and some `Row` messages may follow. `Row` | A single row result. `ParameterStatus` | Announces the value of a session setting. `BackendKeyData` | Information used to cancel queries. #### `ReadyForQuery` Exactly one of these is sent at the end of every request batch. It can be used to synchronize with the server, and means the server is ready for another request. (However, many requests can be made at any time; there is no need to wait for this message before issuing more requests.) The payload is a `string` describing the current transaction state: - `I` for idle: not in a transaction. - `T` for in a transaction. - `E` for a transaction in an error state. A request starting with `ROLLBACK` should be issued to exit it. #### `Notice` A notice can appear at any time and contains diagnostic messages that were generated during execution of the query. The payload has the following structure: ``` { "message": , "code": , "severity": <"warning"|"notice"|"debug"|"info"|"log">, "detail": , "hint": , } ``` #### `Error` Executing a statement resulted in an error. The payload has the following structure: ``` { "message": , "code": , "detail": , "hint": , } ``` #### `CommandStarting` A statement has executed and response data will be returned. This message can be used to know if rows or streaming data will follow. The payload has the following structure: ``` { "has_rows": , "is_streaming": , } ``` The `has_rows` field is `true` if a `Rows` message will follow. The `is_streaming` field is `true` if there is no expectation that a `CommandComplete` message will ever occur. This is the case for `SUBSCRIBE` queries. #### `CommandComplete` Executing a statement succeeded. The payload is a `string` containing the statement's tag. #### `Rows` A rows-returning statement is executing and some number (possibly 0) of `Row` messages will follow. Either a `CommandComplete` or `Error` message will then follow indicating there are no more rows and the final result of the statement. The payload has the following structure: ``` { "columns": [ { "name": , "type_oid": , "type_len": , "type_mod": } ... ] } ``` The inner object's various `type_X` fields are lower-level details that can be used to convert the row results from a string to a more specific data type. `type_oid` is the OID of the data type. `type_len` is the data size type (see `pg_type.typlen`). `type_mod` is the type modifier (see `pg_attribute.atttypmod`). #### `Row` A single row result. Will only occur after a `Rows` message. The payload is an array of JSON values corresponding to the columns from the `Rows` message. Numeric results are converted to strings to avoid possible JavaScript number inaccuracy. #### `ParameterStatus` Announces the value of a session setting. These are sent during startup and when a statement caused a session parameter to change. The payload has the following structure: ``` { "name": , "value": , } ``` #### `BackendKeyData` Information used to cancel queries. The payload has the following structure: ``` { "conn_id": , "secret_key": , } ``` #### TypeScript definition You can model these with the following TypeScript definitions: ```typescript type Auth = | { user: string; password: string; options?: { [name: string]: string } } | { token: string; options?: { [name: string]: string } } ; interface Simple { query: string; } interface ExtendedRequest { query: string; params?: (string | null)[]; } interface Extended { queries: ExtendedRequest[]; } type SqlRequest = Simple | Extended; interface Notice { message: string; severity: string; detail?: string; hint?: string; } interface Error { message: string; code: string; detail?: string; hint?: string; } interface ParameterStatus { name: string; value: string; } interface CommandStarting { has_rows: boolean; is_streaming: boolean; } interface BackendKeyData { conn_id: number; // u32 secret_key: number; // u32 } interface Column { name: string; type_oid: number; // u32 type_len: number; // i16 type_mod: number; // i32 } interface Description { columns: Column[]; } type WebSocketResult = | { type: "ReadyForQuery"; payload: string } | { type: "Notice"; payload: Notice } | { type: "CommandComplete"; payload: string } | { type: "Error"; payload: Error } | { type: "Rows"; payload: Description } | { type: "Row"; payload: any[] } | { type: "ParameterStatus"; payload: ParameterStatus } | { type: "CommandStarting"; payload: CommandStarting } | { type: "BackendKeyData"; payload: BackendKeyData } ; ``` ## Examples ### Run a query ```bash $ echo '{"query": "select 1,2; values (4), (5)"}' | websocat wss:///api/experimental/sql {"type":"CommandStarting","payload":{"has_rows":true,"is_streaming":false}} {"type":"Rows","payload":{"columns":[{"name":"?column?","type_oid":23,"type_len":4,"type_mod":-1},{"name":"?column?","type_oid":23,"type_len":4,"type_mod":-1}]}} {"type":"Row","payload":["1","2"]} {"type":"CommandComplete","payload":"SELECT 1"} {"type":"CommandStarting","payload":{"has_rows":true,"is_streaming":false}} {"type":"Rows","payload":{"columns":[{"name":"column1","type_oid":23,"type_len":4,"type_mod":-1}]}} {"type":"Row","payload":["4"]} {"type":"Row","payload":["5"]} {"type":"CommandComplete","payload":"SELECT 2"} {"type":"ReadyForQuery","payload":"I"} ``` ## See also - [SQL Clients](../sql-clients) [simple-query]: https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.5.7.4 [extended-query]: https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY --- ## Connection Pooling Because Materialize is wire-compatible with PostgreSQL, you can use any PostgreSQL connection pooler with Materialize. In this guide, we’ll cover how to use connection pooling with Materialize, alongside the common tool PgBouncer. ## PgBouncer [PgBouncer](https://www.pgbouncer.org/) is a popular connection pooler for PostgreSQL. It provides a lightweight and efficient way to manage database connections, reducing the overhead of establishing new connections and improving performance. ### Step 1: Install PgBouncer You can [install PgBouncer](https://www.pgbouncer.org/downloads/) on your local machine or on a server. ### Step 2: Create an authentication userlist file The userlist file contains the credentials for your Materialize user/app. The file has the format of: ``` "user@example.com" "mypassword-or-scram-secret" ``` #### If using `auth_type = plain` (Cloud and Self-Managed) Specify the password in plaintext: - **For Cloud**, use the password from an existing [Service Account](https://materialize.com/docs/security/cloud/users-service-accounts/create-service-accounts/) or generate a [new one](https://materialize.com/docs/security/cloud/users-service-accounts/create-service-accounts/). - **For Self-Managed**, use the password associated with the role. Example userlist file: ``` "foo@bar.com" "mypassword" ``` #### If using `auth_type = scram-sha-256` (Self-Managed only) Specify the SCRAM secret. To find the SCRAM secret, run the following query as a superuser: ```sql SELECT rolname, rolpassword FROM pg_authid WHERE rolname = 'your_role_name'; ``` > **Note:** You must be a superuser to access the `pg_authid` table. Once you have the SCRAM secret, add it to the userlist file in the following format: ``` "your_role_name" "the-hash-you-got-from-pg_authid" ``` ### Step 3: Configure PgBouncer to connect to your Materialize instance 1. Locate the configuration file. Refer to the [setup instructions](https://www.pgbouncer.org/downloads/) for your environment. 2. Update the file to: - Define a database named `materialize` with your Materialize connection details. - Configure PgBouncer as needed for your PgBouncer instance. Be sure to specify the `auth_type` and `auth_file` needed to connect to Materialize. For example, the following is a basic configuration example to connect a local PgBouncer to a locally-running Materialize: ```ini [databases] ;; For Cloud, use the connection details from the Console. ;; For Self-Managed, use your Materialize's connection details. materialize = host=localhost port=6877 [pgbouncer] logfile = /var/log/pgbouncer/pgbouncer.log pidfile = /var/run/pgbouncer/pgbouncer.pid ;; Listen on localhost:6432 for incoming connections listen_addr = localhost listen_port = 6432 ;; Set the authentication type ;; Materialize supports both plain and scram-sha-256. ;; Cloud and Self-Managed support plain. ;; auth_type = plain ;; Self-Managed also supports scram-sha-256: auth_type = scram-sha-256 ;; Set the authentication user list file auth_file = /etc/pgbouncer/userlist.txt ``` For additional information on configuring PgBouncer, refer to the [PgBouncer documentation](https://www.pgbouncer.org/config.html). ### Step 4: Start the service and connect After configuring PgBouncer, you can start the service. You can then connect to PgBouncer using the same connection parameters as you would for Materialize, but with the PgBouncer port (default is 6432). For example: ```bash psql -h localhost -p 6432 -U your_role_name -d materialize ``` --- ## Durable subscriptions [//]: # "TODO: Move to Serve results section" [Subscriptions](/sql/subscribe/) allow you to stream changing results from Materialize to an external application programatically. Like any connection over the network, subscriptions might get disrupted for both expected and unexpected reasons. In such cases, it can be useful to have a mechanism to gracefully recover data processing. To avoid the need for re-processing data that was already sent to your external application following a connection disruption, you can: - Adjust the [history retention period](#history-retention-period) for the objects that a subscription depends on, and - [Access past versions of this data](#enabling-durable-subscriptions-in-your-application) at specific points in time to pick up data processing where you left off. ## History retention period By default, all user-defined sources, tables, materialized views, and indexes keep track of the most recent version of their underlying data. To gracefully recover from connection disruptions and enable lossless, _durable subscriptions_, you can configure the sources, tables, and materialized views that the subscription depends on to **retain history**. > **Important:** Configuring indexes to retain history is not recommended. Instead, consider > creating a materialized view for your subscription query and configuring the > history retention period on that view. To configure the history retention period for sources, tables and materialized views, use the `RETAIN HISTORY` option in its `CREATE` statement. This value can also be adjusted at any time using the object-specific `ALTER` statement. ### Semantics #### Increasing the history retention period When you increase the history retention period for an object, both the existing historical data and any subsequently produced historical data are retained for the specified time period. **For sources, tables and materialized views:** increasing the history retention period will not restore older historical data that was already outside the previous history retention period before the change. Configuring indexes to retain history is not recommended. Instead, consider creating a materialized view for your subscription query and configuring the history retention period on that view. See also [Considerations](#considerations). #### Decreasing the history retention period When you decrease the history retention period for an object: * Newly produced historical data is retained for the new, shorter history retention period. * Historical data outside the new, shorter history retention period is no longer retained. If you subsequently increase the history retention period again, the older historical data may already be unavailable. See also [Considerations](#considerations). ### Set history retention period > **Important:** Setting the history retention period for an object will lead to increased > resource utilization. Moreover, for indexes, setting history retention period is > not recommended. Instead, consider creating a materialized view for your > subscription query and configuring the history retention period on that view. > See [Considerations](#considerations). To set the history retention period for [sources](/sql/create-source/), [tables](/sql/create-table/), and [materialized views](/sql/create-materialized-view/), you can either: - Specify the `RETAIN HISTORY` option in the `CREATE` statement. The `RETAIN HISTORY` option accepts positive [interval](/sql/types/interval/) values (e.g., `'1hr'`). For example: ```mzsql CREATE MATERIALIZED VIEW winning_bids WITH (RETAIN HISTORY FOR '1hr') AS SELECT auction_id, bid_id, item, amount FROM highest_bid_per_auction WHERE end_time < mz_now(); ``` - Specify the `RETAIN HISTORY` option in the `ALTER` statement. The `RETAIN HISTORY` option accepts positive [interval](/sql/types/interval/) values (e.g., `'1hr'`). For example: ```mzsql ALTER MATERIALIZED VIEW winning_bids SET (RETAIN HISTORY FOR '2hr'); ``` ### View history retention period for an object To see what history retention period has been configured for an object, look up the object in the [`mz_internal.mz_history_retention_strategies`](/sql/system-catalog/mz_internal/#mz_history_retention_strategies) catalog table. For example: ```mzsql SELECT d.name AS database_name, s.name AS schema_name, mv.name, hrs.strategy, hrs.value FROM mz_catalog.mz_materialized_views AS mv LEFT JOIN mz_schemas AS s ON mv.schema_id = s.id LEFT JOIN mz_databases AS d ON s.database_id = d.id LEFT JOIN mz_internal.mz_history_retention_strategies AS hrs ON mv.id = hrs.id WHERE mv.name = 'winning_bids'; ``` If set, the returning result includes the value (in milliseconds) of the history retention period: ```nofmt database_name | schema_name | name | strategy | value ---------------+-------------+--------------+----------+--------- materialize | public | winning_bids | FOR | 7200000 ``` ### Unset/reset history retention period To disable history retention, reset the history retention period; i.e., specify the `RESET (RETAIN HISTORY)` option in the `ALTER` statement. For example: ```mzsql ALTER MATERIALIZED VIEW winning_bids RESET (RETAIN HISTORY); ``` ### Considerations #### Resource utilization Increasing the history retention period for an object will lead to increased resource utilization in Materialize. **For sources, tables and materialized views:** Increasing the history retention period for these objects increases the amount of historical data that is retained in the storage layer. You can expect storage resource utilization to increase, which may incur additional costs. **For indexes:** Configuring indexes to retain history is not recommended. Instead, consider creating a materialized view for your subscription query and configuring the history retention period on that view. #### Best practices - Because of the increased storage costs and processing time for the additional historical data, consider configuring history retention period on the object directly powering the subscription, rather than all the way through the dependency chain from the source to the materialized view. - Configuring indexes to retain history is not recommended. Instead, consider creating a materialized view for your subscription query and configuring the history retention period on that view. #### Clean-up The history retention period represents the minimum amount of historical data guaranteed to be retained by Materialize. History clean-up is processed in the background, so older history may be accessible for the period of time between when it falls outside the retention period and when it is cleaned up. ## Enabling durable subscriptions in your application 1. In Materialize, configure the history retention period for the object(s) queried in the `SUBSCRIBE`. Choose a duration you expect will allow you to recover in case of connection drops. One hour (`1h`) is a good place to start, though you should be mindful of the [impact](#considerations) of increasing an object's history retention period. 1. In order to restart your application without losing or re-snapshotting data after a connection drop, you need to store the latest timestamp processed for the subscription (either in Materialize, or elsewhere in your application state). This will allow you to resume using the retained history upstream. 1. The first time you start the subscription, run the following continuous query against Materialize in your application code: ```mzsql SUBSCRIBE () WITH (PROGRESS, SNAPSHOT true); ``` If you do not need a full snapshot to bootstrap your application, change this to `SNAPSHOT false`. 1. As results come in continuously, buffer the latest results in memory until you receive a [progress](/sql/subscribe#progress) message. At that point, the data up until the progress message is complete, so you can: 1. Process all the buffered data in your application. 1. Persist the `mz_timestamp` of the progress message. 1. To resume the subscription in subsequent restarts, use the following continuous query against Materialize in your application code: ```mzsql SUBSCRIBE () WITH (PROGRESS, SNAPSHOT false) AS OF ; ``` Note the subtraction of `1` from the last received progress timestamp. By default (i.e., `SNAPSHOT true`), `SUBSCRIBE` begins by emitting a snapshot at the `AS OF` timestamp (if specified) and then emits subsequent updates as they occur. However, when `SNAPSHOT false` is set, the snapshot is not emitted, and `SUBSCRIBE` emits only updates subsequent to the snapshot; i.e., updates with timestamp greater than the `AS OF` timestamp. To include updates that occurred at the last progress timestamp, subtract `1` from the last progress timestamp. If you're subscribing _directly_ to a collection, as in `SUBSCRIBE TO WITH (PROGRESS, SNAPSHOT false) AS OF