CocoIndex quickstart

Build an incremental Rust pipeline that chunks Markdown, batches and caches local embeddings, and keeps a Postgres/pgvector table in sync.

Language
Rust 1.91+
Requires
Postgres with pgvector
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.

Build the Rust version of the text-embedding pipeline: read Markdown files, split them into chunks, embed the chunks, and declare the rows that should exist in Postgres.

The important word is declare. Your code describes the current target state; CocoIndex works out which rows to insert, update, retain, or delete on each run.

Create the project

Create a binary crate with a directory for input files:

cargo new cocoindex-rust-quickstart
cd cocoindex-rust-quickstart
mkdir markdown_files

Until the Rust SDK is published separately, depend on the CocoIndex workspace from GitHub. Add these dependencies to Cargo.toml:

Cargo.toml
[dependencies]
cocoindex = { git = "https://github.com/cocoindex-io/cocoindex", features = [
    "postgres",
    "text",
    "fastembed",
    "fs_live",
] }
dotenvy = "0.15"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }

You also need a Postgres database where the pgvector extension can be created. Set its connection URL:

export POSTGRES_URL=postgres://cocoindex:cocoindex@localhost/cocoindex

Add one or more .md files under markdown_files/.

Define resources and the row type

Start src/main.rs with the imports, constants, and two typed context keys:

main.rs
use std::path::PathBuf;
use std::time::Duration;

use cocoindex::connectors::postgres;
use cocoindex::ops::sentence_transformers::SentenceTransformerEmbedder;
use cocoindex::ops::text::{RecursiveChunkConfig, RecursiveSplitter};
use cocoindex::prelude::*;

const EMBED_MODEL: &str = "sentence-transformers/all-MiniLM-L6-v2";
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>,
}

context_key! gives each provided resource a stable name and type, using the Rust identifier as the key name. Change detection is disabled for the database connection. It is enabled for the embedder, whose type defines the model identity that invalidates dependent memoized work. Runtime handles such as connection pools do not need to be serializable.

SchemaFields derives the database columns from DocEmbedding. The vector dimension is intentionally absent because it will come from the loaded model.

Process one file

Add the per-chunk and per-file processing functions. The per-file function is memoized and takes the table target, and each chunk declares its own row:

main.rs
#[cocoindex::function]
async fn process_chunk(
    ctx: &Ctx,
    chunk: Chunk,
    text: &str,
    filename: &str,
    id_gen: &IdGenerator,
    table: &postgres::TableTarget,
) -> Result<()> {
    let chunk_text = chunk.text(text);
    let id = i64::try_from(id_gen.next_id(ctx, chunk_text).await?)
        .map_err(|_| Error::engine("generated id does not fit in BIGINT"))?;
    table.declare_row(
        ctx,
        &DocEmbedding {
            id,
            filename: filename.to_string(),
            chunk_start: chunk.start.char_offset as i32,
            chunk_end: chunk.end.char_offset as i32,
            text: chunk_text.to_string(),
            embedding: ctx.get_key(&EMBEDDER)?.embed(ctx, chunk_text).await?,
        },
    )
}

#[cocoindex::function(memo, memo_key(table = skip))]
async fn process_file(ctx: &Ctx, file: &FileEntry, table: &postgres::TableTarget) -> Result<()> {
    let text = file.read_text().await?;
    let chunks = RecursiveSplitter::new()?.split_with(
        &text,
        RecursiveChunkConfig {
            chunk_size: 2000,
            min_chunk_size: None,
            chunk_overlap: Some(500),
            language: Some("markdown".to_string()),
        },
    );
    let filename = file.key();
    let id_gen = IdGenerator::new();
    ctx.map(chunks, async |chunk| {
        process_chunk(ctx, chunk, &text, &filename, &id_gen, &table).await
    })
    .await?;
    Ok(())
}

memo caches process_file by its logic and its MemoInput arguments. The file’s content freshness is part of that key, so an unchanged file keeps its rows without re-chunking. The table target is the same for every file, so memo_key(table = skip) leaves it out. SentenceTransformerEmbedder::embed is item-shaped at the call site, but its implementation groups concurrent cache misses into batches of up to 64, and repeated texts are served from CocoIndex’s memo store.

Declare the table and rows

Now add the app’s main processing function:

main.rs
async fn app_main(ctx: Ctx, sourcedir: PathBuf) -> Result<()> {
    // The embedder reports the dimension of the `embedding` column.
    let target_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?;
    target_table.declare_vector_index(
        &ctx,
        "embedding",
        postgres::VectorIndexOptions {
            method: "hnsw",
            ..Default::default()
        },
    )?;

    let files = walk_dir(sourcedir)
        .recursive(true)
        .path_matcher(PatternFilePathMatcher::include(["**/*.md"])?)
        .live()
        .poll_interval(Duration::from_secs(1));
    ctx.mount_each_live(&"process_file", files, move |ctx, file: FileEntry| {
        let target_table = target_table.clone();
        async move { process_file(&ctx, &file, &target_table).await }
    })
    .await
}

mount_each_live gives every file its own processing component, keyed by its relative path. Each component owns the rows it declares, so when a file changes only its rows are reconciled, and when it disappears its rows are removed. The walker is a live source: in live mode (UpdateOptions { live: true, .. }, or cocoindex update -L once the app is registered with the cocoindex CLI as in the full example) it keeps watching the directory and mounts, re-runs or removes a file’s component as the directory changes. In a plain app.run it scans once and returns.

Build the environment and run

Finish src/main.rs by loading the resources and providing them to an environment:

main.rs
fn database_url() -> String {
    std::env::var("POSTGRES_URL")
        .unwrap_or_else(|_| "postgres://cocoindex:cocoindex@localhost/cocoindex".to_string())
}

#[tokio::main]
async fn main() -> Result<()> {
    dotenvy::dotenv().ok();
    let sourcedir = std::env::args()
        .nth(1)
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("markdown_files"));

    let database = postgres::Database::connect(&database_url()).await?;
    let embedder = SentenceTransformerEmbedder::load(EMBED_MODEL).await?;
    let app = Environment::builder()
        .db_path(".cocoindex_db")
        .provide_key(&DB, database)
        .provide_key(&EMBEDDER, embedder)
        .build()
        .await?
        .app("RustTextEmbeddingQuickstart")
        .await?;

    let stats = app.run(move |ctx| app_main(ctx, sourcedir)).await?;
    println!("{stats}");
    Ok(())
}

Run the pipeline:

cargo run

Run it again without changing an input: the target rows and embeddings are skipped. Then edit, add, or delete a Markdown file and rerun; CocoIndex updates only the affected component and reconciles its rows.

Next steps

PreviousInstallation
CocoIndex DocsEdit this pageReport issue