SQL Database Sinks
Relevant source files
The following files were used as context for generating this wiki page:
SQL Database Sinks are specialized components within the Meltano Singer SDK designed to facilitate the loading of data into relational database management systems (RDBMS). They inherit from the base SQLSink class and leverage SQLAlchemy to provide a standardized interface for schema management, data transformation, and bulk loading operations across various SQL dialects like PostgreSQL, MySQL, and SQLite.
These sinks operate by mapping Singer stream schemas to database tables and handling the lifecycle of data records from receipt to persistence. Key capabilities include automatic table creation, schema evolution (adding columns for new properties), and support for high-performance loading methods such as MERGE/UPSERT and bulk inserts.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md, singer_sdk/sql/sink.py
Architecture and Components
The SQL Sink architecture relies on three primary classes working in tandem to manage the connection, the data target, and the specific stream processing logic.
Core Classes
| Class | Description |
|---|---|
SQLTarget | The main entry point for a database target. It handles global configuration and manages the instantiation of sinks for each stream. |
SQLSink | Manages the data for a specific stream. It handles batching, record processing, and coordinates with the connector for DDL/DML operations. |
SQLConnector | Handles the low-level database communication using SQLAlchemy. It manages connection strings, DDL generation (like CREATE TABLE), and dialect-specific logic. |
Sources: singer_sdk/sql/sink.py, tests/sql/test_sink.py:27-55, cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks_sql.py:21-48
System Relationships
The following diagram illustrates the relationship between the Target, Sink, and Connector components during a sync operation.
The diagram shows that a SQLTarget can manage multiple SQLSink instances, each of which relies on a SQLConnector to communicate with the database.
Sources: singer_sdk/sql/sink.py, cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks_sql.py:48-52
Data Flow and Life Cycle
SQL Sinks follow a structured lifecycle to ensure the destination database matches the incoming stream requirements before data is committed.
Initialization and Setup
When a stream starts, the setup() method is invoked. This process ensures the target schema and table exist. If they do not, the SQLConnector generates and executes the necessary DDL based on the JSON Schema provided by the Singer tap.
Record Processing and Batching
SQL Sinks typically use Per batch serialization. Records are accumulated in memory until a threshold (max_size, default 10,000) is reached or the stream ends.
process_record(): Validates and prepares individual records.process_batch(): Triggered when a batch is full. It coordinates the bulk upload.bulk_insert_records(): Executes the actual DML (Data Manipulation Language) statement.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md, cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks_sql.py:61-75
Batch Processing Flow
This flowchart describes the logic used to manage memory and performance by grouping individual records into larger database transactions.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Schema Evolution and Management
A critical feature of SQL Sinks is the ability to handle schema changes mid-stream without losing data.
Key Capabilities
- Automatic Column Addition: If a record contains a property not present in the database table, the connector can issue an
ALTER TABLE ... ADD COLUMNstatement ifallow_column_addis enabled. - Type Mapping: JSON Schema types (string, integer, number, etc.) are automatically mapped to SQL types (VARCHAR, BIGINT, DECIMAL, etc.).
- Preservation of Undrained Records: The SDK is designed to ensure that if a schema change occurs, existing buffered records are still processed correctly against the evolved table structure.
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md, tests/sql/test_sink.py:180-210, cookiecutter/target-template/{{cookiecutter.target_id}}/{{cookiecutter.library_name}}/sinks_sql.py:26-31
Type Mapping Example
SQL targets must define how JSON types translate to the specific SQL dialect.
# Implementation in a custom connector
@property
def type_mapping(self):
return {
"string": "VARCHAR",
"integer": "BIGINT",
"number": "DECIMAL",
"boolean": "BOOLEAN",
"object": "JSONB",
"array": "JSONB",
}
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md
Configuration Options
SQL Sinks are configured via the SQLTarget. Common settings include connection details and behavior flags.
| Setting | Type | Description |
|---|---|---|
sqlalchemy_url | string | The connection string for the database (e.g., postgresql://user:pass@host/db). |
default_target_schema | string | The default database schema (namespace) where tables will be created. |
load_method | string | Determines how data is loaded (e.g., overwrite, append-only, upsert). |
batch_size | integer | Maximum number of records per batch (default 10,000). |
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md, tests/sql/test_sink.py:101-140
Implementation Details
Generating SQL Statements
The SQLSink handles the generation of SQLAlchemy Insert or Update objects. It ensures that reserved words (like "table") are properly quoted and that columns match the stream schema.
def test_generate_insert_statement(self, sink: DummySQLSink, schema: dict):
"""Verifies that the generated SQL correctly quotes reserved keywords."""
stmt = sink.generate_insert_statement("foo", schema=schema)
# Expected: INSERT INTO foo (id, col_ts, "table") VALUES (:id, :col_ts, :table)
assert '"table"' in str(stmt)
Sources: tests/sql/test_sink.py:76-85
Connector Capabilities
Developers can toggle specific database features by overriding boolean flags in the SQLConnector subclass:
allow_column_alter: Allows changing existing column data types.allow_merge_upsert: Enables high-efficiency "Upsert" logic if the dialect supports it.allow_temp_tables: Enables the use of temporary tables for staged loads.
Conclusion
SQL Database Sinks provide a robust, dialect-agnostic framework for loading Singer data into relational databases. By abstracting DDL and DML operations into the SQLConnector and SQLSink classes, the SDK allows developers to focus on dialect-specific optimizations (like native bulk loaders) while benefiting from built-in schema evolution, type mapping, and batch management.