Connecting Applications to Hybrid Tables

Overview

Note: Driver-based connections are the only way to achieve the lowest possible latency with Hybrid Tables. Snowsight adds overhead that is not representative of production performance. Always benchmark using your application driver.

Why Connector Choice and Configuration Matter

What You Will Learn

Prerequisites

Minimum Driver Versions

Setup: Create the Demo Table

CREATE OR REPLACE HYBRID TABLE orders (
    order_id     NUMBER        NOT NULL,
    customer_id  NUMBER        NOT NULL,
    status       VARCHAR(20)   NOT NULL DEFAULT 'PENDING',
    region       VARCHAR(10)   NOT NULL,
    amount       NUMBER(12,2)  NOT NULL,
    created_at   TIMESTAMP_NTZ NOT NULL,
    PRIMARY KEY (order_id),
    INDEX idx_orders_customer (customer_id),
    INDEX idx_orders_status_region (status, region)
);

Step 1: Connection Best Practices

Use Private Key Authentication

-- Create a dedicated service user (run as ACCOUNTADMIN)
CREATE USER ht_app_user
  RSA_PUBLIC_KEY = '<paste_public_key_here>'
  DEFAULT_ROLE = <your_role>
  DEFAULT_WAREHOUSE = <your_warehouse>
  MUST_CHANGE_PASSWORD = FALSE;

openssl genrsa -out rsa_key.pem 2048
openssl pkcs8 -topk8 -inform PEM -in rsa_key.pem -out rsa_key.p8 -nocrypt
openssl rsa -in rsa_key.pem -pubout -out rsa_key_pub.pem

Use Connection Pooling with Long-Lived Connections

Keep AUTOCOMMIT Enabled

Colocate Application and Snowflake

Step 2: JDBC (Java / Spring Boot)

Minimum version: 3.13.31

Connection with Private Key

Properties props = new Properties();
props.put("user", "ht_app_user");
props.put("private_key_file", "/secrets/rsa_key.p8");
props.put("db", "my_database");
props.put("schema", "my_schema");
props.put("warehouse", "my_warehouse");
props.put("role", "my_role");

Connection conn = DriverManager.getConnection(
    "jdbc:snowflake://<account>.snowflakecomputing.com",
    props
);

Batch INSERT with Bound Variables

String sql = "INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) " +
             "VALUES (?, ?, ?, ?, ?, ?)";

try (PreparedStatement stmt = conn.prepareStatement(sql)) {
    for (Order order : orders) {
        stmt.setLong(1, order.getOrderId());
        stmt.setLong(2, order.getCustomerId());
        stmt.setString(3, order.getStatus());
        stmt.setString(4, order.getRegion());
        stmt.setBigDecimal(5, order.getAmount());
        stmt.setTimestamp(6, Timestamp.from(order.getCreatedAt()));
        stmt.addBatch();
    }
    int[] results = stmt.executeBatch();
}

HikariCP Connection Pool

HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:snowflake://<account>.snowflakecomputing.com/" +
    "?warehouse=my_wh&db=my_db&schema=my_schema");
config.setUsername("ht_app_user");
config.addDataSourceProperty("private_key_file", "/secrets/rsa_key.p8");
config.setMaximumPoolSize(20);
config.setMinimumIdle(5);
config.setIdleTimeout(300_000);          // 5 min
config.setConnectionTimeout(30_000);     // 30s — fail fast
config.setKeepaliveTime(60_000);         // 1 min keepalive
config.setMaxLifetime(1_800_000);        // 30 min max lifetime

HikariDataSource dataSource = new HikariDataSource(config);

System.setProperty("net.snowflake.jdbc.ttl", "1800");

Important: The default net.snowflake.jdbc.ttl is -1 (infinite). Without setting this, idle connections may go stale and throw errors when reused. Always align this value with your pool's maxLifetime.

Spring Boot + Kafka: Real-Time Ingest Pattern

Kafka Topic → Spring Boot Consumer (batched, concurrent) → Hybrid Table

spring:
  kafka:
    consumer:
      group-id: ht-order-writer
      max-poll-records: 500          # rows per micro-batch
      bootstrap-servers: localhost:9092

kafka-demo:
  topic:
    name: orders
    partitions: 20                   # concurrency = partitions
  snowflake:
    url: jdbc:snowflake://<account>.snowflakecomputing.com
    user: ht_app_user
    database: my_database
    schema: my_schema
    warehouse: my_warehouse
    private-key-file: /secrets/rsa_key.p8

@KafkaListener(
    id = "order-writer",
    topics = "${kafka-demo.topic.name}",
    groupId = "ht-order-writer",
    concurrency = "${kafka-demo.topic.partitions}",
    batch = "true"
)
public void writeOrders(
    @Payload List<Order> orders,
    @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions
) throws SQLException {
    String sql = "INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) " +
                 "VALUES (?, ?, ?, ?, ?, ?)";

    try (Connection conn = dataSource.getConnection();
         PreparedStatement stmt = conn.prepareStatement(sql)) {
        for (Order o : orders) {
            stmt.setLong(1, o.getOrderId());
            stmt.setLong(2, o.getCustomerId());
            stmt.setString(3, o.getStatus());
            stmt.setString(4, o.getRegion());
            stmt.setBigDecimal(5, o.getAmount());
            stmt.setTimestamp(6, Timestamp.from(o.getCreatedAt()));
            stmt.addBatch();
        }
        stmt.executeBatch();
    }
}

For a detailed walkthrough of this pattern including performance results, see Fresh, Fast, and Value-Effective: Using Snowflake Hybrid Tables to Simplify Kafka Ingestion by Jon Osborn.

Step 3: Python Connector

Minimum version: 3.1.0

Connection with Private Key

import snowflake.connector

conn = snowflake.connector.connect(
    account='<account_identifier>',
    user='ht_app_user',
    authenticator='SNOWFLAKE_JWT',
    private_key_file='/secrets/rsa_key.p8',
    warehouse='my_warehouse',
    database='my_database',
    schema='my_schema',
    autocommit=True,       # default; keep True for HT OLTP
    login_timeout=60,
    network_timeout=30
)

Batch INSERT with executemany

rows = [
    (1001, 5042, 'PENDING', 'US-EAST', 149.99),
    (1002, 3891, 'PENDING', 'EU',      89.50),
    (1003, 7204, 'SHIPPED', 'APAC',    220.00),
]

cursor = conn.cursor()
cursor.executemany(
    "INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) "
    "VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP()::TIMESTAMP_NTZ)",
    rows
)

Critical: Always use ? placeholders (qmark binding). Never construct SQL with Python string formatting (f"...{value}...") — this bypasses plan caching and introduces SQL injection risk.

Point Lookup with Bound Variable

def get_order(conn, order_id: int):
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM orders WHERE order_id = ?",
        (order_id,)
    )
    return cursor.fetchone()

Connection Pooling with SQLAlchemy

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

engine = create_engine(
    URL(
        account='<account>',
        user='ht_app_user',
        database='my_database',
        schema='my_schema',
        warehouse='my_warehouse',
    ),
    connect_args={
        'authenticator': 'SNOWFLAKE_JWT',
        'private_key_file': '/secrets/rsa_key.p8',
        'autocommit': True,
    },
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,          # validate connections before use
    pool_recycle=1800             # recycle connections every 30 min
)

Recommended Batch Size

Step 4: Node.js

Minimum version: 1.9.0

Connection with Private Key

const snowflake = require('snowflake-sdk');
const fs = require('fs');

const connection = snowflake.createConnection({
  account: '<account_identifier>',
  username: 'ht_app_user',
  authenticator: 'SNOWFLAKE_JWT',
  privateKey: fs.readFileSync('/secrets/rsa_key.p8', 'utf8'),
  database: 'my_database',
  schema: 'my_schema',
  warehouse: 'my_warehouse'
});

connection.connect((err, conn) => {
  if (err) throw err;
  console.log('Connected to Snowflake');
});

Batch INSERT with Array Binding

connection.execute({
  sqlText: 'INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) ' +
           'VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP()::TIMESTAMP_NTZ)',
  binds: [
    [1001, 5042, 'PENDING', 'US-EAST', 149.99],
    [1002, 3891, 'PENDING', 'EU',       89.50],
    [1003, 7204, 'SHIPPED', 'APAC',    220.00]
  ],
  complete: (err, stmt, rows) => {
    if (err) console.error('Insert failed:', err.message);
  }
});

Point Lookup with Bound Variable

function getOrder(connection, orderId, callback) {
  connection.execute({
    sqlText: 'SELECT * FROM orders WHERE order_id = ?',
    binds: [orderId],
    complete: (err, stmt, rows) => {
      if (err) return callback(err);
      callback(null, rows[0]);
    }
  });
}

Connection Pool

const pool = snowflake.createPool(
  {
    account: '<account_identifier>',
    username: 'ht_app_user',
    authenticator: 'SNOWFLAKE_JWT',
    privateKey: fs.readFileSync('/secrets/rsa_key.p8', 'utf8'),
    database: 'my_database',
    schema: 'my_schema',
    warehouse: 'my_warehouse'
  },
  {
    max: 10,
    min: 2,
    evictionRunIntervalMillis: 60000,   // run evictor every 60s
    idleTimeoutMillis: 300000           // evict connections idle > 5 min
  }
);

// Use a pooled connection
pool.use(async (conn) => {
  return new Promise((resolve, reject) => {
    conn.execute({
      sqlText: 'INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) ' +
               'VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP()::TIMESTAMP_NTZ)',
      binds: [[1001, 5042, 'PENDING', 'US-EAST', 149.99]],
      complete: (err) => err ? reject(err) : resolve()
    });
  });
});

Important: The default evictionRunIntervalMillis is 0 (eviction disabled). Without setting this, stale idle connections cause errors when reused after network interruptions. Always set evictionRunIntervalMillis in production.

Step 5: Snowpark

Connection

from snowflake.snowpark import Session

session = Session.builder.configs({
    'account': '<account_identifier>',
    'user': 'ht_app_user',
    'authenticator': 'SNOWFLAKE_JWT',
    'private_key_file': '/secrets/rsa_key.p8',
    'database': 'my_database',
    'schema': 'my_schema',
    'warehouse': 'my_warehouse'
}).create()

Single-Row INSERT (Parameterized)

session.sql(
    "INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) "
    "VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP()::TIMESTAMP_NTZ)"
).bind([1001, 5042, 'PENDING', 'US-EAST', 149.99]).collect()

Bulk INSERT from a Staging Table

batch_id = 'batch_20260616_001'

session.sql(
    "INSERT INTO orders (order_id, customer_id, status, region, amount, created_at) "
    "SELECT order_id, customer_id, status, region, amount, created_at "
    "FROM staging_orders "
    "WHERE batch_id = ?"
).bind([batch_id]).collect()

Why Not save_as_table() for HT?

# Correct for HT
session.sql("INSERT INTO orders ... VALUES (?, ?, ...)").bind([...]).collect()

# Not recommended for HT
df.write.mode("append").save_as_table("orders")  # may not honor HT constraints

Step 6: Anti-Patterns

Anti-Pattern 1: String Literals Instead of Bound Variables

# Python — new query plan compiled on every execution
cursor.execute(f"SELECT * FROM orders WHERE order_id = {order_id}")

# Node.js — same problem
connection.execute({ sqlText: `SELECT * FROM orders WHERE order_id = ${orderId}` })

Anti-Pattern 2: Single-Row Inserts in a Loop

for order in orders:
    cursor.execute("INSERT INTO orders VALUES (?, ?, ...)", order)
    # Each call is a separate round-trip — N orders = N network round-trips

Anti-Pattern 3: No Connection Pool (Reconnecting per Request)

def handle_request(order):
    conn = snowflake.connector.connect(...)   # 100-500ms every time
    cursor.execute(...)
    conn.close()

Anti-Pattern 4: Oversized Connection Pool

Anti-Pattern 5: Benchmarking via Snowsight

Anti-Pattern 6: Schema-Qualification Inconsistency

Get Started Faster with Cortex Code

"Review my JDBC connection configuration for Hybrid Table latency. I'm seeing [X]ms end-to-end but the query itself runs in [Y]ms. Diagnose the overhead and suggest fixes."

"Rewrite this Python loop that inserts one row at a time into my Hybrid Table as a proper batch insert with bound variables and connection pooling: [paste code]."

"Set up a Kafka Sink connector for my Hybrid Table. My HT schema is: [paste DDL]. My Kafka topic payload is: [paste schema]. Generate the connector config with appropriate batch size and error handling."

Cleanup

DROP TABLE IF EXISTS orders;

Conclusion and Resources

Need help with your Hybrid Table architecture? Book a 30-minute session with our specialist team to discuss your use case, review your schema design, or troubleshoot performance: Schedule a session

Related Resources

FAQ