SQL Database Streams
Relevant source files
The following files were used as context for generating this wiki page:
SQL Database Streams provide a structured framework within the Meltano Singer SDK for extracting data from and loading data into relational databases using SQLAlchemy. This system abstracts the complexities of database connections, schema discovery, and data type mapping between SQL and JSON Schema, enabling developers to build robust Singer taps and targets for SQL sources.
Sources: CLAUDE.md:88, cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:108-112, cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:70-75
Architecture and Components
The SQL stream architecture relies on three primary base classes that manage the lifecycle of data extraction and loading.
Key Classes
| Class | Role | Description |
|---|---|---|
SQLConnector | Connection Manager | Handles SQLAlchemy engine creation, DDL operations, and schema discovery. |
SQLStream | Data Extractor | Manages record retrieval from a database source, including query filtering and partitioning. |
SQLSink | Data Loader | Handles writing records to a database target, supporting table creation and schema evolution. |
Sources: CLAUDE.md:88-92, cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:11-13
Component Interaction
The following diagram illustrates how these components interact during a discovery and extraction process.
Data Extraction (Taps)
SQL-based taps utilize SQLStream and SQLConnector to automate the discovery of tables and the generation of Singer-compliant schemas.
Schema Discovery and Conversion
The SQLToJSONSchema class is responsible for converting SQLAlchemy column types into JSON Schema definitions. Developers can override to_jsonschema or use singledispatch handlers to support database-specific types.
class MySQLToJSONSchema(SQLToJSONSchema):
@to_jsonschema.register
def custom_type_to_jsonschema(self, column_type: CustomSQLType) -> dict:
return {"type": ["string", "null"]}
Sources: cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:34-45, cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:90-104
Record Retrieval Flow
SQLStream.get_records manages the execution of SELECT queries. It supports:
- Selected Columns: Only fetching columns defined in the stream's schema to avoid
SELECT *. - Replication Keys: Filtering and ordering by a replication key for incremental syncs.
- Query Filters: Applying custom WHERE clauses via
apply_query_filters.
Sources: cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:204-222, cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:233-255
Data Loading (Targets)
SQL targets use SQLSink to handle the insertion of records and the maintenance of destination tables.
Table Maintenance and Schema Evolution
SQLSink provides automatic table creation based on incoming SCHEMA messages. It also handles schema evolution by adding new columns to existing tables without data loss when new properties are detected in the stream.
Load Methods
SQL Sinks typically support bulk operations for higher throughput.
- Bulk Insert: Grouping records for efficient loading.
- MERGE/UPSERT: Using primary keys to update existing records (implementation varies by database dialect).
- Overwrite: Dropping or truncating data before loading (controlled by
load_methodconfig).
Sources: cookiecutter/target-template/{{cookiecutter.target_id}}/AGENTS.md:76-85, tests/sql/test_sink.py:140-145
Configuration
SQL components require specific configuration parameters to establish connectivity and define behavior.
| Parameter | Type | Description |
|---|---|---|
sqlalchemy_url | String | A valid SQLAlchemy connection string (e.g., sqlite:///path/to/db). |
default_target_schema | String | (Target only) The default schema/namespace to use in the destination. |
load_method | String | (Target only) Determines if records are appended or overwritten. |
supports_nulls_first | Boolean | (Stream only) Indicates if the database supports specific null ordering in SQL. |
Sources: cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:118-135, tests/sql/test_sink.py:93-100, cookiecutter/tap-template/{{cookiecutter.tap_id}}/{{cookiecutter.library_name}}/sql-client.py:177
Implementation Details
Type Conformance
The TYPE_CONFORMANCE_LEVEL setting in SQLStream controls how strictly the SDK enforces data types:
ROOT_ONLY: Enforces types only at the top level, which is useful for JSONB columns where internal structures are dynamic.RECURSIVE: Recursively validates and transforms types throughout nested structures.
Custom DDL Operations
The SQLConnector allows for the customization of Data Definition Language (DDL) through methods like get_column_alter_ddl. This is critical for supporting database-specific syntax for altering tables during schema evolution.
Sources: tests/sql/test_sink.py:23-40
Conclusion
SQL Database Streams in the Meltano Singer SDK provide a powerful abstraction over SQLAlchemy, allowing developers to focus on source-specific logic while the framework handles Singer compliance, schema discovery, and evolution. By leveraging the modular design of Connectors, Streams, and Sinks, the SDK ensures high-performance data movement across diverse relational database systems.
Sources: README.md:46-51, CLAUDE.md:86-92