> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mixpeek.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# PostgreSQL

> Sync PostgreSQL table rows into a Mixpeek bucket and run SQL lookups from a retriever.

> [!NOTE]
>
> Each row becomes one bucket object. The same connection also backs the `sql_lookup` retriever stage.

## Overview

A PostgreSQL connection serves two paths:

* **Sync.** Rows from a table land in a bucket as JSON objects, with incremental sync via a watermark column.
* **Lookup.** The [`sql_lookup`](https://docs.mixpeek.com/docs/retrieval/stages/sql-lookup) retriever stage enriches documents mid-pipeline.

## Prerequisites

* PostgreSQL 12 or later.
* A role with `CONNECT` on the database and `USAGE` on the schema.
* `SELECT` on every table you sync or query.
* Network access from Mixpeek to the server.

## Configuration

### Connection-level fields

| Field | Required | Description |
| - | - | - |
| `credentials.username` | Yes | PostgreSQL username |
| `credentials.password` | Yes | Encrypted at rest, never returned in responses |
| `host` | Yes | Hostname or IP |
| `database` | Yes | Database name |
| `port` | No | Defaults to `5432` |
| `schema` | No | Defaults to `public` |
| `ssl_mode` | No | `disable`, `allow`, `prefer` (default), `require`, `verify-ca`, `verify-full` |
| `incremental_column` | No | `TIMESTAMP` or `DATE` column used as the sync watermark |
| `primary_key_columns` | No | Columns that form the key, used to build stable object IDs |
| `query_timeout_seconds` | No | Defaults to `300` |
| `fetch_size` | No | Rows per batch, defaults to `1000` |

### Sync-level fields

| Field | Required | Description |
| - | - | - |
| `source_path` | Yes | `table`, or `schema.table` to override the connection schema |
| `polling_interval_seconds` | No | Defaults to `300` |

> [!WARNING]
>
> `source_path` accepts one or two dot-separated parts. Three or more returns a validation error.

## Setup

**1. Create a read-only role**

```sql theme={null}
CREATE ROLE mixpeek_sync LOGIN PASSWORD 'choose-a-strong-password';
GRANT CONNECT ON DATABASE production TO mixpeek_sync;
GRANT USAGE ON SCHEMA public TO mixpeek_sync;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mixpeek_sync;
```

**2. Create the storage connection**

```python Python theme={null}
from mixpeek import Mixpeek

client = Mixpeek(api_key="your-mixpeek-api-key")

connection = client.organizations.connections.create(
    name="Production Postgres",
    provider_type="postgresql",
    provider_config={
        "credentials": {
            "type": "username_password",
            "username": "mixpeek_sync",
            "password": "choose-a-strong-password",
        },
        "host": "db.example.com",
        "port": 5432,
        "database": "production",
        "schema": "public",
        "ssl_mode": "require",
        "incremental_column": "updated_at",
        "primary_key_columns": ["id"],
    },
)
print(f"Created connection: {connection['connection_id']}")
```

```javascript JavaScript theme={null}
import { Mixpeek } from 'mixpeek-sdk'

const client = new Mixpeek({ apiKey: 'your-mixpeek-api-key' })

const connection = await client.organizations.connections.create({
  name: 'Production Postgres',
  provider_type: 'postgresql',
  provider_config: {
    credentials: {
      type: 'username_password',
      username: 'mixpeek_sync',
      password: 'choose-a-strong-password',
    },
    host: 'db.example.com',
    port: 5432,
    database: 'production',
    schema: 'public',
    ssl_mode: 'require',
    incremental_column: 'updated_at',
    primary_key_columns: ['id'],
  },
})
console.log('Created connection:', connection.connection_id)
```

```bash cURL theme={null}
curl -X POST https://api.mixpeek.com/v1/organizations/connections \
  -H "Authorization: Bearer YOUR_MIXPEEK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Production Postgres",
    "provider_type": "postgresql",
    "provider_config": {
      "credentials": {
        "type": "username_password",
        "username": "mixpeek_sync",
        "password": "choose-a-strong-password"
      },
      "host": "db.example.com",
      "port": 5432,
      "database": "production",
      "schema": "public",
      "ssl_mode": "require",
      "incremental_column": "updated_at",
      "primary_key_columns": ["id"]
    }
  }'
```

**3. Sync a table**

```python Python theme={null}
sync = client.buckets.syncs.create(
    bucket_id="bkt_your_bucket_id",
    connection_id=connection["connection_id"],
    source_path="customers",
    polling_interval_seconds=1800,
)
print(f"Sync created: {sync['sync_config_id']}")
```

```bash cURL theme={null}
curl -X POST https://api.mixpeek.com/v1/buckets/bkt_your_bucket_id/syncs \
  -H "Authorization: Bearer YOUR_MIXPEEK_API_KEY" \
  -H "X-Namespace: ns_your_namespace_id" \
  -H "Content-Type: application/json" \
  -d '{
    "connection_id": "conn_your_connection_id",
    "source_path": "analytics.customers",
    "polling_interval_seconds": 1800
  }'
```

## Incremental sync

Set `incremental_column` to a `TIMESTAMP` or `DATE` column. Mixpeek records the highest value it has read and filters the next run to rows above it.

Set `primary_key_columns` so object IDs stay stable across runs. Without it, a re-read can create a second object for the same row.

## Bulk import and change capture

For a large table, or one with an embedding column, set `provider_filters.import` on the sync. The table is read in primary-key slices, each slice lands in the bucket as one Parquet object, and every row becomes one document in the collection. Each document's ID derives from the source table and primary key, so a re-run, a resumed import or a later change updates the same document.

```json theme={null}
{
  "connection_id": "conn_your_connection_id",
  "source_path": "public.scenes",
  "sync_mode": "continuous",
  "polling_interval_seconds": 60,
  "batch_collection_ids": ["col_your_collection_id"],
  "provider_filters": {
    "import": {
      "mode": "slices",
      "slice_rows": 20000,
      "objects_per_batch": 4,
      "text_column": "caption",
      "vector_columns": {
        "embedding": {"model": "text-embedding-3-large", "dimensions": 3072, "mode": "carry"}
      },
      "replication": "logical",
      "publication": "mixpeek_pub"
    }
  }
}
```

| Field | Description |
| - | - |
| `mode` | `slices` for the bulk import. `rows` keeps one JSON object per row. |
| `slice_rows` | Rows per slice. Defaults to `20000`. |
| `objects_per_batch` | Slices per processing batch. Defaults to `4`. |
| `text_column` | The column the collection's extractor reads as text. |
| `key_columns` | For a view, the columns that key each row. A view has no primary key, so the import needs to be told which columns identify a row. Their values must be unique per row. Ignored for a table. |
| `vector_columns` | One entry per `vector` or `halfvec` column: the model that produced it, its dimension, and `carry` (store the vector as the document's vector, with no embedding call) or `reextract` (drop it and embed `text_column`). |
| `replication` | `logical`, `polling` or `none`. See below. |
| `publication` | The publication the logical slot reads. Required for `logical`. |
| `slot_lag_alert_bytes` | Retained WAL that raises the slot-lag alert. Defaults to 1 GiB. |
| `slot_drop_after_pause_seconds` | A sync paused this long has its slot dropped. Defaults to 24 hours. |

Customer columns keep their names in the document. Mixpeek's own fields, including the source row's lineage (schema, table, primary key and vector origin), live under `_internal`.

A column whose name starts with `_`, or matches a field Mixpeek writes on each imported row (such as `metadata`, `text`, `data`, `document_id`, `collection_id` or `mime_type`), would collide with that field, so the import refuses the table and names the column. Import a view that renames it instead, and set `key_columns` to the view's key:

```sql theme={null}
CREATE VIEW public.scenes_import AS
SELECT scene_id, metadata AS source_metadata, caption FROM public.scenes;
```

```json theme={null}
"import": {"mode": "slices", "key_columns": ["scene_id"], "text_column": "caption"}
```

Logical replication cannot publish a view, so a view imports with `replication: polling` or `none`.

A carried vector is refused at sync create when its model or dimension differs from the collection's feature, or when the model has no registered query-time model, because its documents could not be searched by text. Tables without a vector column can import into a passthrough collection.

### Change capture

With `replication: logical`, the import creates a logical replication slot first and copies every slice inside the slot's exported snapshot. After the copy, each sync run streams the changes committed since, writes them as one change object, and confirms the slot position only after that object's documents are written. Inserts, updates and deletes all reach the collection.

Logical replication needs:

* `wal_level = logical` on the server.
* A role with the `REPLICATION` attribute, connecting directly to Postgres. Transaction pooling (PgBouncer in transaction mode) cannot carry a replication connection.
* A publication that includes the table, created by the table owner:

```sql theme={null}
ALTER ROLE mixpeek_sync REPLICATION;
CREATE PUBLICATION mixpeek_pub FOR TABLE public.scenes;
```

A slot keeps WAL on your server until Mixpeek confirms it. Set `max_slot_wal_keep_size` so a stalled consumer cannot fill the disk, and watch the slot-lag alert on the sync record. Deleting the sync drops its slot.

`replication: polling` reads rows whose `updated_at` (or `updated_at_column`) moved past the last value seen. It needs no replication role, and it misses hard deletes: a row deleted at the source stays in the collection.

Every session that reads rows runs with `default_transaction_read_only = on`, so the connection cannot write to your database. The replication slot is the only object the import creates on the source.

## Related

> [card] [SQL Lookup stage](https://docs.mixpeek.com/docs/retrieval/stages/sql-lookup.md)
>
> Enrich documents from a SQL source inside a retriever.

> [card] [Snowflake](https://docs.mixpeek.com/docs/integrations/snowflake-warehouse.md)
>
> Warehouse connection with the same two paths.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.