PostgreSQL

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 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`

source_path accepts one or two dot-separated parts. Three or more returns a validation error.

Setup

Create a read-only role

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;

Create the storage connection

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']}")

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)

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"]
    }
  }'

Sync a table

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']}")

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.

{
  "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.
`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 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:

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.

SQL Lookup stage

Enrich documents from a SQL source inside a retriever.

Snowflake

Warehouse connection with the same two paths.

Get started

Connect your data

Extract features

Build retrievers

Enrich & organize

Integrate & operate

Resources

Overview

Prerequisites

Configuration

Connection-level fields

Sync-level fields

Setup

Incremental sync

Bulk import and change capture

Change capture

SQL Lookup stage

Snowflake

​Overview

​Prerequisites

​Configuration

​Connection-level fields

​Sync-level fields

​Setup

​Incremental sync

​Bulk import and change capture

​Change capture

​Related

SQL Lookup stage

Snowflake

Overview

Prerequisites

Configuration

Connection-level fields

Sync-level fields

Setup

Incremental sync

Bulk import and change capture

Change capture

Related