---
title: "Turn a Postgres table into a semantic index"
description: "Use an existing PostgreSQL table as a CocoIndex source: derive fields, embed each row, and store the vectors in Postgres with pgvector, incrementally."
last_updated: 2025-09-01
doc_version: "2025-09-01"
canonical: https://cocoindex.io/blogs/postgres-source/
---
# Turn a Postgres table into a semantic index

> Use an existing PostgreSQL table as a CocoIndex source: derive fields, embed each row, and store the vectors in Postgres with pgvector, incrementally.

Published: 2025-09-01 · Canonical: https://cocoindex.io/blogs/postgres-source/

Most of your data already lives in a database. This example takes an existing Postgres table of products, reads it row by row, derives a couple of fields, embeds each row, and writes the result, including the vector, back into Postgres with [pgvector](https://github.com/pgvector/pgvector). Structured data in, searchable-by-meaning index out, no separate export step.

You declare the transformation in native Python and your own types, `target_state = transformation(source_state)`, and a Rust engine handles the heavy lifting underneath: incremental processing, change tracking, and managed targets. Only the rows that changed get re-embedded and re-upserted.

The full code is in the [CocoIndex v1 repo](https://github.com/cocoindex-io/cocoindex/tree/main/examples/postgres_source). A star on [GitHub](https://github.com/cocoindex-io/cocoindex) is appreciated if this helps.

## What we'll build

The pipeline reads product rows from `source_products`, computes a `total_value` and a combined `full_description`, embeds that description, and stores the enriched rows plus their vectors in a second Postgres table.

Both ends are Postgres here, but they need not be: point the source at one database and the target at another by setting a different connection string. The source can be any table you already have.

## Define the row shapes

Each source row maps into a plain dataclass, and each output row into another. The `embedding` field is annotated with the embedder so CocoIndex infers the vector's size automatically:

```python title=main.py
from dataclasses import dataclass
from typing import Annotated
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder", detect_change=True)

@dataclass
class SourceProduct:
    product_category: str
    product_name: str
    description: str
    price: float
    amount: int

@dataclass
class OutputProduct:
    product_category: str
    product_name: str
    description: str
    price: float
    amount: int
    total_value: float
    embedding: Annotated[NDArray, EMBEDDER]
```

The database pools and the embedder are created once, in the app lifespan. Set `SOURCE_DATABASE_URL` to read from a separate database, or leave it pointing at the same instance:

```python title=main.py
import asyncpg
from cocoindex.connectors import postgres

PG_DB = coco.ContextKey[asyncpg.Pool]("postgres_source_db")
SOURCE_POOL = coco.ContextKey[asyncpg.Pool]("source_pool")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with (
        asyncpg.create_pool(DATABASE_URL) as target_pool,
        asyncpg.create_pool(SOURCE_DATABASE_URL) as source_pool,
    ):
        builder.provide(PG_DB, target_pool)
        builder.provide(SOURCE_POOL, source_pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder(EMBED_MODEL))
        yield
```

## Transform and embed each row

`process_product` runs once per source row. It computes the derived fields, embeds the composed description, and declares the output row. `memo=True` caches this work, so a row whose content and code are unchanged is skipped on the next run:

```python title=main.py
@coco.fn(memo=True)
async def process_product(
    product: SourceProduct,
    table: postgres.TableTarget[OutputProduct],
) -> None:
    full_description = (
        f"Category: {product.product_category}\n"
        f"Name: {product.product_name}\n\n{product.description}"
    )
    total_value = product.price * product.amount
    embedding = await coco.use_context(EMBEDDER).embed(full_description)
    table.declare_row(
        row=OutputProduct(
            product_category=product.product_category,
            product_name=product.product_name,
            description=product.description,
            price=product.price,
            amount=product.amount,
            total_value=total_value,
            embedding=embedding,
        ),
    )
```

Embeddings carry more meaning when they see more context. We embed the *composed* description, category and name included, so a search for "wireless audio" matches even when the body text never says it.

The AI step and the plain arithmetic sit side by side in the same function. `total_value = price * amount` is an ordinary Python expression; `embed(...)` is a model call. To CocoIndex both are just transforms that derive an output field from the source row.

## Wire up the app

`app_main` mounts the target table, opens the source table, and runs one processing component per source row:

```python title=main.py
@coco.fn
async def app_main() -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name="output",
        table_schema=await postgres.TableSchema.from_class(
            OutputProduct,
            primary_key=["product_category", "product_name"],
        ),
        pg_schema_name="coco_examples_v1",
    )

    source = postgres.PgTableSource(
        coco.use_context(SOURCE_POOL),
        table_name="source_products",
        row_type=SourceProduct,
    )

    await coco.mount_each(
        process_product,
        source.fetch_rows().items(lambda p: (p.product_category, p.product_name)),
        target_table,
    )

app = coco.App(
    coco.AppConfig(name="PostgresSourceV1"),
    app_main,
)
```

`PgTableSource` reads the existing table directly, and `row_type=SourceProduct` maps each row straight into the dataclass. `items(...)` tags each row with its `(product_category, product_name)` composite key, which becomes the output row's primary key. `mount_table_target` creates and manages the target table for you: schema, idempotent upserts, and cleanup of rows whose source disappeared.

Because the output key derives from the source row, incremental updates are exact. Change one product and only that row is re-embedded and re-upserted; delete a source row and its output row is removed. That is what `memo=True` and the derived key buy you: the expensive embedding work scales with how much actually changed, not with the size of the table.

## Run it

Start Postgres with pgvector (the repo ships a compose file), configure the connection strings, and seed the source table:

```sh
docker compose -f ../../dev/postgres.yaml up -d
cp .env.example .env     # set POSTGRES_URL and SOURCE_DATABASE_URL
pip install -e .
psql "$SOURCE_DATABASE_URL" -f ./prepare_source_data.sql
```

Build the index. The Postgres source runs as a one-shot catch-up: it scans the source table, syncs the target, and exits.

```sh
cocoindex update main
```

Re-run it any time; only new or changed rows are processed. To keep the index fresh on a schedule, run `cocoindex update` from cron.

## Query the index

Searching is the mirror image of indexing: embed the query with the same model, then ask Postgres for the nearest vectors. The `<=>` operator is pgvector's cosine distance, so smaller means more similar:

```python title=main.py
async def query_once(pool, embedder, query: str, *, top_k: int = 5) -> None:
    query_vec = await embedder.embed(query)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """
            SELECT product_category, product_name, description, amount,
                   total_value, embedding <=> $1 AS distance
            FROM "coco_examples_v1"."output"
            ORDER BY distance ASC
            LIMIT $2
            """,
            query_vec,
            top_k,
        )
    for r in rows:
        score = 1.0 - float(r["distance"])
        print(f"[{score:.3f}] {r['product_category']} | {r['product_name']}")
        print(f"    {r['description']}")
```

Run it and ask in plain language:

```sh
python main.py "wireless headphones"
```

The closest products come back ranked, even when they share none of the words in your query. That is the point of a vector index over your structured data.

## Where to go next

You now have a semantic index over an existing Postgres table, kept in sync incrementally. The same shape extends a long way: swap the source for [Amazon S3](https://cocoindex.io/blogs/s3-incremental-etl), [Azure Blob](https://cocoindex.io/docs/connectors/azure_blob/), or Google Drive; swap the target for a dedicated vector database like Qdrant; or feed the retrieved rows to an LLM for [RAG](https://cocoindex.io/docs/). Because the pipeline is declarative and incremental, none of that changes how you keep the index fresh.

## Support us

We are constantly adding examples and improving the runtime. If this guide helped, please [star CocoIndex on GitHub](https://github.com/cocoindex-io/cocoindex) and let us know what you are building.

## Sitemap

- [Blog index](https://cocoindex.io/blogs/)
- [Site index (llms.txt)](https://cocoindex.io/llms.txt)
- [Full blog corpus](https://cocoindex.io/llms-full.txt)
- [Markdown sitemap](https://cocoindex.io/sitemap.md)
- [XML sitemap](https://cocoindex.io/sitemap.xml)
- [RSS feed](https://cocoindex.io/blogs/rss.xml)