What is cocoindex.io? A Declarative Framework for Incremental AI Data Pipelines

cocoindex.io is an open-source framework that enables developers to build declarative, incremental data pipelines using a Rust-powered core and Python-first API, automatically syncing AI-augmented applications with heterogeneous data sources while tracking lineage down to the byte level.

cocoindex.io is the open-source engine behind the cocoindex-io/cocoindex repository, designed to transform static data sources into live, queryable context for large language model (LLM) applications. Unlike traditional batch ETL systems that reprocess entire datasets, cocoindex.io adopts a declarative model where you define target states—such as rows in a vector database—and the framework automatically propagates only the changes (the "Δ") from source to sink.

Core Architecture: Declarative Models and Incremental Execution

The architecture separates intent from implementation, allowing developers to specify what the target data should look like while the engine handles how to maintain it efficiently.

Declarative Data Definitions

At the heart of cocoindex.io lies the coco.App class, defined in python/cocoindex/_internal/app.py, which serves as the entry point for pipeline definitions. You declare transformations using the @coco.fn decorator, optionally enabling memoization with @coco.fn(memo=True) to cache results by input hash plus code hash. The coco.mount family of functions, including coco.mount_each, connects these transformations to data sources without imperative boilerplate.

Rust-Powered Incremental Engine

The heavy lifting occurs in the Rust core located at rust/core/src/engine/, responsible for parallel chunking, zero-copy transforms, and robust failure isolation. This engine tracks state changes precisely, ensuring only modified data triggers downstream processing. The Python bridge uses PyO3 to expose async-friendly handles such as UpdateHandle and DropHandle, defined in python/cocoindex/_internal/app.py, allowing Python code to monitor and control the lifecycle of index updates.

Live Lineage and Provenance Tracking

Every record processed by cocoindex.io maintains detailed provenance metadata linking it back to the exact source byte. This lineage system enables sub-second freshness guarantees and cost-effective re-embedding strategies, as the engine can identify precisely which embeddings require updates when source documents change. The provenance tracking implementation supports full auditability, making it straightforward to trace any generated vector or knowledge graph entry back to its origin in files, databases, or streaming sources.

Pluggable Connectors for Enterprise Data

The framework ships with adapters for heterogeneous enterprise environments, implemented in the python/cocoindex/connectors/ directory. Built-in integrations include:

• PostgreSQL with vector index support via python/cocoindex/connectors/postgres/_target.py
• Vector databases including Qdrant and Turbopuffer
• Graph databases such as SurrealDB
• Streaming sources via Kafka
• Object storage including S3 and Google Drive
• Local filesystem access through the localfs connector

Each connector implements consistent mounting semantics, allowing targets to be declared with methods like postgres.mount_table_target() before invoking table.declare_vector_index() for embedding storage.

Building Your First Pipeline with cocoindex.io

Below is a minimal pipeline that indexes a directory into PostgreSQL, demonstrating the memoization, mounting, and declaration patterns found in the project README (lines 81-98):

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter

# 1️⃣ Declare a memoized processing function

@coco.fn(memo=True)                     # cached by input hash + code hash

async def index_file(file, table):
    # Split the file into semantic chunks

    for chunk in RecursiveSplitter().split(await file.read_text()):
        # Declare each chunk as a row in the target table

        table.declare_row(text=chunk.text, embedding=embed(chunk.text))

# 2️⃣ Top-level async entry point

@coco.fn
async def main(src):
    # Mount a Postgres table target (creates it if needed)

    table = await postgres.mount_table_target(PG, table_name="docs")
    table.declare_vector_index(column="embedding")
    # Mount the indexing function over every file in the source directory

    await coco.mount_each(index_file, localfs.walk_dir(src).items(), table)

# 3️⃣ Run the app – the engine will back-fill once and then only re-process changed files

coco.App(coco.AppConfig(name="docs"), main, src="./docs").update_blocking()

The RecursiveSplitter class imported from python/cocoindex/ops/text.py handles semantic text chunking, while update_blocking() initiates the incremental sync. The CLI entry point in python/cocoindex/cli.py provides additional command-line control for running such applications.

Summary

• cocoindex.io provides a declarative alternative to imperative ETL, letting developers define target states rather than transformation scripts.
• The Rust core (rust/core/src/engine/) delivers production-grade performance with parallel processing and failure isolation, bridged to Python via PyO3.
• Lineage tracking stores provenance down to the source byte, enabling efficient incremental updates and full auditability for RAG and agentic applications.
• The connector ecosystem supports Postgres, Qdrant, Kafka, S3, Google Drive, and local files through a unified mounting API.
• Memoization via @coco.fn(memo=True) automatically caches expensive operations based on content and code hashing.

Frequently Asked Questions

How does cocoindex.io differ from traditional ETL frameworks?

Traditional ETL tools typically require developers to specify step-by-step data transformations and often reprocess entire datasets on each run. cocoindex.io inverts this model by using a declarative approach where you declare the desired target state—such as rows in a vector database—and the framework automatically computes and applies only the necessary changes (the delta) to maintain synchronization.

What programming languages does cocoindex.io support?

While the incremental engine and performance-critical components are implemented in Rust, cocoindex.io exposes a Python-first API that supports async/await patterns. The Rust core is bridged to Python using PyO3, exposing handles like UpdateHandle and DropHandle in python/cocoindex/_internal/app.py, allowing Python developers to build high-performance pipelines without managing Rust code directly.

How does the incremental processing work under the hood?

The Rust engine tracks state changes across all mounted sources and maintains a dependency graph of transformations. When a source changes, the engine identifies exactly which downstream records are affected through byte-level lineage tracking, then reprocesses only those specific chunks. This architecture supports zero-copy transforms and parallel chunking, minimizing computational overhead for large-scale AI workloads.

Which databases and storage systems are compatible with cocoindex.io?

The framework includes built-in connectors for PostgreSQL (with vector index support via python/cocoindex/connectors/postgres/_target.py), Qdrant, Turbopuffer, SurrealDB, Kafka, S3, Google Drive, and local filesystems. Each connector implements the mounting interface, allowing targets to be declared with methods like declare_vector_index() for embedding storage in vector databases.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}