Postgres connector

Read Postgres tables as a row stream via read_table and write with declare_table_target() — schema derivation from Rust structs, pgvector index management (ivfflat / hnsw), column type overrides, and SQL attachments for triggers and indexes.

Version: v 1.0.14
Last reviewed: Nov 4, 2026

ExperimentalThe Rust SDK is experimental. It is outside CocoIndex's compatibility promise: any release, including a minor one, may change or remove its API. It is not yet published on crates.io.

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

use cocoindex::connectors::postgres;

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

Connection setup

Open a postgres::Database:

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

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.

Different from Python

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:

/// 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:

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

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 before renaming.

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.

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).
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.

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). 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:

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.

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:

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.

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:

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:

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:

/// 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.

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:

#[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:

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.

Different from Python

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

// 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:

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

Example:

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

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.