sqlconnect

Sqlconnect provides a uniform client interface for accessing multiple warehouses:

• bigquery (configuration)
• clickhouse (configuration)
• databricks (configuration)
• Microsoft Fabric (configuration); unlike other dialects, its deprecated FormatTableName method preserves identifier case because Fabric's default collation is case-sensitive
• mysql (configuration)
• postgres (configuration)
• redshift using data API driver (configuration)
• redshift using postgres driver (configuration)
• snowflake (configuration)
• trino (configuration)

Installation

go get github.com/rudderlabs/sqlconnect-go

API

All available DB methods can be found here

ClickHouse setup

The driver needs a working database for validation and working tables. The driver has no default name for it. rudder-sources owns the name and passes it in ValidationOptions.WorkingDatabase. Validation refuses an empty name with CH_CONFIG_INVALID (field workingDatabase). The rudder-sources ClickHouse client docs give the default name and the rudderSchema override.

The customer creates the working database before connection validation. The driver does not create this database. Run these statements as an administrator. Replace <working_database>, rudder_retl and analytics with your working database, sync user and customer database:

CREATE DATABASE IF NOT EXISTS <working_database>;
GRANT SELECT ON analytics.* TO rudder_retl;
GRANT SELECT, INSERT, CREATE TABLE, DROP TABLE ON <working_database>.* TO rudder_retl;

Sync-log pruning also requires GRANT ALTER DELETE ON <working_database>.sync_log TO rudder_retl after that table exists. The driver rejects rudderSchema and scratchDatabase account keys. Validation checks database existence, its engine, and the required privileges. The driver needs no grant on a system table and never reads system.processes or system.query_log. When a read's context ends, a read with readonly > 0 and cancel_http_readonly_queries_on_client_close = 1 stops at its next input read, or at max_execution_time; the driver's own read settings set both. A CPU-bound read, such as a large join, can run on for many seconds. A server that ends TLS itself sees the closed request only when it next writes output, so a read with little output runs until max_execution_time. A write is stopped by its query id. Validation fails at stage 2 when the user's profile pins replace_running_query or replace_running_query_max_wait_ms, because every stop sets both.

To revoke access and remove working tables, run:

REVOKE SELECT, INSERT, CREATE TABLE, DROP TABLE ON <working_database>.* FROM rudder_retl;
DROP DATABASE IF EXISTS <working_database> SYNC;

Revoke ALTER DELETE on <working_database>.sync_log first if you granted it. Drop the database only when no sync uses its tables.

Usage

Loading all necessary db drivers

import _ "github.com/rudderlabs/sqlconnect-go/sqlconnect/config"

Creating a new DB client

db, err := sqlconnect.NewDB("postgres", []byte(`{
    "host": "postgres.example.com",
    "port": 5432,
    "dbname": "dbname",
    "user": "user",
    "password": "password"

}`))

if err != nil {
    panic(err)
}

Creating a new DB client using legacy mappings for backwards compatibility

db, err := sqlconnect.NewDB("postgres", []byte(`{
    "host": "postgres.example.com",
    "port": 5432,
    "dbname": "dbname",
    "user": "user",
    "password": "password",
    "legacyMappings": useLegacyMappings

}`))

if err != nil {
    panic(err)
}

Performing admin operations

{ // schema admin
    exists, err := db.SchemaExists(ctx, sqlconnect.SchemaRef{Name: "schema"})
    if err != nil {
        panic(err)
    }
    if !exists {
        err = db.CreateSchema(ctx, sqlconnect.SchemaRef{Name: "schema"})
        if err != nil {
            panic(err)
        }
    }
}

// table admin
{
    exists, err := db.TableExists(ctx, sqlconnect.NewRelationRef("table", sqlconnect.WithSchema("schema")))
    if err != nil {
        panic(err)
    }
    if !exists {
        err = db.CreateTestTable(ctx, sqlconnect.RelationRef{Schema: "schema", Name: "table"})
        if err != nil {
            panic(err)
        }
    }
}

Using the async query API

table := sqlconnect.NewRelationRef("table", sqlconnect.WithSchema("schema"))

ch, leave := sqlconnect.QueryJSONAsync(ctx, db, "SELECT * FROM " + db.QuoteTable(table))
defer leave()
for row := range ch {
    if row.Err != nil {
        panic(row.Err)
    }
    _ = row.Value
}

Utilities

SplitStatements: Splits a string of SQL statements separated with semicolons into individual statements

import sqlconnectutil "github.com/rudderlabs/sqlconnect-go/sqlconnect/util"

func main() {
    statements := sqlconnectutil.SplitStatements("SELECT * FROM table; SELECT * FROM table;")
}

IsObjectNotFound: Reports whether a query error says that the table, schema, database or catalog it addressed does not exist, for every driver except ClickHouse, whose errors carry their own codes (clickhousequery.Describe). Snowflake (2003) and SQL Server (208) use the same error for an object the role cannot see, so on those two a true result can also be a permission problem

import "github.com/rudderlabs/sqlconnect-go/sqlconnect/sqlerr"

func main() {
    _, err := db.QueryContext(ctx, "SELECT 1 FROM missing_schema.missing_table")
    if sqlerr.IsObjectNotFound(err) {
        // the table, its schema or the database is not there for this connection
    }
}