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

Note: Inline Stored Procedures cannot be invoked from inside an open transaction — they must be called with autocommit enabled. If you plan to call them, keeping AUTOCOMMIT = TRUE is a requirement rather than only a recommendation.

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)",
    params=[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 = ?",
    params=[batch_id]
).collect()

Why Not save_as_table() for HT?

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

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

Step 6: Inline Stored Procedures

Note: Inline Stored Procedures are in Public Preview and operate exclusively on Hybrid Tables. Statements that reference standard Snowflake tables, Iceberg tables, or other table types are not supported inside the body.

Create the Procedure

CREATE OR REPLACE HYBRID TABLE order_audit (
    audit_id    NUMBER        NOT NULL,
    order_id    NUMBER        NOT NULL,
    old_status  VARCHAR(20)   NOT NULL,
    new_status  VARCHAR(20)   NOT NULL,
    changed_at  TIMESTAMP_NTZ NOT NULL,
    PRIMARY KEY (audit_id)
);

CREATE OR REPLACE INLINE PROCEDURE update_order_status(
  p_order_id   NUMBER,
  p_new_status VARCHAR,
  p_audit_id   NUMBER,
  p_changed_at TIMESTAMP_NTZ
)
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
  v_old_status VARCHAR;
BEGIN ATOMIC
  SELECT status INTO :v_old_status
    FROM orders
    WHERE order_id = :p_order_id;

  UPDATE orders
    SET status = :p_new_status
    WHERE order_id = :p_order_id;

  INSERT INTO order_audit (audit_id, order_id, old_status, new_status, changed_at)
    VALUES (:p_audit_id, :p_order_id, :v_old_status, :p_new_status, :p_changed_at);

  RETURN 'Updated order ' || :p_order_id
         || ' from ' || :v_old_status
         || ' to ' || :p_new_status;
END;
$$;

Important: Inline Stored Procedures are compiled at CALL time, not at CREATE time. A CREATE statement can succeed and the first CALL still fail on an unsupported construct. Call each new procedure once in a test environment before you deploy it.

Note: This section covers the constraints most likely to affect application code. It is not the complete list — see Inline Stored Procedures for Hybrid Tables for the full set, including restrictions on table functions, OUT arguments, CONTINUE handlers, session variables, and referencing more than one database.

Call the Procedure with Bound Variables

Note: The Python example below was executed against a Hybrid Table. The JDBC, Node.js, and Snowpark examples translate the same bind pattern into each driver's own API — run them against your own account before relying on them in production.

Python

cursor = conn.cursor()
cursor.execute(
    "CALL update_order_status(?, ?, ?, ?)",
    (1001, 'SHIPPED', 9001, '2026-09-23 11:00:00')
)
print(cursor.fetchone()[0])     # Updated order 1001 from PENDING to SHIPPED

JDBC

String sql = "CALL update_order_status(?, ?, ?, ?)";

try (PreparedStatement stmt = conn.prepareStatement(sql)) {
    stmt.setLong(1, 1001L);
    stmt.setString(2, "SHIPPED");
    stmt.setLong(3, 9001L);
    stmt.setTimestamp(4, Timestamp.valueOf("2026-09-23 11:00:00"));

    try (ResultSet rs = stmt.executeQuery()) {
        if (rs.next()) {
            System.out.println(rs.getString(1));
        }
    }
}

Note: Timestamp.valueOf produces a wall-clock value, which matches the TIMESTAMP_NTZ column used here. If you build the value from an Instant instead, convert it to the intended local time first so the stored value is not shifted.

Node.js

connection.execute({
  sqlText: 'CALL update_order_status(?, ?, ?, ?)',
  binds: [1001, 'SHIPPED', 9001, '2026-09-23 11:00:00'],
  complete: (err, stmt, rows) => {
    if (err) return console.error('Call failed:', err.message);
    console.log(rows[0].UPDATE_ORDER_STATUS);
  }
});

Snowpark

session.sql(
    "CALL update_order_status(?, ?, ?, ?)",
    params=[1001, 'SHIPPED', 9001, '2026-09-23 11:00:00']
).collect()

Important: Inline Stored Procedures must be called with autocommit enabled and cannot run inside an open transaction. AUTOCOMMIT = TRUE is already the recommended setting for Hybrid Table workloads (see Step 1), so most applications need no change. If your framework or connection pool disables autocommit, commit or roll back the open transaction before calling.

Returning a Result Set

CREATE OR REPLACE INLINE PROCEDURE get_order(p_order_id NUMBER)
RETURNS TABLE(
  order_id    NUMBER,
  customer_id NUMBER,
  status      VARCHAR(20),
  region      VARCHAR(10),
  amount      NUMBER(12,2)
)
LANGUAGE SQL
AS
$$
BEGIN ATOMIC
  LET res RESULTSET := (
    SELECT order_id, customer_id, status, region, amount
      FROM orders
      WHERE order_id = :p_order_id
  );
  RETURN TABLE(res);
END;
$$;

Error Handling and Atomicity

import logging

try:
    cursor.execute(
        "CALL update_order_status(?, ?, ?, ?)",
        (1001, 'RETURNED', 9001, '2026-09-23 12:00:00')
    )
except snowflake.connector.errors.ProgrammingError as e:
    # The server rejected the call, so neither the UPDATE nor the audit INSERT
    # took effect. Re-raise or route to your retry logic - do not swallow it.
    logging.warning("update_order_status failed: %s", e)
    raise

Note: A client-side timeout is not the same as a server-side failure. If the connection drops after the server committed, your application sees an error for a call that actually succeeded, so catch connection and timeout errors separately and re-check state before retrying rather than assuming a rollback.

Important: A handler that returns makes the CALL succeed. Because the call completed normally, statements that already ran in that call are committed rather than rolled back — you have not caught and rolled back, you have converted a failure into a commit of whatever finished first. A procedure that updates a row and then fails to write its audit record leaves the update in place and still returns a value. A returned error string is therefore not a rollback signal. Use a handler only where partial completion is acceptable; when you need all-or-nothing behavior, leave the error unhandled so it reaches the caller.

Note: Handlers only run for errors raised after compilation succeeds. Because the body is compiled at CALL time, statically detectable problems — an unknown column, a return-type mismatch, or an unsupported construct such as CURRENT_TIMESTAMP — are reported before any statement executes and cannot be caught by an EXCEPTION handler.

When Not to Use One

Step 7: 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."

"Convert this multi-statement transaction against my Hybrid Tables into an Inline Stored Procedure, and show me the parameterized CALL from [Python / Java / Node.js]: [paste code]. Flag anything in it that Inline Stored Procedures don't support."

Cleanup

DROP PROCEDURE IF EXISTS update_order_status(NUMBER, VARCHAR, NUMBER, TIMESTAMP_NTZ);
DROP PROCEDURE IF EXISTS get_order(NUMBER);
DROP TABLE IF EXISTS order_audit;
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