Home  /  Serve results  /  Sink results

turbopuffer

View as Markdown

This guide shows how to send results from Materialize to turbopuffer. A Kafka sink writes the results to a Kafka topic. mz-tpuf-sink reads that topic and writes the documents to a turbopuffer namespace.

Use this pipeline to keep a turbopuffer namespace up to date, just using SQL. Materialize maintains the search document as an incrementally updated view over your operational data, and pushes precise deltas to turbopuffer as upstream data changes, so only the affected documents are rewritten.

Vectors come from a transform, a Python function that declares the columns it reads and the attributes it produces. The sink calls a transform only for the documents whose source columns actually changed, so embedding costs scale with what changed.

Before you begin

  • A turbopuffer API key, and the region that holds your namespace.

  • Python 3.12 or later on the host that runs the sink, and a way to install packages. The examples below use uv.

  • Materialize SQL credentials to read catalog metadata.

  • A Kafka or Redpanda cluster. Materialize and the destination system must both connect to this cluster.

  • A materialized view, source, or table to export. A sink cannot read from a plain view.

  • A cluster to run the sink. Name this cluster with IN CLUSTER. See CREATE CLUSTER. When a sink starts, it loads a full snapshot of the relation into memory. Size the cluster for the snapshot, not for the steady-state rate of change.

  • The Kafka ACLs that the sink needs.

You also need these privileges in Materialize:

  • CREATE privileges on the containing schema.
  • SELECT privileges on the item being written out to an external system.
    • NOTE: if the item is a materialized view, then the view owner must also have the necessary privileges to execute the view definition.
  • CREATE privileges on the containing cluster if the sink is created in an existing cluster.
  • CREATECLUSTER privileges on the system if the sink is not created in an existing cluster.
  • USAGE privileges on all connections and secrets used in the sink definition.
  • USAGE privileges on the schemas that all connections and secrets in the statement are contained in.

Step 1. Set up the sink in Materialize

The examples in this guide build a search document for an article catalog tracking its content and page views.

Create the connections

CREATE SECRET kafka_password AS '<BROKER_PASSWORD>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER '<BROKER_HOST>:9092',
    SASL MECHANISMS = 'SCRAM-SHA-256',
    SASL USERNAME = '<BROKER_USERNAME>',
    SASL PASSWORD = SECRET kafka_password
);

CREATE SECRET csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
    URL '<CSR_URL>',
    USERNAME = '<CSR_USERNAME>',
    PASSWORD = SECRET csr_password
);

The embedding transform compares structured records to find changes, so this pipeline needs Avro with a schema registry. For other authentication methods, see CREATE CONNECTION.

Create the search document

Create a materialized view that builds the document you want to search.

CREATE MATERIALIZED VIEW articles AS
    SELECT a.id, a.title, a.body, count(p.article_id) AS views
    FROM article_content a
    LEFT JOIN page_views p ON a.id = p.article_id
    GROUP BY 1, 2, 3;

Create the sink

CREATE SINK articles_sink_v1
  IN CLUSTER sinks_cluster
  FROM articles
  INTO KAFKA CONNECTION kafka_connection (
    TOPIC 'articles_v1',
    TOPIC PARTITION COUNT 6
  )
  KEY (id) NOT ENFORCED
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection
  ENVELOPE DEBEZIUM;

ENVELOPE DEBEZIUM wraps each change in a {"before": ..., "after": ...} value. The transform compares these two fields to find the columns that changed. The transform also converts each delete into a tombstone. The connector applies this tombstone as a document delete. For the full list of options, see CREATE SINK ... INTO KAFKA.

KEY (id) must name exactly one column. That column’s value becomes the turbopuffer document id, unchanged. Integer, string, and uuid columns work, and a string id holds at most 64 bytes. To key on something wider, add a hashed key column to the view and sink that column instead.

Step 2. Install the sink library

Install mz-tpuf-sink and the client for your embedding provider. This example uses OpenAI:

uv add "mz-tpuf-sink @ git+https://github.com/MaterializeInc/mz-turbopuffer-sink"
uv add openai

Step 3. Configure and run the sink

The sink writes documents to a turbopuffer namespace. Do not create the namespace first. turbopuffer creates it on the first write, and the sink declares the attribute schema on every write. Column types come from the Avro schema in the schema registry, so numbers stay numbers and timestamps stay timestamps, filterable and sortable. Adding a column to the view needs no change to the program below.

WARNING!

A new sink’s snapshot inserts only the rows that exist when it starts. The snapshot does not remove old documents from the destination.

Do not point a new sink at a destination that already holds documents. Those documents stay in the destination. No later write removes them.

Write a program that declares the transforms and runs the sink:

import os

from mz_tpuf_sink import FunctionTransform, SinkConfig, run_sink
from openai import OpenAI

client = OpenAI()


def embed_column(column, rows):
    """Embed one column for a batch of documents, in one API call."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=[row[column] for row in rows],
    )
    return [{f"{column}_embedding": item.embedding} for item in response.data]


def embedding_transform(column):
    return FunctionTransform(
        name=f"{column}_embedding",
        sources=(column,),
        schema={f"{column}_embedding": {"type": "[1536]f32", "ann": True}},
        distance_metric="cosine_distance",
        batch_size=256,
        compute=lambda rows: embed_column(column, rows),
    )


run_sink(
    SinkConfig(
        kafka_bootstrap_servers="<BROKER_HOST>:9092",
        kafka_topic="articles_v1",
        schema_registry_url="<CSR_URL>",
        schema_registry_auth=f"<CSR_USERNAME>:{os.environ['CSR_PASSWORD']}",
        materialize_dsn=os.environ["MATERIALIZE_DSN"],
        materialize_sink="materialize.public.articles_sink_v1",
        turbopuffer_api_key=os.environ["TURBOPUFFER_API_KEY"],
        turbopuffer_region="aws-us-east-1",
        namespace="articles_v1",
    ),
    transforms=[embedding_transform("title"), embedding_transform("body")],
)

The OpenAI() client reads its key from the OPENAI_API_KEY environment variable.

Each transform keeps one vector in sync with one text column:

  • sources names the columns the transform reads. An update that leaves every source column unchanged never reaches compute. Editing an article’s title re-embeds the title. Changing its views a thousand times embeds nothing.
  • schema declares the turbopuffer attributes the transform produces. A vector needs ann: True and a distance_metric, and the sink refuses to start without them. A namespace holds at most two vector attributes, so the two transforms above are at the limit.
  • batch_size bounds how many documents reach one compute call. The sink batches the calls, so one API request covers many documents.
  • distance_metric applies to the whole namespace. Two vector transforms cannot declare different metrics.

A transform is ordinary Python, so it can call any model, local or hosted, and it can produce anything, not just vectors. A slug, a sentiment score, or a translated title all work the same way.

SinkConfig names the two ends of the pipeline:

  • kafka_topic is the topic the Materialize sink writes to.
  • materialize_sink is the sink’s fully qualified name, as database.schema.sink. A bare name could match sinks in several schemas, so the sink rejects one.
  • namespace is the turbopuffer namespace to write.

Run the program:

uv run python sink.py

Run one process per topic, writing to one namespace. run_sink blocks until stopped. To shut down cleanly, pass a threading.Event as its second argument and set that event from a signal handler.

Embedding calls make a flush long-running, and a Kafka consumer that does not poll within max.poll.interval.ms is evicted, which drops buffered state and replays the work. Raise kafka_max_poll_interval_ms in SinkConfig if the sink logs a slow flush warning.

Step 4. Validate the pipeline

  1. Confirm that the documents arrived with their vectors:

    from turbopuffer import Turbopuffer

namespace = Turbopuffer(
    api_key="<TURBOPUFFER_API_KEY>",
    region="aws-us-east-1",
).namespace("articles_v1")

response = namespace.query(
    rank_by=("id", "asc"),
    top_k=1,
    include_attributes=["id", "title", "views", "title_embedding"],
    consistency={"level": "strong"},
)
for row in response.rows:
    print(row.id, row.title, row.views, row.title_embedding[:4])

    The output should resemble the following:

    1 Storage engines 42 [0.021, -0.118, 0.043, 0.009]

  2. Delete the row with id = 1 from article_content in Materialize:

    DELETE FROM article_content WHERE id = 1;

    Confirm that the document is gone:

    response = namespace.query(
    rank_by=("id", "asc"),
    filters=("id", "Eq", 1),
    top_k=1,
    consistency={"level": "strong"},
)
print(response.rows)

    The output is an empty list.

Back to top ↑