DB-API Extensions

The confluent-sql driver extends the standard DB-API v2 interface with features specifically designed for stream processing and event-driven applications with Confluent Cloud Flink SQL.

Understanding Snapshot vs Streaming Modes

By default, the driver operates in SNAPSHOT mode, producing behavior very similar to traditional SQL databases:

Queries execute and block until complete
Results are finite (complete as of query start time)
Standard DB-API methods (fetchone(), fetchall(), iteration) work as expected
Perfect for ad-hoc data exploration and analysis
No special streaming knowledge required for simple queries

However, Confluent Cloud Flink is fundamentally a streaming database. The driver provides optional streaming mode for applications that need to work with continuous, unbounded data:

Queries run indefinitely, producing results as they arrive
Non-blocking fetch methods (polling pattern with may_have_results)
Stateful queries return changelog events tracking row modifications
Automatic state compression available via changelog_compressor()

This design choice means:

✅ Familiar to traditional database users - Snapshot mode works like PostgreSQL/MySQL/SQLite
✅ Opt-in streaming - Enable it only when you need continuous data
✅ Pure DB-API compliance - Snapshot queries don't require any extensions

All extensions are backward compatible and opt-in—standard DB-API code works unchanged.

For comprehensive details on streaming queries, polling patterns, and changelog handling, see STREAMING.md.

Quick Navigation

Result Format Extensions - Dictionary rows, custom types
Streaming Query Support - Comprehensive streaming guide
Statement Lifecycle Management - DDL, naming, stopping, deletion
Tableflow Lifecycle - Enable, read, and disable Iceberg/Delta sinks
Introspection and Metadata - Properties for query state
Performance Monitoring - Fetch metrics
Type System Extensions - Flink type support
Custom Exceptions - Streaming-specific errors
Extended Execute Parameters - Additional cursor.execute() options

Result Format Extensions

These extensions control how results are returned by cursors, and work with both snapshot queries (finite results) and streaming queries (continuous results).

Dictionary Result Rows (`as_dict=True`)

By default, cursors return rows as tuples (standard DB-API). Use as_dict=True to return dictionaries with column names as keys, improving readability and enabling column access by name.

Available on:

connection.cursor(as_dict=True)
connection.streaming_cursor(as_dict=True)
connection.closing_cursor(as_dict=True)
connection.closing_streaming_cursor(as_dict=True)

Example:

# Tuple access (default)
cursor = connection.cursor()
cursor.execute("SELECT customer_id, name, email FROM customers")
row = cursor.fetchone()
print(row[0])  # customer_id by position
print(row[1])  # name

# Dictionary access
cursor = connection.cursor(as_dict=True)
cursor.execute("SELECT customer_id, name, email FROM customers")
row = cursor.fetchone()
print(row["customer_id"])  # Access by column name
print(row["name"])

Use dictionary rows when:

✅ Column names improve code readability
✅ Working with many columns (easier to track positions)
✅ Passing rows to functions expecting dicts
✅ JSON serialization needed
✅ Code is less brittle to schema changes

Use tuple rows when:

✅ Performance is critical (tuples are marginally faster)
✅ Position-based access is clearer for your use case
✅ Minimal memory overhead needed

Auto-Closing Cursor (`closing_cursor()`)

Convenience context manager that creates and automatically closes a cursor, simplifying resource management.

Signature:

with connection.closing_cursor(as_dict=False, mode=ExecutionMode.SNAPSHOT) as cursor:
    cursor.execute("SELECT * FROM users")
    for row in cursor:
        print(row)
# cursor automatically closed

Equivalent to:

cursor = connection.cursor(as_dict=False, mode=ExecutionMode.SNAPSHOT)
try:
    cursor.execute("SELECT * FROM users")
    for row in cursor:
        print(row)
finally:
    cursor.close()
# cursor is explicitly closed even if an error occurs

Benefits:

✅ Guarantees cursor cleanup even if exceptions occur
✅ Cleaner, more Pythonic code
✅ Works with streaming and snapshot cursors

Example with Streaming Cursor:

import time
from confluent_sql.execution_mode import ExecutionMode

with connection.closing_cursor(mode=ExecutionMode.STREAMING_QUERY, as_dict=True) as cursor:
    cursor.execute("SELECT * FROM orders_stream WHERE total > %s", (1000,))

    while cursor.may_have_results:
        rows = cursor.fetchmany(10)
        if rows:
            for row in rows:
                print(f"Order: {row['order_id']}, Total: {row['total']}")
        else:
            time.sleep(0.1)
# cursor automatically closed; cleanup for terminal phases only
# For RUNNING streaming queries, explicitly call delete_statement() to stop them

Auto-Closing Streaming Cursor (`closing_streaming_cursor()`)

Convenience context manager specifically for streaming cursors—equivalent to closing_cursor(mode=ExecutionMode.STREAMING_QUERY, as_dict=as_dict). This is the recommended way to create auto-closing streaming cursors.

Signature:

with connection.closing_streaming_cursor(as_dict=True) as cursor:
    cursor.execute("SELECT * FROM orders_stream WHERE total > %s", (1000,))
    while cursor.may_have_results:
        rows = cursor.fetchmany(10)
        if rows:
            for row in rows:
                process(row)
# cursor automatically closed

Equivalent to:

with connection.closing_cursor(as_dict=True, mode=ExecutionMode.STREAMING_QUERY) as cursor:
    cursor.execute("SELECT * FROM orders_stream WHERE total > %s", (1000,))
    while cursor.may_have_results:
        rows = cursor.fetchmany(10)
        if rows:
            for row in rows:
                process(row)
# cursor is automatically closed

Benefits:

✅ More concise than using closing_cursor() with an explicit mode parameter
✅ Intent is immediately clear: "streaming cursor"
✅ No need to import ExecutionMode
✅ Guarantees cursor cleanup even if exceptions occur

Complete Example:

import time
import confluent_sql

connection = confluent_sql.connect(...)

min_amount = 1000
with connection.closing_streaming_cursor(as_dict=True) as cursor:
    cursor.execute("""
        SELECT order_id, customer_id, amount FROM orders
        WHERE amount > %s
    """, (min_amount,))

    start_time = time.time()
    timeout_seconds = 60

    while cursor.may_have_results:
        rows = cursor.fetchmany(10)
        if rows:
            for row in rows:
                print(f"Order {row['order_id']}: ${row['amount']} from customer {row['customer_id']}")
        else:
            if time.time() - start_time > timeout_seconds:
                print("No new orders for 60 seconds, exiting")
                break
            time.sleep(0.5)
# cursor automatically closed; statement cleanup happens for terminal phases
# For long-running queries still RUNNING on the server, use delete_statement() to stop them

When to use:

✅ Preferred for all streaming cursor usage (cleaner than closing_cursor())
✅ Processing continuous data from Flink SQL
✅ Non-blocking event consumption with automatic resource cleanup

For detailed streaming query patterns and examples, see STREAMING.md.

Custom ROW Type Mapping (`register_row_type()`)

Register custom namedtuple, typing.NamedTuple, or @dataclass classes to be used when deserializing Flink ROW types in query results.

Signature:

connection.register_row_type(MyRowClass)

Example:

from typing import NamedTuple
from collections import namedtuple

# Define custom row type
class OrderRow(NamedTuple):
    order_id: int
    customer_name: str
    total_amount: float

# Register with connection
connection.register_row_type(OrderRow)

# Now ROW results will use this type
with connection.closing_cursor(as_dict=True) as cursor:
    cursor.execute("SELECT order_row FROM orders")
    row = cursor.fetchone()
    order = row["order_row"]
    assert isinstance(order, OrderRow)
    print(f"Order {order.order_id}: {order.customer_name}")

When to use:

✅ Type safety for ROW columns
✅ IDE autocomplete for ROW field access
✅ Validation and custom methods on ROW types

For detailed type support documentation and more complex examples, see TYPES.md.

Streaming Query Support

While the driver defaults to SNAPSHOT mode (traditional DB-API behavior), Confluent Cloud Flink is fundamentally a streaming event database. The driver provides full support for continuous streaming queries through extended cursor types and execution modes.

Key Distinction:

SNAPSHOT mode is designed for users migrating from PostgreSQL, MySQL, etc. Write standard DB-API code and get traditional database behavior.
STREAMING_QUERY mode unlocks Flink's streaming capabilities: continuous results, non-blocking fetches, changelog streams, and state management.

Execution Modes

Mode	Method	Use Case	Result Behavior	Fetch Pattern
SNAPSHOT (default)	`connection.cursor()`	Point-in-time queries (bounded)	Finite result set	Blocking: `fetchall()` works, `for row in cursor` blocks
STREAMING_QUERY	`connection.streaming_cursor()`	Continuous queries (unbounded)	Results arrive over time	Non-blocking: Poll with `fetchone()`/`fetchmany()`, check `may_have_results`

Quick Example

import time
from contextlib import closing

# Create streaming cursor
cursor = connection.streaming_cursor(as_dict=True)

# Execute streaming query
cursor.execute("SELECT order_id, amount FROM orders WHERE amount > %s", (1000,))

# Poll for results (non-blocking)
while cursor.may_have_results:
    rows = cursor.fetchmany(10)
    if rows:
        for row in rows:
            print(f"Order {row['order_id']}: ${row['amount']}")
    else:
        time.sleep(0.5)  # Wait before next poll

Complete Streaming Documentation

For comprehensive documentation of streaming queries, see STREAMING.md.

Statement Lifecycle Management

DDL Execution Convenience Methods

Execute Data Definition Language statements (CREATE TABLE, ALTER, DROP, etc.) with appropriate execution modes and resource management.

execute_snapshot_ddl() - Bounded DDL

statement = connection.execute_snapshot_ddl(
    "CREATE TABLE users AS SELECT * FROM source_users WHERE created_date > %s",
    (date(2024, 1, 1),),
    timeout=3000  # Wait up to 3000 seconds
)
print(f"Created table, statement: {statement.name}")

Use for DDL operations that complete after processing finite data:

CREATE TABLE (not as SELECT)
CREATE TABLE AS SELECT (snapshot mode, bounded source)
DROP TABLE
ALTER TABLE
CREATE VIEW

execute_streaming_ddl() - Unbounded/Open-Ended Statements

Despite the name, this method is not limited to DDL statements. Use it for any open-ended statement that produces no results back to the client and runs indefinitely—including DDL operations and data ingestion/transformation jobs.

# Create a table from streaming source
statement = connection.execute_streaming_ddl(
    "CREATE TABLE orders_stream AS SELECT * FROM kafka_orders",
    timeout=3000  # Wait for job to start (doesn't wait for completion)
)
print(f"Started streaming job: {statement.name}")

# Continuously ingest and transform data
statement = connection.execute_streaming_ddl(
    """
    INSERT INTO filtered_orders
    SELECT order_id, customer_id, amount
    FROM orders_kafka_source
    WHERE amount > %s
    """,
    (100,),
    timeout=3000
)
print(f"Started data pipeline: {statement.name}")

Use for statements that produce unbounded/continuous results:

CREATE TABLE AS SELECT from streaming sources
INSERT INTO ... SELECT from continuous sources (data pipelines)
Any DDL or DML producing indefinitely running Flink jobs

Benefits vs manual cursor approach:

✅ Clearer intent in code
✅ Automatically sets correct execution mode
✅ No need to create/manage cursor for one-off DDL
✅ Returns Statement object for management

Extended Parameters for DDL Methods

Both execute_snapshot_ddl() and execute_streaming_ddl() also accept these parameters for controlling statement identity and lifecycle:

statement_name (str | None): Custom statement identifier (defaults to auto-generated UUID)
statement_labels (list[str] | None): List of labels for grouping related statements

Example with statement naming and labeling:

statement = connection.execute_streaming_ddl(
    "CREATE TABLE orders_stream AS SELECT * FROM kafka_orders",
    statement_name="orders-stream-job",
    statement_labels=["data-pipelines", "streaming"]
)

# Later, find all statements with any of the labels
statements = connection.list_statements(label="data-pipelines")

For more details on managing named and labeled statements, see the Statement Naming and Labeling section.

Statement Naming and Labeling

Control statement identity and grouping for tracking and management.

Statement Names - Unique identifier for each statement

cursor.execute(
    "SELECT * FROM users WHERE active = %s",
    (True,),
    statement_name="active-users-query"
)
# Later, delete by name
connection.delete_statement("active-users-query")

Statement Labels - Group related statements for batch operations

from confluent_sql import HIDDEN_LABEL

# Execute multiple statements with labels
for source in ["kafka_topic_a", "kafka_topic_b", "kafka_topic_c"]:
    cursor.execute(
        f"CREATE TABLE {source}_backup AS SELECT * FROM {source}",
        statement_labels=["daily-backups", "batch-job"]
    )

# Mark a background job as hidden
cursor.execute(
    "SELECT * FROM `INFORMATION_SCHEMA`.`TABLES`",
    statement_labels=[HIDDEN_LABEL]
)

# Later, list and delete all backups
statements = connection.list_statements(label="daily-backups")
for stmt in statements:
    connection.delete_statement(stmt)

When to use:

✅ Long-running streaming jobs (track and manage)
✅ Batch operations (group related statements)
✅ Error recovery (find and clean up failed jobs)

Finding and Deleting Statements

list_statements() - Find statements, optionally filtered

Called with no arguments, returns every statement in the environment. Three optional, server-side filters narrow the results and combine with AND semantics:

label — statements carrying the given end-user label.
compute_pool_id — statements in a specific compute pool.
name_contains — statements whose name contains the given substring (case-sensitive).

# Every statement in the environment
statements = connection.list_statements()

# Narrow by label, compute pool, and/or name substring (filters AND together)
statements = connection.list_statements(label="daily-backups", page_size=100)
statements = connection.list_statements(compute_pool_id="lfcp-789012")
statements = connection.list_statements(name_contains="orders-stream")

for statement in statements:
    print(f"Statement: {statement.name}")
    print(f"Phase: {statement.phase}")  # RUNNING, COMPLETED, FAILED, etc.
    print(f"Created: {statement.created_at}")

get_statement() - Retrieve statement by exact name

# Get statement by name
stmt = connection.get_statement("my-statement-name")
print(f"Status: {stmt.phase}")  # PENDING, RUNNING, COMPLETED, FAILED, etc.

# Check if results are ready
if stmt.can_fetch_results(ExecutionMode.SNAPSHOT):
    # Results are available
    ...

# Refresh a Statement object with latest server state
stmt = cursor.statement
time.sleep(5)
stmt = connection.get_statement(stmt)  # Fetch updated state
print(f"Updated phase: {stmt.phase}")

# Raises StatementNotFoundError if statement not found
try:
    stmt = connection.get_statement("non-existent-statement")
except StatementNotFoundError as e:
    print(f"Statement '{e.statement_name}' does not exist")
except OperationalError as e:
    print(f"Other error: {e}")

stop_statement() - Halt a statement without deleting it

Stops a running statement (for example, a long-lived streaming query) while keeping the statement resource around for inspection — unlike delete_statement(), which stops and destroys the statement and its results. The stop is issued as a JSON Patch flipping spec.stopped to true; the statement then transitions through STOPPING to the terminal STOPPED phase.

# Stop from connection (by name or Statement object), blocking until STOPPED (the default)
stopped = connection.stop_statement("active-users-query")
print(stopped.is_stopped)  # True

# Non-blocking: return as soon as the stop is accepted
stmt = connection.stop_statement(statement_obj, wait_for_stopped=False)
print(stmt.stop_requested)  # True (the stop was accepted)
# Note: stmt.phase may still be RUNNING at this instant -- the server transitions the phase
# to STOPPED asynchronously. Poll get_statement() if you need to observe STOPPED:
while not connection.get_statement(stmt).is_stopped:
    time.sleep(0.5)

# Bound the blocking wait (seconds); raises OperationalError on timeout
stopped = connection.stop_statement("active-users-query", timeout=60)

# Stop from cursor (current statement); updates the cursor's tracked statement
stopped = cursor.stop_statement()

Behavior notes:

Accepts a statement name (string) or a Statement object. A Statement already in a terminal phase (STOPPED/COMPLETED/FAILED/DELETED) is returned unchanged without an API call.
wait_for_stopped=True (default) blocks until the statement reaches a terminal phase — normally STOPPED, but COMPLETED if a bounded query finished before the stop landed — so the caller knows the statement is no longer running. wait_for_stopped=False returns once the stop is accepted — confirm acceptance via Statement.stop_requested rather than the phase.
Raises StatementNotFoundError if the statement does not exist, or OperationalError on other API errors, on timeout, or if the statement transitions to FAILED while stopping.
cursor.stop_statement() raises InterfaceError when the cursor has no executed statement to stop (it is not a silent no-op like cursor.delete_statement()).

delete_statement() - Stop and remove a statement

# Delete from connection (by name or Statement object)
connection.delete_statement("active-users-query")
connection.delete_statement(statement_obj)

# Delete from cursor (current statement)
cursor.delete_statement()

When to delete statements:

✅ Long-running streaming jobs no longer needed
✅ Freeing compute pool resources (required before closing connection)
✅ Cleanup during error handling
⚠️ Deletion stops the statement immediately (may cause errors if still in use)

Tableflow Lifecycle

Tableflow materializes the Kafka topic backing a Flink table into an Iceberg or Delta table. Three Connection methods manage that sink. Enabling it also unlocks efficiency gains for snapshot queries against the table.

In Confluent Flink a table is backed by a like-named Kafka topic, so the table_name you pass is both the Flink table and the topic — no escaping or casing translation.

A runnable example covering the full enable/get/disable lifecycle is in examples/tableflow_lifecycle_example.py.

Not available under BYOIDC. Tableflow is a control-plane surface, and Confluent's authorization model accepts no BYOIDC bearer token there. A connection authenticated with external_access_token / identity_pool_id (see the README's BYOIDC bearer-token authentication) fails closed on these methods — use an API-key connection for Tableflow.

Selecting formats: `TableFormat`

There is a single format vocabulary, TableFormat (ICEBERG / DELTA), on both the request and response sides. A topic can carry both formats at once (there is no per-format config), so enable_tableflow's tableflow_formats argument accepts either a single TableFormat for the common case or a collection for both:

connection.enable_tableflow("orders", tableflow_formats=TableFormat.ICEBERG, storage=...)
connection.enable_tableflow(
    "orders", tableflow_formats={TableFormat.ICEBERG, TableFormat.DELTA}, storage=...
)

Responses name the same TableFormats (topic.spec.table_formats, topic.status.failing_table_formats), so checking what you got against what you asked for is a plain set comparison:

topic = connection.enable_tableflow(
    "orders", tableflow_formats={TableFormat.ICEBERG, TableFormat.DELTA},
    storage=ManagedStorage(), wait_for_running=True,
)
assert set(topic.spec.table_formats) == {TableFormat.ICEBERG, TableFormat.DELTA}

Storage variants

enable_tableflow requires an explicit, frozen storage spec — no silent default:

ManagedStorage() — Confluent-managed bucket, zero config.
ByobAwsStorage(bucket_name=..., provider_integration_id=...) — bring-your-own AWS S3 bucket.
AzureAdlsStorage(storage_account_name=..., container_name=..., provider_integration_id=...) — customer-owned Azure Data Lake Storage Gen2.

Cluster-id resolution

The Tableflow API addresses the cluster by its lkc-… id, which the connection must know. Either:

Pass database_kafka_cluster_id to connect() (works with only a tableflow_api_key pair), or
Let it resolve lazily from database (the cluster name) via CMK on first use — this path requires a global API key, and the resolved id is cached for the connection's life. A name that matches more than one cluster raises, listing the candidate ids so you can disambiguate with database_kafka_cluster_id.

`enable_tableflow()` — add an Iceberg/Delta sink

from confluent_sql import ManagedStorage, TableFormat, TableflowPhase

topic = connection.enable_tableflow(
    "orders",
    tableflow_formats=TableFormat.ICEBERG,
    storage=ManagedStorage(),
)
assert topic.phase is TableflowPhase.RUNNING   # blocked to RUNNING by default

Behavior notes:

tableflow_formats and storage are required (no defaults); tableflow_formats must name at least one format. config is an optional TableflowTopicConfig (retention, error-handling) shared across all enabled formats.
Blocks until RUNNING by default (wait_for_running=True), raising OperationalError on FAILED (surfacing status.error_message and failing_table_formats) — consistent with stop_statement's wait-by-default. Pass wait_for_running=False to return as soon as the create is accepted (topic in PENDING).
Raises TableflowTopicAlreadyExistsError if Tableflow is already enabled (HTTP 409), or ProgrammingError if no management credential is available or the cluster id can't be resolved.

`get_tableflow()` — read current state

topic = connection.get_tableflow("orders")
print(topic.phase)                       # TableflowPhase.PENDING / RUNNING / FAILED
print(topic.spec.table_formats)          # [TableFormat.ICEBERG, ...]

Raises TableflowTopicNotFoundError if Tableflow is not enabled for the topic (HTTP 404). There is no separate health check — health is read off get_tableflow(...).phase.

`disable_tableflow()` — tear down the sink

connection.disable_tableflow("orders")   # blocks until confirmed gone by default

Behavior notes:

All-or-nothing in v1: removes the entire Tableflow topic. (Removing just one of two enabled formats needs a future API and is not yet supported.)
Deletion is asynchronous. Blocks until removal is confirmed by default (wait_for_removal=True), polling get_tableflow until it 404s. This is why a following DROP TABLE is safe by default — dropping the Flink table drops its backing topic, so Tableflow must be confirmed gone first to avoid racing an active materialization. Pass wait_for_removal=False to return as soon as the DELETE is accepted.
Raises TableflowTopicNotFoundError if Tableflow was not enabled (HTTP 404).

Reusing format and config across many tables

Every input is a reusable value — the selection is an enum member, and storage/config are frozen — so hoist them out of the loop. The cluster-id lookup resolves once and is cached, so the loop hits CMK at most once:

from confluent_sql import ManagedStorage, TableFormat, TableflowTopicConfig

storage = ManagedStorage()
config = TableflowTopicConfig(retention_ms="604800000")

for table in ("orders", "shipments", "returns"):
    connection.enable_tableflow(
        table,
        tableflow_formats=TableFormat.ICEBERG,
        storage=storage,
        config=config,
    )

Introspection and Metadata

Connection Properties

Property	Type	Description
`is_closed`	`bool`	Check if connection has been closed
`http_user_agent`	`str`	Get or set User-Agent header for HTTP requests (1-100 characters)

Example:

# Check if connection is still active
if not connection.is_closed:
    cursor = connection.cursor()
    cursor.execute("SELECT 1")

# Customize User-Agent
connection.http_user_agent = "MyApp/1.0 (custom agent)"

Cursor Properties

Property	Type	When Available	Description
`statement`	`Statement`	After `execute()`	Full statement metadata and lifecycle info
`may_have_results`	`bool`	After `execute()`	More data may arrive (streaming), or results exhausted
`is_closed`	`bool`	Always	Check if cursor is closed
`execution_mode`	`ExecutionMode`	Always	`SNAPSHOT` or `STREAMING_QUERY` (from cursor config)
`is_streaming`	`bool`	Always	Convenience: `execution_mode == STREAMING_QUERY`
`returns_changelog`	`bool`	After `execute()`	Results are ChangeloggedRow with operations
`as_dict`	`bool`	Always	Rows returned as dicts vs tuples
`metrics`	`FetchMetrics`	After first fetch call	Accumulated fetch performance statistics

Usage Examples:

cursor = connection.streaming_cursor(as_dict=True)
cursor.execute("SELECT * FROM orders GROUP BY customer_id")

# Introspect query characteristics
if cursor.returns_changelog:
    print("Query is stateful - use changelog compressor")
    compressor = cursor.changelog_compressor()
else:
    print("Query is append-only - iterate directly")

# Check execution mode
if cursor.is_streaming:
    print("Non-blocking fetch pattern")
    while cursor.may_have_results:
        row = cursor.fetchone()
        if row:
            process(row)
        else:
            time.sleep(0.1)
else:
    print("Blocking fetch pattern")
    for row in cursor:
        process(row)

Statement Object

The cursor.statement property provides detailed metadata about the executed query.

Key Statement Properties:

Property	Type	Description
`name`	`str`	Statement identifier (auto-generated UUID or custom name)
`phase`	`StatementPhase`	Execution phase: `PENDING`, `RUNNING`, `COMPLETED`, `FAILED`, `STOPPED`
`is_append_only`	`bool`	Query produces only inserts (vs changelog with updates/deletes)
`is_bounded`	`bool`	Query has finite result set (snapshot) vs unbounded (streaming)
`is_deletable`	`bool`	Statement can be deleted
`schema`	`Schema`	Result schema with column names and types
`sql_kind`	`str`	Query type: `SELECT`, `INSERT`, `CREATE`, etc.

Example:

cursor.execute("SELECT product_id, COUNT(*) as sales FROM orders GROUP BY product_id")

stmt = cursor.statement
print(f"Statement: {stmt.name}")
print(f"SQL Kind: {stmt.sql_kind}")
print(f"Schema: {stmt.schema}")

# Check query characteristics
print(f"Append-only: {stmt.is_append_only}")
print(f"Bounded: {stmt.is_bounded}")

Understanding Changelog Snapshots

When working with streaming non-append-only queries (aggregations, joins), the changelog compressor yields snapshots of results—complete accumulated result sets at specific points in time. It's important to understand what a snapshot represents.

What is a Snapshot of Results?

A snapshot of results is a self-consistent, complete result set of the accumulated state at a point in time. It represents all rows that exist at that moment, built from the entire history of INSERT, UPDATE, and DELETE operations processed so far.

Key Characteristics of a Snapshot:

Complete Result Set: Contains all rows that exist at that moment (accumulated from all INSERT/UPDATE/DELETE operations processed since the query started)
Self-Consistent: All currently available changelog events have been consumed and applied. No pending operations are awaiting completion
Point-in-Time: Represents the state after processing all events up to that moment
May Be Identical: Two consecutive snapshots can show the same results if no new events arrived between them. This is normal and expected

Important: Snapshot of Results vs Snapshot Query

Do not confuse a snapshot of results from the changelog compressor with a snapshot query (an execution mode):

Snapshot Query (execution mode): A bounded point-in-time query that returns finite results and completes
Snapshot of Results (from compressor): The accumulated, complete result set of an ongoing streaming query showing the current state

Example: Understanding Snapshots of Results

Consider a streaming GROUP BY query counting users by first letter:

cursor = connection.streaming_cursor()
cursor.execute("SELECT first_letter, COUNT(*) as user_count FROM users GROUP BY first_letter")
compressor = cursor.changelog_compressor()

# Each snapshot of results is the COMPLETE count of users by first letter at that moment
for snapshot in compressor.snapshots():
    # snapshot[0] might be: ('A', 5)  - 5 users with first letter A
    # snapshot[1] might be: ('B', 3)  - 3 users with first letter B
    # snapshot[2] might be: ('C', 2)  - 2 users with first letter C
    print(f"Current user counts: {snapshot}")
    time.sleep(5)

As users are added, updated, or deleted, each new snapshot of results shows the updated complete count:

Snapshot 1: [('A', 5), ('B', 3), ('C', 2)]
Snapshot 2: [('A', 5), ('B', 4), ('C', 2)] (B count increased due to new user)
Snapshot 3: [('A', 5), ('B', 4), ('C', 2)] (No change, no new events arrived)
Snapshot 4: [('A', 6), ('B', 4), ('C', 2)] (A count increased)

Each snapshot is the complete result set showing all aggregated rows at that moment—not just the rows that changed.

Performance Monitoring

Fetch Metrics

Monitor and analyze result fetching performance using the cursor.metrics property. Useful for identifying bottlenecks and optimizing polling patterns in streaming applications.

Basic Example:

cursor = connection.streaming_cursor()
cursor.execute("SELECT * FROM high_volume_stream")

# Consume some results
for _ in range(100):
    row = cursor.fetchone()
    if not row:
        time.sleep(0.1)

# Analyze performance
metrics = cursor.metrics
print(f"Total page fetches: {metrics.total_page_fetches}")
print(f"Rows returned: {metrics.rows_returned}")
print(f"Time in fetches: {metrics.fetch_request_secs:.2f}s")
print(f"Total pause time: {metrics.paused_secs:.2f}s")
print(f"Bytes received: {metrics.bytes_received}")

Available Metrics:

Metric	Type	Description
`total_page_fetches`	`int`	Total result pages fetched from server
`total_changelog_rows_fetched`	`int`	Total changelog rows received
`empty_page_fetches`	`int`	Pages with no rows (polling overhead)
`fetch_request_secs`	`float`	Total time spent in fetch requests
`paused_times`	`int`	Number of times pause delay occurred
`paused_secs`	`float`	Total time paused between fetches
`bytes_received`	`int`	Total bytes received from server
`rows_returned`	`int`	Total rows returned to caller

Use cases:

Debugging slow queries
Monitoring streaming query throughput
Detecting inefficient polling patterns
Tuning result_page_fetch_pause_millis parameter

Type System Extensions

The driver supports all Flink SQL types with automatic parameter interpolation and result decoding. Some types require driver-specific Python types.

Special Types

SqlNone - Typed NULL values

Flink requires NULL values to have explicit types. Use SqlNone to specify NULL values with a Flink type:

from confluent_sql import SqlNone

cursor.execute(
    "INSERT INTO users (id, name, age) VALUES (%s, %s, %s)",
    (
        1,
        "Alice",
        SqlNone.INTEGER  # NULL with INTEGER type
    )
)

# For non-scalar types, pass the Flink type string
null_array = SqlNone("Array<int>")
cursor.execute("INSERT INTO data_column VALUES (%s)", (null_array,))

YearMonthInterval - Year-month intervals

Flink's INTERVAL YEAR TO MONTH type requires the driver's YearMonthInterval dataclass:

from confluent_sql import YearMonthInterval
from datetime import timedelta

interval = YearMonthInterval(years=1, months=6)  # 1 year, 6 months
cursor.execute("SELECT * FROM orders WHERE created > (NOW() - %s)", (interval,))

Complete Type Reference

For comprehensive type support documentation, see TYPES.md:

Complete type mapping table - Flink types ↔ Python types
Parameter interpolation - Encoding Python values to Flink SQL
Result decoding - Decoding Flink values to Python types
Type conversion caveats - Edge cases and limitations
ARRAY and MAP examples - Complex nested types
ROW type registration - Custom type mapping
Examples from tests - Real, tested code samples

Custom Exceptions

Standard DB-API Exceptions

All standard DB-API v2 exceptions are available:

Warning - Warning messages
Error - Base exception class
InterfaceError - Interface-related errors
DatabaseError - Database-related errors
DataError - Data value-related errors
OperationalError - Database operation errors
IntegrityError - Relational integrity errors
InternalError - Internal database errors
ProgrammingError - Programming/SQL errors
NotSupportedError - Unsupported feature

Streaming-Specific Exceptions

Additional exceptions for streaming query lifecycle:

Exception	Inherits From	When Raised
`StatementStoppedError`	`OperationalError`	Streaming statement stops unexpectedly during iteration
`StatementDeletedError`	`StatementStoppedError`	Statement was deleted (404 from server)
`ComputePoolExhaustedError`	`OperationalError`	Compute pool has no available resources

StatementStoppedError Attributes:

statement_name - Name of the stopped statement
statement - Statement object (if available)
phase - Terminal phase (STOPPED, FAILED, etc.)

Exception Handling Example:

from confluent_sql import StatementStoppedError, ComputePoolExhaustedError
import time

cursor = connection.streaming_cursor()

try:
    cursor.execute("SELECT * FROM streaming_orders")
except ComputePoolExhaustedError:
    # Raised at submission time if compute pool has no resources
    print("Compute pool exhausted, cannot submit query")
    raise

try:
    while cursor.may_have_results:
        row = cursor.fetchone()
        if row:
            process(row)
        else:
            time.sleep(0.1)
except StatementStoppedError as e:
    # Raised during result processing if statement stops unexpectedly
    print(f"Query stopped: {e.statement_name}")
    print(f"Final phase: {e.phase}")

Extended Execute Parameters

The cursor.execute() method accepts additional parameters beyond standard DB-API for controlling statement execution and management.

Signature

cursor.execute(
    statement_text: str,
    parameters: tuple | list | None = None,
    *,
    timeout: int = 3000,
    statement_name: str | None = None,
    statement_labels: list[str] | None = None,
    properties: dict[str, str | int | bool] | StatementProperties | None = None,
    compute_pool_id: str | None = None,
) -> None

Parameter Reference

Parameter	Type	Default	Description
`statement_text`	`str`	(required)	SQL statement to execute
`parameters`	`tuple \| list \| None`	None	Parameter values for parameterized statements
`timeout`	`int`	3000	Max seconds to wait for statement to reach RUNNING/COMPLETED phase
`statement_name`	`str \| None`	None	Custom statement identifier (defaults to UUID)
`statement_labels`	`list[str] \| None`	None	List of labels for grouping related statements
`properties`	`dict[str, str \| int \| bool] \| StatementProperties \| None`	None	Statement properties to set for execution
`compute_pool_id`	`str \| None`	None	Compute pool to run this statement on, overriding the connection's default

Statement Properties

The properties parameter sets Flink SQL statement properties at query execution time — the same properties Flink SQL SET statements control. There are two ways to provide them:

StatementProperties (recommended) — a frozen, keyword-only dataclass covering the curated options below, discoverable via autocomplete and validated at construction time instead of at the server. A wrong-property enum value (e.g. a SnapshotMode passed to scan_startup_mode), a field of the wrong Python type, or an extra key that duplicates a modeled field all raise immediately:
```
from confluent_sql import Property, ScanStartupMode, SnapshotWriteMode, StatementProperties
from datetime import timedelta

cursor.execute(
    "SELECT * FROM orders WHERE status = %s",
    ("pending",),
    properties=StatementProperties(
        state_ttl=timedelta(hours=1),                  # -> "3600 s"
        snapshot_write_mode=SnapshotWriteMode.FAST_WRITE,
        scan_startup_mode=ScanStartupMode.EARLIEST_OFFSET,
        # `extra` escape hatch for a property not yet a typed field; Keys can
        # either be Property enums or strings.
        extra={Property.SCAN_IDLE_TIMEOUT: "30 s"},
    ),
)
```
The set of modeled fields grows over time -- see the StatementProperties docstring/source for the current list -- but the shape is uniform: only fields you actually set are emitted, so an unset field never pins a server default or collides with the driver's own overlay, and each enum-typed field also accepts a bare str, so a Flink value newer than this driver's enum can still be passed through without waiting for a driver release.
A raw dict[str, str | int | bool] — the original, open-ended form. Any sql.* key is accepted, keyed by the string from the SET-options reference, which is useful for options StatementProperties doesn't model yet (equivalent to extra above, without needing to go via the dataclass):
```
cursor.execute(query, properties={"sql.state-ttl": "3600 s"})
```
confluent_sql.Property enumerates the known sql.* keys (e.g. Property.STATE_TTL) if you want autocomplete on the keys without adopting the full dataclass; members are plain str instances, so they drop straight into the dict with no .value unwrapping.

Both forms are validated identically — a StatementProperties is downgraded to a dict internally before the same checks run, so a reserved key smuggled through extra is rejected the same way a raw dict would be.

Important Precedence Rules:

System properties are always applied and cannot be overridden by the caller: the connection's catalog/database and the cursor's execution mode (e.g. sql.snapshot.mode for snapshot queries).
The connection-level local_time_zone default (see Connection.local_time_zone below) fills in sql.local-time-zone only when the call's own properties didn't already set it.
User-provided properties in the properties parameter can set anything not covered by the two rules above, but attempting to set a system property (e.g. sql.current-catalog) raises InterfaceError rather than being silently overridden.

Accessing Properties After Execution: The properties are stored in the cursor-captured Statement object and can be accessed via statement.properties, a dict[str, str | int | bool]:

cursor.execute(query, properties={"sql.state-ttl": "100 ms"})
props = cursor.statement.properties
assert props["sql.state-ttl"] == "100 ms"

Connection-level `local_time_zone` default

Connection.local_time_zone (also settable via connect(local_time_zone=...)) is a read/write property that seeds sql.local-time-zone for every statement the connection executes, so you don't have to repeat it on each execute() call:

connection.local_time_zone = "America/Chicago"
cursor.execute("SELECT CURRENT_TIMESTAMP")  # runs with sql.local-time-zone = America/Chicago

# A statement can still override it for itself:
cursor.execute(
    "SELECT CURRENT_TIMESTAMP",
    properties=StatementProperties(local_time_zone="America/Los_Angeles"),
)

The connection-level value only fills in where a statement's own properties (dict or StatementProperties) didn't already set sql.local-time-zone — it never overrides an explicit per-call value. Set it to None to stop emitting a default.

A runnable example covering both property forms and the connection-level default is in examples/statement_properties_example.py.

General Usage Examples:

# Basic execution
cursor.execute("SELECT * FROM users")

# With parameters
cursor.execute("SELECT * FROM users WHERE age > %s", (18,))

# With custom timeout
cursor.execute(
    "SELECT * FROM users",
    timeout=100  # Wait up to 100 seconds
)

# With statement naming
cursor.execute(
    "SELECT * FROM orders WHERE status = %s",
    ("completed",),
    statement_name="completed-orders-daily"
)

# With statement labeling
cursor.execute(
    "CREATE TABLE orders_backup AS SELECT * FROM orders",
    statement_labels=["daily-backups", "batch-job"]
)

# All together
cursor.execute(
    "SELECT product_id, COUNT(*) as sales FROM orders GROUP BY product_id",
    (),
    timeout=3000,
    statement_name="product-sales-hourly",
    statement_labels=["analytics", "hourly"]
)

# With statement properties (raw dict)
cursor.execute(
    "SELECT * FROM orders WHERE status = %s",
    ("pending",),
    statement_name="pending-orders-query",
    properties={"sql.state-ttl": "100 ms"}
)

# With statement properties (StatementProperties)
cursor.execute(
    "SELECT * FROM orders WHERE status = %s",
    ("pending",),
    statement_name="pending-orders-query",
    properties=StatementProperties(state_ttl=timedelta(milliseconds=100)),
)

Uh oh!

FilesExpand file tree

DBAPI_EXTENSIONS.md

Latest commit

History

DBAPI_EXTENSIONS.md

File metadata and controls

DB-API Extensions

Understanding Snapshot vs Streaming Modes

Quick Navigation

Result Format Extensions

Dictionary Result Rows (as_dict=True)

Auto-Closing Cursor (closing_cursor())

Auto-Closing Streaming Cursor (closing_streaming_cursor())

Custom ROW Type Mapping (register_row_type())

Streaming Query Support

Execution Modes

Quick Example

Complete Streaming Documentation

Statement Lifecycle Management

DDL Execution Convenience Methods

Statement Naming and Labeling

Finding and Deleting Statements

Tableflow Lifecycle

Selecting formats: TableFormat

Storage variants

Cluster-id resolution

enable_tableflow() — add an Iceberg/Delta sink

get_tableflow() — read current state

disable_tableflow() — tear down the sink

Reusing format and config across many tables

Introspection and Metadata

Connection Properties

Cursor Properties

Statement Object

Understanding Changelog Snapshots

What is a Snapshot of Results?

Important: Snapshot of Results vs Snapshot Query

Example: Understanding Snapshots of Results

Performance Monitoring

Fetch Metrics

Type System Extensions

Special Types

Complete Type Reference

Custom Exceptions

Standard DB-API Exceptions

Streaming-Specific Exceptions

Extended Execute Parameters

Signature

Parameter Reference

Statement Properties

Connection-level local_time_zone default

See Also

Dictionary Result Rows (`as_dict=True`)

Auto-Closing Cursor (`closing_cursor()`)

Auto-Closing Streaming Cursor (`closing_streaming_cursor()`)

Custom ROW Type Mapping (`register_row_type()`)

Selecting formats: `TableFormat`

`enable_tableflow()` — add an Iceberg/Delta sink

`get_tableflow()` — read current state

`disable_tableflow()` — tear down the sink

Connection-level `local_time_zone` default