BigQuery connector

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

Version

v 1.0.14

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

use cocoindex::connectors::bigquery::{self, BigQueryConfig, BigQueryConnection, TableSchema};

Connection setup

Declare a context key for a BigQueryConnection with cocoindex::context_key!, then provide the connection when you build the environment:

use cocoindex::connectors::bigquery::{BigQueryConfig, BigQueryConnection};

cocoindex::context_key!(static BIGQUERY: BigQueryConnection, key = "bigquery_demo");

let optional = |name: &str| std::env::var(name).ok().filter(|v| !v.is_empty());
let conn = BigQueryConnection::connect(BigQueryConfig {
    project: optional("BIGQUERY_PROJECT"),
    credentials_path: optional("GOOGLE_APPLICATION_CREDENTIALS").map(Into::into),
    location: optional("BIGQUERY_LOCATION"),
})?;
let app = cocoindex::Environment::builder()
    .db_path(".cocoindex_db")
    .provide_key(&BIGQUERY, conn)
    .build()
    .await?
    .app("BigQueryTarget")
    .await?;

BigQueryConnection::connect(config)? resolves the credentials and makes no request until the first query. The connection is Clone; conn.query(sql).await? runs a query and returns its rows as JSON objects.

If credentials_path is omitted, Application Default Credentials are used.

BigQueryConfig

pub struct BigQueryConfig {
    pub project: Option<String>,
    pub credentials_path: Option<PathBuf>,
    pub location: Option<String>,
}

BigQueryConfig implements Default. When project is unset, the GOOGLE_CLOUD_PROJECT / GCLOUD_PROJECT environment variable, then the credentials file’s project, is used.

Parameters:

• project - Optional Google Cloud project that runs the query jobs.
• credentials_path - Optional path to a service account JSON key file. If omitted, Application Default Credentials are used.
• location - Optional BigQuery job location, for example US or EU.

As target

The bigquery 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 async fn mount_table_target(
    ctx: &Ctx,
    conn: &ContextKey<BigQueryConnection>,
    dataset: impl Into<String>,
    table_name: impl Into<String>,
    table_schema: TableSchema,
) -> Result<TableTarget>

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

pub struct BigQueryTableOptions {
    /// Project that owns the table. `None` means the connection's project.
    pub project: Option<String>,
    pub managed_by: ManagedBy,
}

declare_table_target and declare_table_target_with_options take the same arguments and declare the table in the current component.

Parameters:

• conn - The context key of the BigQueryConnection to use.
• dataset - BigQuery dataset name.
• table_name - Name of the table.
• table_schema - Schema definition including columns and primary key.
• options.project - Optional project that owns the target table. If None, the connection’s project is used. None and naming the connection’s project explicitly are two different keys for the same table (see the caution below).
• options.managed_by - Whether CocoIndex manages the table lifecycle (ManagedBy::System, the default) or assumes it exists (ManagedBy::User).

With ManagedBy::System, CocoIndex creates the dataset and table if needed. Table changes use BigQuery DDL, and row changes use BigQuery MERGE for upserts.

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.

Every primary-key column must be present in the row.

Table schema: from a Rust struct

Define the table structure with a struct that derives SchemaFields:

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

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

let schema = TableSchema::from_row::<BigQueryOrder>(["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 JSON on its own.

Rust field types are automatically mapped to BigQuery column types:

JSON 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 = "…")] on a field to specify a custom BigQuery type:

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

#[coco(type = "…")] sets the column type verbatim. Explicit columns are also possible: TableSchema::new([("id", ColumnDef::new("INT64").not_null()), ("payload", ColumnDef::new("JSON").parse_json())], ["id"])?. A ColumnDef is nullable unless .not_null() is called, and .parse_json() writes the value through PARSE_JSON.

let schema = TableSchema::from_row::<Row>(["id"])?
    .with_column_type("payload", ColumnDef::new("JSON").parse_json())?;

Example

let schema = TableSchema::from_row::<BigQueryOrder>(["order_id"])?;
let table = bigquery::mount_table_target(
    &ctx,
    &BIGQUERY,
    "analytics",
    "product_index",
    schema,
)
.await?;
let rows = use_mount_each!(orders, |order| process_order(ctx, order)).await?;
for row in &rows {
    table.declare_row(&ctx, row)?;
}

To name the project that owns the table, use mount_table_target_with_options with BigQueryTableOptions { project: Some("my-project".to_string()), ..Default::default() }.

See examples/rust/bigquery_target for a runnable project.

Identifier handling

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