> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # PostgreSQL Loader The PostgreSQL loader writes query results into a PostgreSQL table. Choose it when your application already reads from PostgreSQL, or when you need to join Amp's data with your own tables in SQL. It creates the table for you, from the columns of your query, and it undoes [reorgs](/get-started/concepts/reorgs) with a fast indexed delete. ## Prerequisites * **The Python SDK**, [installed](/sdks/python#installation), and the PostgreSQL driver it uses, `psycopg2-binary`: :::code-group ```bash [pip] pip install psycopg2-binary ``` ```bash [uv] uv add psycopg2-binary ``` ::: * **A running Amp with the Quickstart's dataset**, `acme/eth_mainnet`. [Quickstart](/get-started/quickstart) sets it up. * **A PostgreSQL database you can write to.** To run one locally in Docker, set a password, then start PostgreSQL 17 with it: ```bash export POSTGRES_PASSWORD=postgres docker run --name amp-loader-postgres -e POSTGRES_PASSWORD -p 5432:5432 -d postgres:17 ``` If port 5432 is taken, for example by the PostgreSQL you run for Amp, publish another one, like `-p 5433:5432`, and set `port` to it in the configuration below. ## Load data This script names a connection to your database, `local_postgres`, then loads the Quickstart's 100 blocks into a `blocks` table, in the `public` schema: ```python [load_postgresql.py] import os from amp import Client client = Client('grpc://localhost:1602') client.configure_connection( 'local_postgres', 'postgresql', { 'host': 'localhost', 'database': 'postgres', 'user': 'postgres', 'password': os.environ['POSTGRES_PASSWORD'], 'connection_params': {'options': '-c search_path=public'}, }, ) result = client.sql( 'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks' ).load(connection='local_postgres', destination='blocks', read_all=True) print(result) ``` ```bash python load_postgresql.py ``` `destination` is the table's name, and `connection_params` sets the connection's `search_path` to `public`, the schema the table goes in (see [The table it creates](/sdks/python/loaders/postgresql#the-table-it-creates)). `read_all=True` loads the whole result at once and returns one `LoadResult`. Without it, `load()` returns an iterator with one `LoadResult` per batch of rows, which loads nothing until you iterate over it. [Loaders](/sdks/python/loaders) covers both, and the other options `load()` takes. `configure_connection()` saves the connection, password included, so later scripts can use `local_postgres` without configuring it again. [Loaders](/sdks/python/loaders) explains where it saves it. Query the table: :::terminal ```bash docker exec amp-loader-postgres psql -U postgres -c 'SELECT block_num, timestamp, gas_used FROM blocks ORDER BY block_num DESC LIMIT 3' ``` ``` block_num | timestamp | gas_used -----------+------------------------+---------- 24000099 | 2025-12-13 00:25:59+00 | 10751338 24000098 | 2025-12-13 00:25:47+00 | 36233422 24000097 | 2025-12-13 00:25:35+00 | 52877862 (3 rows) ``` ::: ## The table it creates When the table doesn't exist, the loader creates it, with a column for each column of your query, then an `_amp_batch_id` column and an index on it. `_amp_batch_id` identifies the block range each row came from, so the loader can delete the rows of a reorged range. It stays empty when you load all at once, like above, and is filled when you [stream](/sdks/python/loaders/postgresql#streaming-and-reorgs). :::terminal ```bash docker exec amp-loader-postgres psql -U postgres -c '\d blocks' ``` ``` Table "public.blocks" Column | Type | Collation | Nullable | Default ---------------+--------------------------+-----------+----------+--------- block_num | numeric(20,0) | | not null | timestamp | timestamp with time zone | | not null | hash | bytea | | not null | gas_used | numeric(20,0) | | not null | _amp_batch_id | text | | | Indexes: "idx_blocks_amp_batch_id" btree (_amp_batch_id) ``` ::: Each column gets the PostgreSQL type that holds its values: | Query column type | PostgreSQL type | | ------------------------------------------ | ------------------------------------------- | | 8- and 16-bit integers, unsigned 8-bit | `SMALLINT` | | 32-bit integers, unsigned 16-bit | `INTEGER` | | 64-bit integers, unsigned 32-bit | `BIGINT` | | Unsigned 64-bit integers, like `block_num` | `NUMERIC(20,0)` | | 16- and 32-bit floats | `REAL` | | 64-bit floats | `DOUBLE PRECISION` | | Decimals | `NUMERIC` with the same precision and scale | | Strings, lists, and structs | `TEXT` | | Binary, like hashes and addresses | `BYTEA` | | Booleans | `BOOLEAN` | | Timestamps with a time zone | `TIMESTAMPTZ` | | Timestamps without one | `TIMESTAMP` | | Dates and times | `DATE` and `TIME` | | Anything else | `TEXT` | Columns that can't be null get `NOT NULL`. The loader names the table without a schema, so the connection's [`search_path`](https://www.postgresql.org/docs/current/ddl-schemas.html#DDL-SCHEMAS-PATH) decides where it goes: PostgreSQL creates it in the first schema of the search path that exists, and the loader writes to the first table of that name the search path finds. The default search path, `"$user", public`, starts with a schema named after the user: without `connection_params`, a user `amp` in a database with an `amp` schema gets `amp.blocks`. Setting `search_path` to `public`, as above, keeps the table in `public` whatever schemas the database has. To load into another schema, name it instead, like `{'options': '-c search_path=analytics'}`. The loader only looks for an existing table in `public`, though: if `public` has a table of the same name, it doesn't create one in your schema, and the load fails (see [Troubleshooting](/sdks/python/loaders/postgresql#insert-operation-failed-relation-blocks_live-does-not-exist)). When the table already exists, the loader leaves it as it is, and inserts into the columns with your query's column names. The table must have them all, with compatible types. The loader writes with `COPY` when the rows have no binary columns, and with `INSERT` statements when they do, which is slower. To load faster, convert binary columns to text in your query, like `evm_decode_hex(hash) AS hash` (see [EVM functions](/querying/functions/evm)). ## Configuration The third argument of `configure_connection()`: | Key | Type | Default | Description | | ------------------- | ------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `host` | string | required | The server's host name or IP address, like `'localhost'`. | | `database` | string | required | The database to load into, like `'postgres'`. | | `user` | string | required | The user to connect as, like `'postgres'`. It needs to create tables, and to insert into and delete from them. | | `password` | string | required | The user's password. Read it from an environment variable, like `os.environ['POSTGRES_PASSWORD']`. | | `port` | integer | `5432` | The server's port, like `5433`. | | `max_connections` | integer | `10` | The most connections the loader opens to the server at once, like `20`. | | `connection_params` | dict | `{}` | Other [connection parameters](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-PARAMKEYWORDS), passed to the driver as they are, like `{'sslmode': 'require', 'connect_timeout': 10}`. | | `batch_size` | integer | `10000` | Not used. Pass `batch_size` to `load()` instead (see [Loaders](/sdks/python/loaders)). | The `state` and `resilience` keys, which every loader takes, are on [Loaders](/sdks/python/loaders). `destination` is a table name, without a schema, quoted as it is: `'Blocks'` and `'blocks'` are different tables, and `'analytics.blocks'` is a table named `analytics.blocks`, not a table in the `analytics` schema. The connection's `search_path` picks the schema (see [The table it creates](/sdks/python/loaders/postgresql#the-table-it-creates)). ## Load modes Pass `mode` to `load()` ([Loaders](/sdks/python/loaders#load-modes) explains the modes every loader shares): * **`append`** (the default) inserts the rows, keeping the table's existing rows. * **`overwrite`** is meant to empty the table first, but doesn't work reliably: * With `read_all=True`, it only empties a table that the same `load()` call created, so on an existing table it appends. * Without `read_all`, it empties the table before each batch, so only the last batch stays, and the last batch is often empty. To replace a table's rows, empty it yourself, then load in `append` mode: ```bash docker exec amp-loader-postgres psql -U postgres -c 'TRUNCATE blocks' ``` `upsert` and `merge` aren't supported: they fail with `Unsupported mode LoadMode.UPSERT`. ## Streaming and reorgs (Unstable) With `stream=True`, `load()` runs a [streaming query](/querying/streaming-queries) and keeps loading new rows as Amp extracts new blocks. [Streaming](/sdks/python/streaming) explains how it works. Two things are specific to PostgreSQL: * **A stream doesn't create its table.** Create it first, with an empty load that isn't `read_all`, as this script does. Without it, every batch fails with `relation "blocks_live" does not exist`. * **Each batch is loaded in one transaction**, so a batch is either loaded completely or not at all. ```python [stream_postgresql.py] from amp import Client client = Client('grpc://localhost:1602') query = 'SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks' # A stream doesn't create its table, so create it with an empty load first for _ in client.sql(f'{query} LIMIT 0').load(connection='local_postgres', destination='blocks_live'): pass results = client.sql(query).load( connection='local_postgres', destination='blocks_live', stream=True, with_reorg_detection=False ) for result in results: print(result) ``` :::terminal ```bash python stream_postgresql.py ``` ``` Failed to load checkpoint, starting from beginning: 'PostgreSQLLoader' object has no attribute 'checkpoint_store' ✅ Loaded 100 rows to blocks_live in 0.03s ✅ Loaded 0 rows to blocks_live in 0.00s ``` ::: The first line is a warning every stream prints, and `with_reorg_detection=False` works around a problem with reorg detection: both are explained in [Known issues](/sdks/python/streaming#known-issues). The stream ends 30 seconds after it starts, even on a dataset that follows its chain: Amp sends a message every 30 seconds to keep the connection open, and pyarrow, which the SDK reads streams with, takes it for the end of the stream (see [Arrow Flight](/querying/arrow-flight#keep-alive-messages)). To keep loading, start the stream again in a loop, as [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds) shows. Streamed rows have their `_amp_batch_id` set. When Amp reports a reorg, the loader looks up the batches it loaded from the reorged blocks onward, and deletes their rows: ```sql DELETE FROM blocks WHERE _amp_batch_id LIKE ANY (ARRAY['%7dce8312375e3370%', …]) ``` It deletes in chunks of 1,000 batches, and the index on `_amp_batch_id` keeps each delete fast. The loader remembers which batches it loaded in memory, for as long as the `load()` call runs. When a stream restarts, it starts from the beginning of the query, and can't undo reorgs of the batches it loaded before. ## Troubleshooting ### `connection to server at "localhost" (127.0.0.1), port 5432 failed: Connection refused` PostgreSQL isn't running, or listens on another port. Start it, or set `port` to the port it listens on. The message may list each address `localhost` stands for, like `::1` and `127.0.0.1`. ### `password authentication failed for user "postgres"` The password doesn't match the user's. Check `POSTGRES_PASSWORD` in the terminal you run the script from. ### `Unsupported mode LoadMode.UPSERT` The PostgreSQL loader only supports the `append` and `overwrite` modes (see [Load modes](/sdks/python/loaders/postgresql#load-modes)). ### `INSERT operation failed: relation "blocks_live" does not exist` The table doesn't exist, and the load doesn't create it: you passed `create_table=False` to `load()`, you're streaming, or `search_path` starts with another schema while `public` has a table of that name. Create the table first (see [Streaming and reorgs](/sdks/python/loaders/postgresql#streaming-and-reorgs)), or drop or rename the one in `public`. Without binary columns, the message is `Table 'blocks_live' does not exist. Set create_table=True to auto-create.`, followed by the same cause. ### `INSERT operation failed: column "…" of relation "…" does not exist` The table already exists, without a column your query returns. Add the column to the table, rename it in your query with `AS`, or load into a new table.