sqlx_support

SQLx Support for Kinetica via PostgreSQL Wire Protocol

This module provides helpers for using SQLx with Kinetica’s PostgreSQL wire protocol support.

​

Overview

Kinetica supports the PostgreSQL wire protocol on port 5432, which means you can use SQLx’s PostgreSQL driver directly. This module provides:

• KineticaPool - A type alias for PgPool with Kinetica-specific configuration
• KineticaPoolOptions - Builder for creating connection pools with sensible defaults
• connect, connect_with_url and connect_with_options - Convenience functions for establishing connections
• KineticaText, KineticaTimestamp, KineticaBytes, KineticaJson, KineticaVector and KineticaArray - for the Kinetica types the protocol does not carry as PostgreSQL does

​

Quick Start

use gpudb_api::sqlx_support::{KineticaPoolOptions, connect};
use sqlx::Row;

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    // Simple connection
    let pool = connect("localhost", "admin", "password").await?;

    // Or with full options
    let pool = KineticaPoolOptions::new()
        .host("localhost")
        .port(5432)
        .username("admin")
        .password("password")
        .max_connections(10)
        .connect()
        .await?;

    // Execute queries using standard SQLx
    let rows = sqlx::query("SELECT id, name FROM users")
        .fetch_all(&pool)
        .await?;

    for row in rows {
        let id: i64 = row.get("id");
        let name: String = row.get("name");
        println!("User: id={}, name={}", id, name);
    }

    Ok(())
}

​

Type Mappings

Kinetica’s PostgreSQL protocol carries its types as below, as measured against Kinetica 7.2. Where the PostgreSQL type does not carry the value whole, read and write it through the wrappers this module provides (see KineticaText and its siblings).

This crate enables SQLx’s chrono and json features, so NaiveDate and serde_json::Value read as they are. Nulls read as Option<_> of any of these, and bind as None.

What goes wrong without the wrappers:

• SQLx’s Vec<T> refuses Kinetica’s arrays, which are numbered from 0.
• A DATETIME read as a NaiveDateTime, or bound as one, loses its milliseconds.
• Kinetica refuses uuid::Uuid, NaiveTime, decimal and Vec<u8> (BYTEA) parameters.
• SQLx binds serde_json::Value and sqlx::types::Json as JSONB, which Kinetica refuses as invalid JSON.
• A plain INSERT refuses a vector as text, and everything else refuses one as raw floats.

​

Parameters

Kinetica reads every parameter after a statement’s first as text, whatever type SQLx sends it as. A second integer, float or date parameter therefore fails to convert in an INSERT. In a WHERE clause it matches nothing, so an UPDATE or DELETE reports success and changes no rows. A literal beside such a parameter can be stored as NULL. Text parameters work in every position, and Kinetica converts each to its column’s type, so bind every parameter through a wrapper, which sends text:

use gpudb_api::sqlx_support::{KineticaPool, KineticaText, KineticaTimestamp};

async fn record_login(pool: &KineticaPool, user_id: i64, active: bool) -> Result<u64, sqlx::Error> {
    let result = sqlx::query("UPDATE users SET active = $1, last_login = $2 WHERE id = $3")
        .bind(KineticaText(active))
        .bind(KineticaTimestamp(chrono::Utc::now()))
        .bind(KineticaText(user_id))
        .execute(pool)
        .await?;
    Ok(result.rows_affected())
}

​

JSON, Arrays and Vectors

Each is inserted, updated, filtered on and deleted by with ordinary SQL, binding the wrappers above. Measured against Kinetica 7.2:

• JSON. Bind documents with KineticaJson, which takes any Serialize value; read them as KineticaJson<T> for any DeserializeOwned T. JSON_VALUE(col, $1)</code> (whose path may be a parameter), <code>JSON\_QUERY</code>, <code>JSON\_EXISTS</code>, <code>JSON\_LENGTH</code>, <code>JSON\_TYPE</code>, <code>IS\_JSON</code>, <code>JSON\_OBJECT</code> and <code>JSON\_ARRAY</code> work, with nested paths in <code>GROUP BY</code> and <code>JOIN</code> too. A SQL <code>NULL</code> and the JSON <code>null</code> stay distinct. Declare JSON columns as <code>JSON</code>: <code>VARCHAR(json)</code> and <code>STRING(json)</code> are plain strings on the server.</li> <li><strong>Full-text search.</strong> A <code>JSON(text\_search)</code> column, on a server with <code>enable\_text\_search</code>, is searched by key with <code>text\_match(col, $1) (bool) and ranked with text_match_bm25 or text_match_bm25_global (f64), the query bound as KineticaText(“customer.tier:Enterprise”). Do not UPDATE such a column: on Kinetica 7.2.3.21 that crashes a worker rank. Delete the row and insert it again.
• Arrays. Bind with KineticaArray, and use KineticaArray<Option<T>> for arrays holding nulls. ARRAY_CONTAINS, ARRAY_CONTAINS_ANY, ARRAY_CONTAINS_ALL, ARRAY_LENGTH, col[i] (numbered from 1), ARRAY_APPEND, ARRAY_CONCAT, ARRAY_SLICE and ARRAY_TO_STRING work; arrays they compute arrive as JSON text, which KineticaArray reads too. A numeric ARRAY_CONTAINS(col, $1)</code> finds nothing whatever is bound: write <code>ARRAY\_CONTAINS\_ANY(col, $1) with a one-element KineticaArray. A fixed size, as in INTEGER[3], is not enforced, and an infinite float is refused.
• Vectors. A plain INSERT … VALUES ($n)</code> takes <a href="/content/api/rust/sqlx_support/struct.KineticaVector" title="struct gpudb_api::sqlx_support::KineticaVector"><code>KineticaVector</code></a>; <code>UPDATE ... SET v =$n, VECTOR(n, dims)</code> and function arguments take <code>KineticaText(vector.as\_slice())</code>. <code>L1\_DISTANCE</code>, <code>L2\_DISTANCE</code>, <code>COSINE\_DISTANCE</code>, <code>DOT\_PRODUCT</code> and <code>L2\_NORM</code> return <code>f32</code>, as does <code>v &lt;-&gt; CAST(1 AS VECTOR(n)); a bare v <-> $1 is refused.

use gpudb_api::sqlx_support::{
    KineticaArray, KineticaJson, KineticaPool, KineticaText, KineticaVector,
};
use serde::{Deserialize, Serialize};
use sqlx::Row;

#[derive(Serialize, Deserialize)]
struct Profile {
    name: String,
    score: f64,
}

async fn nearest_tagged(pool: &KineticaPool, tag: &str, probe: &[f32]) -> Result<Vec<(i64, Profile)>, sqlx::Error> {
    sqlx::query("INSERT INTO docs (id, profile, tags, embedding) VALUES ($1, $2, $3, $4)")
        .bind(KineticaText(1_i64))
        .bind(KineticaJson(Profile { name: "first".into(), score: 0.5 }))
        .bind(KineticaArray(vec![tag.to_string()]))
        .bind(KineticaVector(probe.to_vec()))
        .execute(pool)
        .await?;

    let rows = sqlx::query(
        "SELECT id, profile FROM docs WHERE ARRAY_CONTAINS(tags, $1)
         ORDER BY COSINE_DISTANCE(embedding, $2) LIMIT 5",
    )
    .bind(KineticaText(tag))
    .bind(KineticaText(probe))
    .fetch_all(pool)
    .await?;
    Ok(rows
        .iter()
        .map(|row| (row.get("id"), row.get::<KineticaJson<Profile>, _>("profile").0))
        .collect())
}

Read arrays through sqlx::query: in the text form sqlx::raw_sql returns, Kinetica writes null in capitals wherever it occurs in a string element.

​

Compile-Time Checked Queries

SQLx supports compile-time checked queries with the query! macro. However, this requires a live database connection during compilation. For Kinetica, you can:

1. Use the offline mode with sqlx prepare
2. Use runtime queries with sqlx::query()

use gpudb_api::sqlx_support::KineticaPool;

// Runtime query (always works)
async fn get_users(pool: &KineticaPool) -> Result<Vec<(i64, String)>, sqlx::Error> {
    let rows = sqlx::query_as::<_, (i64, String)>("SELECT id, name FROM users")
        .fetch_all(pool)
        .await?;
    Ok(rows)
}

// With a custom struct
#[derive(sqlx::FromRow)]
struct User {
    id: i64,
    name: String,
    email: Option<String>,
}

async fn get_users_typed(pool: &KineticaPool) -> Result<Vec<User>, sqlx::Error> {
    sqlx::query_as::<_, User>("SELECT id, name, email FROM users")
        .fetch_all(pool)
        .await
}

​

Transactions

Kinetica’s PostgreSQL protocol does not support transactions: pool.begin() fails (“got unexpected connection status after attempting to begin transaction”), and each statement applies when it runs.

​

Connection Pooling Best Practices

For optimal performance with Kinetica:

• Set max_connections based on your workload (default: 10)
• Use min_connections for warm connection pools
• Set appropriate timeouts for your network latency

The server serves one session per thread, up to its postgres_proxy.max_threads (64 by default), shared by every client. It ends a session when its client sends PostgreSQL’s Terminate message, which pool.close() sends. A connection closed without it, as when a process exits or a task panics with the connection open, keeps its session until the server’s postgres_proxy.idle_connection_timeout (300 seconds by default). Pools that outgrow the server’s sessions wait, then fail with sqlx::Error::PoolTimedOut. SQLx 0.8’s pool.close() can return while a connection on its way back to the pool is still open: before a program ends, close again once pool.size() shows it is back (see SQLX_API.md).

use gpudb_api::sqlx_support::KineticaPoolOptions;
use std::time::Duration;

let pool = KineticaPoolOptions::new()
    .host("kinetica-cluster.example.com")
    .username("admin")
    .password("secure_password")
    .max_connections(20)
    .min_connections(5)
    .acquire_timeout(Duration::from_secs(30))
    .idle_timeout(Some(Duration::from_secs(600)))
    .connect()
    .await?;

​

Limitations

Besides the types and parameters above:

• Transactions are not supported.
• Sessions are limited by the server, and outlive connections that are not closed (see above).
• PostgreSQL-specific features, such as LISTEN/NOTIFY, are not available.
• Kinetica-specific SQL extensions should be written as raw SQL strings.

For full fidelity on every Kinetica type, and for Kinetica-specific operations (bulk insert, type management, KiFS), use the native API of the Kinetica client. It carries each value with its column’s properties.