Valkey connector

Write documents to Valkey search indexes with vector similarity search — configured via VectorDef for dimension, distance metric, and index algorithm (HNSW or FLAT).

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The valkey connector provides utilities for writing documents to Valkey with vector search support, using the valkey-search module for similarity search indexes.

use cocoindex::connectors::valkey::{
    self, Distance, Document, FieldDef, FieldType, IndexSchema, VectorAlgorithm, VectorDef,
};

The module is compiled with the valkey Cargo feature of the cocoindex crate. It uses the redis crate.

Connection setup

Valkey::connect() and Valkey::connect_with_config() open a connection to Valkey.

pub struct ClientConfig {
    pub host: String,                 // default "localhost"
    pub port: u16,                    // default 6379
    pub password: Option<String>,
    pub use_tls: bool,
    pub client_name: Option<String>,  // default Some("cocoindex_vector_store")
}

impl Valkey {
    pub async fn connect(uri: &str) -> Result<Self>
    pub async fn connect_with_config(config: ClientConfig) -> Result<Self>
    pub async fn from_client(state_id: impl Into<String>, client: &redis::Client) -> Result<Self>
}

Valkey::connect("redis://localhost:6379") takes a URI. Valkey::connect_with_config(ClientConfig { .. }) takes the fields below. For anything ClientConfig does not cover (client certificates, a username, a database number), build a redis::Client yourself and pass it to from_client.

ClientConfig fields:

• host — Valkey server host (default: "localhost").
• port — Valkey server port (default: 6379).
• password — Optional authentication password. The Debug output of ClientConfig shows it as redacted.
• use_tls — Whether to use TLS for the connection (default: false).
• client_name — Client name for the connection, set with CLIENT SETNAME and visible in CLIENT LIST and monitoring dashboards (default: Some("cocoindex_vector_store")). Pass None to set no name.

Returns: A Valkey handle. It is cheap to clone: the multiplexed connection is shared, and every command or pipeline runs on a clone of it.

The handle’s state_id() is the server’s address and database number as a URI, for example redis://localhost:6379. It never holds the username or password, so a credential does not leak into memo dependencies or the target-state store, and rotating a credential does not invalidate them. With from_client(state_id, client), you choose the state_id: name the server, not a secret.

Example:

let vk = valkey::Valkey::connect(&uri).await?;

// With a password, TLS and a client name
let from_config = valkey::Valkey::connect_with_config(valkey::ClientConfig {
    host: host.clone(),
    port,
    password: info.redis.password.clone(),
    client_name: Some(client_name.clone()),
    ..valkey::ClientConfig::default()
})
.await?;

As target

The valkey connector provides target state APIs for writing documents to search indexes. CocoIndex tracks what documents should exist and automatically handles upserts and deletions.

Declaring target states

Setting up a connection

Declare a context key for a valkey::Valkey to identify your Valkey connection, then provide the connection on the environment builder:

use cocoindex::connectors::valkey;

cocoindex::context_key!(static VK: valkey::Valkey, key = "valkey_target_conn");

let vk = valkey::Valkey::connect(&uri).await?;
// Provide it on the environment builder, or in a `#[cocoindex::lifespan]`:
let builder = Environment::builder().provide_key(&VK, vk);

context_key! declares the key. Its stable name is the identifier, or the key = "…" string when given. There is no close(): drop the Valkey handle.

Indexes (parent state)

Declares a search index as a target state. Returns an IndexTarget for declaring documents.

pub fn declare_index_target(
    ctx: &Ctx,
    db: &ContextKey<Valkey>,
    index_name: impl Into<String>,
    schema: IndexSchema,
) -> Result<IndexTarget>

pub fn declare_index_target_with_options(
    ctx: &Ctx,
    db: &ContextKey<Valkey>,
    index_name: impl Into<String>,
    schema: IndexSchema,
    options: ManagedTargetOptions,
) -> Result<IndexTarget>

pub async fn mount_index_target(
    ctx: &Ctx,
    db: &ContextKey<Valkey>,
    index_name: impl Into<String>,
    schema: IndexSchema,
) -> Result<IndexTarget>

pub async fn mount_index_target_with_options(
    ctx: &Ctx,
    db: &ContextKey<Valkey>,
    index_name: impl Into<String>,
    schema: IndexSchema,
    options: ManagedTargetOptions,
) -> Result<IndexTarget>

declare_index_target* returns a handle whose documents resolve when the current component commits; mount_index_target*(...).await? returns a handle that accepts documents at once. managed_by is set with ManagedTargetOptions::user_managed() on the *_with_options forms. index_target and index_target_with_options return the raw TargetState, to compose with mount_target or declare_target_state_with_child.

let index = valkey::mount_index_target_with_options(
    &ctx,
    &VK,
    index_name,
    schema().await,
    ManagedTargetOptions::user_managed(),
)
.await?;

Parameters:

• db — A &ContextKey<Valkey> identifying the Valkey client to use.
• index_name — Name of the search index.
• schema — Schema definition specifying vector configuration (see Index Schema).
• managed_by — ManagedBy::System (default): CocoIndex creates and drops the index automatically. ManagedBy::User: assumes the index already exists and never drops it.

Documents (child states)

Once an IndexTarget is resolved, declare documents to be upserted:

impl IndexTarget {
    pub fn declare_document(&self, ctx: &Ctx, document: Document) -> Result<()>
}

pub struct Document {
    pub id: String,
    pub vector: Vec<f32>,
    pub payload: Option<BTreeMap<String, PayloadValue>>,
}

pub enum PayloadValue {
    Str(String),
    Int(i64),
    Float(f64),
}

let mut payload = BTreeMap::new();
payload.insert("text".to_string(), text.to_string());
index.declare_document(&ctx, Document::new(id, vector).with_payload(payload))?;

// A numeric payload value for a NUMERIC field:
index.declare_document(
    &ctx,
    Document::new("d3", vec![0.7, 0.8, 0.9]).with_payload([
        ("text", PayloadValue::from("priced")),
        ("price", PayloadValue::from(42i64)),
    ]),
)?;

with_payload takes any iterator of (key, value) pairs whose values convert into PayloadValue (String, &str, i64 or f64).

Parameters:

• document — A valkey::Document containing:
  • id — Document ID (string)
  • vector — Vector data (Vec<f32>). It must have the index’s dimension and finite elements: the search module stores a hash whose vector it cannot read but never indexes it, so declare_document returns an error instead.
  • payload — Optional metadata map (values are text, i64 or f64). The key vector is reserved for the vector field (see Data storage), so declare_document returns an error for a payload with that key.

Index schema

Define vector configuration for an index using IndexSchema.

impl IndexSchema {
    pub async fn create(vectors: VectorDef<'_>, fields: Vec<FieldDef>) -> Result<Self>
}

IndexSchema::create(
    VectorDef {
        schema: &VectorSchema::f32(3),
        distance: Distance::Cosine,
        algorithm: VectorAlgorithm::Hnsw,
    },
    vec![FieldDef::new("text", FieldType::Text)],
)
.await?

Pass vec![] for no indexed payload fields. FieldDef::new(name, FieldType::Text | Tag | Numeric) and .sortable() describe an indexed payload field.

Parameters:

• vectors — A VectorDef specifying the vector field configuration.
• fields — List of FieldDef for indexed payload fields (enables search/filtering). A FieldDef named vector is an error: that name is reserved for the vector field.

VectorDef

Specifies vector configuration including dimension, distance metric, and algorithm:

pub struct VectorDef<'a> {
    pub schema: &'a (dyn VectorSchemaProvider + 'a),
    pub distance: Distance,          // Cosine (default), L2, Ip
    pub algorithm: VectorAlgorithm,  // Hnsw (default), Flat
}

impl<'a> VectorDef<'a> {
    pub fn new(schema: &'a (dyn VectorSchemaProvider + 'a)) -> Self
}

VectorDef::new(&embedder) uses the defaults. schema borrows the provider; IndexSchema::create resolves the dimension at once.

Parameters:

• schema — A VectorSchemaProvider that defines vector dimensions.
• distance — Distance metric for similarity search (default: Distance::Cosine).
• algorithm — Vector index algorithm (default: VectorAlgorithm::Hnsw).

VectorSchemaProvider

The schema field of VectorDef borrows a VectorSchemaProvider to specify the vector dimension and dtype: an explicit &VectorSchema::f32(n) or a reference to an embedder. For an embedder held in the context, pass ctx.get_key(&EMBEDDER)?. See Vector Schema for details.

Distance metrics

The distance parameter in VectorDef specifies the similarity metric:

• Distance::Cosine — Cosine similarity (default)
• Distance::L2 — Euclidean distance (L2)
• Distance::Ip — Inner product (dot product)

Index algorithms

The algorithm parameter specifies the vector indexing strategy:

• VectorAlgorithm::Hnsw — Hierarchical Navigable Small World graph (default, best for most use cases)
• VectorAlgorithm::Flat — Brute-force flat index (exact results, slower for large datasets)

Data storage

Documents are stored as Valkey HASH keys with a prefix pattern:

• Key format: {index_name}:{document_id}
• Vector field: stored as binary float32 blob in a vector hash field
• Payload fields: stored as individual hash fields (string values)

The search index is created with FT.CREATE ON HASH PREFIX 1 {index_name}: to automatically index all documents with the matching prefix.

A batch of document changes is sent in pipelines: one DEL for every deleted key, then the upserts in pipelines of up to 1,000 documents. Each document is written as its own MULTI DEL HSET EXEC transaction, so no stale payload field survives an update and a document is never half written. A batch of n documents costs 1 + ceil(n / 1000) round trips, not n.

Example

use std::collections::BTreeMap;

use cocoindex::connectors::valkey::{
    self, Distance, Document, FieldDef, FieldType, IndexSchema, VectorAlgorithm, VectorDef,
};
use cocoindex::prelude::*;

cocoindex::context_key!(static VK: valkey::Valkey, key = "valkey_target_conn");

async fn schema() -> IndexSchema {
    IndexSchema::create(
        VectorDef {
            schema: &VectorSchema::f32(3),
            distance: Distance::Cosine,
            algorithm: VectorAlgorithm::Hnsw,
        },
        vec![FieldDef::new("text", FieldType::Text)],
    )
    .await
    .unwrap()
}

fn doc(id: &str, vector: Vec<f32>, text: &str) -> Document {
    let mut payload = BTreeMap::new();
    payload.insert("text".to_string(), text.to_string());
    Document::new(id, vector).with_payload(payload)
}

// In the app:
let index = valkey::mount_index_target(&ctx, &VK, index_name, schema().await).await?;
index.declare_document(&ctx, doc("d1", vec![0.1, 0.2, 0.3], "hello"))?;
index.declare_document(&ctx, doc("d2", vec![0.4, 0.5, 0.6], "world"))?;

There is no example program for Valkey yet; rust/sdk/cocoindex/tests/valkey_target.rs is the end-to-end reference.

Vector search

The connector focuses on writing documents to Valkey and has no search helper. For vector search queries, run FT.SEARCH through the redis crate directly.