Snowflake connector

Write CocoIndex target rows into Snowflake tables with managed DDL, MERGE upserts, deletes, and automatic mapping from Rust row types.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The snowflake connector provides target state APIs for writing rows to Snowflake tables. CocoIndex tracks the rows that should exist and applies table creation, upserts, updates, and deletes incrementally.

use cocoindex::connectors::snowflake::{
    self, SnowflakeConfig, SnowflakeConnection, SnowflakeTableOptions, TableSchema,
};

The module is compiled with the snowflake Cargo feature of the cocoindex crate. Snowflake has no official Rust client, so the connector speaks the HTTP protocol of the official connectors directly: a password login that yields a session token, then statements with server-side ? bindings.

Connection setup

Declare a context key for a SnowflakeConnection to identify the Snowflake connection, then provide the connection on the environment builder:

use cocoindex::connectors::snowflake::{SnowflakeConfig, SnowflakeConnection};

cocoindex::context_key!(static SNOWFLAKE: SnowflakeConnection, key = "snowflake_demo");

let mut config = SnowflakeConfig::new(
    required("SNOWFLAKE_ACCOUNT")?,
    required("SNOWFLAKE_USER")?,
    required("SNOWFLAKE_PASSWORD")?,
)
.warehouse(env_or("SNOWFLAKE_WAREHOUSE", "COCOINDEX_DEMO_WH"));
if let Some(role) = std::env::var("SNOWFLAKE_ROLE")
    .ok()
    .filter(|v| !v.is_empty())
{
    config = config.role(role);
}
let conn = SnowflakeConnection::new(config);
let app = cocoindex::Environment::builder()
    .db_path(".cocoindex_db")
    .provide_key(&SNOWFLAKE, conn)
    .build()
    .await?
    .app("SnowflakeTarget")
    .await?;

SnowflakeConnection::new(config) makes no request until the first statement. Each batch of statements logs in, runs and closes its own session. The connection is cheap to clone; conn.query(sql).await? runs a statement in a fresh session and returns its rows as JSON objects keyed by column name.

SnowflakeConfig

pub struct SnowflakeConfig {
    pub account: String,
    pub user: String,
    pub password: String,
    pub warehouse: Option<String>,
    pub role: Option<String>,
}

pub fn SnowflakeConfig::new(
    account: impl Into<String>,
    user: impl Into<String>,
    password: impl Into<String>,
) -> Self
pub fn SnowflakeConfig::warehouse(self, warehouse: impl Into<String>) -> Self
pub fn SnowflakeConfig::role(self, role: impl Into<String>) -> Self

Parameters:

• account - Snowflake account identifier, such as ORGNAME-ACCOUNTNAME, without the .snowflakecomputing.com suffix.
• user - Snowflake username.
• password - Password for the user.
• warehouse - Optional warehouse to use for DDL and DML.
• role - Optional role for the session.

As target

The snowflake connector provides target state APIs for writing rows to tables.

Tables

Declares a table as a target state. Returns a TableTarget for declaring rows.

pub fn declare_table_target(
    ctx: &Ctx,
    conn: &ContextKey<SnowflakeConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
) -> Result<TableTarget>

pub fn declare_table_target_with_options(
    ctx: &Ctx,
    conn: &ContextKey<SnowflakeConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    options: SnowflakeTableOptions,
) -> Result<TableTarget>

pub async fn mount_table_target(
    ctx: &Ctx,
    conn: &ContextKey<SnowflakeConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
) -> Result<TableTarget>

pub async fn mount_table_target_with_options(
    ctx: &Ctx,
    conn: &ContextKey<SnowflakeConnection>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
    options: SnowflakeTableOptions,
) -> Result<TableTarget>

pub struct SnowflakeTableOptions {
    /// Database holding the table (created if missing). `None` uses the session's current database.
    pub database: Option<String>,
    /// Schema holding the table (created if missing). `None` uses the session's current schema.
    pub schema: Option<String>,
    pub managed_by: ManagedBy,
}

declare_table_target* declares the table in the current component; its rows resolve when the component commits. mount_table_target* mounts the table first, so you can declare rows at once. table_target and table_target_with_options build the table target state without declaring it.

Parameters:

• conn - A &ContextKey<SnowflakeConnection> identifying the connection to use.
• table_name - Name of the table.
• table_schema - Schema definition including columns and primary key.
• database - Optional Snowflake database name.
• schema - Optional Snowflake schema name.
• managed_by - Whether CocoIndex manages the table lifecycle (ManagedBy::System, the default) or assumes it exists (ManagedBy::User).

When managed_by is ManagedBy::System, CocoIndex creates the database, schema, and table if needed. Table changes use Snowflake DDL, and row changes use MERGE for upserts.

A batch of row changes runs in one session: rows are MERGEd and deleted by primary key in statements of up to 4,000 bindings each, so a batch takes a few statements, not one per row.

Rows

Once a TableTarget is resolved, declare rows to be upserted:

pub fn TableTarget::declare_row<R: serde::Serialize>(&self, ctx: &Ctx, row: &R) -> Result<()>

The row serializes to an object whose keys are column names; keys outside the schema are ignored.

Parameters:

• row - Any Serialize value whose entries are the table’s columns. Must include all primary key columns.

Table schema: from a Rust struct

Define the table structure using a Rust struct that derives SchemaFields:

use cocoindex::prelude::*;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;

#[derive(Clone, Serialize, Deserialize, SchemaFields)]
struct SnowflakeOrder {
    order_id: String,
    customer: String,
    quantity: i64,
    unit_price: f64,
    #[coco(json)]
    attributes: JsonValue,
}

let schema = TableSchema::from_row::<SnowflakeOrder>(["order_id"])?;

TableSchema::from_row::<Row>(primary_key) takes the primary-key columns as an iterator of names. A serde_json::Value field needs #[coco(json)]; a Vec<T>, map or nested struct maps to VARIANT on its own.

Rust types are automatically mapped to Snowflake column types:

VARIANT and ARRAY values are JSON-serialized and written with PARSE_JSON. An Option<T> field is a nullable column; every other field is NOT NULL.

Custom column types

Use #[coco(type = "…")] to specify a custom Snowflake type:

#[derive(Serialize, SchemaFields)]
struct ProductRow {
    #[coco(type = "NUMBER(38, 0)")]
    id: i64,
    #[coco(type = "ARRAY")]
    embedding: Vec<f64>,
}

#[coco(type = "…")] sets the column type verbatim. A value is written as it serializes with serde; for another form, put #[serde(serialize_with = "…")] on the field. Explicit columns are also possible with TableSchema::new(columns, primary_key)? and ColumnDef::new("VARIANT").parse_json(); a ColumnDef is nullable unless .not_null() is called.

You can also replace the type of one column on the derived schema, for this schema only. with_column_type keeps the column’s derived nullability and takes the parse_json flag of the ColumnDef you pass:

use cocoindex::connectors::snowflake::{ColumnDef, TableSchema};

let schema = TableSchema::from_row::<ProductRow>(["id"])?
    .with_column_type("id", ColumnDef::new("NUMBER(38, 0)"))?;

Example

let schema = TableSchema::from_row::<SnowflakeOrder>(["order_id"])?;
let table = snowflake::mount_table_target_with_options(
    &ctx,
    &SNOWFLAKE,
    "product_index",
    schema,
    SnowflakeTableOptions {
        database: Some("ANALYTICS".to_string()),
        schema: Some("PUBLIC".to_string()),
        ..Default::default()
    },
)
.await?;
let rows = use_mount_each!(orders, |order| process_order(ctx, order)).await?;
for row in &rows {
    table.declare_row(&ctx, row)?;
}

See examples/rust/snowflake_target for a runnable project.

Identifier handling

Database, schema, table, and column names must be simple Snowflake identifiers containing letters, numbers, and underscores, and must not start with a number. The connector quotes identifiers when generating SQL.