# Postgres connector

> **CocoIndex v1, Rust SDK (experimental).** This page documents the CocoIndex **v1** Rust SDK (the `cocoindex` crate). The Rust SDK is experimental: any release may change or remove its API, and it is not yet published on crates.io. The Python SDK pages are at https://cocoindex.io/docs/.
>
> Source: https://cocoindex.io/docs/rust/connectors/postgres/ · Rust docs index: https://cocoindex.io/docs/rust/llms.txt · Rust SDK tour: https://github.com/cocoindex-io/cocoindex/blob/main/rust/sdk/SHOWCASE.md · Rust examples: https://github.com/cocoindex-io/cocoindex/tree/main/examples/rust

The `postgres` connector provides utilities for reading rows from and writing rows to PostgreSQL databases, with built-in support for pgvector.

```rust
use cocoindex::connectors::postgres;
```

The module is compiled with the `postgres` Cargo feature of the `cocoindex` crate.

## Connection setup

Open a `postgres::Database`:

```rust
let db = postgres::Database::connect("postgres://user:pass@localhost/dbname").await?;
let pool = db.pool().clone();  // the underlying sqlx::PgPool
```

`postgres::Database` wraps an [`sqlx`](https://docs.rs/sqlx) `PgPool`. `Database::connect(url)` opens a pool of up to 8 connections; `Database::from_pool(state_id, pool)` wraps a pool you built yourself.

The connector handles pgvector extension setup automatically when a table uses vector columns — no special pool initialization is needed.

## As source

Use `read_table` to read rows from a PostgreSQL table. It returns a stream of rows, read from one consistent snapshot of the table.

### read_table

```rust
pub async fn read_table<T>(
    db: &Database,
    table_name: &str,
) -> Result<impl Stream<Item = Result<T>> + Send + 'static>
where
    T: serde::de::DeserializeOwned + Send + 'static

pub async fn read_table_with_options<T>(
    db: &Database,
    table_name: &str,
    options: ReadTableOptions,
) -> Result<impl Stream<Item = Result<T>> + Send + 'static>
where
    T: serde::de::DeserializeOwned + Send + 'static

pub async fn read_table_items<T, K, S>(
    db: &Database,
    table_name: &str,
    key_fn: K,
) -> Result<impl Stream<Item = Result<(StableKey, T)>> + Send + 'static>
where
    T: serde::de::DeserializeOwned + Send + 'static,
    K: Fn(&T) -> S + Send + 'static,
    S: IntoStableKey

impl ReadTableOptions {
    pub fn new() -> Self
    pub fn pg_schema_name(self, schema: impl Into<String>) -> Self
    pub fn columns(self, columns: impl IntoIterator<Item = impl Into<String>>) -> Self
}
```

`read_table_with_options` and `read_table_items_with_options` take a `ReadTableOptions` as well. `read_table_items` pairs each row with the key `key_fn` derives from it, ready for `ctx.mount_each` once collected.

**Parameters:**

- `db` — The `postgres::Database` to read from.
- `table_name` — Name of the table to read from.
- `T` — The row type. Each row is deserialized into `T`.
- `ReadTableOptions::columns` — Column names to select. When not set, the query is `SELECT *`.
- `ReadTableOptions::pg_schema_name` — PostgreSQL schema name. When not set, the table name isn't schema-qualified, so the connection's `search_path` picks the schema (normally `public`).
- `key_fn` (`read_table_items` only) — Derives the stable key of a row.

### Row mapping

The row type is the type parameter of `read_table::<T>`. `T` must implement `serde::Deserialize`; column names map to struct fields, and unknown columns are ignored. Use `serde_json::Value` as `T` to get each row as a JSON object.

There is no `row_factory`. Read the rows into a struct, then map them with the stream's or an iterator's `map`.

#### Choosing the row type

Name the row type on the result, or as `read_table::<T>`. Select fewer columns with `ReadTableOptions::columns`:

```rust
/// One row of the `source_products` source table (extra columns are ignored).
#[derive(Clone, Serialize, Deserialize, MemoInput)]
struct SourceProduct {
    product_category: String,
    product_name: String,
    description: String,
    price: f64,
    amount: i64,
}

let products: Vec<SourceProduct> = postgres::read_table(source_db, "source_products")
    .await?
    .try_collect()
    .await?;

// Only some columns
let mut rows: Vec<SourceNameOnly> = postgres::read_table_with_options(
    &db,
    &src_table,
    postgres::ReadTableOptions::new().columns(["name"]),
)
.await?
.try_collect()
.await?;
```

`try_collect` comes from `futures::TryStreamExt`.

Derive `MemoInput` as well when the row is an argument of a memoized function.

### Iterating rows

`read_table` returns a stream of `Result<T>`. Consume it with the `futures` stream adapters, or collect it:

```rust
use futures::TryStreamExt as _;

let products: Vec<SourceProduct> = postgres::read_table(source_db, "source_products")
    .await?
    .try_collect()
    .await?;
println!("read {} source product(s)", products.len());
```

`read_table` reads the whole table in one `REPEATABLE READ, READ ONLY` transaction, so the rows come from one consistent snapshot. The rows stream from the server as the stream is consumed: at most 256 rows are buffered ahead of the consumer. Dropping the stream ends the read.

### Example

```rust
use cocoindex::connectors::postgres;
use cocoindex::prelude::*;
use futures::TryStreamExt as _;

// Source database.
cocoindex::context_key!(static SOURCE_DB: postgres::Database);

/// One row of the `source_products` source table (extra columns are ignored).
#[derive(Clone, Serialize, Deserialize, MemoInput)]
struct SourceProduct {
    product_category: String,
    product_name: String,
    description: String,
    price: f64,
    amount: i64,
}

/// Derive and embed one product, declaring its output row. Memoized on the
/// source row: an unchanged row keeps its output without re-running.
#[cocoindex::function(memo, memo_key(table = skip))]
async fn process_product(
    ctx: &Ctx,
    product: SourceProduct,
    table: postgres::TableTarget,
) -> Result<()> {
    // ...
}

async fn app_main(ctx: Ctx) -> Result<()> {
    let target =
        postgres::mount_table_target(&ctx, &DB, TABLE, output_schema()?, Some(PG_SCHEMA)).await?;

    let source_db = ctx.get_key(&SOURCE_DB)?;
    let products: Vec<SourceProduct> = postgres::read_table(source_db, "source_products")
        .await?
        .try_collect()
        .await?;

    // One processing component per source row.
    mount_each!(
        products
            .into_iter()
            .map(|p| (format!("{}|{}", p.product_category, p.product_name), p)),
        |product| process_product(ctx, product, target)
    )
    .await?;
    Ok(())
}
```

`mount_each!` takes `(key, row)` pairs; the key is the component subpath. See `examples/rust/postgres_source` for the full program.

## As target

The `postgres` connector provides target state APIs for writing rows to tables. With it, CocoIndex tracks what rows should exist and automatically handles upserts and deletions.

### Declaring target states

#### Setting up a connection

Declare a context key for the `postgres::Database`, then provide the database in your lifespan:

**Note**
The key name is load-bearing across runs — it's the stable identity CocoIndex uses to track managed rows. See [ContextKey as stable identity](/docs/rust/programming_guide/context/#contextkey-as-stable-identity) before renaming.

```rust
use cocoindex::connectors::postgres;
use cocoindex::prelude::*;

cocoindex::context_key!(static DB: postgres::Database);

#[cocoindex::lifespan]
async fn lifespan(builder: EnvironmentBuilder) -> Result<EnvironmentBuilder> {
    let db = postgres::Database::connect(&database_url()).await?;
    let pool = db.pool().clone();
    Ok(builder
        .db_path(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&DB, db)
        .on_shutdown(move || async move { pool.close().await }))
}
```

`context_key!(static DB: postgres::Database)` declares the key; its name (`DB`) is the stable identity. `provide_key(&DB, db)` provides the connection. Without a lifespan, call `Environment::builder().provide_key(&DB, db)` directly.

#### Tables (parent state)

Declares a table as a target state. Returns a `TableTarget` for declaring rows.

```rust
pub fn declare_table_target(
    ctx: &Ctx,
    db: &ContextKey<Database>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    pg_schema_name: Option<&str>,
) -> Result<TableTarget>

pub async fn mount_table_target(
    ctx: &Ctx,
    db: &ContextKey<Database>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    pg_schema_name: Option<&str>,
) -> Result<TableTarget>

pub async fn mount_table_target_with_options(
    ctx: &Ctx,
    db: &ContextKey<Database>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    pg_schema_name: Option<&str>,
    options: ManagedTargetOptions,
) -> Result<TableTarget>
```

Each function has a `_with_options` variant (`declare_table_target_with_options`, `mount_table_target_with_options`, `table_target_with_options`) that takes a `ManagedTargetOptions`. `ManagedTargetOptions::user_managed()` sets a user-managed table; the default is system-managed.

**Parameters:**

- `ctx` — The current component's context.
- `db` — The `ContextKey<postgres::Database>` identifying the database to use.
- `table_name` — Name of the table.
- `table_schema` — Schema definition including columns and primary key (see [Table Schema](#table-schema-from-a-rust-struct)).
- `pg_schema_name` — Optional PostgreSQL schema name. When `None`, the table name isn't schema-qualified, so the connection's `search_path` picks the schema (normally `public`).
- `options` (`_with_options` variants) — `ManagedTargetOptions`: whether CocoIndex manages the table lifecycle (the default) or assumes it exists (`ManagedTargetOptions::user_managed()`).

**Returns:** A `TableTarget`. From `declare_table_target`, its rows resolve when the current component commits. From `mount_table_target(...).await?`, it accepts rows at once.

**Caution — One table, one key**
CocoIndex identifies a table by its key: the `db` key, `pg_schema_name` and `table_name`. Two different keys are two different tables to CocoIndex, even when PostgreSQL resolves them to the same table (see [ContextKey as stable identity](/docs/rust/programming_guide/context/#contextkey-as-stable-identity)). Two cases are easy to miss:

- `pg_schema_name: None` and `Some("public")` are the same key, even when `search_path` makes `None` reach a table in another schema. Two targets in that case collide on one key.
- PostgreSQL truncates a table name longer than 63 bytes, so two long names that share their first 63 bytes are one table.

Declare each table through one key only. Switching a table from one form to the other changes its key: CocoIndex then cleans up the table it tracked under the old key, which is the same table.

#### Rows (child states)

Once a `TableTarget` is resolved, declare rows to be upserted:

```rust
impl TableTarget {
    pub fn declare_row<R: Serialize>(&self, ctx: &Ctx, row: &R) -> Result<()>
}

table.declare_row(&ctx, &row)?;
```

**Parameters:**

- `row` — Any `Serialize` value that serializes to an object whose entries are the table's columns. Extra entries are ignored, and a missing entry is written as NULL. It must include all primary key columns.

CocoIndex writes rows in batches: each upsert or delete statement carries up to 2,000 rows, and the statements of a batch run concurrently over the pool.

#### Vector indexes (attachment)

Declare a pgvector index on a vector column of the table. CocoIndex tracks the index spec and automatically creates, recreates, or drops the index as needed.

```rust
impl TableTarget {
    pub fn declare_vector_index(
        &self,
        ctx: &Ctx,
        column: &str,
        options: VectorIndexOptions,
    ) -> Result<()>
}

pub struct VectorIndexOptions {
    pub name: Option<String>,
    pub metric: &'static str,    // "cosine" (default), "l2", "ip"
    pub method: &'static str,    // "ivfflat" (default), "hnsw"
    pub lists: Option<u32>,
    pub m: Option<u32>,
    pub ef_construction: Option<u32>,
}
```

The actual PostgreSQL index is named `{table_name}__vector__{name}`. PostgreSQL truncates a name longer than 63 bytes. `declare_vector_index` does not check for two names that become the same after truncation: keep the full index names distinct in their first 63 bytes.

**Parameters:**

- `name` — Logical index name (defaults to `column`).
- `column` — Column to index (must be a vector column).
- `metric` — Distance metric: `"cosine"`, `"l2"`, or `"ip"` (inner product).
- `method` — Index method: `"ivfflat"` or `"hnsw"`.
- `lists` — Number of lists (ivfflat only).
- `m` — Maximum number of connections per layer (hnsw only).
- `ef_construction` — Size of the dynamic candidate list for construction (hnsw only).

**Example:**

```rust
table.declare_vector_index(
    &ctx,
    "embedding",
    postgres::VectorIndexOptions {
        method: "hnsw",
        ..Default::default()
    },
)?;
```

#### SQL command attachments

Declare an arbitrary SQL command that CocoIndex manages alongside the table. The setup SQL runs when the attachment is created or changed; the optional teardown SQL runs when the attachment is removed or before re-running setup on change.

```rust
impl TableTarget {
    pub fn declare_sql_command_attachment(
        &self,
        ctx: &Ctx,
        name: &str,
        setup_sql: impl Into<String>,
        teardown_sql: Option<String>,
    ) -> Result<()>
}
```

**Parameters:**

- `name` — Stable identifier for the attachment.
- `setup_sql` — SQL to execute on creation or change.
- `teardown_sql` — SQL to execute on removal or before re-running setup (optional). If omitted, no cleanup is performed when the attachment is removed.

**Example:**

```rust
table.declare_sql_command_attachment(
    &ctx,
    "label_index",
    format!("CREATE INDEX IF NOT EXISTS rows_label_idx ON \"{schema}\".\"rows\" (label)"),
    Some(format!("DROP INDEX IF EXISTS \"{schema}\".rows_label_idx")),
)?;
```

### Table schema: from a Rust struct

Define the table structure with a struct that derives `SchemaFields` and `Serialize`:

```rust
impl TableSchema {
    pub fn from_row<T: SchemaFields>(
        primary_key: impl IntoIterator<Item = impl Into<String>>,
    ) -> Result<Self>

    pub fn with_vector_dim(self, field_name: &str, dim: usize) -> Result<Self>
    pub async fn with_vector_schema(
        self,
        field_name: &str,
        provider: &(impl VectorSchemaProvider + ?Sized),
    ) -> Result<Self>
    pub fn with_column_type(self, field_name: &str, pg_type: impl Into<String>) -> Result<Self>
}
```

`TableSchema::from_row::<Row>(primary_key)` derives the columns. Builder calls on the derived schema override single columns: `with_vector_schema` or `with_vector_dim` for a vector column, `with_column_type` for a type override.

**Parameters:**

- `T` — A struct whose fields define table columns.
- `primary_key` — Column names forming the primary key.

**Example:**

```rust
/// One row written to the output table.
#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct OutputProduct {
    product_category: String,
    product_name: String,
    description: String,
    price: f64,
    amount: i64,
    total_value: f64,
    #[coco(vector)]
    embedding: Vec<f32>,
}

fn output_schema() -> Result<postgres::TableSchema> {
    postgres::TableSchema::from_row::<OutputProduct>(["product_category", "product_name"])?
        .with_vector_dim("embedding", EMBED_DIM)
}
```

`#[coco(vector)]` marks a vector column whose dimension is set later with `with_vector_dim` or `with_vector_schema`; `#[coco(vector = 384)]` fixes it in the struct. `Option<T>` fields become nullable columns; all others are `NOT NULL`.

Rust field types are automatically mapped to PostgreSQL types:

| Rust Type | PostgreSQL Type |
|-----------|-----------------|
| `bool` | `boolean` |
| `i64` (`i16`: `smallint`, `i32`: `integer`) | `bigint` |
| `f64` (`f32`: `real`) | `double precision` |
| `Decimal` | `numeric` |
| `String` | `text` |
| `Vec<u8>` | `bytea` |
| `Uuid` | `uuid` |
| `NaiveDate` | `date` |
| `NaiveTime` | `time with time zone` |
| `DateTime`, `NaiveDateTime` | `timestamp with time zone` |
| `Duration` | `interval` |
| other `Vec<T>`, maps, nested structs, `#[coco(json)]` | `jsonb` |
| `Vec<f32>` with `#[coco(vector)]` (`half` for `halfvec`) | `vector(n)` or `halfvec(n)` |

Unsigned integers map to the next signed width (`u32` to `bigint`, `u64` to `bigint`); a value above `i64::MAX` is rejected when the row is written.

**Note — U+0000 (NUL) in strings**
U+0000 (NUL) is a valid Unicode codepoint, but Postgres cannot store it — neither in `text`-family columns nor inside strings in `jsonb` (the `\u0000` escape is rejected at parse time). CocoIndex automatically strips U+0000 from strings before writing to Postgres, recursively for nested strings and dict keys in `jsonb` payloads. For example, `"Hello\0World"` is written as `"HelloWorld"`.

To override the default mapping, use:

- **Field attribute** — `#[coco(type = "…")]` or `#[coco(vector = N)]` on the struct field
- **Schema builder calls** — `with_column_type`, `with_vector_schema` or `with_vector_dim` on the `TableSchema` derived with `from_row`

#### Column types

Set a custom PostgreSQL type on a field:

```rust
#[derive(Serialize, SchemaFields)]
struct MyRow {
    #[coco(type = "integer")]
    id: i64,
    #[coco(type = "real")]
    value: f64,
    #[coco(type = "timestamp")]
    created_at: DateTime<Utc>,
}
```

Or on the derived schema, for this schema only:

```rust
let table_schema = postgres::TableSchema::from_row::<ProviderRow>(["id"])?
    .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
    .await?
    .with_column_type("title", "varchar(40)")?;
```

`#[coco(type = "…")]` applies to every connector the struct is used with; `with_column_type` applies to one schema and keeps the column's nullability.

There is no `PgType` encoder. A field's `Serialize` impl produces the value that is bound: `uuid`, `chrono`, `rust_decimal`, `std::net::IpAddr` (for `inet`) and `Vec<f32>` (for pgvector) serialize to what their columns take. For another form, use `#[serde(serialize_with = "…")]` on the field.

#### VectorSchemaProvider

For a `#[coco(vector)]` field, a [`VectorSchemaProvider`](/docs/rust/common_resources/vector_schema/#vectorschemaprovider) passed to `with_vector_schema` specifies the vector dimension and element type. The connector has built-in pgvector support and automatically creates the extension when needed.

```rust
// The embedder reports the dimension of the `embedding` column.
postgres::TableSchema::from_row::<DocEmbedding>(["id"])?
    .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
    .await?
```

`with_vector_schema(field, provider)` takes any `VectorSchemaProvider`: an embedder, or a `VectorSchema`. For an embedder held in the context, pass `ctx.get_key(&EMBEDDER)?`. An `f16` provider selects `halfvec(n)`.

### Table schema: explicit column definitions

Define columns directly using `ColumnDef`:

```rust
impl TableSchema {
    pub fn new(
        columns: impl IntoIterator<Item = (impl Into<String>, ColumnDef)>,
        primary_key: impl IntoIterator<Item = impl Into<String>>,
    ) -> Result<Self>
}
```

**Example:**

```rust
postgres::TableSchema::new(
    [
        ("category", postgres::ColumnDef::new("text")),
        ("name", postgres::ColumnDef::new("text")),
        ("total_value", postgres::ColumnDef::new("double precision")),
        ("title", postgres::ColumnDef::new("text").nullable()),
    ],
    ["category", "name"],
)?
```

`ColumnDef::new(pg_type)` is `NOT NULL`; `.nullable()` opts in to nulls.

### Example

```rust
use cocoindex::connectors::postgres;
use cocoindex::ops::sentence_transformers::SentenceTransformerEmbedder;
use cocoindex::prelude::*;

const PG_SCHEMA: &str = "coco_examples";
const TABLE: &str = "doc_embeddings";

cocoindex::context_key!(static DB: postgres::Database);
cocoindex::context_key!(
    static EMBEDDER: SentenceTransformerEmbedder,
    detect_change
);

#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct DocEmbedding {
    id: i64,
    filename: String,
    chunk_start: i32,
    chunk_end: i32,
    text: String,
    #[coco(vector)]
    embedding: Vec<f32>,
}

#[cocoindex::lifespan]
async fn lifespan(builder: EnvironmentBuilder) -> Result<EnvironmentBuilder> {
    let db = postgres::Database::connect(&database_url()).await?;
    let pool = db.pool().clone();
    let embedder = load_embedder().await?;
    Ok(builder
        .db_path(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&DB, db)
        .provide_key(&EMBEDDER, embedder)
        .on_shutdown(move || async move { pool.close().await }))
}

async fn index(ctx: Ctx, sourcedir: PathBuf) -> Result<()> {
    // The embedder reports the dimension of the `embedding` column.
    let table = postgres::mount_table_target(
        &ctx,
        &DB,
        TABLE,
        postgres::TableSchema::from_row::<DocEmbedding>(["id"])?
            .with_vector_schema("embedding", ctx.get_key(&EMBEDDER)?)
            .await?,
        Some(PG_SCHEMA),
    )
    .await?;
    table.declare_vector_index(
        &ctx,
        "embedding",
        postgres::VectorIndexOptions {
            method: "hnsw",
            ..Default::default()
        },
    )?;

    // One component per markdown file; each declares its own rows.
    let files = walk_dir(sourcedir)
        .recursive(true)
        .path_matcher(PatternFilePathMatcher::new(
            ["**/*.md"],
            std::iter::empty::<&str>(),
        )?)
        .live()
        .poll_interval(Duration::from_secs(1));
    ctx.mount_each_live(&"files", files, move |ctx, file: FileEntry| {
        let table = table.clone();
        async move {
            for row in process_file(&ctx, &file).await? {
                table.declare_row(&ctx, &row)?;
            }
            Ok(())
        }
    })
    .await
}
```

See `examples/rust/text_embedding` for the full program, including `process_file` and the `#[cocoindex::app]` entry point.