<!-- source: sling-docs/sling-cli/getting-started.md -->
# Getting Started with Sling CLI

## Getting Started with Sling CLI

{% embed url="https://f.slingdata.io/videos/sling.cli.demo.2023.10.720.mp4" %}
Sling CLI Demo
{% endembed %}

## Getting Started

`sling` CLI is a free tool that allows data extraction and loading from / into many popular databases / storage platforms. Follow the instructions below to install it on your respective operating system.

### Installation

All commands below will install the latest version of sling. If you'd like to determine which version is latest/current, you can check out the [releases page](https://github.com/slingdata-io/sling-cli/releases) on github.

#### One-liner on Mac / Linux

The install script auto-detects your OS and architecture, downloads the latest binary, and adds `sling` to your `PATH`.

```shell
curl -fsSL https://slingdata.io/install.sh | bash

# You're good to go!
sling -h
```

By default sling installs to `~/.sling/bin`. Set `SLING_INSTALL` to override the prefix, or pass a version: `bash -s -- v1.5.17`.

#### One-liner on Windows (PowerShell)

```powershell
irm https://slingdata.io/install.ps1 | iex

# You're good to go!
sling -h
```

The script installs to `$HOME\.sling\bin` and updates your user `PATH`. Set `$env:SLING_INSTALL` before running to override.

#### Brew on Mac

Follow these [directions](https://brew.sh/) to install HomeBrew for mac if not already installed.

```shell
# Install with brew first
brew install slingdata-io/sling/sling

# Once, installed, `sling` should be available
sling -h
```

#### Scoop on Windows

```powershell
scoop bucket add sling https://github.com/slingdata-io/scoop-sling.git
scoop install sling

# You're good to go!
sling -h
```

#### Manual binary download (Linux example)

```shell
# download latest binary
curl -LO 'https://github.com/slingdata-io/sling-cli/releases/latest/download/sling_linux_amd64.tar.gz' \
  && tar xf sling_linux_amd64.tar.gz \
  && rm -f sling_linux_amd64.tar.gz \
  && chmod +x sling

# You're good to go!
./sling -h
```

#### Docker

```shell
docker pull slingdata/sling

docker run --rm -i slingdata/sling --help
```

#### Other Binary Downloads

See [Releases](https://github.com/slingdata-io/sling-cli/releases) on Github.

### Setting up your Connections

Sling looks for credentials in several places:

* Environment Variables
* Sling Env File (located at `~/.sling/env.yaml`)
* DBT Profiles Files (located at `~/.dbt/profiles.yml`)

Please see [environment](environment.md) for more details.

## Using Sling

`sling` CLI is designed to be easy to use. Say we want to load a CSV file into a PostgreSQL database. We could run the following command:

{% tabs %}
{% tab title="Linux" %}
{% code overflow="wrap" %}
```bash
export MY_PG='postgresql://user:mypassw@pg.host:5432/db1'

sling run --src-stream file:///path/to/myfile.csv --tgt-conn MY_PG --tgt-object public.my_new_data

# OR pipe it in
cat /path/to/myfile.csv | sling run --tgt-conn MY_PG --tgt-object public.my_new_data
```
{% endcode %}
{% endtab %}

{% tab title="Mac" %}
{% code overflow="wrap" %}
```bash
export MY_PG='postgresql://user:mypassw@pg.host:5432/db1'

sling run --src-stream file:///path/to/myfile.csv --tgt-conn MY_PG --tgt-object public.my_new_data

# OR pipe it in
cat /path/to/myfile.csv | sling run --tgt-conn MY_PG --tgt-object public.my_new_data
```
{% endcode %}
{% endtab %}

{% tab title="Windows" %}
{% code overflow="wrap" %}
```powershell
# using windows Powershell
$env:MY_PG = 'postgresql://user:mypassw@pg.host:5432/db1'
sling run --src-stream file://C:/path/to/myfile.csv --tgt-conn MY_PG --tgt-object public.my_new_data

# OR pipe it in
cat C:\path\to\myfile.csv | sling run --tgt-conn MY_PG --tgt-object public.my_new_data
```
{% endcode %}
{% endtab %}

{% tab title="Docker" %}
{% code overflow="wrap" %}
```bash
export MY_PG='postgresql://user:mypassw@pg.host:5432/db1'

docker run --rm -i -e MY_PG -v /path/to/myfile.csv slingdata/sling run --src-stream file:///path/to/myfile.csv --tgt-conn MY_PG --tgt-object public.my_new_data
```
{% endcode %}
{% endtab %}
{% endtabs %}


### With Python

If you have Python `pip` installed, you can simply run:

```bash
pip install sling
```

You call also check out the [Python wrapper](https://github.com/slingdata-io/sling-python) library on github.

```python
from sling import Replication, ReplicationStream

replication = Replication(
  source='MY_PG',
  target='MY_AWS_S3',
  steams={
    "my_table": ReplicationStream(
      sql="select * from my_table",
      object='my_folder/new_file.csv',
    ),
  }
)

replication.run()
```

<!-- source: sling-docs/sling-cli/commands.md -->
# CLI Commands

## CLI Commands

Sling CLI provides several powerful commands to manage data operations, connections, and workflows. Here's a comprehensive overview of the available commands.

## Core Commands

### `sling run`
The primary command for executing data movement operations. It allows you to perform extractions and loads between different sources and targets. The command supports various modes including full refresh, incremental, and snapshot replications.

Example:
```bash
cat my_file.csv | sling run --tgt-conn MYDB --tgt-object my_schema.my_table
```

### `sling conns`
Manages connection configurations for various data sources and targets. This command allows you to set up, list, test, and remove connection credentials for databases, file storage systems, and other supported platforms.

Example:
```bash
sling conns set POSTGRES type=postgres host=localhost database=mydb user=myuser
```

### sling agent
Controls Sling agents, which are worker processes that execute data operations. This command helps manage agent deployment, configuration, and monitoring in your infrastructure.

### sling project
Handles project-level operations and configurations. Projects help organize and manage related data operations and their configurations.

## Command Structure

Each Sling command follows a consistent structure:
```bash
sling <command> <subcommand> [options]
```

Where:
- `command` is the main operation category (run, conns, agent, project)
- `subcommand` specifies the specific action to perform
- `options` are additional parameters to customize the operation

## Common Options

Most Sling commands support these general options:
- `--help`: Display help information for any command
- `--verbose`: Enable detailed logging output
- `--quiet`: Suppress non-essential output
- `--version`: Show the Sling CLI version

## Environment Variables

Sling CLI respects several environment variables for configuration:
- Connection strings (e.g., `POSTGRES`, `MYSQL`, `SNOWFLAKE`)
- AWS credentials (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`)
- Authentication tokens (`MOTHERDUCK_TOKEN`, `GC_KEY_BODY`)

## Configuration Files

Sling CLI can also read configurations from:
- `env.yaml`: Contains connection configurations and environment settings
- `.env`: Traditional environment variable file
- Project configuration files for specific workflow settings

## Best Practices

1. **Connection Management**
   - Store sensitive credentials securely
   - Use environment variables for production deployments
   - Test connections before running operations

2. **Data Operations**
   - Start with small datasets when testing new configurations
   - Use appropriate replication modes based on data volume
   - Monitor operations for performance and errors

3. **Project Organization**
   - Group related operations within projects
   - Maintain consistent naming conventions
   - Document configurations and workflows

## Getting Help

For detailed information about any command, use the help option:
```bash
sling <command> --help
sling <command> <subcommand> --help
```

For additional support:
- Join the Discord community
- Contact support@slingdata.io
- Visit the documentation website
- Open issues on GitHub

<!-- source: sling-docs/sling-cli/conns.md -->
# Conns

Here's a structured summary of the `sling conns` command:

## Connection Management with `sling conns`

The `sling conns` command provides comprehensive functionality for managing database and storage connections in Sling. It allows users to set up, test, list, and execute queries against various data sources.

## Core Functionality

### Setting Connections
Users can create and modify connections using the `set` subcommand. Each connection requires a unique name and specific configuration parameters based on the connection type. For example:
```bash
sling conns set POSTGRES type=postgres host=localhost database=mydb user=myuser
```

### Testing Connections
The `test` subcommand verifies if a connection is properly configured and accessible:
```bash
sling conns test CONNECTION_NAME
```

### Listing Connections
Users can view all configured connections using the `list` subcommand, which displays connection names and their associated types.

### Query Execution
The `exec` subcommand allows users to run SQL queries against database connections:
```bash
sling conns exec CONNECTION_NAME "SELECT * FROM table"
```

## Supported Connection Types

### Databases
- Relational Databases: PostgreSQL, MySQL, Oracle, SQL Server, MariaDB
- Data Warehouses: Snowflake, BigQuery, Redshift
- NoSQL Databases: MongoDB
- Specialized Databases: DuckDB, ClickHouse, SQLite
- Time-Series Databases: Prometheus

### Cloud Storage
- AWS S3
- Azure Storage
- Google Cloud Storage
- Other S3-Compatible Storage: Cloudflare R2, DigitalOcean Spaces, MinIO

## Connection Configuration

### Common Parameters
Most database connections support standard parameters:
- `host`: Server hostname or IP address
- `port`: Connection port
- `user`: Username for authentication
- `password`: Authentication password
- `database`: Target database name
- `schema`: Default schema (where applicable)

### Security Features
- SSH Tunnel support via `ssh_tunnel` parameter
- TLS/SSL configuration options
- Support for various authentication methods (password, key-based, token-based)

### Advanced Options
- Connection pooling configuration
- Timeout settings
- Custom parameters specific to each database type
- Support for connection URLs as an alternative to individual parameters

## Best Practices

1. **Security**
   - Store sensitive credentials securely using environment variables
   - Use appropriate authentication methods for production environments
   - Implement least-privilege access for database users

2. **Configuration**
   - Test connections before using them in production workflows
   - Use meaningful connection names for better organization
   - Document connection configurations for team reference

3. **Maintenance**
   - Regularly verify connection health
   - Update credentials before expiration
   - Monitor connection usage and performance

For additional support or troubleshooting, users can reach out through:
- Email: support@slingdata.io
- Discord community
- GitHub Issues: https://github.com/slingdata-io/sling-cli/issues

<!-- source: sling-docs/sling-cli/run.md -->
# Running Sling

## Running Sling

The `sling run` command is the primary mechanism for executing data movement operations in Sling CLI. It provides a flexible interface for transferring data between various sources and targets, with support for different replication modes and configuration options.

There are 2 primary ways to configure and run sling, using:

* [**CLI Flags**](run.md#cli-flags-overview): quick ad-hoc runs from your terminal shell or script.
* [**Replication**](../concepts/replication.md): streams defined in a YAML or JSON file.

***

Furthermore, you'll find plenty of examples on how to use Sling:

* [Database to Database](run/examples/database-to-database.md)
* [Database to File](run/examples/database-to-file.md)
* [File to Database](run/examples/file-to-database.md)

## CLI Flags Overview

For quickly running ad-hoc operations from the terminal, using CLI flags is often best. Here are some examples using:

{% code overflow="wrap" %}
```bash
# Load all tables in a schema in with 3 threads
$ export SLING_THREADS=3
$ sling run \
    --src-conn MY_SOURCE_DB \
    --src-stream 'source_schema.*' \
    --tgt-conn MY_TARGET_DB \
    --tgt-object 'target_schema.{stream_table}'
    --mode full-refresh
    
# Pipe in your json file and flatten the nested keys into their own columns
$ cat /tmp/my_file.json | sling run --src-options '{"flatten": "true"}' --tgt-conn MY_TARGET_DB --tgt-object 'target_schema.target_table' --mode full-refresh

# Read folder containing many CSV files
$ sling run \
    --src-stream 'file:///tmp/my_csv_folder/' \
    --tgt-conn MY_TARGET_DB --tgt-object 'target_schema.target_table' \
    --mode full-refresh

# Load only latest data from one source DB to another.
$ sling run \
    --src-conn MY_SOURCE_DB \
    --src-stream 'source_schema.source_table' \
    --tgt-conn MY_TARGET_DB \
    --tgt-object 'target_schema.target_table' \
    --mode incremental \
    --primary-key 'id' --update-key 'last_modified_dt' 

# Export / Backup database tables to JSON files
$ sling run \
    --src-conn MY_SOURCE_DB \
    --src-stream 'source_schema.source_table' \
    --tgt-conn MY_S3_BUCKET \
    --tgt-object 's3://my-bucket/my_json_folder/' \
    --tgt-options '{"file_max_rows": 100000, "format": "jsonlines"}'
```
{% endcode %}

## Interface Specifications

<table data-full-width="false"><thead><tr><th width="215.4264102691841">CLI Flag</th><th>Description</th></tr></thead><tbody><tr><td><code>--src-conn</code></td><td>The source database connection (name, conn string or URL).</td></tr><tr><td><code>--tgt-conn</code></td><td>The target database connection (name, conn string or URL).</td></tr><tr><td><code>--src-stream</code></td><td>The source table (schema.table), local / cloud file path. Can also be the path of sql file or in-line text to use as query. Use <code>file://</code> for local paths.</td></tr><tr><td><code>--tgt-object</code></td><td>The target table (schema.table) or local / cloud file path. Use <code>file://</code> for local paths. See <a href="../concepts/replication/runtime-variables.md">here</a> for details on runtime variables.</td></tr><tr><td><code>--mode</code></td><td>The target load <a href="../concepts/replication/modes.md">mode</a> to use: <code>incremental</code>, <code>truncate</code>, <code>full-refresh</code>, <code>backfill</code> or <code>snapshot</code>. Default is <code>full-refresh</code>.</td></tr><tr><td><code>--primary-key</code></td><td>The column(s) to use as primary key (for <code>incremental</code> mode). If composite key, use a comma-delimited string.</td></tr><tr><td><code>--update-key</code></td><td>The column to use as update key (for <code>incremental</code> mode).</td></tr><tr><td><code>--src-options</code></td><td>In-line options to further configure source (JSON or YAML). See <a href="../concepts/replication/source-options.md">here</a> for details.</td></tr><tr><td><code>--tgt-options</code></td><td>In-line options to further configure target (JSON or YAML). See <a href="../concepts/replication/target-options.md">here</a> for details.</td></tr><tr><td><code>--stdout</code></td><td>Output the stream to standard output (STDOUT).</td></tr><tr><td><code>--select</code></td><td>Select or exclude specific columns from the source stream. (comma separated). Use <code>-</code> prefix to exclude.</td></tr><tr><td><code>--transforms</code></td><td>An object/map, or array/list of built-in transforms to apply to records (JSON or YAML).</td></tr><tr><td><code>--columns</code></td><td>An object/map to specify the type that a column should be cast as (JSON or YAML).</td></tr><tr><td><code>--streams</code></td><td>Only run specific streams from a replication (comma separated). See <a href="../concepts/replication/tags-wildcards.md">here</a> for details.</td></tr></tbody></table>

## Features

* **Flexible Data Sources**: Supports databases, files, cloud storage, and standard input
* **Multiple Load Modes**: Includes full refresh, incremental, snapshot, and truncate modes
* **Data Transformations**: Allows column selection, type casting, and custom transformations
* **Progress Tracking**: Monitors row counts, bytes transferred, and constraint violations
* **Error Handling**: Provides detailed error reporting and validation

The `sling run` command is designed to be both powerful and flexible, accommodating various data movement scenarios while maintaining ease of use through consistent parameter patterns and comprehensive documentation.

<!-- source: sling-docs/sling-cli/environment.md -->
# Environment

## Environment

Sling looks for connection credentials in several places:

* [Sling Env File](environment.md#sling-env-file-env.yaml) (located at `~/.sling/env.yaml`)
* [Project Env File](environment.md#project-env-file-.env.sling) (`.env.sling` in the current working directory)
* [DBT Profiles Files](environment.md#dbt-profiles-dbt-profiles.yml) (located at `~/.dbt/profiles.yml`)
* [Environment Variables](environment.md#environment-variables)

One of the easiest ways is to manage your connections is to use the `sling conns` sub-command. Follow in the next section.

## Managing Connections

Sling makes it easy to **set**, **list** and **test** connections. You can even see the available streams in a connection by using the **discover** sub-command.

```bash
$ sling conns -h
conns - Manage and interact with local connections

See more details at https://docs.slingdata.io/sling-cli/

  Usage:
    conns [discover|list|set|test]

  Subcommands:
    discover   list available streams in connection
    list       list local connections detected
    test       test a local connection
    unset      remove a connection from the sling env file
    set        set a connection in the sling env file
    exec       execute a SQL query on a Database connection

  Flags:
       --version   Displays the program version string.
    -h --help      Displays help with available flag, subcommand, and positional value parameters.
```

### Set Connections

Here we can easily set a connection with the `sling conns set` command and later refer to them by their name. This ensures credentials are not visible by other users when using process monitors, for example.

{% code overflow="wrap" %}
```bash
# set a connection by providing the key=value pairs
$ sling conns set AWS_S3 type=s3 bucket=sling-bucket access_key_id=ACCESS_KEY_ID secret_access_key="SECRET_ACCESS_KEY"

# we set a database connection with just the url
$ sling conns set MY_PG url='postgresql://postgres:myPassword@pghost:5432/postgres'
```
{% endcode %}

To see what credential keys are necessary/accepted for each type of connector, click below:

* File/Storage Connections (see [here](../connections/file-connections/))
* Database Connections (see [here](../connections/database-connections/))

### List Connections

Once connections are set, we can run the `sling conns list` command to list our detected connections:

```bash
$ sling conns list
+--------------------------+-----------------+-------------------+
| CONN NAME                | CONN TYPE       | SOURCE            |
+--------------------------+-----------------+-------------------+
| AWS_S3                   | FileSys - S3    | sling env yaml    |
| FINANCE_BQ               | DB - BigQuery   | sling env yaml    |
| DO_SPACES                | FileSys - S3    | sling env yaml    |
| LOCALHOST_DEV            | DB - PostgreSQL | dbt profiles yaml |
| MSSQL                    | DB - SQLServer  | sling env yaml    |
| MYSQL                    | DB - MySQL      | sling env yaml    |
| ORACLE_DB                | DB - Oracle     | env variable      |
| MY_PG                    | DB - PostgreSQL | sling env yaml    |
+--------------------------+-----------------+-------------------+
```

### Test Connections

We can also test a connection by running the `sling conns test` command:

```bash
$ sling conns test LOCALHOST_DEV
9:04AM INF success!
```

### Discover Connections

We can easily discover streams available in a connection with the `sling conns discover` command:

```bash
$ sling conns discover postgres --pattern public.work*
+---+--------+-------------------+-------+---------+
| # | SCHEMA | NAME              | TYPE  | COLUMNS |
+---+--------+-------------------+-------+---------+
| 1 | public | worker_heartbeats | table |      14 |
| 2 | public | workers           | table |      20 |
| 3 | public | workspaces        | table |       9 |
+---+--------+-------------------+-------+---------+

$ sling conns discover aws_s3
+---+------------------+-----------+---------+-------------------------------+
| # | NAME             | TYPE      | SIZE    | LAST UPDATED (UTC)            |
+---+------------------+-----------+---------+-------------------------------+
| 1 | logging/         | directory | -       | -                             |
| 2 | sling_test/      | directory | -       | -                             |
| 3 | work/            | directory | -       | -                             |
| 4 | temp/            | directory | -       | -                             |
| 5 | records.json     | file      | 442 KiB | 2022-12-07 11:05:01 (1y ago)  |
| 6 | test.sqlite.db   | file      | 4.8 MiB | 2022-12-14 21:00:48 (1y ago)  |
| 7 | test1.parquet    | file      | 48 KiB  | 2024-03-31 22:54:52 (29d ago) |
| 8 | test_1000.csv    | file      | 99 KiB  | 2024-02-23 09:53:13 (67d ago) |
+---+------------------+-----------+---------+-------------------------------+
```

Show column level information:

```bash
$ sling conns discover postgres -p public.workspaces --columns
+----------+--------+------------+----+--------------+--------------------------+--------------+
| DATABASE | SCHEMA | TABLE      | ID | COLUMN       | NATIVE TYPE              | GENERAL TYPE |
+----------+--------+------------+----+--------------+--------------------------+--------------+
| postgres | public | workspaces |  1 | id           | bigint                   | bigint       |
| postgres | public | workspaces |  2 | account_id   | bigint                   | bigint       |
| postgres | public | workspaces |  3 | name         | text                     | text         |
| postgres | public | workspaces |  4 | short_name   | varchar                  | string       |
| postgres | public | workspaces |  5 | token        | text                     | text         |
| postgres | public | workspaces |  6 | settings     | jsonb                    | json         |
| postgres | public | workspaces |  7 | created_dt   | timestamp with time zone | timestampz   |
| postgres | public | workspaces |  8 | updated_dt   | timestamp with time zone | timestampz   |
| postgres | public | workspaces |  9 | deleted_dt   | timestamp with time zone | timestampz   |
+----------+--------+------------+----+--------------+--------------------------+--------------+

$ sling conns discover aws_s3 -p test1.parquet --columns
+---------------------------------+----+------------------+----------------+--------------+
| FILE                            | ID | COLUMN           | NATIVE TYPE    | GENERAL TYPE |
+---------------------------------+----+------------------+----------------+--------------+
| s3://my-bucket/test1.parquet    |  1 | id               | INT_64         | bigint       |
| s3://my-bucket/test1.parquet    |  2 | first_name       | UTF8           | string       |
| s3://my-bucket/test1.parquet    |  3 | last_name        | UTF8           | string       |
| s3://my-bucket/test1.parquet    |  4 | email            | UTF8           | string       |
| s3://my-bucket/test1.parquet    |  5 | target           | BOOLEAN        | bool         |
| s3://my-bucket/test1.parquet    |  6 | create_dt        | Timestamp      | datetime     |
| s3://my-bucket/test1.parquet    |  7 | date             | Timestamp      | datetime     |
| s3://my-bucket/test1.parquet    |  8 | rating           | DECIMAL        | decimal      |
| s3://my-bucket/test1.parquet    |  9 | code             | DECIMAL        | decimal      |
| s3://my-bucket/test1.parquet    | 10 | json_data        | UTF8           | string       |
| s3://my-bucket/test1.parquet    | 11 | _sling_loaded_at | INT_64         | bigint       |
+---------------------------------+----+------------------+----------------+--------------+
```

## Credentials Location

### Sling Env File (`env.yaml`)

The Sling Env file is the primary way sling reads connections globally. It needs to be saved in the path `~/.sling/env.yaml` where the `~` denotes the path of the user Home folder, which can have different locations depending on the operating system (see [here for Windows](https://stackoverflow.com/a/42966089/2295355), [here for Mac](https://apple.stackexchange.com/a/51282) and [here for Linux](https://www.linuxshelltips.com/find-user-home-directory-linux/)). Sling automatically creates the `.sling` folder in the user home directory, which is typically as shown below:

* Linux: `/home/<username>/.sling`, or `/root/.sling` if user is `root`
* Mac: `/Users/<username>/.sling`
* Windows: `C:\Users\<username>\.sling`

Once in the user home directory, setting the Sling Env File (named `env.yaml`) is easy, and adheres to the structure below. Running `sling` the first time will auto-create it. You can alternatively provide the environment variable `SLING_HOME_DIR`.

To see what credential keys are necessary/accepted for each type of connector, click below:

* File/Storage Connections (see [here](../connections/file-connections/))
* Database Connections (see [here](../connections/database-connections/))

```yaml
# Holds all connection credentials for Extraction and Loading
connections:
  marketing_pg:
    url: 'postgres://...' 
    ssh_tunnel: 'ssh://...' # optional
  
  # or dbt profile styled
  marketing_pg:
    type: postgres        
    host: [hostname]      
    user: [username]      
    password: ${PASSWORD} # you can pass in environment variables as well
    port: [port]          
    dbname: [database name]
    schema: [dbt schema]  
    ssh_tunnel: 'ssh://...' 
  
  finance_bq:
    type: bigquery
    method: service-account
    project: [GCP project id]
    dataset: [the name of your dbt dataset]
    keyfile: [/path/to/bigquery/keyfile.json]

# Global variables for specific settings, available to all connections at runtime (Optional)
variables:
  SLING_CLI_TOKEN: xxxxxxxxxxxxxxxx  # picked up machine wide
  SLING_LOG_DIR: ~/.sling/logs       # write debug logs here
  aws_access_key: '...'
  aws_secret_key: '...'
```

### Dot-Env File (`.env.sling`)

Sling automatically loads a `.env.sling` file from the **current working directory** when it starts. This lets you define per-project connections and variables without modifying the global `env.yaml`. This applies to version *1.5.10+*.

This is useful when you have multiple projects, each with their own connections or credentials. Simply place a `.env.sling` file in the project directory, and Sling will pick it up automatically when run from that directory.

The file uses a simple `KEY=VALUE` format (one per line). Comments and blank lines are supported. Values can optionally be wrapped in single or double quotes.

```bash
# .env.sling - project-specific connections

# Database connections
MY_PG='postgresql://user:password@localhost:5432/mydb'
STAGING_PG='postgresql://user:password@staging-host:5432/mydb'

# File storage
MY_S3='{type: s3, bucket: my-project-bucket, access_key_id: AKID, secret_access_key: SECRET}'

# API Connection
SALESFORCE='{ type: api, spec: salesforce, secrets: { client_id: "xxxxxxxx", client_secret: "xxxxxxxx", instance: "mycompany.my.salesforce.com" } }'

# Other variables
SLING_LOG_DIR=/tmp/logs
SLING_LOADED_AT_COLUMN=timestamp
SLING_CLI_TOKEN=xxxxxxxxxxxxxxxx
```

{% hint style="info" %}

**Existing environment variables are not overwritten.** If a variable is already set in the shell environment, the value from `.env.sling` will be ignored. This means shell-level overrides always take precedence.

{% endhint %}

{% hint style="warning" %}

Since `.env.sling` may contain sensitive credentials, make sure to add it to your `.gitignore` to avoid accidentally committing secrets to version control.

{% endhint %}

### Environment Variables

Sling also reads environment variables. Simply export a connection URL (or YAML payload) to the current shell environment to use them.

To see examples of setting environment variables for each type of connector, click below:

* File/Storage Connections (see [here](../connections/file-connections/))
* Database Connections (see [here](../connections/database-connections/))

{% tabs %}
{% tab title="Mac / Linux" %}
{% code overflow="wrap" %}
```bash
$ export MY_PG='postgresql://user:mypassw@pg.host:5432/db1'
$ export MY_PG='{type: postgres, host: "pg.host", user: user, database: "db1", password: "mypassw", port: 5432}'

$ export MY_SNOWFLAKE='snowflake://user:mypassw@sf.host/db1'
$ export MY_SNOWFLAKE='{type: snowflake, host: "<host>", user: "<user>", database: "<database>", password: "<password>", role: "<role>"}'

$ export ORACLE_DB='oracle://user:mypassw@orcl.host:1521/db1'

$ export BIGQUERY_DB='{type: bigquery, dataset: public, key_file: /path/to/service.json, project: my-google-project}' # yaml or json form is also accepted

$ sling conns list
+---------------+------------------+-----------------+
| CONN NAME     | CONN TYPE        | SOURCE          |
+---------------+------------------+-----------------+
| MY_PG         | DB - PostgreSQL  | env variable    |
| MY_SNOWFLAKE  | DB - Snowflake   | env variable    |
| ORACLE_DB     | DB - Oracle      | env variable    |
| BIGQUERY_DB   | DB - Big Query   | env variable    |
+---------------+------------------+-----------------+
```
{% endcode %}
{% endtab %}

{% tab title="Windows" %}
{% code overflow="wrap" %}
```powershell
$ $env:MY_PG='postgresql://user:mypassw@pg.host:5432/db1'
$ $env:MY_PG='{type: postgres, host: "pg.host", user: user, database: "db1", password: "mypassw", port: 5432}'

$ $env:MY_SNOWFLAKE='snowflake://user:mypassw@sf.host/db1'
$ $env:MY_SNOWFLAKE='{type: snowflake, host: "sf.host", user: user, database: db1, password: "mypassw", role: "<role>"}'

$ $env:ORACLE_DB='oracle://user:mypassw@orcl.host:1521/db1'

$ $env:BIGQUERY_DB='{type: bigquery, dataset: public, key_file: /path/to/service.json, project: my-google-project}' # yaml or json form is also accepted

$ sling conns list
+---------------+------------------+-----------------+
| CONN NAME     | CONN TYPE        | SOURCE          |
+---------------+------------------+-----------------+
| MY_PG         | DB - PostgreSQL  | env variable    |
| MY_SNOWFLAKE  | DB - Snowflake   | env variable    |
| ORACLE_DB     | DB - Oracle      | env variable    |
+---------------+------------------+-----------------+
```
{% endcode %}
{% endtab %}
{% endtabs %}

### DBT Profiles (`~/dbt/profiles.yml`)

Sling also reads dbt profiles connections! If you're already set up with dbt cli locally, you don't need to create additional duplicate connections.

See [here](https://docs.getdbt.com/dbt-cli/configure-your-profile) for more details.

```bash
$ sling conns list
+------------------+------------------+-------------------+
| CONN NAME        | CONN TYPE        | SOURCE            |
+------------------+------------------+-------------------+
| SNOWCASTLE_DEV   | DB - Snowflake   | dbt profiles yaml |
| SNOWCASTLE_PROD  | DB - Snowflake   | dbt profiles yaml |
+------------------+------------------+-------------------+
```

## Location String

The location string is a way to describe where sling should look for a file object or database object. It is used in a few places, such as the [`SLING_STATE`](../sling-cli/cli-pro.md#file--state-based-incremental-loading) env var, as well as [Hooks](../concepts/hooks.md) such as [`delete`](../concepts/hooks/delete.md), [`copy`](../concepts/hooks/copy.md) and [`inspect`](../concepts/hooks/inspect.md). 

The proper input format is `CONN_NAME/path/to/key` for storage connections, or `CONN_NAME/[database.]schema.table` for database objects.

Local location examples:
- `local/relative/path`
- `local/../parent/relative/path`
- `local//absolute/linux/path`
- `local/C:/absolute/windows/path`

For cloud or remote storage connections (with defined `AWS_S3`, `GCP`, `AZURE`, `SFTP` connections):

- `aws_s3/path/to/folder`
- `gcp/path/to/folder/file.parquet`
- `azure/path/to/folder/file.log`
- `sftp/relative/path/to/folder/file.log`
- `sftp//absolute/path/to/folder/file.log` 

Database location examples (for hooks like `inspect`):

- `postgres/public.users` - PostgreSQL table in public schema
- `mysql_db/analytics.events` - MySQL table in analytics schema  
- `snowflake/DATABASE.SCHEMA.TABLE` - Snowflake table with explicit database
- `bigquery/project.dataset.table_name` - BigQuery table
- `oracle_db/HR.EMPLOYEES` - Oracle table in HR schema
- `mssql/dbo.customers` - SQL Server table in dbo schema

{% hint style="warning" %}

For **file storage connections** (`local`, `ftp` and `sftp`), you can specify a relative or absolute path. For FTP connections, it will be relative to the default folder of the username connecting.

**Relative Path**: You use the typical single slash (`/`) after the connection name:

- `local/relative/path`
- `sftp/relative/path`
- `ftp/relative/path`

**Absolute Path**: You need to add 2 slashes (`//`) after the connection name:

- `local//absolute/path`
- `local/C:/absolute/path`
- `sftp//absolute/path`
- `ftp//absolute/path`

For **database connections**, use the standard database object naming convention: `connection_name/[database.]schema.table_name`

{% endhint %}

<!-- source: sling-docs/sling-cli/variables.md -->
# Global Variables

## Global Variables

Sling utilizes the following global environment variables to further configure the load behavior. You can simply define them in your environment, the `env.yaml` file or the `env` section in a task or replication.

<table><thead><tr><th width="295">Variable Name</th><th>Description</th></tr></thead><tbody><tr><td><code>SLING_HOME_DIR</code></td><td>The sling home directory, which contains <code>env.yaml</code>. Will use <a href="../environment.md#sling-env-file-env.yaml">default</a> if not provided.</td></tr><tr><td><code>SLING_LOADED_AT_COLUMN</code></td><td>Whether to add an audit timestamp column named <code>_sling_loaded_at</code> in target object. Accepts values <code>true</code>, <code>false</code>, <code>unix</code> (for <a href="https://www.epochconverter.com/">epoch</a> integer values) or <code>timestamp</code>. <code>true</code> defaults to <code>unix</code>.</td></tr><tr><td><code>SLING_SYNCED_AT_COLUMN</code></td><td>Whether to add sync tracking columns <code>_sling_synced_at</code> (timestamp) and <code>_sling_synced_op</code> (operation type) in target object. The <code>_sling_synced_op</code> column tracks the last operation: <code>I</code> = Insert, <code>U</code> = Update, <code>D</code> = Delete (soft delete). This is useful when streaming from staging to multiple destinations to track when records were last touched. To enable, set to <code>true</code>. When enabled, <code>_sling_synced_at</code> replaces <code>_sling_deleted_at</code> for soft deletes.</td></tr><tr><td><code>SLING_STREAM_URL_COLUMN</code></td><td>If source is file, whether to add a column <code>_sling_stream_url</code> with the source file path / url in target object. To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_TIMEOUT</code></td><td>The maximum number of minutes the sling replication should run. Once reached, it will kill the process. To enable, set a number (<code>SLING_TIMEOUT=10.5</code>)</td></tr><tr><td><code>SLING_RECURSIVE_LIMIT</code></td><td>The number limit of file names to pull, when listing from cloud file systems such as S3, GCP and Azure Storage.</td></tr><tr><td><code>SLING_ROW_ID_COLUMN</code></td><td>Whether to add a column named <code>_sling_row_id</code> in the target object, which will have a random UUIDv7 value. This will be unique. To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_ROW_NUM_COLUMN</code></td><td>If source is file, whether to add a column named <code>_sling_row_num</code> in the target object, which will be the row number of the stream (incremented by record processed). To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_EXEC_ID_COLUMN</code></td><td>Whether to add a column named <code>_sling_exec_id</code> in the target object, which will have the run / execution string (a random UUIDv7 value). This will be unique per run. To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_STATE</code></td><td>The <a href="environment.md#location-string">location</a> to read/write information such as incremental values. Proper input format is <code>CONN_NAME/key</code>. For example: <code>POSTGRES/sling_state.state_table</code> , <code>AWS_S3/my/folder</code> or <code>MY_SFTP/my/folder</code>. </td></tr><tr><td><code>SLING_ALLOW_EMPTY</code></td><td>This is useful to create tables / files using the stream columns structure, even if there is no data. To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_DIRECT_INSERT</code></td><td>Tells sling to insert directly into the final table (not create a temp table before). To enable, set to <code>true</code>.</td></tr><tr><td><code>SLING_THREADS</code></td><td>sets the maximum number of concurrent stream runs. Accepts an integer value, default is <code>1</code>.</td></tr><tr><td><code>SLING_RETRIES</code></td><td>sets the maximum number of retries for a failed stream run. Accepts an integer value, default is <code>0</code>.</td></tr><tr><td><code>SLING_KEEP_TEMP</code></td><td>Tells sling to keep any temporary files or tables created in the load process. To enable, set to <code>true</code></td></tr><tr><td><code>SLING_ENV_YAML</code></td><td>Provide the body of the <code>env.yaml</code> file as an environment variable.</td></tr><tr><td><code>SLING_DISABLE_TELEMETRY</code></td><td>this disables any anonymous usage reporting. These are used to improve sling. To disable, set this to <code>true</code>.</td></tr><tr><td><code>SLING_SHOW_PROGRESS</code></td><td>Whether the progress of the stream should be displayed (<code>true</code> or <code>false</code>).</td></tr><tr><td><code>SLING_LOGGING</code></td><td>How sling formats the log lines. Accepts values <code>JSON</code>, <code>NO_COLOR</code> or <code>CONSOLE</code> (default).</td></tr><tr><td><code>SLING_LOG_DIR</code></td><td>Directory for automatic date-based debug log files. When set, Sling creates a <code>sling_debug_YYYY_MM_DD.log</code> file in the specified directory and automatically cleans up old log files, keeping the latest 15. Supports <code>~</code> for home directory (e.g., <code>~/.sling/logs</code>).</td></tr><tr><td><code>SLING_SAMPLE_SIZE</code></td><td>The number of records to process in order to infer column types (especially for file sources). Default is <code>900</code>.</td></tr><tr><td><code>SLING_DUCKDB_COMPUTE</code></td><td>Whether to use DuckDB for writing to parquet files and partitioned parquet/CSV files. DuckDB provides optimized performance for these formats. To disable DuckDB compute, set to <code>false</code>. Default is <code>true</code>.</td></tr><tr><td><code>SLING_OTEL_ENDPOINT</code></td><td>The OpenTelemetry HTTP endpoint URL to export logs to (e.g., <code>http://otel.host.ip:4318/v1/logs</code>). When set, Sling will send structured logs to the specified OTLP endpoint with execution attributes including <code>project_id</code>, <code>job_id</code>, and <code>exec_id</code> if configured. Requires <a href="cli-pro.md">CLI Pro</a>.</td></tr><tr><td><code>SLING_PROXY</code></td><td>Route database connections through a SOCKS5 proxy. Accepts a SOCKS5 URL (e.g., <code>socks5://user:pass@host:1080</code>). To proxy a specific connection instead, set <code>proxy_url</code> in its <code>env.yaml</code> connection entry. Useful for reaching databases behind VPNs or private networks (e.g., Tailscale). Available in v1.5.9+</td></tr></tbody></table>

<!-- source: sling-docs/sling-cli/project.md -->
# sling project

## sling project

The `sling project` command helps manage projects on the Sling Data Platform. It provides functionality to initialize, monitor, and synchronize project files and configurations.

## Core Commands

### `sling project init`
Initializes a new Sling project in the current directory or links to an existing one. This creates a `.sling.json` configuration file that defines the project's root directory and any nested paths to monitor.

### `sling project status`
Displays detailed information about the current project, including:
- Project ID and name
- Organization details
- Owner information
- Project folder location
- Configured project paths
- Additional status details from the platform

### `sling project sync`
Synchronizes local project files with the Sling Data Platform. Features include:
- Detects new and modified files
- Shows a preview of changes before pushing
- Supports force push with `-f` or `--force` flag
- Maintains file modification timestamps
- Provides detailed sync status and confirmation prompts

### `sling project jobs`

#### List Jobs (`sling project jobs list`)
Displays all jobs configured in the project with details such as:
- Job ID
- Name
- File name
- Status
- Active state
- Execution history
- Schedule information

#### Job History (`sling project jobs history`)
Shows the execution history of jobs including:
- Execution ID
- Job name
- Start time
- Row counts
- Data volume
- Execution status

#### Trigger Jobs (`sling project jobs trigger <job-id>`)
Manually triggers a specific job execution using its job ID. Returns an execution ID that can be used to track the job's progress.

## Configuration

Projects are configured using a `.sling.json` file which contains:
- Project ID
- Root directory path
- Nested paths for file monitoring
- Platform endpoint configurations

## Authentication

The command requires a valid Sling Project Token for authentication, which should be set in the `SLING_PROJECT_TOKEN` environment variable. Different features may require specific plan levels.

## Best Practices

1. Initialize projects at the root of your repository
2. Regularly sync changes to maintain consistency
3. Use nested paths to organize different components
4. Monitor job executions through the history command
5. Maintain proper authentication tokens and permissions

For additional details and examples, visit the [Sling Documentation](https://docs.slingdata.io/).

<!-- source: sling-docs/sling-cli/mcp.md -->
# Sling MCP Server

## Sling MCP Server

## Overview

The Sling CLI includes a built-in MCP (Model Context Protocol) server that enables AI assistants like Claude, ChatGPT, and GitHub Copilot to interact with your data infrastructure through a standardized interface. MCP is an open protocol that allows AI models to connect to external tools and data sources safely and efficiently.

By running `sling mcp`, you expose Sling's powerful data movement and transformation capabilities to AI assistants, enabling them to:
- Query and explore databases across 30+ database systems
- Manage files across cloud storage providers (S3, Azure, GCS, etc.)
- Execute data replications and pipelines
- Create and test API specifications
- Discover schemas, tables, and columns

{% embed url="https://f.slingdata.io/videos/mcp.demo.20251201.mp4" %}
Sling MCP Demo
{% endembed %}


## Core Capabilities

The Sling MCP server exposes six main tools that AI assistants can use:

### 1. Connection Tool
Manages connections to databases, file systems, and APIs.

**Actions:**
- `list` - List all configured connections
- `discover` - Discover tables, files, or endpoints in a connection
- `test` - Test connection validity
- `set` - Create or update a connection
- `docs` - Fetch connection documentation

### 2. Database Tool
Provides database-specific operations for querying and schema exploration.

**Actions:**
- `docs` - Fetch database documentation
- `query` - Execute SQL queries (read-only by default)
- `get_schemata` - Get detailed schema information (databases, schemas, tables, columns)
- `get_schemas` - List available schemas
- `get_columns` - Get column metadata for specific tables

### 3. File System Tool
Manages files across local and cloud storage systems.

**Actions:**
- `list` - List files and directories
- `copy` - Copy files between connections
- `inspect` - Get file metadata and statistics
- `docs` - Fetch file system documentation

### 4. API Spec Tool
Creates and manages API specifications for REST APIs.

**Actions:**
- `parse` - Parse and validate API specification files
- `test` - Test API endpoints defined in specifications
- `docs` - Fetch API specification documentation

### 5. Replication Tool
Executes data replication configurations.

**Actions:**
- `parse` - Parse replication YAML files
- `compile` - Compile and validate replications
- `run` - Execute replications
- `docs` - Fetch replication documentation

### 6. Pipeline Tool
Manages and executes data pipelines.

**Actions:**
- `parse` - Parse pipeline configurations
- `run` - Execute pipelines
- `docs` - Fetch pipeline documentation

## Installation

### Prerequisites

1. **Install Sling CLI**: Follow the [installation guide](./getting-started.md)
2. **Verify installation**: Run `sling --version`
3. **Set up connections**: Configure your database and storage connections using [environment variables](./environment.md)


### VSCode with GitHub Copilot

GitHub Copilot in VSCode supports MCP servers through workspace or user configuration:

#### Workspace Configuration
Create `.vscode/mcp.json` in your project root:
```json
{
  "servers": {
    "sling": {
      "type": "stdio",
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "your-token-here"
      }
    }
  }
}
```

#### User Configuration (Global)
1. Open Command Palette (`Cmd+Shift+P` or `Ctrl+Shift+P`)
2. Run `MCP: Add Server`
3. Select "Global"
4. Enter configuration:
```json
{
  "name": "sling",
  "type": "stdio",
  "command": "sling",
  "args": ["mcp"],
  "env": {
    "SLING_CLI_TOKEN": "your-token-here"
  }
}
```

#### Using with Copilot
1. Open Chat view (`Ctrl+Alt+I` or `Cmd+Alt+I`)
2. Select "Agent mode" from the dropdown
3. Click "Tools" button to see available Sling tools
4. Start using Sling commands in your prompts

![Sling MCP on VSCode](image-4.png)

![Sling MCP on VSCode](image-5.png)

![Sling MCP on VSCode](image-6.png)

### Claude Desktop

Claude Desktop supports MCP servers through a configuration file. Here's how to set up Sling:

{% tabs %}
{% tab title="macOS" %}
1. Open the configuration file:
```bash
open ~/Library/Application\ Support/Claude/claude_desktop_config.json
```

2. Add the Sling MCP server configuration:
```json
{
  "mcpServers": {
    "sling": {
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "your-token-here"
      }
    }
  }
}
```

3. Restart Claude Desktop
4. Look for the MCP indicator (🔌) in the bottom-right corner of the chat input
{% endtab %}

{% tab title="Windows" %}
1. Open the configuration file at:
```
%APPDATA%\Claude\claude_desktop_config.json
```

2. Add the Sling MCP server configuration:
```json
{
  "mcpServers": {
    "sling": {
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "your-token-here"
      }
    }
  }
}
```

3. Restart Claude Desktop
4. Look for the MCP indicator (🔌) in the bottom-right corner of the chat input
{% endtab %}
{% endtabs %}

![Sling MCP on Claude Desktop](image-7.png)

![Sling MCP on Claude Desktop](image-8.png)

![Sling MCP on Claude Desktop](image-9.png)

### Claude Code

Claude Code supports MCP servers at three configuration scopes:

#### Local Scope (Project-specific)
```bash
# Add for current project only
claude mcp add sling --args "mcp" --env SLING_CLI_TOKEN=your-token-here
```

#### Project Scope (Shared with team)
Create `.mcp.json` in your project root:
```json
{
  "servers": {
    "sling": {
      "type": "stdio",
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "${SLING_CLI_TOKEN}"
      }
    }
  }
}
```

#### User Scope (Global)
```bash
# Add globally for all projects
claude mcp add sling --scope user --args "mcp" --env SLING_CLI_TOKEN=your-token-here
```

Alternatively, edit `~/.claude.json` directly:
```json
{
  "mcpServers": {
    "sling": {
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "your-token-here"
      }
    }
  }
}
```

### ChatGPT Desktop

{% hint style="warning" %}
**Note:** As of 2025, OpenAI has announced plans to add native MCP support to ChatGPT Desktop, but implementation is pending. Check the [OpenAI Developer Community](https://community.openai.com/) for updates.
{% endhint %}

For now, you can use bridge solutions like the [chatgpt-mcp server](https://github.com/xncbf/chatgpt-mcp) that enables MCP interaction through the ChatGPT macOS app:

```json
{
  "mcpServers": {
    "chatgpt-sling-bridge": {
      "command": "uvx",
      "args": ["chatgpt-mcp"],
      "env": {
        "SLING_MCP_COMMAND": "sling mcp",
        "SLING_CLI_TOKEN": "your-token-here"
      }
    }
  }
}
```


## Usage Examples

### Querying a Database

**Simple Analysis Prompt to AI Assistant:**

{% code overflow="wrap" %}
```
Use sling connection `postgres_prod` to query the sales table in my warehouse connection and show me the top 10 revenue generating products this month
```
{% endcode %}

The assistant will construct and execute:

{% code overflow="wrap" %}
```json
{
  "action": "query",
  "input": {
    "connection": "postgres_prod",
    "query": "SELECT product_id, SUM(revenue) as total_revenue FROM sales WHERE date >= '2025-01-01' GROUP BY product_id ORDER BY total_revenue DESC LIMIT 10"
  }
}

```
{% endcode %}


**Table Comparison Prompt to AI Assistant:**

{% code overflow="wrap" %}
```
Use sling in connection `snowflake_dw` to compare the tables: dbt_dev.core_transactions (dev table) and finance.core_transactions (prod table). Compare the counts and null counts as well as distinct counts.
```
{% endcode %}

The assistant will construct and execute multiple queries and return  a summary.


### Discovering Database Tables

**Prompt to AI Assistant:**

{% code overflow="wrap" %}
```
Using Sling, show me all tables in my `postgres_rds` connection that start with "customer_"
```
{% endcode %}

The assistant will use:
```json
{
  "action": "discover",
  "input": {
    "connection": "postgres_rds",
    "pattern": "*.customer_*"
  }
}
```

### Copying Files Between Storage Systems

**Prompt to AI Assistant:**

{% code overflow="wrap" %}
```
Use sling to copy all CSV files from connection `aws_s3` folder "raw/2025/" to connection `AZURE_PROD` "processed/" folder
```
{% endcode %}

The assistant will execute:
```json
{
  "action": "copy",
  "input": {
    "source_location": "aws_s3/raw/2025/*.csv",
    "target_location": "azure_prod/processed/",
    "recursive": true
  }
}
```

## MCP Prompts

The Sling MCP server provides specialized prompts that guide AI assistants through complex API specification workflows. These prompts are pre-built conversation templates that help with creating, extending, and debugging API integrations.

### api_spec_create_spec

Creates a complete Sling API specification from scratch by analyzing API documentation and building endpoints with authentication, pagination, and data extraction configuration.

**Arguments:**
| Argument | Required | Description |
|----------|----------|-------------|
| `spec_name` | Yes | Name for the API specification (used as filename) |
| `spec_file_path` | No | Full file path for the spec file |
| `connection_name` | Yes | Name for the API connection to create and test |
| `api_docs_url` | Yes | URL to the API documentation website |
| `endpoint_names` | Yes | Comma-separated list of endpoint names to include |
| `additional_info` | No | Additional instructions or requirements |

**Workflow:**
1. Fetches Sling API spec documentation
2. Analyzes the target API documentation (using browser if available)
3. Creates the specification file
4. Creates or uses existing connection
5. Tests and iterates until endpoints work correctly

### api_spec_add_endpoint

Adds a new endpoint to an existing Sling API specification by analyzing endpoint documentation and implementing proper configuration.

**Arguments:**
| Argument | Required | Description |
|----------|----------|-------------|
| `spec_file_path` | Yes | Full file path to the existing spec file |
| `endpoint_name` | Yes | Name of the new endpoint to add |
| `endpoint_docs_url` | No | URL to the specific endpoint documentation |
| `additional_info` | No | Additional instructions for the implementation |

**Workflow:**
1. Fetches Sling API spec documentation
2. Loads and parses the existing specification
3. Analyzes endpoint documentation
4. Implements the new endpoint following existing patterns
5. Tests until the endpoint returns data successfully

### api_spec_debug_endpoint

Debugs and fixes issues with an existing endpoint in a Sling API specification by analyzing errors and adjusting configuration.

**Arguments:**
| Argument | Required | Description |
|----------|----------|-------------|
| `spec_file_path` | Yes | Full file path to the spec file |
| `endpoint_name` | Yes | Name of the endpoint to debug and fix |
| `additional_info` | No | Additional context about the issues |

**Workflow:**
1. Fetches Sling API spec documentation
2. Loads and examines the current specification
3. Analyzes API documentation for verification
4. Tests and diagnoses issues
5. Fixes configuration and iterates until successful

**Common issues checked:**
- Authentication (token format, headers, auth type)
- URL construction (base URLs, parameter encoding)
- Data extraction (JMESPath expressions)
- Pagination (next page logic, stop conditions)
- Rate limiting (request rates, backoff strategies)

## Troubleshooting

### Log Locations

- **Claude Desktop**: `~/Library/Logs/Claude/` (macOS) or `%APPDATA%\Claude\logs\` (Windows)
- **Claude Code**: View logs with `claude mcp logs sling`
- **VSCode**: Use Command Palette > "MCP: Show Logs"

### Debug Output

Enable trace logging for detailed debugging:
```json
{
  "mcpServers": {
    "sling": {
      "command": "sling",
      "args": ["mcp"],
      "env": {
        "SLING_CLI_TOKEN": "your-token-here",
        "DEBUG": "TRACE"
      }
    }
  }
}
```

### Resources

- [Model Context Protocol Documentation](https://modelcontextprotocol.io/)
- [Sling Environment Configuration](./environment.md)
- [Sling Replications Guide](../concepts/replication.md)
- [Sling Pipelines Guide](../concepts/pipeline.md)

{% hint style="success" %}
**Pro Tip:** Start with simple operations like listing connections and discovering tables before moving to complex replications and pipelines. This helps you understand how the AI assistant interacts with your data infrastructure.
{% endhint %}

<!-- source: sling-docs/sling-cli/cli-pro.md -->
# CLI Pro Features

## CLI Pro Features

Sling CLI Pro extends the core functionality with advanced features designed for production environments and complex data operations.

- ✅ [API Sources](#api-sources) (extract data from any REST API by using Specs)
- ✅ [Parallel Stream Processing](#stream-chunking-and-parallel-processing) (run streams in parallel)
- ✅ [Stream Chunking](#stream-chunking-and-parallel-processing) (split large streams into smaller ones)
- ✅ [Pipelines & Hooks](#pipelines-and-hooks) (such as `http`, `query`, `check` and more)
- ✅ [OpenTelemetry Logging](#opentelemetry-logging) (export structured logs to any OTLP endpoint)
- ✅ [Capture Deletes](#capture-deletes) (similar to CDC)
- ✅ [Staged Transforms](#staged-transforms) (advanced multi-stage transformations with expressions and functions)
- ✅ [File Target Incremental Mode](#state-based-incremental-loading)
- ✅ [State Based Incremental](#state-based-incremental-loading)
- ✅ [ODBC Connections](#odbc-connections) (connect to any database via ODBC drivers)
- ✅ Support Sling and its continuous development

**CLI Pro Max Features**

- ✅ [Change Capture (CDC)](#change-capture-cdc) (continuously replicate row-level changes from transaction logs)
- ✅ [Schema Migration](#schema-migration) (migrate primary keys, foreign keys, indexes, defaults, and more)
- ✅ Priority Support (direct access to the Sling team for faster issue resolution)


{% hint style="success" %}

You can obtain a token for free at [https://dash.slingdata.io](https://dash.slingdata.io). There is 7-day trial (no credit card needed).

Once you have a token, just put the value into the `SLING_CLI_TOKEN` environment variable before running sling (make sure the version is 1.4+).

For Pricing details see [here](https://slingdata.io/cli-pro).

{% endhint %}

## API Sources

Extract data from any REST API with powerful YAML-based specifications called `API Specs`:
- Define authentication methods (Bearer, Basic, OAuth2)
- Configure endpoints with pagination strategies
- Process responses with JMESPath extraction
- Manage state for incremental synchronization
- Support for queues and dependent requests
- Built-in retry and error handling

In your `env.yaml`:

```yaml
connections:
  stripe_api:
    type: api
    spec: stripe  # Use official spec or custom YAML (e.g. file://path/to/stripe.spec.yaml)
    secrets:
      api_key: sk_live_xxxxxx
```

In your replication:

```yaml
source: stripe_api
target: ducklake

defaults:
  object: stripe.{stream_name}

streams:
  customers:
    mode: incremental
```

See [API Specs](../concepts/api-specs.md) for complete documentation and examples.

## Stream Chunking & Parallel Processing

Process large datasets efficiently with automatic chunking and parallel execution:
- Break down data into manageable chunks for various modes (`full-refresh`, `truncate`,`incremental`, `backfill`)
- Support for time-based (hours, days, months), numeric, count-based, and expression-based chunks
- Run multiple streams concurrently with automatic retry mechanisms
- Configurable concurrency and retry settings

```yaml
streams:
  my_schema.events:
    mode: full-refresh  # works with various modes
    primary_key: [id]
    update_key: event_date
    source_options:
      chunk_count: 8  # Process in 8 equal sized chunks

  my_schema.orders:
    mode: incremental  # works with various modes
    update_key: order_date
    source_options:
      chunk_size: 7d  # Process in 7-day chunks

env:
  SLING_THREADS: 3   # maximum of 3 streams concurrently
  SLING_RETRIES: 1   # maximum of 1 retry per failed stream
```

Environment variables:
- `SLING_THREADS` sets the maximum number of concurrent stream runs. Accepts an integer value, default is `1`.
- `SLING_RETRIES` sets the maximum number of retries for a failed stream run. Accepts an integer value, default is `0`.

See [Chunking](../examples/database-to-database/chunking.md) for detailed examples.

## Pipelines & Hooks

Extend functionality with hooks and pipelines to create complex workflows. Hooks are used within replications to execute custom logic before/after operations, while Pipelines are standalone workflows that execute multiple steps in sequence.

Available action types:

| Step Type   | Description                                                | Documentation                    |
| ----------- | ---------------------------------------------------------- | -------------------------------- |
| Check       | Validate conditions and control flow                       | [Check Hook](../concepts/hooks/check.md)     |
| Command     | Run any command/process                                    | [Command Hook](../concepts/hooks/command.md)       |
| Copy        | Transfer files between local or remote storage connections | [Copy Hook](../concepts/hooks/copy.md)       |
| Delete      | Remove files from local or remote storage connections      | [Delete Hook](../concepts/hooks/delete.md)   |
| Group       | Run sequences of steps or loop over values                 | [Group Hook](../concepts/hooks/group.md)   |
| HTTP        | Make HTTP requests to external services                    | [HTTP Hook](../concepts/hooks/http.md)       |
| Inspect     | Inspect a file or folder                                   | [Inspect Hook](../concepts/hooks/inspect.md) |
| List        | List files in folder                                       | [List Hook](../concepts/hooks/list.md) |
| Log         | Output custom messages and create audit trails             | [Log Hook](../concepts/hooks/log.md)         |
| Query       | Execute SQL queries against any defined connection         | [Query Hook](../concepts/hooks/query.md)     |
| Read        | Read contents of files from storage connections            | [Read Hook](../concepts/hooks/read.md)       |
| Replication | Run a Replication                                          | [Replication Hook](../concepts/hooks/replication.md) |
| Routine     | Execute reusable step sequences from external files        | [Routine Hook](../concepts/hooks/routine.md) |
| Store       | Store values for later in-process access                   | [Store Hook](../concepts/hooks/store.md)     |
| Write       | Write content to files in storage connections              | [Write Hook](../concepts/hooks/write.md)     |

See [Hooks](../concepts/hooks.md) and [Pipelines](../concepts/pipeline.md) for usage examples and patterns.

## Staged Transforms

Transform data with advanced multi-stage processing using expressions and functions:
- Apply transformations in sequential stages with cross-column references
- Create new columns dynamically without modifying source schemas
- Use 50+ built-in functions for string, numeric, date, and conditional operations
- Build complex logic with `if/then/else` conditions and record references

```yaml
streams:
  customers:
    transforms:
      # Stage 1: Clean and normalize data
      - first_name: "trim_space(value)"
        last_name: "trim_space(value)" 
        email: "lower(value)"
      
      # Stage 2: Create computed columns
      - full_name: 'record.first_name + " " + record.last_name'
        email_hash: 'hash(record.email, "md5")'
      
      # Stage 3: Add business logic
      - customer_type: 'record.total_orders >= 50 ? "vip" : "regular"'
        discount_rate: 'record.customer_type == "vip" ? 0.15 : 0.05'
```

See [Transforms](../concepts/replication/transforms.md) for detailed examples and [Available Functions](../concepts/functions.md) for all available functions.

## State Based Incremental Loading

Maintain state across file & database loads with intelligent incremental processing:
- Track and resume file processing from last successful position
- Support for incremental writes to databases and files
- Automatic file partitioning and truncation management

See [Database to Database Incremental Loading](../examples/database-to-database/incremental.md#using-sling_state), [Database to File Incremental Loading](../examples/database-to-file/incremental.md) and [File to Database Incremental Loading](../examples/file-to-database/incremental.md) for detailed examples.


## Capture Deletes

Track deleted records using a `_sling_deleted_at` column:
- Automatically detect and mark deleted records
- Maintain historical record states
- Support for soft deletes in target systems

See [Delete Missing Records](../examples/database-to-database/capture_deletes.md) for implementation details.

## OpenTelemetry Logging

Export structured logs to any OpenTelemetry-compatible endpoint for centralized logging and observability:

- Send logs to any OTLP HTTP endpoint (Grafana, Datadog, Honeycomb, etc.)
- Automatic enrichment with execution attributes (such as `exec_id`, `stream_name`, `object_name`, `row_count`, `status`, `duration`, etc.)
- Structured log records with severity levels and timestamps
- Seamless integration with existing observability infrastructure

Set the `SLING_OTEL_ENDPOINT` environment variable to enable:

```bash
export SLING_OTEL_ENDPOINT='http://otel-collector:4318/v1/logs'
```

## Schema Migration

Migrate database schema attributes along with your data to preserve structure and relationships:
- Primary keys, foreign keys, and indexes
- Auto-increment/identity columns with seed and increment values
- NOT NULL constraints and default values
- Column and table descriptions/comments
- Automatic topological sorting for foreign key dependencies

```yaml
source: mssql
target: postgres

defaults:
  mode: full-refresh
  object: public.{stream_table}

streams:
  dbo.categories:
  dbo.customers:
  dbo.products:    # FK to categories
  dbo.orders:      # FK to customers
  dbo.order_items: # FK to orders and products

env:
  # Enable all schema attributes
  SLING_SCHEMA_MIGRATION: all

  # Or enable specific attributes
  # SLING_SCHEMA_MIGRATION: description, primary_key, foreign_key, indexes
```

Available options: `all`, `primary_key`, `foreign_key`, `indexes`, `auto_increment`, `nullable`, `default_value`, `description`

See [Schema Migration](../examples/database-to-database/schema-migration.md) for detailed examples and supported databases.

## Change Capture (CDC)

Continuously replicate row-level changes (inserts, updates, deletes) from a source database's transaction log:
- Automatic initial snapshot with chunked, resumable loading
- Incremental change capture from transaction logs (binlog, WAL, etc.)
- Soft delete support to preserve deleted rows with timestamps
- Bounded runs with configurable event limits and duration
- Replay/backfill from earlier positions when needed

```yaml
source: MY_MYSQL
target: MY_POSTGRES

defaults:
  mode: change-capture
  primary_key: [id]
  object: public.{stream_table}
  change_capture_options:
    run_max_events: 10000
    run_max_duration: 10m

streams:
  my_database.customers:
  my_database.orders:
    change_capture_options:
      soft_delete: true
```

See [Change Capture (CDC)](../concepts/change-capture.md) for complete documentation, supported sources, and examples.

## ODBC Connections

Connect to any database using Open Database Connectivity (ODBC) drivers:
- Access databases that may not have a dedicated Sling connector
- Use standardized ODBC interface for maximum compatibility
- Support for SQL Server, PostgreSQL, MySQL, DB2, SAP HANA, Teradata, and more
- Create custom SQL templates for unsupported database dialects

```yaml
connections:
  my_odbc:
    type: odbc
    conn_string: "Driver={ODBC Driver 18 for SQL Server};Server=myserver;Database=mydb;Uid=myuser;Pwd=mypassword"
```

See [ODBC Connections](../connections/database-connections/odbc.md) for complete documentation, driver installation, and custom template examples.

# Frequently Asked Questions

**How are tokens validated?**

Tokens are validated through CloudFlare's global network, ensuring high reliability and fast response times worldwide. This validation occurs when the Sling CLI process initializes. If you'd like to confirm validation, run sling in debug mode (with flag `-d`), and you should see a log message: `CLI Pro token validated`.

**Can I get an offline/air-gapped token?**

For air-gapped or high-security environments, we offer offline license tokens. These require yearly renewal by default, but perpetual licenses are also available. Please contact [support@slingdata.io](mailto:support@slingdata.io) to request a quote.

**How many subscriptions do I need?**

Each CLI Pro subscription includes 2 tokens:
- 1 Production token: For use in production environments
- 1 Development token: For development and testing

Each subscription is designed for **a single team** within your organization. A "team" refers to a cohesive group managing separate data pipelines, configurations, or business objectives that benefit from isolation for security, governance, or operational independence.

This per-team structure enables us to maintain predictable flat-rate pricing that's sustainable and fair for everyone, allowing us to deliver high-performance features and priority support without usage-based metering.

**What this means for you:**
- Each team or distinct project needs its own subscription
- Subscriptions are for internal use within your organization only
- You can use the production token across all your production environments (servers, containers, etc.)
- Team members can share the development token for testing and collaboration

**Company-wide licensing:** If you prefer a single license for your entire organization, perpetual licenses are available that cover company-wide usage. Contact [support@slingdata.io](mailto:support@slingdata.io) to request a quote.

**Examples:**
- A data engineering team handling customer data → 1 subscription
- A separate analytics team working on reporting → 1 subscription  
- Multiple independent teams in your organization → 1 subscription per team
- Consultants or freelancers serving multiple clients → 1 subscription per client

**Important Licensing Restrictions:**

{% hint style="warning" %}
**Reselling and Commercial Redistribution Prohibited**

CLI Pro subscriptions are licensed for use by the subscribing organization only. You are prohibited from:
- Reselling or redistributing access to CLI Pro features
- Acting as a service provider offering CLI Pro to third parties
- White-labeling or rebranding CLI Pro as your own service
- Providing commercial access to CLI Pro without proper licensing

**For Consultants and Service Providers:**
If you wish to use CLI Pro in a consulting capacity or provide it to your clients, each client organization should have their own CLI Pro subscription. Contact us at [support@slingdata.io](mailto:support@slingdata.io) to discuss partner licensing arrangements.

**For System Integrators:**
We offer specific partner licensing programs for system integrators and technology partners. Contact us to discuss appropriate licensing for your use case.

Unauthorized reselling or redistribution will result in immediate termination of your subscription and may subject you to legal action.
{% endhint %}

Please use tokens responsibly and in accordance with our [Terms of Service](https://slingdata.io/terms). Each subscription is intended for use within a single organization or team, not for redistribution to external parties.

<!-- source: sling-docs/concepts/overview.md -->
# What is Sling?

## What is Sling?

Sling is a modern data movement and transformation platform designed to simplify and streamline data operations. It provides both a powerful CLI tool and a comprehensive platform for managing data workflows between various sources and destinations.

## Core Features

- **Data Movement**: Transfer data between different storage systems and databases efficiently
- **Flexible Connectivity**: Support for numerous databases, data warehouses, and file storage systems
- **Transformation Capabilities**: Built-in data transformation features during transfer
- **Multiple Operation Modes**: Support for various replication modes including full-refresh, incremental, and snapshot
- **Enterprise-Grade**: Production-ready with monitoring, scheduling, and error handling

## Key Components

### 1. Sling CLI

The command-line interface provides direct access to Sling's capabilities, perfect for:
- Local development and testing
- CI/CD pipeline integration
- Automated data operations
- Quick data transfers and transformations

### 2. Sling Platform

The web-based platform offers:
- Visual interface for creating and managing data workflows
- Team collaboration features
- Monitoring and alerting
- Centralized connection management
- Job scheduling and orchestration

### 3. Sling Agents

Agents are the workers that execute your data operations:
- Run in your own infrastructure
- Secure access to your data sources
- Support for both development and production environments

## Common Use Cases

- Database replication and synchronization
- Data warehouse loading and ETL operations
- File system to database ingestion
- Cross-platform data migration
- Backup and archival operations
- Real-time data copying and transformation

## Getting Started

To begin using Sling, you can either:
1. Install the CLI tool for local development and testing
2. Sign up for the Sling Platform for a managed experience
3. Use both in combination for a complete data operations solution

Choose the approach that best fits your needs and scale up as your requirements grow.

<!-- source: sling-docs/concepts/replication.md -->
# Replication

## Replication

## Overview

Replications are the best way to use sling in a reusable manner. The `defaults` key allows reusing your inputs with the ability to override any of them in a particular stream. Both YAML or JSON files are accepted. When you run a replication, internally, Sling auto-generates many tasks (one per stream) and runs them in order.

See these pages for more details:

* [Structure](./replication/structure.md)
* [Modes](./replication/modes.md)
* [Source Options](./replication/source-options.md)
* [Target Options](./replication/target-options.md)
* [Columns & Constraints](./replication/columns.md)
* [Transformations](./replication/transforms.md)
* [Hooks](../concepts/hooks.md)

Here is a basic example, where all PostgreSQL tables in the schema `my_schema` will be loaded into Snowflake. The `my_schema.*` notation as the stream name is a feature possible **only in Replications**. Also notice how `defaults.object` uses [runtime variables](./replication/runtime-variables.md).

{% code title="replication.yaml" %}
```yaml
source: MY_POSTGRES
target: MY_SNOWFLAKE

# default config options which apply to all streams
defaults:
  mode: full-refresh
  object: new_schema.{stream_schema}_{stream_table}

streams:
  my_schema.*:

env:
  SLING_THREADS: 3
```
{% endcode %}

Another example:

{% code title="replication.yaml" %}
```yaml
source: MY_MYSQL
target: MY_BIGQUERY

defaults:
  mode: incremental
  object: '{target_schema}.{stream_schema}_{stream_table}'
  primary_key: [id]
  
  source_options:
    empty_as_null: false
    
  target_options:
    column_casing: snake

streams:
  finance.accounts:
  finance.users:
    disabled: true
  
  finance.departments:
    object: '{target_schema}.finance_departments_old' # overwrite default object
    source_options:
      empty_as_null: false

  finance."Transactions":
    mode: incremental # overwrite default mode
    primary_key: [other_id]
    update_key: last_updated_at
  
  finance.all_users.custom:
    sql: |
      select col1, col2
      from finance."all_Users"
    object: finance.all_users # need to add 'object' key for custom SQL

env:
  # adds the _sling_loaded_at timestamp column
  SLING_LOADED_AT_COLUMN: true 
  
  # if source is file, adds a _sling_stream_url column with file path / url
  SLING_STREAM_URL_COLUMN: true

  # parallel stream runs
  SLING_THREADS: 3

  # retry failing stream runs
  SLING_RETRIES: 1
```
{% endcode %}

We can use a replication config with: `sling run -r /path/to/replication.yaml`

<!-- source: sling-docs/concepts/pipeline.md -->
# Pipeline

## Pipeline

A Pipeline in Sling allows you to execute multiple steps in sequence. Each step can be a different type of operation, enabling you to create complex workflows by chaining together various actions like running replications, executing queries, making HTTP requests, and more.

{% hint style="success" %}

Sling Pipelines integrate seamlessly with the [Sling VSCode Extension](../sling-cli/vscode.md). The extension provides schema validation, auto-completion, hover documentation, and diagnostics for your pipeline configurations, making it easier to author and debug complex workflows.

{% endhint %}

## Pipeline Configuration

A pipeline is defined in YAML format with a `steps` key at the root level containing an array of steps. Each step supports the same types and configurations as [Hooks](hooks.md).

```yaml
steps:
  - type: log
    message: "Starting pipeline execution"

  - type: replication
    path: path/to/replication.yaml
    id: my_replication

  - type: query
    if: state.my_replication.status == "success"
    connection: my_database
    query: "UPDATE status SET completed = true"

env:
  MY_KEY: VALUE
```

## Available Step Types

Pipelines support all the same types as Hooks:

| Step Type   | Description                                                | Documentation                    |
| ----------- | ---------------------------------------------------------- | -------------------------------- |
| Check       | Validate conditions and control flow                       | [Check Step](hooks/check.md)     |
| Command     | Run any command/process                                    | [Command Step](hooks/command.md)       |
| Copy        | Transfer files between local or remote storage connections | [Copy Step](hooks/copy.md)       |
| Delete      | Remove files from local or remote storage connections      | [Delete Step](hooks/delete.md)   |
| Group       | Run sequences of steps or loop over values                 | [Group Step](hooks/group.md)   |
| HTTP        | Make HTTP requests to external services                    | [HTTP Step](hooks/http.md)       |
| Inspect     | Inspect a file or folder                                   | [Inspect Step](hooks/inspect.md) |
| List        | List files in folder                                       | [List Step](hooks/list.md) |
| Log         | Output custom messages and create audit trails             | [Log Step](hooks/log.md)         |
| Query       | Execute SQL queries against any defined connection         | [Query Step](hooks/query.md)     |
| Read        | Read contents of files from storage connections            | [Read Step](hooks/read.md)       |
| Replication | Run a Replication                                          | [Replication Step](hooks/replication.md) |
| Routine     | Execute reusable step sequences from external files        | [Routine Step](hooks/routine.md) |
| Store       | Store values for later in-process access                   | [Store Step](hooks/store.md) |
| Write       | Write content to files in storage connections              | [Write Step](hooks/write.md)     |

## Common Step Properties

Each step shares the same common properties as hooks:

| Property     | Description                                                                        | Required                 |
| ------------ | ---------------------------------------------------------------------------------- | ------------------------ |
| `type`       | The type of step (`query`/ `http`/ `check`/ `copy` / `delete`/  `log` / `inspect`) | Yes                      |
| `if`         | Optional condition to determine if the step should execute                         | No                       |
| `id`         | A specific identifier to refer to the step output data                             | No                       |
| `on_failure` | What to do if the step fails (see [On Failure Behaviors](#on-failure-behaviors))   | No (defaults to `abort`) |

### On Failure Behaviors

The `on_failure` property controls what happens **when a step fails**. It only takes effect if the step errors — on success it has no impact. The following values are supported:

| Value             | Behavior                                                                                                                                                                          |
| ----------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `abort` (default) | Stops the pipeline immediately and fails the run with the error. This is the behavior when `on_failure` is not set.                                                              |
| `warn`            | Logs a warning with the error message and **continues to the next step**. The run is not failed.                                                                                 |
| `quiet`           | Silently swallows the error (no log output) and **continues to the next step**. The run is not failed.                                                                           |
| `break`           | Stops the current sequence of steps gracefully (without failing the run) and moves on. Useful to stop early without marking the pipeline as failed.                              |
| `retry`           | Retries the failed step once. If it fails again, the error is raised.                                                                                                            |
| `defer`           | Only meaningful inside a [group](hooks/group.md): records the error but lets the remaining steps in the group finish, then surfaces the error at the end of the group.           |

## Variables Available

Pipeline steps have access to the runtime state which includes various variables that can be referenced using curly braces `{variable}`. The available variables include:

* `runtime_state` - Contains all state variables available
* `env.*` - All variables defined in the `env`
* `timestamp.*` - Various timestamp parts information
* `steps.*` - Output data from previous steps (referenced by their `id`)

You can view all available variables by using a log step:

```yaml
steps:
  - type: log
    message: '{runtime_state}'
```

## Example Pipeline

Here's a complete example that demonstrates various pipeline capabilities:

```yaml
env:
  DATABASE: production
  NOTIFY_URL: https://api.example.com/webhook

steps:
  # Log the start of execution
  - type: log
    message: "Starting pipeline execution"

  # Run a replication
  - type: replication
    path: replications/daily_sync.yaml
    id: daily_sync
    on_failure: warn

  # Validate the results
  - type: check
    check: state.daily_sync.status == "success"
    message: "Daily sync failed"
    on_failure: abort

  # Update status in database
  - type: query
    connection: "{env.DATABASE}"
    query: |
      UPDATE pipeline_status 
      SET last_run = current_timestamp
      WHERE name = 'daily_sync'
    on_failure: warn

  # Send notification
  - type: http
    url: "{env.NOTIFY_URL}"
    method: POST
    payload: |
      {
        "pipeline": "daily_sync",
        "status": "success",
      }

  # Log completion
  - type: log
    message: "Pipeline completed successfully"
```

## Best Practices

1. **Error Handling**: Use appropriate `on_failure` behaviors for each step
2. **Validation**: Include check steps to validate critical conditions
3. **Logging**: Add log steps for better observability
4. **Modularity**: Break down complex operations into multiple steps
5. **Conditions**: Use `if` conditions to control step execution
6. **Variables**: Leverage environment variables and runtime state for dynamic configuration
7. **Identifiers**: Use meaningful `id`s for steps when you need to reference their output later

## Running a Pipeline

You can run a pipeline using the Sling CLI:

```bash
sling run --pipeline path/to/pipeline.yaml
```

<!-- source: sling-docs/concepts/hooks.md -->
# Hooks

## Hooks

Hooks are powerful mechanisms in Sling that allow you to execute custom actions before (pre-hooks) or after (post-hooks) a replication stream, as well as at the start or end of the replication parent cycle (before first stream and/or after last stream). They enable you to extend and customize your data pipeline with various operations such as data validation, notifications, file management, and custom processing.

{% hint style="success" %}

Hooks are the same as Steps, when using sling in [Pipeline](../concepts/pipeline.md) mode.

Furthermore, Sling Hooks integrate seamlessly with the [Sling VSCode Extension](../sling-cli/vscode.md). The extension provides schema validation, auto-completion, hover documentation, and diagnostics for your hooks configurations, making it easier to author and debug complex workflows.

{% endhint %}

Some typical operations include:

## Stream Level

**Pre-Hooks**: Execute before the replication stream run starts

* Validate prerequisites
* Download necessary files
* Set up configurations
* Perform cleanup operations

**Post-Hooks**: Execute after the replication stream run completes

* Validate results
* Send notifications
* Upload processed files
* Clean up temporary files
* Log completion status

**Pre/Post-Merge-Hooks**: Execute in transaction session, before/after data is loaded/merged into final table

* Set specific session setting and configuration
* Alter table holding the temporary data, prior to merge
* Run Specific SQL queries on other tables

## Available Hook Types

| Hook Type | Description                                                | Documentation                    |
| --------- | ---------------------------------------------------------- | -------------------------------- |
| Check     | Validate conditions and control flow                       | [Check Hook](hooks/check.md)     |
| Command   | Run any command/process                                    | [Command Hook](hooks/command.md)       |
| Copy      | Transfer files between local or remote storage connections | [Copy Hook](hooks/copy.md)       |
| Delete    | Remove files from local or remote storage connections      | [Delete Hook](hooks/delete.md)   |
| Group     | Run sequences of steps or loop over values                 | [Group Hook](hooks/group.md)   |
| HTTP      | Make HTTP requests to external services                    | [HTTP Hook](hooks/http.md)       |
| Inspect   | Inspect a file or folder                                   | [Inspect Hook](hooks/inspect.md) |
| List      | List files in folder                                       | [List Hook](hooks/list.md) |
| Log       | Output custom messages and create audit trails             | [Log Hook](hooks/log.md)         |
| Query     | Execute SQL queries against any defined connection         | [Query Hook](hooks/query.md)     |
| Replication   | Run a Replication                                      | [Replication Hook](hooks/replication.md) |
| Routine   | Execute reusable step sequences from external files        | [Routine Hook](hooks/routine.md) |
| Store     | Store values for later in-process access                   | [Store Hook](hooks/store.md)     |
| Read        | Read contents of files from storage connections          | [Read Step](hooks/read.md)       |
| Write       | Write content to files in storage connections            | [Write Step](hooks/write.md)     |

## Hook Configuration

Hooks can be configured in two locations:

* At the `defaults` level (applies to all streams)
* At the individual `stream` level (overrides defaults)

**Stream level Hooks**

We can use the following structure to decare hooks with the `hooks` key, under the `defaults` branch or under any stream branch.

```yaml
defaults:
  ...

streams:
  my_stream:
    hooks:
      # Prior to stream beginning
      pre:
        - type: log
          # hook configuration...

      # Inside session/transaction, before merge, right after BEGIN
      pre_merge:
        - type: query
          # hook configuration...

      # Inside session/transaction, after merge, right before COMMIT
      post_merge:
        - type: query
          # hook configuration...

      # After stream finishes
      post:
        - type: http
          # hook configuration...
```

**Replication level Hooks**

We can also define hooks to run at the replication file level, meaning before any of the streams run and/or after all the streams have ran. For replication level hooks, we must declare the `start` and `end` hooks at the root of the YAML configuration.

```yaml
# replication level hooks need to be set at the root of the YAML
hooks:
  start:
    - type: query
      # hook configuration...

  end:
    - type: http
      # hook configuration...

defaults:
  ...

streams:
  ...
```

## Common Hook Properties

All hook types share some common properties:

| Property     | Description                                                                        | Required                 |
| ------------ | ---------------------------------------------------------------------------------- | ------------------------ |
| `type`       | The type of hook (`query`/ `http`/ `check`/ `copy` / `delete`/  `log` / `inspect`) | Yes                      |
| `if`         | Optional condition to determine if the hook should execute                         | No                       |
| `id`         | a specify identifier to refer to the hook output data.                             | No                       |
| `on_failure` | What to do if the hook fails (`abort`/ `warn`/ `quiet`/`skip`/`break`)             | No (defaults to `abort`) |

## Variables Available

* `runtime_state` - Contains all state variables available
* `state.*` - All hooks output state information (keyed by hook id)
* `store.*` - All stored values from previous hooks
* `env.*` - All variables defined in the `env`
* `timestamp.*` - Various timestamp parts information
* `execution.*` - Replication run level information
* `source.*` - Source connection information
* `target.*` - Target connection information
* `stream.*` - Current source stream info
* `object.*` - Current target object info
* `runs.*` - All runs information (keyed by stream run id)
* `run.*` - Current stream run information

### Nested Fields

<details>

<summary><code>timestamp.*</code> Fields</summary>

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `timestamp` | datetime | Full timestamp object | `2025-01-19T08:27:31.473303-05:00` |
| `unix` | integer | Unix epoch timestamp | `1737286051` |
| `file_name` | string | Timestamp formatted for file names | `2025_01_19_082731` |
| `rfc3339` | string | RFC3339 formatted timestamp | `2025-01-19T08:27:31-05:00` |
| `date` | string | Date only | `2025-01-19` |
| `datetime` | string | Date and time | `2025-01-19 08:27:31` |
| `YYYY` | string | Four-digit year | `2025` |
| `YY` | string | Two-digit year | `25` |
| `MMM` | string | Three-letter month abbreviation | `Jan` |
| `MM` | string | Two-digit month | `01` |
| `DD` | string | Two-digit day | `19` |
| `DDD` | string | Three-letter day abbreviation | `Sun` |
| `HH` | string | Two-digit hour (24-hour format) | `08` |

</details>

<details>

<summary><code>execution.*</code> Fields</summary>

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `id` | string | Unique execution identifier | `2rxeplXz2UqdIML1NncvWKNQuwD` |
| `file_path` | string | Path to the replication configuration file | `/path/to/replication.yaml` |
| `file_name` | string | Name to the replication configuration file | `replication.yaml` |
| `total_bytes` | integer | Total bytes processed across all runs | `6050` |
| `total_rows` | integer | Total rows processed across all runs | `34` |
| `status.count` | integer | Total number of streams | `1` |
| `status.success` | integer | Number of successful streams | `1` |
| `status.running` | integer | Number of running streams | `0` |
| `status.skipped` | integer | Number of skipped streams | `0` |
| `status.cancelled` | integer | Number of cancelled streams | `0` |
| `status.warning` | integer | Number of streams with warnings | `0` |
| `status.error` | integer | Number of errored streams | `0` |
| `start_time` | datetime | Execution start time | `2025-01-19T08:27:22.988403-05:00` |
| `end_time` | datetime | Execution end time | `2025-01-19T08:27:31.472684-05:00` |
| `duration` | integer | Execution duration in seconds | `8` |
| `error` | string/null | Error message if execution failed | `null` |

</details>

<details>

<summary><code>source.*</code> / <code>target.*</code> Connection Fields</summary>

| Field | Type | Description | Example (Source) | Example (Target) |
|-------|------|-------------|------------------|------------------|
| `name` | string | Connection name | `aws_s3` | `postgres` |
| `type` | string | Connection type | `s3` | `postgres` |
| `kind` | string | Connection kind | `file` | `database` |
| `bucket` | string | S3/GCS bucket name | `my-bucket-1` | `` |
| `container` | string | Azure container name | `` | `` |
| `database` | string | Database name | `` | `postgres` |
| `instance` | string | Database instance | `` | `` |
| `schema` | string | Default schema | `` | `public` |

</details>

<details>

<summary><code>stream.*</code> Fields</summary>

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `file_folder` | string | Parent folder of the file | `update_dt_year=2018` |
| `file_name` | string | Name of the file | `update_dt_month=11` |
| `file_ext` | string | File extension | `parquet` |
| `file_path` | string | Full file path | `test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11` |
| `name` | string | Stream name pattern | `test/public_test1k_postgres_pg_parquet/{part_year}/{part_month}/` |
| `description` | string | Stream description (if provided) | `` |
| `schema` | string | Schema name (for database sources) | `` |
| `schema_lower` | string | Schema name in lowercase | `` |
| `schema_upper` | string | Schema name in uppercase | `` |
| `table` | string | Table name (for database sources) | `` |
| `table_lower` | string | Table name in lowercase | `` |
| `table_upper` | string | Table name in uppercase | `` |
| `full_name` | string | Full stream identifier | `s3://my-bucket-1/test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11/` |

</details>

<details>

<summary><code>object.*</code> Fields</summary>

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `schema` | string | Target schema name | `public` |
| `table` | string | Target table name | `test1k_postgres_pg_parquet` |
| `name` | string | Quoted object name | `"public"."test1k_postgres_pg_parquet"` |
| `full_name` | string | Full quoted object name | `"public"."test1k_postgres_pg_parquet"` |
| `temp_schema` | string | Temporary schema name | `public` |
| `temp_table` | string | Temporary table name | `test1k_postgres_pg_parquet_tmp` |
| `temp_full_name` | string | Full temporary table name | `"public"."test1k_postgres_pg_parquet_tmp"` |

</details>

<details>

<summary><code>run.*</code> Fields</summary>

| Field | Type | Description | Example |
|-------|------|-------------|---------|
| `id` | string | Run identifier | `test_public_test1k` |
| `stream.*` | object | Stream information (see stream fields above) | `{...}` |
| `object.*` | object | Object information (see object fields above) | `{...}` |
| `total_bytes` | integer | Total bytes processed in this run | `6050` |
| `total_rows` | integer | Total rows processed in this run | `34` |
| `status` | string | Run status | `success` |
| `start_time` | datetime | Run start time | `2025-01-19T08:27:22.988403-05:00` |
| `end_time` | datetime | Run end time | `2025-01-19T08:27:31.472684-05:00` |
| `duration` | integer | Run duration in seconds | `8` |
| `incremental_value` | any | The incremental value used | `2025-01-19T08:27:31.472684-05:00` |
| `range` | string | The start/end range values used | `2025-01-01,2025-02-01` |
| `error` | string/null | Error message if run failed | `null` |
| `config.mode` | string | Replication mode | `incremental` |
| `config.object` | string | Target object | `"public"."test1k_postgres_pg_parquet"` |
| `config.primary_key` | array | Primary key columns | `["id"]` |
| `config.update_key` | string | Update key column | `update_dt` |
| `config.source_options` | object | Source-specific options | `{}` |
| `config.target_options` | object | Target-specific options | `{}` |

</details>


### `runtime_state` Payload

The best way to view any available variables is to print the `runtime_state` variable.

For example, using the `log` hook as shown below will print all available variables.

```yaml
- type: log
  message: '{runtime_state}'
```

Shows something like below JSON payload.

<details>

<summary><code>runtime_state</code> Payload</summary>

```json
{
  "state": {
    "end-01": {},
    "start-01": {
      "level": "info",
      "message": "{...}",
      "status": "success"
    },
    "start-02": {
      "path": "sling-state/test/r.19",
      "status": "success"
    }
  },
  "store": {
    "my_key": "my_value"
  },
  "env": {
    "RESET": "true",
    "SLING_STATE": "aws_s3/sling-state/test/r.19"
  },
  "timestamp": {
    "timestamp": "2025-01-19T08:27:31.473303-05:00",
    "unix": 1737286051,
    "file_name": "2025_01_19_082731",
    "rfc3339": "2025-01-19T08:27:31-05:00",
    "date": "2025-01-19",
    "datetime": "2025-01-19 08:27:31",
    "YYYY": "2025",
    "YY": "25",
    "MMM": "Jan",
    "MM": "01",
    "DD": "19",
    "DDD": "Sun",
    "HH": "08"
  },
  "source": {
    "name": "aws_s3",
    "type": "s3",
    "kind": "file",
    "bucket": "my-bucket-1",
    "container": "",
    "database": "",
    "instance": "",
    "schema": ""
  },
  "target": {
    "name": "postgres",
    "type": "postgres",
    "kind": "database",
    "bucket": "",
    "container": "",
    "database": "postgres",
    "instance": "",
    "schema": "public"
  },
  "stream": {
    "file_folder": "update_dt_year=2018",
    "file_name": "update_dt_month=11",
    "file_ext": "parquet",
    "file_path": "test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11",
    "name": "test/public_test1k_postgres_pg_parquet/{part_year}/{part_month}/",
    "schema": "",
    "schema_lower": "",
    "schema_upper": "",
    "table": "",
    "table_lower": "",
    "table_upper": "",
    "full_name": "s3://my-bucket-1/test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11/"
  },
  "object": {
    "schema": "public",
    "table": "test1k_postgres_pg_parquet",
    "name": "\"public\".\"test1k_postgres_pg_parquet\"",
    "full_name": "\"public\".\"test1k_postgres_pg_parquet\"",
    "temp_schema": "public",
    "temp_table": "test1k_postgres_pg_parquet_tmp",
    "temp_full_name": "\"public\".\"test1k_postgres_pg_parquet_tmp\""
  },
  "runs": {
    "test_public_test1k": {
      "id": "test_public_test1k",
      "stream": {
        "file_folder": "update_dt_year=2018",
        "file_name": "update_dt_month=11",
        "file_ext": "parquet",
        "file_path": "test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11",
        "name": "test/public_test1k_postgres_pg_parquet/{part_year}/{part_month}/",
        "schema": "",
        "schema_lower": "",
        "schema_upper": "",
        "table": "",
        "table_lower": "",
        "table_upper": "",
        "full_name": "s3://my-bucket-1/test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11/"
      },
      "object": {
        "schema": "public",
        "table": "test1k_postgres_pg_parquet",
        "name": "\"public\".\"test1k_postgres_pg_parquet\"",
        "full_name": "\"public\".\"test1k_postgres_pg_parquet\"",
        "temp_schema": "public",
        "temp_table": "test1k_postgres_pg_parquet_tmp",
        "temp_full_name": "\"public\".\"test1k_postgres_pg_parquet_tmp\""
      },
      "total_bytes": 6050,
      "total_rows": 34,
      "status": "success",
      "start_time": "2025-01-19T08:27:22.988403-05:00",
      "end_time": "2025-01-19T08:27:31.472684-05:00",
      "duration": 8,
      "error": null,
      "config": {
        "mode": "incremental",
        "object": "\"public\".\"test1k_postgres_pg_parquet\"",
        "primary_key": [
          "id"
        ],
        "update_key": "update_dt",
        "source_options": {},
        "target_options": {},
        "single": false,
        "hooks": {}
      }
    }
  },
  "execution": {
    "id": "2rxeplXz2UqdIML1NncvWKNQuwD",
    "string": "/path/to/replication.yaml",
    "total_bytes": 6050,
    "total_rows": 34,
    "status": {
      "count": 1,
      "success": 1,
      "running": 0,
      "skipped": 0,
      "cancelled": 0,
      "warning": 0,
      "error": 0
    },
    "start_time": "2025-01-19T08:27:22.988403-05:00",
    "end_time": "2025-01-19T08:27:31.472684-05:00",
    "duration": 8,
    "error": null
  },
  "run": {
    "id": "test_public_test1k",
    "stream": {
      "file_folder": "update_dt_year=2018",
      "file_name": "update_dt_month=11",
      "file_ext": "parquet",
      "file_path": "test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11",
      "name": "test/public_test1k_postgres_pg_parquet/{part_year}/{part_month}/",
      "schema": "",
      "schema_lower": "",
      "schema_upper": "",
      "table": "",
      "table_lower": "",
      "table_upper": "",
      "full_name": "s3://my-bucket-1/test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11/"
    },
    "object": {
      "schema": "public",
      "table": "test1k_postgres_pg_parquet",
      "name": "\"public\".\"test1k_postgres_pg_parquet\"",
      "full_name": "\"public\".\"test1k_postgres_pg_parquet\"",
      "temp_schema": "public",
      "temp_table": "test1k_postgres_pg_parquet_tmp",
      "temp_full_name": "\"public\".\"test1k_postgres_pg_parquet_tmp\""
    },
    "total_bytes": 6050,
    "total_rows": 34,
    "status": "success",
    "start_time": "2025-01-19T08:27:22.988403-05:00",
    "end_time": "2025-01-19T08:27:31.472684-05:00",
    "duration": 8,
    "error": null,
    "config": {
      "mode": "incremental",
      "object": "\"public\".\"test1k_postgres_pg_parquet\"",
      "primary_key": [
        "id"
      ],
      "update_key": "update_dt",
      "source_options": {},
      "target_options": {},
      "single": false,
      "hooks": {}
    }
  }
}
```

</details>

Furthermore, we can access any data-point using a [`jmespath` expression](https://jmespath.org/):

* `state["start-02"].status` - Gets the status of a hook (returns `success`)
* `store.my_key` - Gets a stored value from the store (returns `my_value`)
* `run.total_rows` - Gets the number of rows processed in the current run (returns `34`)
* `run.duration` - Gets the duration of the current run in seconds (returns `8`)
* `timestamp.unix` - The epoch/unix timestamp (returns `1737286051`)
* `source.bucket` - Gets the source S3 bucket name (returns `my-bucket-1`)
* `target.database` - Gets the target database name (returns `postgres`)
* `run.config.primary_key[0]` - Gets the first primary key column (returns `id`)
* `stream.file_path` - Gets the current stream's file path (returns `test/public_test1k_postgres_pg_parquet/update_dt_year=2018/update_dt_month=11`)
* `stream.file_ext` - Gets the file extension (returns `parquet`)
* `stream.schema_lower` - Gets the stream schema name in lowercase
* `stream.table_upper` - Gets the stream table name in uppercase
* `object.temp_full_name` - Gets the temporary table full name (returns `"public"."test1k_postgres_pg_parquet_tmp"`)
* `execution.status.error` - Gets the count of errored streams (returns `0`)
* `execution.total_bytes` - Gets the total bytes processed across all runs (returns `6050`)
* `runs["test_public_test1k"].status` - Gets the status of a specific run by ID (returns `success`)

## Complete Example

```yaml
# replication level hooks need to be set at the root of the YAML
hooks:
  # runs in order before replication starts.
  start:
      - type: query
        query: select ....
        id: my_query # can use `{state.my_query.result[0].col1}` later
        on_failure: abort
      
  # runs in order after all streams have completed.
  end:
      # check for any errors. if errored, do not proceed (break)
      - type: check
        check: execution.status.error == 0
        on_failure: break
      
      - type: query
        query: update ....
        into: result # can use `{result.col1}` later
        on_failure: abort

defaults:
  hooks:
    pre:
      - type: query
        connection: source_db
        query: "UPDATE status SET running = true"
        on_failure: abort

    post:
      - type: check
        check: "run.total_rows > 0"
        on_failure: warn
      
      - type: http
        if: run.status == "success"
        url: "https://api.example.com/webhook"
        method: POST
        payload: | # my_query.result will serialize into an array of objects
          {
            "status": "complete",
            "name": "{ my_query.result[0].name }"
            "records": {my_query.result}}
          }

streams:
  public.users:
    hooks:
      # runs in order before stream run
      pre:
        - type: query
          query: update ....
          on_failure: abort

      # runs in order after stream run
      post:
        - type: http
          url: https://my.webhook/path
          method: POST
          payload: |
            {"result": "{run.status}"}
```

## Best Practices

1. **Error Handling**: Specify appropriate `on_failure` behavior. The default value is `abort`.
2. **Validation**: Use `check` hooks to validate prerequisites and results
3. **Logging**: Implement `log` hooks for better observability
4. **Cleanup**: Use `delete` hooks to manage temporary / old files
5. **Modularity**: Break down complex operations into multiple hooks
6. **Conditions**: Use `if` conditions to control hook execution
7. **Environment Awareness**: Consider different environments in hook configurations

<!-- source: sling-docs/concepts/source-target.md -->
# Source and Target

## Source and Target

In Sling, every replication involves two key components: a source (where data comes from) and a target (where data goes to). Understanding how to configure these components is essential for successful data movement.

## Source Configuration

A source defines the origin of your data and how Sling should read it. Sources can be databases, files, or other data stores.

### Basic Source Structure
```yaml
source:
  type: "postgres"                # The type of source
  stream: "public.users"          # What to read
  options:
    batch_size: 1000             # How many records to read at once
    where: "created_at > '2023'" # Optional filter
```

### Common Source Types

1. **Databases**
   - Relational databases (PostgreSQL, MySQL, etc.)
   - NoSQL databases (MongoDB)
   - Data warehouses (Snowflake, BigQuery)

2. **File Systems**
   - Local files
   - Cloud storage (S3, GCS, Azure Blob)
   - FTP/SFTP servers

### Source Options

#### General Options
- `batch_size`: Number of records to read in each batch
- `where`: SQL-style filter condition
- `columns`: List of columns to include/exclude
- `timeout`: Connection timeout settings

#### Database-Specific Options
```yaml
source:
  type: "postgres"
  stream: "users"
  options:
    batch_size: 1000
    fetch_size: 2000
    cursor_name: "mycursor"
```

#### File-Specific Options
```yaml
source:
  type: "s3"
  stream: "data/*.csv"
  options:
    format: "csv"
    delimiter: ","
    header: true
```

## Target Configuration

A target defines where your data should be written and how it should be stored.

### Basic Target Structure
```yaml
target:
  type: "mysql"                # The type of target
  object: "users"             # Where to write
  options:
    create_table: true        # Auto-create table if doesn't exist
    truncate: false          # Whether to clear existing data
```

### Common Target Types

1. **Databases**
   - Same database types as sources
   - Different write optimization options
   - Schema management capabilities

2. **File Systems**
   - Various file formats (CSV, JSON, Parquet)
   - Compression options
   - Directory organization

### Target Options

#### General Options
- `create_table`: Automatically create target table
- `truncate`: Clear existing data before writing
- `batch_size`: Number of records to write at once
- `on_error`: Error handling behavior

#### Database-Specific Options
```yaml
target:
  type: "postgres"
  object: "users"
  options:
    create_table: true
    primary_key: ["id"]
    indexes: ["email", "created_at"]
```

#### File-Specific Options
```yaml
target:
  type: "s3"
  object: "output/users"
  options:
    format: "parquet"
    compression: "snappy"
    partition_by: ["date"]
```

## Advanced Configurations

### Source Filtering
```yaml
source:
  type: "mysql"
  stream: "orders"
  options:
    where: "status = 'completed' AND created_at >= '2023-01-01'"
    order_by: "id ASC"
```

### Target Schema Management
```yaml
target:
  type: "postgres"
  object: "users"
  options:
    create_table: true
    schema:
      id: "bigint"
      name: "varchar(255)"
      email: "varchar(255)"
      created_at: "timestamp"
```

### Error Handling
```yaml
source:
  type: "postgres"
  stream: "users"
  options:
    on_error: "skip"  # Skip problematic records

target:
  type: "mysql"
  object: "users"
  options:
    on_error: "fail"  # Stop on any error
```

## Best Practices

1. **Source Configuration**
   - Use appropriate batch sizes for your data volume
   - Include only necessary columns
   - Set reasonable timeouts
   - Use filters to minimize data transfer

2. **Target Configuration**
   - Define proper schema and constraints
   - Configure appropriate indexes
   - Set up error handling
   - Use optimized write settings

3. **Performance Optimization**
   - Balance batch sizes between source and target
   - Use appropriate data types
   - Configure proper indexes
   - Monitor resource usage

<!-- source: sling-docs/concepts/functions.md -->
# Available Functions

## Available Functions

Unlock the power of your data workflows with these versatile built-in functions. They enable sophisticated data transformations, validations, and manipulations, and can be seamlessly integrated into your [pipelines](./pipeline.md), [hooks](./hooks.md), and [transforms](./replication/transforms.md).


{% hint style="success" %}

**CLI Pro Required**: Functions require a [CLI Pro token](../sling-cli/cli-pro.md) or [Platform Plan](./sling-platform/platform.md).

{% endhint %}

## String Functions

| Function                            | Description                          | Parameters                                                       | Returns          | Example                                            |
| ----------------------------------- | ------------------------------------ | ---------------------------------------------------------------- | ---------------- | -------------------------------------------------- |
| `contains(string, substring)`       | Checks if string contains substring  | `string`, `substring`                                            | Boolean          | `contains("hello world", "world")` → `true`        |
| `join(array, separator)`            | Joins array elements into string     | `array`, `separator`                                             | String           | `join(["a", "b", "c"], ", ")` → `"a, b, c"`        |
| `length(string\|array\|map)`        | Gets length of string/array/map      | `value`: String, array or map                                    | Number           | `length("hello")` → `5`, `length([1,2])` → `2`       |
| `lower(string)`                     | Converts string to lowercase         | `string`: Input string                                           | Lowercase string | `lower("HELLO")` → `"hello"`                       |
| `snake(string)`                     | Converts string to snake_case (spaces become underscores) | `string`: Input string                              | snake_case string | `snake("Hello World")` → `"hello_world"`          |
| `slugify(string)`                   | Converts string to identifier / URL friendly slug (lowercase with hyphens, alphanumeric only) | `string`: Input string | URL slug string | `slugify("Hello, World!")` → `"hello-world"` |
| `replace(string, pattern, replace)` | Replaces occurrences of pattern      | `string`, `pattern`, `replacement`                               | Modified string  | `replace("hello", "l", "x")` → `"hexxo"`           |
| `split_part(string, sep, index)`    | Gets part of split string by index   | `string`, `separator`, `index` (0-based)                         | String part      | `split_part("a,b,c", ",", 1)` → `"b"`              |
| `split(string, separator)`          | Splits string into array             | `string`, `separator`                                            | Array of strings | `split("a,b,c", ",")` → `["a", "b", "c"]`      |
| `substring(string, start[, end])`   | Extracts part of a string (0-indexed) | `string`: Input string<br>`start`: Start index<br>`end`: Optional end index | Substring        | `substring("hello world", 0, 5)` → `"hello"`       |
| `trim(string)`                      | Removes whitespace from start/end    | `string`: Input string                                           | Trimmed string   | `trim(" hello ")` → `"hello"`                      |
| `upper(string)`                     | Converts string to uppercase         | `string`: Input string                                           | Uppercase string | `upper("hello")` → `"HELLO"`                       |
| `parse_ms_uuid(string)`             | Parses Microsoft UUID format         | `string`: 16-byte UUID string                                    | Parsed UUID      | `parse_ms_uuid(binary_uuid)` → `"12345678-..."`   |
| `replace_0x00(string)`              | Removes null bytes from string       | `string`: Input string                                           | Cleaned string   | `replace_0x00("hello\x00world")` → `"helloworld"` |
| `remove_diacritics(string)`         | Removes accents/diacritics           | `string`: Input string                                           | ASCII string     | `remove_diacritics("café")` → `"cafe"`            |
| `replace_non_printable(string)`     | Removes non-printable characters     | `string`: Input string                                           | Cleaned string   | `replace_non_printable("hello\x01world")` → `"helloworld"` |

## Numeric Functions

| Function                         | Description                      | Parameters                               | Returns           | Example                                    |
| -------------------------------- | -------------------------------- | ---------------------------------------- | ----------------- | ------------------------------------------ |
| `bool_parse(value)`              | Converts value to boolean        | `value`: Value to convert                | Boolean or error  | `bool_parse("true")` → `true`                |
| `float_format(value, format)`    | Formats float (Go format)        | `value`: Number<br>`format`: Format string | Formatted string  | `float_format(42.5678, "%.2f")` → `"42.57"`  |
| `float_parse(value)`             | Converts value to float          | `value`: Value to convert                | Float or error    | `float_parse("42.5")` → `42.5`               |
| `greatest(array\|val1, val2...)` | Finds maximum value              | `array` or multiple values               | Maximum value     | `greatest(1, 5, 3)` → `5`                    |
| `int_format(value, format)`      | Formats integer (Go format)      | `value`: Number<br>`format`: Format string | Formatted string  | `int_format(42, "%05d")` → `"00042"`         |
| `int_parse(value)`               | Converts value to integer        | `value`: Value to convert                | Integer or error  | `int_parse("42")` → `42`                     |
| `int_range(start, end[, step])`  | Generates integer range          | `start`, `end`: Integers, `step`: Optional integer (default 1) | Array of integers | `int_range(1, 5)` → `[1,2,3,4,5]`            |
| `is_greater(val1, val2)`         | Checks if `val1 > val2`          | `val1`, `val2`: Values to compare        | Boolean           | `is_greater(5, 3)` → `true`                  |
| `is_less(val1, val2)`            | Checks if `val1 < val2`          | `val1`, `val2`: Values to compare        | Boolean           | `is_less(3, 5)` → `true`                     |
| `least(array\|val1, val2...)`    | Finds minimum value              | `array` or multiple values               | Minimum value     | `least(1, 5, 3)` → `1`                       |

## Date Functions

Uses Go's `time` package and `strftime` conventions via [timefmt-go](https://github.com/itchyny/timefmt-go). Please refer to [`man 3 strftime`](https://linux.die.net/man/3/strftime) and [`man 3 strptime`](https://linux.die.net/man/3/strptime) for formatter syntax.

| Function                           | Description                    | Parameters                                              | Returns              | Example                                                                    |
| ---------------------------------- | ------------------------------ | ------------------------------------------------------- | -------------------- | -------------------------------------------------------------------------- |
| `now()`                            | Gets current date and time     | None                                                    | Time object          | `now()`                                                                    |
| `date_parse(string[, format])`     | Parses string to time object   | `string`: Date string<br>`format`: "auto" or `strftime` | Time object or error | `date_parse("2022-01-01T10:00:00Z", "auto")`                              |
| `date_format(date, format)`        | Formats time object to string  | `date`: Time object<br>`format`: `strftime` format        | Formatted string     | `date_format(now(), "%Y-%m-%d")` → `"2023-10-27"` (example date)        |
| `date_add(date, duration[, unit])` | Adds duration to time object   | `date`, `duration` (int), `unit` (string, default "s")  | Time object          | `date_add(now(), -7, "day")`                                               |
| `date_diff(date1, date2[, unit])`  | Time between dates             | `date1`, `date2`, `unit` (string, default "s")          | Number (float)       | `date_diff(date_add(now(), 1, "day"), now(), "hour")` → `24.0`            |
| `date_trunc(date, unit)`           | Truncates date to unit start   | `date`, `unit` ("year", "month", "day", "hour", etc.)   | Time object          | `date_trunc(now(), "month")` → First day of current month at 00:00:00      |
| `date_extract(date, part)`         | Extracts part from date        | `date`, `part` ("year", "month", "day", "hour", etc.)   | Number               | `date_extract(now(), "year")` → `2023` (example year)                     |
| `date_last(date[, period])`        | Gets last day of period        | `date`, `period` ("month", "year", default "month")     | Time object          | `date_last(now())` → Last day of current month                             |
| `date_first(date[, period])`       | Gets first day of period       | `date`, `period` ("month", "year", default "month")     | Time object          | `date_first(now())` → First day of current month                           |
| `date_range(start, end[, step])` | Creates array of dates           | `start`, `end`: Dates, `step`: Duration string or int+unit | Array of dates    | `date_range("2023-01-01", "2023-01-03", "1d")` → `["2023-01-01", "2023-01-02", "2023-01-03"]` |
| `date_timezone(date, timezone)`  | Converts time to specified [IANA Timezone](https://en.wikipedia.org/wiki/List_of_tz_database_time_zones) | `date`: Time object<br>`timezone`: Timezone string (e.g., "UTC", "America/New_York") | Time object | `date_timezone(now(), "America/New_York")` → Time in NY timezone |

*Date function `unit`/`part`/`period` parameters often accept: "year", "month", "week", "day", "hour", "minute", "second".*
*`range` function with dates requires time objects as start/end.*


## Value Handling Functions

| Function                     | Description                        | Parameters                                       | Returns                | Example                                                   |
| ---------------------------- | ---------------------------------- | ------------------------------------------------ | ---------------------- | --------------------------------------------------------- |
| `cast(value, type)`          | Converts value to type             | `value`, `type` ("int", "float", "string", "bool") | Converted value or error | `cast("42", "int")` → `42`                                |
| `coalesce(val1, val2, ...)`  | Returns first non-null value       | Multiple values                                  | First non-null value   | `coalesce(null, sync.val, state.val, "default")`        |
| `element(array, index)`      | Gets element at 0-based index      | `array`, `index` (integer)                       | Element or error       | `element(["a", "b"], 1)` → `"b"`                            |
| `equals(val1, val2)`         | Checks deep equality             | `val1`, `val2`                                   | Boolean                | `equals(response.status, 200)` → `true`                   |
| `first_valid(val1, val2, ...)`     | Returns first non-null and non-empty value      | Multiple values                                  | First valid value  | `first_valid("", state.val, "default")` → `state.val` (if not empty) |
| `if(condition, then, else)`  | Conditional expression             | `condition` (bool), `then_val`, `else_val`       | Selected value         | `if(state.count > 0, "has items", "empty")`             |
| `is_empty(value)`            | Checks if value is empty           | `value` (string, array, map)                     | Boolean                | `is_empty("")` → `true`, `is_empty([])` → `true`            |
| `is_null(value)`             | Checks if value is null            | `value`                                          | Boolean                | `is_null(state.optional_param)` → `true` or `false`         |
| `null_if(value, sentinel, ...)` | Returns null when value equals any sentinel, else the value | `value`, one or more `sentinel` values | Value or null | `null_if(record.amount, false)` → `null` (if value is `false`) |
| `require(val[, error_msg])`  | Ensures value is not null or error | `val`, `error_msg` (optional)                    | Value or error         | `require(secrets.api_key, "API Key is required")`         |
| `try_cast(value, type)`      | Tries conversion, returns null     | `value`, `type`                                  | Converted value or null | `try_cast("abc", "int")` → `null`                         |

## Collection Functions

| Function                       | Description                     | Parameters                                | Returns           | Example                                                        |
| ------------------------------ | ------------------------------- | ----------------------------------------- | ----------------- | -------------------------------------------------------------- |
| `array(val1, val2, ...)`       | Creates array from values       | Multiple values                           | Array             | `array(1, 2, 3)` → `[1, 2, 3]`                                 |
| `chunk(array\|queue, size)`    | Splits array/queue into chunks  | `array` or `queue.name`, `size` (int)     | Channel of arrays | Used in `iterate`: `over: chunk(queue.ids, 50)`                |
| `filter(array, expression)`    | Filters array using expression| `array`, `expression` (string condition) | Filtered array    | `filter([1, 2, 3], "element > 1")` → `[2, 3]`                  |
| `get_path(object, path)`       | Gets value using dot notation   | `object`, `path` (string, e.g., "a.b[0]") | Value at path     | `get_path(response.json, "user.profile.email")`              |
| `jmespath(object, expression)` | Evaluates JMESPath expression   | `object`, `expression` (string)           | Query result      | `jmespath(response.json, "data.items[?age > 30]")`            |
| `jq(object, expression)`      | Evaluates jq expression         | `object`, `expression` (jq filter string) | Array of results  | `jq(response.json, ".data.items[] \| select(.age > 30)")`     |
| `keys(map)`                    | Gets keys from map              | `map` object                              | Array of keys     | `keys({"a": 1, "b": 2})` → `["a", "b"]`                       |
| `exists(collection, item)`     | Checks if key exists in map or value exists in array | `collection`: Map or array, `item`: Key or value to find | Boolean           | `exists({"a": 1}, "a")` → `true`, `exists([1, 2], 2)` → `true`   |
| `object(k1, v1, k2, v2, ...)`  | Creates object from key/value pairs | Even number of arguments (key, value pairs) | Map/object        | `object("name", "John", "age", 30)` → `{"name": "John", "age": 30}` |
| `pluck(data, column_name)`     | Extracts values from a column   | `data`: Array of maps, `column_name`: String | Array of values   | `pluck([{"name": "John", "age": 30}, {"name": "Jane", "age": 25}], "name")` → `["John", "Jane"]` |
| `explode(record, field_name)`  | Expands a record by unnesting an array field | `record`: Map with a nested array field, `field_name`: Key of the array to explode | Array of maps | `explode({"id": 1, "items": [{"k": "a"}, {"k": "b"}]}, "items")` → `[{"id": 1, "k": "a"}, {"id": 1, "k": "b"}]` |
| `range(start, end[, step])`    | Generates range (auto-detects type) | `start`, `end`: Numbers or dates, `step`: Optional | Array or channel  | `range(1, 5)` → `[1,2,3,4,5]`, delegates to `int_range` or `date_range` |
| `sort(array[, descending])`    | Sorts array elements            | `array`, `descending` (optional bool)     | Sorted array      | `sort([3, 1, 2])` → `[1, 2, 3]`                               |
| `values(map)`                  | Gets values from map            | `map` object                              | Array of values   | `values({"a": 1, "b": 2})` → `[1, 2]`                         |
| `object_rename(map, old, new, ...)` | Renames keys in map          | `map`: Object, followed by pairs of `old_key`, `new_key` | Modified map      | `object_rename({"a": 1, "b": 2}, "a", "x")` → `{"x": 1, "b": 2}` |
| `object_delete(map, key1, ...)`  | Deletes keys from map           | `map`: Object, followed by keys to delete | Modified map      | `object_delete({"a": 1, "b": 2}, "a")` → `{"b": 2}`             |
| `object_casing(map, casing)`  | Transforms keys in map to specified casing | `map`: Object, `casing`: "snake", "camel", "upper", "lower" | Modified map      | `object_casing({"firstName": "John"}, "snake")` → `{"first_name": "John"}` |
| `object_merge(map1, map2, ...)`       | Merges multiple maps together   | Two or more maps to merge (later maps override earlier) | Merged map        | `object_merge({"a": 1}, {"b": 2}, {"c": 3})` → `{"a": 1, "b": 2, "c": 3}` |
| `object_zip(keys, values)`            | Creates object from key/value arrays | `keys`: Array of strings, `values`: Array of values (matched by position) | Map/object        | `object_zip(["a", "b"], [1, 2])` → `{"a": 1, "b": 2}` |

> **`jq()` vs `jmespath()`:** Both query JSON data, but use different syntax. `jmespath()` (alias: `jp()`) uses JMESPath syntax and returns a single value. `jq()` uses [jq filter syntax](https://jqlang.github.io/jq/manual/) (dot-prefix paths, pipe operators, `select()` for filtering) and always returns an array of results. In processor expressions, append `[0]` to get a single value from `jq()`: `jq(record, ".user.name")[0]`. In `response.records.jq`, the array is handled natively.

## Encoding/Decoding Functions

| Function                    | Description             | Parameters                        | Returns         | Example                                          |
| --------------------------- | ----------------------- | --------------------------------- | --------------- | ------------------------------------------------ |
| `encode_url(string)`        | URL encodes a string    | `string`                          | Encoded string  | `encode_url("a b")` → `"a%20b"`                     |
| `decode_url(string)`        | URL decodes a string    | `string`                          | Decoded string  | `decode_url("a%20b")` → `"a b"`                    |
| `encode_base64(string)`     | Base64 encodes string   | `string`                          | Encoded string  | `encode_base64("user:pass")` → `"dXNlcjpwYXNz"`    |
| `decode_base64(string)`     | Base64 decodes string   | `string`                          | Decoded string  | `decode_base64("dXNlcjpwYXNz")` → `"user:pass"`   |
| `hash(string[, algorithm])` | Creates hash of string  | `string`, `algorithm` ("md5", "sha1", "sha256", default "md5") | Hash string (hex) | `hash("hello", "md5")` → `"5d4..."`              |
| `json_parse(string)`        | Parses JSON string      | `string`: JSON string             | Object/Array/Value | `json_parse('{"name": "John"}')` → `{"name": "John"}` |

## Utility Functions

| Function                               | Description                       | Parameters                            | Returns                | Example                                                       |
| -------------------------------------- | --------------------------------- | ------------------------------------- | ---------------------- | ------------------------------------------------------------- |
| `uuid()`                               | Generates random UUID v4          | None                                  | UUID string            | `uuid()` → `"..."`                                              |
| `log(message)`                         | Logs a message during evaluation  | `message`                             | The message (passthru) | `log("Processing record: " + record.id)`                      |
| `regex_match(string, pattern)`         | Checks if string matches pattern  | `string`, `pattern` (Go regex)        | Boolean                | `regex_match("img_123.jpg", "^img_.*\\.jpg$")` → `true`        |
| `regex_extract(string, pattern[, idx])`| Extracts matches using regex      | `string`, `pattern`, `idx` (optional) | Matches/group or null  | `regex_extract("id=123", "id=(\\d+)", 1)` → `"123"`           |
| `regex_replace(string, pattern, replacement)` | Replaces pattern matches    | `string`, `pattern` (Go regex), `replacement` | Modified string        | `regex_replace("hello123world", "\\d+", "-")` → `"hello-world"` |
| `pretty_table(data)`             | Renders array of maps as table   | `data`: Array of maps                            | Formatted table string | `pretty_table([{"name": "John", "age": 30}])` → formatted table |
| `type_of(value)`                 | Gets type of value               | `value`                                          | Type string            | `type_of(42)` → `"integer"`                               |
| `conn_property(connection, key)` | Gets property value from connection | `connection`: Connection name (string)<br>`key`: Property name (string) | Property value | `conn_property("my_db", "host")` → `"localhost"` |
| `machine_stats()`                | Gets machine resource statistics | None                                             | Object with stats      | `machine_stats()` → `{"memory_percent": 45.2, "cpu_percent": 23.5}` |

```

<!-- source: sling-docs/concepts/api-specs.md -->
# Sling API Specs

## Sling API Specs

Sling API specs are YAML files that define how to interact with REST APIs. They provide a structured way to specify authentication methods, define endpoints, configure request parameters, handle pagination, process responses, manage state for incremental loads, and more.

{% hint style="success" %}

**CLI Pro Required**: APIs require a [CLI Pro token](../sling-cli/cli-pro.md) or [Platform Plan](./sling-platform/platform.md).

{% endhint %}

## Quick Start to build a Spec

Here's a complete, working example of a Sling API spec that fetches user data from a REST API:

```yaml
name: "Example API"
description: "Simple API to get user data"

defaults:
  state:
    base_url: https://api.example.com/v1

  request:
    headers:
      Accept: "application/json"
      Authorization: "Bearer {secrets.api_token}"  # read from env.yaml secrets section

endpoints:
  users:
    description: "Retrieve list of users"
    docs: https://docs.example.com/users # docs url for endpoint
    disabled: false                      # Set to true to temporarily disable
    
    # State variables control request parameters and track values between runs
    state:
      page: 1          # Start at page 1
      limit: 100       # Fetch 100 records per request
    
    request:
      # Will resolve to https://api.example.com/v1/users
      url: '{state.base_url}/users'
      parameters:
        page: '{state.page}'
        limit: '{state.limit}'
    
    # Control how to fetch next pages
    pagination:
      next_state:
        page: '{state.page + 1}'  # Increment page number for next request
      stop_condition: "length(response.records) < state.limit"  # Stop when page isn't full
    
    # Define how to extract and process response data
    response:
      records:
        jmespath: "data.users[]"  # Extract array of users from response
        primary_key: ["id"]       # Use 'id' field to deduplicate records
```

To run this spec with Sling:

1. Save your spec somewhere (local disk, S3, SFTP, http). You can access your spec by the [location string](../../sling-cli/environment.md#location-string) convention.

2. Create a connection in your [env.yaml](../../sling-cli/environment.md#sling-env-file-envyaml) file, like this:

```yaml
connections:
  my_api:
    type: api
    spec: file:///path/to/my_api.spec.yaml  # or Github repo file, HTTP URL
    secrets:
      api_key: xxxxxxxxxxxxxxxxxx
  
  my_postgres:
    url: postgres://....
```


3. Create a replication

```yaml
source: my_api
target: my_postgres

streams:
   # '*' will read from all endpoints (endpoint name = stream name)
   # or you can simply input the specific endpoint names
  '*':
    object: my_schema.{stream_name}
    mode: full-refresh
```

4.  Run Replication.

```bash
sling run -r replication.api_postgres.yaml
```

That's it!

{% hint style="info" %}
**Build Specs with AI:** You can use AI to automatically research APIs and build specs for you. See [Using AI to build API specs](../sling-cli/ai.md#building-a-custom-api-connector) for a guided workflow.
{% endhint %}


## Official Specs

We are actively building the number of "official" Sling API Specs offered, such as:

- [Airtable](../connections/api-connections/airtable.md)
- [HubSpot](../connections/api-connections/hubspot.md)
- [Stripe](../connections/api-connections/stripe.md)
- [Github](../connections/api-connections/github.md)

Go [here](../connections/api-connections/README.md) to see the full list.

Official specs don't require a full URL or Path. They are maintained by the Sling team and can be fetched by specifying the corresponding ID, such as `stripe`, `salesforce` or `github`.

```yaml
connections:
  stripe_api:
    type: api
    spec: stripe  # Use official stripe spec
    secrets:
      api_key: sk_live_xxxxxx
```
{% hint style="info" %}

If you'd like us to build a new spec, please submit a request by filling out this [Google Form](https://docs.google.com/forms/d/e/1FAIpQLScIgkpTC6C-nWaW6atc5eLFl3uUNiIakw37WL69HuPUks08aQ/viewform?usp=dialog), or by submitting a new [Github issue](https://github.com/slingdata-io/sling-cli/issues) (choosing the API Spec option).

{% endhint %}

### Forking Official Specs

When you run a connection test or use an official spec, Sling automatically downloads and caches the spec file locally at:

```
$SLING_HOME_DIR/api/specs/{spec_id}.yaml
```

For example, if you're using the `stripe` spec, after running `sling conns test my_stripe_conn`, you'll find the spec at `~/.sling/api/specs/stripe.yaml`.

To fork and customize an official spec:

1. Run a connection test to download the spec:
   ```bash
   sling conns test my_stripe_conn
   ```

2. Copy the spec from the cache folder:
   ```bash
   cp ~/.sling/api/specs/stripe.yaml ./my_custom_stripe.spec.yaml
   ```

3. Modify the spec as needed for your use case

4. Update your connection to use your custom spec:
   ```yaml
   connections:
     my_stripe:
       type: api
       spec: file://./my_custom_stripe.spec.yaml
       secrets:
         api_key: sk_live_xxxxxx
   ```

## API Spec Documentation

This section covers the details of building Sling API specifications:

*   **[Structure](api/structure.md):** Understand the fundamental YAML structure of API specs, including endpoints, state management, sync variables, stream overrides, and lifecycle sequences (setup/teardown).

*   **[Authentication](api/authentication.md):** Configure authentication methods including Bearer tokens, Basic Auth, OAuth2, AWS Signature V4, and custom sequence-based authentication workflows.

*   **[Requests & Iteration](api/request.md):** Define HTTP requests with URLs, methods, headers, parameters, and payloads. Configure iteration to loop requests over data sets or queues, and use setup/teardown sequences for multi-step workflows.

*   **[Response Processing](api/response.md):** Process API responses in multiple formats (JSON, CSV, XML, JSON Lines), extract records with JMESPath, configure deduplication strategies, and access response state for pagination and rules.

*   **[Queues](api/queues.md):** Pass data between endpoints using queues for multi-step extraction workflows, such as collecting IDs from one endpoint to use in detail requests from another endpoint.

*   **[Dynamic Endpoints](api/dynamic-endpoints.md):** Programmatically generate multiple endpoint configurations based on runtime data, perfect for APIs where available resources aren't known until queried.

*   **[Advanced Features](api/advanced.md):** Master pagination strategies (cursor, offset, page-based), response processors for data transformation, sync state for incremental loads, and rules for error handling with intelligent retry and backoff strategies.

*   **[Expression Functions](./functions.md):** Leverage built-in functions within expressions (`{...}`) for data manipulation, date operations, type casting, string operations, and control flow throughout your API spec configuration.

*   **[Testing & Debugging](api/testing-debug.md):** Test your API specs using the `sling conns test` command with `--debug` and `--trace` flags to inspect request/response details, troubleshoot issues, and optimize your configuration.

*   **[Troubleshooting](api/troubleshooting.md):** Common error messages, debugging techniques, and solutions for authentication, pagination, and JMESPath issues.

## API Workflow Overview

```mermaid
graph TD
    A[Define API Spec YAML] --> B[Configure Authentication]
    B --> C[Define Endpoints]
    C --> D[Configure Requests]
    D --> E[Setup Response Processing]
    E --> F[Handle Pagination]
    F -->|Optional| G[Setup Queues]
    G -->|Optional| H[Configure Additional Endpoints]
    E -->|Optional| I[Define Incremental Sync]

    style A fill:#4a9eff,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style B fill:#ff8c42,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style C fill:#7cb342,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style D fill:#ffd54f,stroke:#ffffff,stroke-width:2px,color:#000000
    style E fill:#26c6da,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style F fill:#ab47bc,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style G fill:#5c6bc0,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style H fill:#7cb342,stroke:#ffffff,stroke-width:2px,color:#ffffff
    style I fill:#ab47bc,stroke:#ffffff,stroke-width:2px,color:#ffffff
```

For practical examples, see the [API Examples](../../examples/api/README.md) section.

## Common Use Cases

| Use Case | Key Features |
|----------|-------------|
| Simple data extraction | Basic endpoint definition, JMESPath extraction |
| Paginated data | Pagination configuration with `next_state` and `stop_condition` |
| Incremental updates | `sync` state variables, timestamp filtering |
| Dependent requests | Queues, iteration over IDs |
| Authentication | Bearer tokens, Basic auth, OAuth2 |
| Rate limiting | Response rules, backoff strategies |

> 💡 **Tip:** Start with a minimal working spec and gradually add more advanced features as needed.

<!-- source: sling-docs/concepts/data-quality.md -->
# Data Quality

Sling provides several powerful features to ensure and maintain data quality throughout your data pipeline:

## Constraints

[Constraints](./constraints.md) allow you to validate data at ingestion time using SQL-like syntax. They're specified at the column level and can prevent invalid data from entering your system.

```yaml
streams:
  my_stream:
    columns:
      # Ensure email is valid
      email: string | value ~ '^[^@]+@[^@]+\.[^@]+$'
      
      # Status can only be specific values
      status: string | value in ('active', 'pending', 'inactive')
      
      # Amount must be positive
      amount: decimal | value > 0
```

## Check Hooks

[Check](../hooks/check.md) hooks enable you to implement custom validation logic at any point in your pipeline. They're particularly useful for:
- Validating row counts
- Ensuring data freshness
- Implementing complex business rules

```yaml
hooks:
  post:
    # Ensure minimum row count
    - type: check
      check: "run.total_rows >= 1000"
      on_failure: abort
    
    # Verify recent data
    - type: check
      check: "state.freshness_check.result.last_update >= timestamp.unix - 3600"
      on_failure: warn
```

## 3. Query Hooks

[Query](../hooks/query.md) hooks allow you to run SQL-based quality checks and store results:

```yaml
hooks:
  post:
    - type: query
      connection: target_db
      query: |
        INSERT INTO quality_metrics (
          stream_name,
          check_time,
          null_rate,
          duplicate_rate
        )
        SELECT 
          '{run.stream.name}',
          CURRENT_TIMESTAMP,
          SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
          COUNT(*) - COUNT(DISTINCT id) * 100.0 / COUNT(*)
        FROM {run.object.full_name}
```

## Failure Handling

All data quality features support flexible failure handling:
- `abort`: Stop processing immediately
- `warn`: Continue but emit a warning
- `skip`: Skip the problematic record (for constraints)
- `quiet`: Continue silently

This allows you to implement the appropriate level of strictness for your use case.

By combining these features, you can build robust data quality checks throughout your pipeline, from ingestion to final delivery.

<!-- source: sling-docs/concepts/change-capture.md -->
# Change Capture (CDC)

## Change Capture (CDC)

Change Data Capture (CDC) continuously replicates row-level changes (inserts, updates, deletes) from a source database to a target database by reading the database's transaction log. Unlike `incremental` mode which polls for new or updated rows, CDC captures every change as it happens, including deletes.

{% hint style="success" %}

**Advanced Plan Required**: CDC requires a [CLI Pro Max](../sling-cli/cli-pro.md) token or an [Advanced Platform Plan](../sling-platform/platform.md).

{% endhint %}

## Supported Sources

| Source | Transaction Log | Status |
|--------|----------------|--------|
| [MySQL](change-capture/mysql.md) | Binary log (binlog) | Available |
| [MariaDB](change-capture/mysql.md) | Binary log (binlog) | Available |
| [PostgreSQL](change-capture/postgres.md) | Write-Ahead Log (WAL) | Available |
| [SQL Server](change-capture/sql-server.md) | CDC change tables | Available |
| [Oracle](change-capture/oracle.md) | GoldenGate Data Streams | Available |
| [MongoDB](change-capture/mongodb.md) | Change Streams (oplog) | Available |

## How It Works

CDC operates in two phases: an **initial load** that copies existing data, followed by **incremental change capture** that streams ongoing changes.

### Phase 1: Initial Load

On the first run for a given stream, Sling performs a full table copy from source to target:

1. **Position capture** — Sling records the current transaction log position (e.g., binlog file + offset) _before_ the snapshot begins. This ensures no changes are lost between the snapshot and the first incremental run.
2. **Chunked reading** — Large tables are automatically split into primary-key-range chunks (configurable via `snapshot_chunk_size`). Each chunk is read, written, and checkpointed independently.
3. **Resumability** — If the process is interrupted (crash, timeout, kill), the next run detects the in-progress snapshot and resumes from the last completed chunk. No data is re-read.
4. **Completion** — Once all chunks are written, Sling marks the initial load as complete in the state store.

{% hint style="info" %}
Chunked mode requires an integer-like primary key for range splitting. If the table has no primary key or the PK is non-numeric, Sling falls back to a single-shot full table read automatically.
{% endhint %}

### Phase 2: Incremental Changes

On subsequent runs, Sling reads the source database's transaction log from the last saved position:

1. **Read changes** — Reads inserts, updates, and deletes from the transaction log starting at the saved position, up to `run_max_events` or `run_max_duration`.
2. **Merge to target** — Applies changes to the target table using a merge strategy that handles inserts, updates, and deletes.
3. **Save position** — Persists the new log position in the state store so the next run picks up where this one left off.

Each run is bounded and exits after processing its batch. This makes CDC safe to schedule on a recurring interval (e.g., every 30 seconds or every 5 minutes) via cron or the Sling Platform.

### What Is a CDC Event?

A CDC event corresponds to a single statement in the transaction log, not a single row. A bulk insert like `INSERT INTO t VALUES (...), (...), (...)` produces **one event** containing multiple rows. Similarly, an `UPDATE ... WHERE status = 'old'` that modifies 1,000 rows is a single event with 1,000 row changes.

This means `run_max_events: 10000` does not necessarily equal 10,000 rows — it could represent significantly more rows depending on how the source application writes data. Keep this in mind when tuning `run_max_events` for high-throughput workloads.

### Lifecycle Diagram

```
First Run                          Subsequent Runs
─────────                          ────────────────
┌─────────────────────┐            ┌──────────────────────┐
│ Record log position │            │ Read from saved      │
│ (before snapshot)   │            │ log position         │
└─────────┬───────────┘            └──────────┬───────────┘
          │                                   │
┌─────────▼───────────┐            ┌──────────▼───────────┐
│ Read chunk 1        │            │ Capture changes      │
│ Write to target     │            │ (inserts, updates,   │
│ Checkpoint          │            │  deletes)            │
└─────────┬───────────┘            └──────────┬───────────┘
          │                                   │
┌─────────▼───────────┐            ┌──────────▼───────────┐
│ Read chunk 2...N    │            │ Merge into target    │
│ Write + checkpoint  │            │ table                │
└─────────┬───────────┘            └──────────┬───────────┘
          │                                   │
┌─────────▼───────────┐            ┌──────────▼───────────┐
│ Mark initial load   │            │ Save new log         │
│ complete            │            │ position             │
└─────────────────────┘            └──────────────────────┘
```

## Replication Structure

CDC is configured using `mode: change-capture` in a standard Sling replication file. CDC-specific options go under `change_capture_options`:

```yaml
source: MY_SOURCE
target: MY_TARGET

defaults:
  mode: change-capture
  primary_key: [id]
  object: public.{stream_table}
  change_capture_options:
    run_max_events: 10000       # max change events per run
    run_max_duration: 10m       # max wall-clock time per run
    soft_delete: false          # keep deleted rows marked with _sling_synced_op='D'
    snapshot_start: now         # 'now' or 'beginning' for the very first run
    snapshot_chunk_size: 100000 # rows per chunk during initial snapshot
    snapshot_run_duration: 30m  # cap time spent on initial snapshot per run
    # replay_from: "2025-01-01T00:00:00Z"  # rewind to re-process from a point in time (beta)
    # slot_level: shared        # 'shared' (one slot for all streams) or 'stream' (one per table)
    retry_attempts: 3           # retries on transient failures
    retry_delay: 5s             # delay between retries

streams:
  my_database.users:

  my_database.orders:
    change_capture_options:
      soft_delete: true         # per-stream override: preserve deleted rows
      run_max_events: 50000     # higher event budget for a busy table
      run_max_duration: 5m      # shorter run window

  my_database.products:
    change_capture_options:
      snapshot_start: beginning         # capture full history from earliest log position
      snapshot_chunk_size: 50000        # smaller chunks for a wide table
      snapshot_run_duration: 30m        # resume snapshot across runs
```

Options set in `defaults.change_capture_options` apply to all streams. Per-stream `change_capture_options` override the defaults.

## Options Reference

<table data-full-width="false"><thead><tr><th width="240px">Key</th><th>Description</th></tr></thead><tbody>
<tr><td><code>run_max_events</code></td><td>Maximum number of change events to process per run. When this limit is reached, Sling saves the position and exits. Default is <code>10000</code>.</td></tr>
<tr><td><code>run_max_duration</code></td><td>Maximum duration per run (e.g., <code>30s</code>, <code>10m</code>, <code>1h</code>). If no events arrive within this window, the run completes with zero changes. Default is <code>10m</code>.</td></tr>
<tr><td><code>soft_delete</code></td><td>When <code>true</code>, DELETE events mark the row with <code>_sling_synced_op = 'D'</code> and update <code>_sling_synced_at</code> instead of removing the row. Default is <code>false</code>.</td></tr>
<tr><td><code>snapshot_start</code></td><td>Where to start reading the transaction log on the very first run. Default is <code>now</code>. Use <code>beginning</code> to read from the earliest available log position.</td></tr>
<tr><td><code>snapshot_chunk_size</code></td><td>Number of rows per chunk during the initial snapshot. Default is <code>100000</code>.</td></tr>
<tr><td><code>snapshot_run_duration</code></td><td>Maximum time to spend on the initial snapshot per run (e.g., <code>30m</code>, <code>1h</code>). When the budget is exhausted, Sling exits cleanly after the current chunk and resumes on the next run. Default: no limit.</td></tr>
<tr><td><code>replay_from</code></td><td>Rewind the CDC position to re-process changes from an earlier point. Accepts source-specific formats (e.g., RFC 3339 timestamp, binlog position, GTID set). Applied once per unique value.</td></tr>
<tr><td><code>slot_level</code></td><td>Controls how the source's replication reader is scoped across the streams in a replication. <code>shared</code> uses a single replication slot (PostgreSQL) or binlog reader (MySQL) for all streams in the group, so every stream converges to the same unified transaction-log position — giving a point-in-time-consistent view across tables and reading the log only once. <code>stream</code> uses one slot/reader per table (independent positions). Defaults to <code>shared</code> on sources that support a shared reader (<strong>PostgreSQL</strong>, <strong>MySQL</strong>/MariaDB) and <code>stream</code> on all others. Setting <code>shared</code> on a source without a shared reader (e.g. SQL Server, Oracle, MongoDB) is ignored and falls back to per-stream.</td></tr>
<tr><td><code>retry_attempts</code></td><td>Number of retry attempts on transient failures. Default is <code>3</code>.</td></tr>
<tr><td><code>retry_delay</code></td><td>Delay between retries. Default is <code>5s</code>.</td></tr>
</tbody></table>

## CDC Metadata Columns

Sling adds three metadata columns to every CDC-managed target table:

| Column | Type | Description |
|--------|------|-------------|
| `_sling_synced_at` | `timestamptz` | Timestamp when the row was last synced. |
| `_sling_synced_op` | `varchar` | The operation type: `S` (snapshot), `I` (insert), `U` (update), `D` (delete). When `soft_delete: true`, deleted rows are preserved with `_sling_synced_op = 'D'`. |
| `_sling_cdc_seq` | `bigint` | Monotonically increasing sequence number for ordering events within and across runs. |

## State Management

CDC state is stored in the connection specified by the `SLING_STATE` environment variable. This tracks:

- The current transaction log position
- Whether the initial snapshot is complete
- Checkpoint progress for in-progress snapshots
- Total rows captured

```bash
# Store state in a PostgreSQL table
export SLING_STATE='MY_POSTGRES/sling_state'

# Or store state in a file
export SLING_STATE='MY_AWS/sling_state'
```

{% hint style="warning" %}
The `SLING_STATE` connection must be configured before running CDC replications. Without it, Sling cannot track positions and each run would repeat the initial snapshot. See location string details [here](../sling-cli/environment.md#location-string).
{% endhint %}

## Scheduling

CDC is designed to be run repeatedly on a schedule. Each run processes a bounded batch of changes (controlled by `run_max_events` and `run_max_duration`) and exits.

Configure the replication in the [Sling Platform](../sling-platform/platform.md) UI with a schedule interval. The platform handles orchestration, monitoring, and alerting automatically.

![Sling Platform UI](../sling-platform/sling-platform-ui.png)

## Comparison with Other Modes

| Feature | `change-capture` | `incremental` | `full-refresh` |
|---------|-----------------|---------------|----------------|
| Captures inserts | Yes | Yes | Yes |
| Captures updates | Yes | Yes (with `update_key`) | Yes |
| Captures deletes | Yes | With `delete_missing` | Yes |
| Reads from | Transaction log | Table query | Table query |
| State tracking | Log position | Max update_key value | None |
| Source load | Minimal (reads log) | Queries table | Full table scan |
| Initial setup | Automatic snapshot | Manual first load | N/A |

<!-- source: sling-docs/connections/database-connections/postgres.md -->
# Postgres

## Postgres

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` **(required)** -> The username to access the instance
* `database` **(required)** -> The database name of the instance
* `schema` (optional)  -> The default schema to use
* `password` (optional) -> The password to access the instance
* `port` (optional) -> The port of the instance. Default is `5432`.
* `role` (optional) -> The role to access the instance
* `statement_timeout` (optional in *v1.4.21+*) -> The timeout to use for a postgres query, input as millisecond (`10000` for 10 seconds).
* `sslmode` (optional) -> The sslmode of the instance (`disable`, `allow`, `prefer`, `require`, `verify-ca` or `verify-full`). Default is `disable`.
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

### Google Cloud SQL IAM Authentication (*v1.4.25+*)

For Google Cloud SQL PostgreSQL instances with IAM authentication enabled, you can use the following properties:

* `gcp_instance` **(required)** -> The Cloud SQL instance name
* `gcp_project` **(required)** -> The GCP project ID
* `gcp_region` **(required)** -> The GCP region (e.g., `us-central1`)
* `user` **(required)** -> The IAM user or service account email (will be converted to lowercase and trimmed)
* `gcp_key_file` (optional) -> Path to the GCP service account credentials JSON file
* `gcp_key_body` (optional) -> The GCP service account credentials JSON content as a string
* `gcp_lazy_refresh` (optional) -> Enable lazy refresh for serverless environments (`true` or `false`). Default is `false`.
* `gcp_use_private_ip` (optional) -> Use private IP for the connection (`true` or `false`). Default is `false`.

**Authentication Methods** (in priority order):
1. If `gcp_key_body` is provided, uses the JSON credentials directly
2. If `gcp_key_file` is provided, uses the credentials file
3. Otherwise, uses Application Default Credentials (ADC)

**Important Notes**:
- When using IAM authentication, **do NOT provide a `password`** - authentication is handled via IAM tokens
- The `user` field should match the user identifier in the Users panel of the Cloud SQL instance (e.g., `my-sa@my-project`). 
- The service account must have the `Cloud SQL Instance User` role
- IAM authentication must be enabled on the Cloud SQL instance: `gcloud sql instances patch INSTANCE --database-flags=cloudsql.iam_authentication=on`

{% hint style="warning" %}

If you're getting error: `FATAL: password authentication failed`, this error returned from CloudSQL is misleading. It usually means the user is not found, or the user identifier provided is wrong. Make sure it matches what shows in the Cloud SQL Users panel (on GCP's website). Also ensure the IAM user is actually created. You **must** create a user even if providing a service account.

{% endhint %}

### Using `sling conns`

Here are examples of setting a connection named `POSTGRES`. We must provide the `type=postgres` property:

{% code overflow="wrap" %}
```bash
$ sling conns set POSTGRES type=postgres host=<host> user=<user> database=<database> password=<password> port=<port> 

# OR use url
$ sling conns set POSTGRES url="postgresql://myuser:mypass@host.ip:5432/mydatabase?sslmode=require&role=<role>"

# For Google Cloud SQL with IAM authentication
$ sling conns set POSTGRES type=postgres \
  gcp_instance=my-instance \
  gcp_project=my-project \
  gcp_region=us-central1 \
  user=my-sa@my-project.gserviceaccount.com \
  database=mydatabase \
  gcp_key_file=/path/to/credentials.json
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export POSTGRES='postgresql://myuser:mypass@host.ip:5432/mydatabase?sslmode=require&role=<role>'

export POSTGRES='{ type: postgres, user: "myuser", password: "mypass", host: "host.ip", port: 5432, database: "mydatabase", sslmode: "require", role: "<role>" }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  POSTGRES:
    type: postgres
    host: <host>
    user: <user>
    password: <password>
    port: <port>
    database: <database>
    sslmode: <sslmode>
    schema: <schema>
    role: <role>

  POSTGRES_URL:
    url: "postgresql://myuser:mypass@host.ip:5432/mydatabase?sslmode=require"

  POSTGRES_CLOUDSQL:
    type: postgres
    gcp_instance: my-instance
    gcp_project: my-project
    gcp_region: us-central1
    user: my-sa@my-project.gserviceaccount.com
    database: mydatabase
    gcp_key_file: /path/to/credentials.json
    # gcp_lazy_refresh: true  # Enable for serverless environments
    # gcp_use_private_ip: true  # Enable for private IP connections

  POSTGRES_CLOUDSQL:
    type: postgres
    gcp_instance: my-instance
    gcp_project: my-project
    gcp_region: us-central1
    user: my-sa@my-project.gserviceaccount.com
    database: mydatabase
    gcp_key_body: |
      { "type": "service_account", ... }
```

## Database user creation

To allow Sling to access your database, we need to create a user with the proper privileges. Please follow the steps below:

1.  Create a user `sling` (or whatever you prefer) by running:

    ```sql
    CREATE USER sling WITH PASSWORD '<password>';
    ```
2.  If you are planning to load data into this connection, grant the following privileges to that user:

    ```sql
    GRANT CREATE ON DATABASE <database_name> TO sling;
    ```
3.  If you are planning to extract data from this connection, you need to give permission to read the tables you'd like Sling to extract.

    ```sql
    -- Need this to read table & column names 
    GRANT SELECT ON ALL TABLES IN SCHEMA information_schema TO sling;
    GRANT SELECT ON ALL TABLES IN SCHEMA pg_catalog TO sling;

    -- run this to grant SELECT permission to all tables in schema `marketing` to user sling
    GRANT SELECT ON ALL TABLES IN SCHEMA marketing TO sling;
    ```


If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/snowflake.md -->
# Snowflake

## Snowflake

## Setup

The following credentials keys are accepted:

* `account` **(required)** -> The hostname or account the instance (eg. `pua90768.us-east-11`)
* `user` **(required)** -> The username to access the instance
* `database` **(required)** -> The database name of the instance
* `password` (optional)  -> The password to access the instance
* `schema` (optional)  -> The default schema to use
* `role` (optional) -> The role to access the instance
* `warehouse` (optional) -> The warehouse to use
* `passcode` (optional) -> Specifies the passcode provided by Duo when using multi-factor authentication (MFA) for login.
* `authenticator` (optional) -> Specifies the authenticator to use to login (e.g. `snowflake`, `snowflake_jwt`, `externalbrowser`, `oauth`, `programmatic_access_token`, `username_password_mfa`).
* `token` (optional since *v1.5.1*) -> Specifies the token for OAuth or PAT authentication. Required when using `authenticator=programmatic_access_token`.
* `private_key` (optional) -> Specifies the private key to use for `snowflake_jwt` authentication. Accepts a PEM-encoded key body, a Base64-encoded DER key (convenient for CI/CD environment variables, no PEM wrapping needed), or a file path to a key file.
* `private_key_passphrase` (optional) -> Specifies the private key file passphrase.
* `max_chunk_download_workers` (optional) -> Specifies the Maximum Number of Result Set Chunk Downloader (`integer`).
* `custom_json_decoder_enabled` (optional) -> Specifies to use the Custom JSON Decoder for Parsing Result Set (`true` or `false`).
* `internal_stage` (optional) -> Specifies a custom internal stage to use for bulk operations. If not provided, Sling will attempt to create a stage in the default schema named `SLING_SCHEMA.SLING_STAGING`.
* `copy_method` (optional) -> Specifies to use the platform to use for loading/unloading (`DEFAULT`, `AWS`, `AZURE`). For `AWS` or `AZURE`, you'll need to provide the necessary credentials, such as `aws_bucket`, `aws_access_key_id` and `aws_secret_access_key`, for AWS, or `azure_account`, `azure_container` and `azure_sas_svc_url` for AZURE.
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

### Using `sling conns`

Here are examples of setting a connection named `SNOWFLAKE`. We must provide the `type=snowflake` property:

{% code overflow="wrap" %}
```bash
$ sling conns set SNOWFLAKE type=snowflake account=<account> user=<user> database=<database> password=<password> role=<role>

# Or use url
$ sling conns set SNOWFLAKE url="snowflake://myuser:mypass@host.account/mydatabase?schema=<schema>&role=<role>"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export SNOWFLAKE='snowflake://myuser:mypass@host.account/mydatabase?schema=<schema>&role=<role>'

# use JSON format
export SNOWFLAKE_CONN='{ "type": "snowflake", "account": ..., "private_key": "<private-key-file-path>", "private_key_passphrase": "<passphrase>" }'

# use YAML format (with new lines)
export SNOWFLAKE='
type: snowflake
account: <account>
user: <user>
password: <password>
database: <database>
schema: <schema>
role: <role>
warehouse: <warehouse>
private_key: |
    -----BEGIN PRIVATE KEY-----
    MIIEvgIBADANBgkqhkiG7w0BAQEFAASCBKgwggSkAgEAAoIBAQDFWDdPxN7sKH/i
    ......
    SxUARJ4Rd2euQIEMqSY2UVPlNSaZK4wEq12jhXEM98cINVyKomJcThOHblz5IbV6
    +5I2kK6DCYSY2zm0xzYqeGFN
    -----END PRIVATE KEY-----
'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  # Standard password authentication
  SNOWFLAKE:
    type: snowflake
    account: <account>
    user: <user>
    password: <password>
    database: <database>
    schema: <schema>
    role: <role>
    warehouse: <warehouse>

  # Key-pair authentication
  SNOWFLAKE_JWT:
    type: snowflake
    account: <account>
    user: <user>
    database: <database>
    schema: <schema>
    role: <role>
    warehouse: <warehouse>
    authenticator: snowflake_jwt
    private_key: |
        -----BEGIN PRIVATE KEY-----
        MIIEvgIBADANBgkqhkiG7w0BAQEFAASCBKgwggSkAgEAAoIBAQDFWDdPxN7sKH/i
        ......
        SxUARJ4Rd2euQIEMqSY2UVPlNSaZK4wEq12jhXEM98cINVyKomJcThOHblz5IbV6
        +5I2kK6DCYSY2zm0xzYqeGFN
        -----END PRIVATE KEY-----

  # Programmatic Access Token (PAT) authentication
  SNOWFLAKE_PAT:
    type: snowflake
    account: <account>
    user: <user>
    database: <database>
    schema: <schema>
    role: <role>
    warehouse: <warehouse>
    authenticator: programmatic_access_token
    token: <token>  # Your PAT token

  # MFA authentication (with token caching)
  SNOWFLAKE_MFA:
    type: snowflake
    account: <account>
    user: <user>
    password: <password>
    database: <database>
    schema: <schema>
    role: <role>
    warehouse: <warehouse>
    authenticator: username_password_mfa

  # URL format
  SNOWFLAKE_URL:
    url: "snowflake://myuser:mypass@host.account/mydatabase?schema=<schema>&role=<role>"
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/bigquery.md -->
# BigQuery

## BigQuery

## Setup

The following credentials keys are accepted:

* `project` **(required)** -> The GCP project ID for the project
* `dataset` **(required)** -> The default dataset (like a schema)
* `gc_bucket` (optional) -> The Google Cloud Storage Bucket to use for loading (Recommended)
* `key_file` (optional) -> The path of the Service Account JSON. If not provided, the Google [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials) will be used.
* `key_body` (optional) -> The Service Account JSON key content as a string. You can also provide the JSON content in env var `GC_KEY_BODY`.
* `location` (optional) -> The location of the account, such as `US` or `EU`. Default is `US`.
* `extra_scopes` (optional) -> An array of strings, which represent scopes to use in addition to `https://www.googleapis.com/auth/bigquery`. e.g. `["https://www.googleapis.com/auth/drive", "https://www.googleapis.com/auth/spreadsheets"]`
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

{% hint style="warning" %}

If you'd like to have sling use the machine's Google Cloud Application Default Credentials (usually with `cloud auth application-default login`), don't specify a `key_file` (or the env var `GC_KEY_BODY`).

{% endhint %}

### Using `sling conns`

Here are examples of setting a connection named `BIGQUERY`. We must provide the `type=bigquery` property:

{% code overflow="wrap" %}
```bash
$ sling conns set BIGQUERY type=bigquery project=<project> dataset=<dataset> gc_bucket=<gc_bucket> key_file=/path/to/service.account.json location=<location>
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export BIGQUERY='{type: bigquery, project: my-google-project, gc_bucket: my_gc_bucket, dataset: public, location: US, key_file: /path/to/service.account.json}'
```
{% endcode %}

You can also provide Sling the Service Account JSON in `key_body` as a string, or via environment variable `GC_KEY_BODY`, instead of a `key_file`.

{% code overflow="wrap" %}
```bash
export GC_KEY_BODY='{"type": "service_account","project_id": ...........}'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  BIGQUERY:
    type: bigquery
    project: <project>
    dataset: <dataset>
    gc_bucket: <gc_bucket>
    key_file: '<key_file>'

  # using with `key_body` instead of `key_file`
  BIGQUERY:
    type: bigquery
    project: <project>
    dataset: <dataset>
    gc_bucket: <gc_bucket>
    key_body: |
      { "type": "service_account", ... } 
```

### BigQuery Table Partitioning

```yaml
streams:
  my_schema.another_table:
    object: my_dataset.{stream_table}
    target_options:
      table_keys:
        partition: [ DATE_TRUNC(transaction_date, MONTH) ]

# OR
streams:
  my_schema.another_table:
    object: my_dataset.{stream_table}
    target_options:
      table_ddl: |
         CREATE TABLE my_dataset.{stream_table} ({col_types}) 
          PARTITION BY
            DATE_TRUNC(transaction_date, MONTH)
            OPTIONS (
              partition_expiration_days = 3,
              require_partition_filter = TRUE)
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/mysql.md -->
# MySQL

## MySQL

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` **(required)** -> The username to access the instance
* `database` **(required)** -> The database name of the instance
* `schema` (optional)  -> The default schema to use
* `password` (optional) -> The password to access the instance. **If you're having issues with a special character in the password, try using the url by [encoding](https://www.w3schools.com/tags/ref_urlencode.ASP) the special character.**
* `port` (optional) -> The port of the instance. Default is `3306`.
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.
* `innodb_lock_wait_timeout` -> The timeout in seconds for InnoDB row lock waits (integer, default: `50`). Increase this value if you encounter `Lock wait timeout exceeded` errors during upsert/merge operations with concurrent writes.
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

### Google Cloud SQL IAM Authentication (*v1.4.25+*)

For Google Cloud SQL MySQL instances with IAM authentication enabled, you can use the following properties:

* `gcp_instance` **(required)** -> The Cloud SQL instance name
* `gcp_project` **(required)** -> The GCP project ID
* `gcp_region` **(required)** -> The GCP region (e.g., `us-central1`)
* `user` **(required)** -> The IAM user or service account name (short format, e.g., `my-sa`)
* `gcp_key_file` (optional) -> Path to the GCP service account credentials JSON file
* `gcp_key_body` (optional) -> The GCP service account credentials JSON content as a string
* `gcp_use_iam_auth` **(required)** -> Must be set to `true` to enable IAM authentication
* `gcp_lazy_refresh` (optional) -> Enable lazy refresh for serverless environments (`true` or `false`). Default is `false`.
* `gcp_use_private_ip` (optional) -> Use private IP for the connection (`true` or `false`). Default is `false`.

**Authentication Methods** (in priority order):
1. If `gcp_key_body` is provided, uses the JSON credentials directly
2. If `gcp_key_file` is provided, uses the credentials file
3. Otherwise, uses Application Default Credentials (ADC)

**Important Notes**:
- When using IAM authentication, **do NOT provide a `password`** - authentication is handled via IAM tokens
- The `user` field should be the **short format** (e.g., `my-sa`, not `my-sa@my-project.iam.gserviceaccount.com`)
- MySQL has a 32-character username limit
- The service account must have the `Cloud SQL Client` role

Additional Parameters. See [here](https://github.com/go-sql-driver/mysql?tab=readme-ov-file#parameters) for more details:

* `allow_all_files` -> Allows using LOCAL DATA INFILE without restrictions (`true` or `false`)
* `allow_cleartext_passwords` -> Permits sending passwords in clear text (`true` or `false`)
* `allow_fallback_to_plaintext` -> Allows fallback to unencrypted connection if server doesn't support TLS (`true` or `false`)
* `allow_native_passwords` -> Enables the native password authentication method (`true` or `false`)
* `allow_old_passwords` -> Allows the old insecure password method (`true` or `false`)
* `charset` -> Sets the charset for server-side prepared statements (e.g., `utf8mb4`)
* `check_conn_liveness` -> Checks connection liveness before using it (`true` or `false`)
* `collation` -> Sets the collation for server-side prepared statements (e.g., `utf8mb4_general_ci`)
* `client_found_rows` -> Returns number of matching rows instead of rows changed (`true` or `false`)
* `columns_with_alias` -> Prepares result columns as if they have an alias (`true` or `false`)
* `interpolate_params` -> Interpolates placeholders instead of using prepared statements (`true` or `false`)
* `loc` -> Sets the location for time.Time values (e.g., `Local`, `UTC`, or a time zone name)
* `time_truncate` -> Truncates time values to the given precision (`true` or `false`)
* `max_allowed_packet` -> Max packet size allowed (integer value in bytes)
* `multi_statements` -> Allows multiple statements in one query (`true` or `false`)
* `parse_time` -> Converts TIME/DATE/DATETIME to time.Time (`true` or `false`)
* `read_timeout` -> I/O read timeout (duration string, e.g., `30s`, `0.5m`, `1h`)
* `reject_read_only` -> Rejects read-only connections (`true` or `false`)
* `server_pub_key` -> Server public key name (string)
* `timeout` -> Timeout for establishing connections (duration string, e.g., `30s`, `0.5m`, `1h`)
* `write_timeout` -> I/O write timeout (duration string, e.g., `30s`, `0.5m`, `1h`)
* `connection_attributes` -> Connection attributes to send to MySQL (comma-separated list of key-value pairs)
* `tls` -> TLS configuration name (`true`, `false`, `skip-verify`, or `custom` when providing `cert_*` keys below)

Custom TLS Certificates:
* `cert_file` (optional) -> the client certificate to use to access the instance via TLS (file path or raw)
* `cert_key_file` (optional) -> the client key to use to access the instance via TLS (file path or raw)
* `cert_ca_file` (optional) -> the client CA certificate to use to access the instance via TLS (file path or raw)

### Using `sling conns`

Here are examples of setting a connection named `MYSQL`. We must provide the `type=mysql` property:

{% code overflow="wrap" %}
```bash
$ sling conns set MYSQL type=mysql host=<host> user=<user> database=<database> password=<password> port=<port> 

# OR use url (especially for special characters in the password)
$ sling conns set MYSQL url="mysql://myuser:mypass%24@host.ip:3306/mydatabase?tls=skip-verify"

# For Google Cloud SQL with IAM authentication
$ sling conns set MYSQL type=mysql \
  gcp_instance=my-instance \
  gcp_project=my-project \
  gcp_region=us-central1 \
  user=my-sa \
  database=mydatabase \
  gcp_use_iam_auth=true \
  gcp_key_file=/path/to/credentials.json
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export MYSQL='mysql://myuser:mypass@host.ip:3306/mydatabase?tls=skip-verify'
export MYSQL='{ type: mysql, user: "myuser", password: "mypass", host: "host.ip", port: 3306, database: "mydatabase", tls: "skip-verify" }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  MYSQL:
    type: mysql
    host: <host>
    user: <user>
    port: <port>
    database: <database>
    schema: <schema>
    password: <password>

  # use url (especially for special characters in the password)
  MYSQL_URL:
    url: "mysql://myuser:mypass%24@host.ip:3306/mydatabase?tls=skip-verify"

  MYSQL_CLOUDSQL:
    type: mysql
    gcp_instance: my-instance
    gcp_project: my-project
    gcp_region: us-central1
    user: my-sa  # Short format without domain suffix
    database: mydatabase
    gcp_use_iam_auth: true
    gcp_key_file: /path/to/credentials.json
    # gcp_lazy_refresh: true  # Enable for serverless environments
    # gcp_use_private_ip: true  # Enable for private IP connections
```

## Database user creation

To allow Sling to access your database, we need to create a user with the proper privileges. Please follow the steps below:

1.  First you'll need to login as a user with `CREATE USER` and `GRANT OPTION` privileges. Create a user `sling` (or whatever you prefer) by running :

    ```sql
    CREATE USER 'sling'@'%' IDENTIFIED BY '<password>';
    ```
2.  If you are planning to load data into this connection, you need to grant the following privileges to that user so we can create tables in schema _sling_:

    ```sql
    CREATE SCHEMA sling;
    GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, ALTER, CREATE TEMPORARY TABLES, CREATE VIEW ON sling.* TO 'sling'@'%';
    ```
3.  If you are planning to extract data from this connection, you need to give permission to read the tables you'd like Sling to extract.

    ```sql
    -- To give read access to all tables in a specific schema
    GRANT SELECT ON <schema_name>.* TO  'sling'@'%';
    ```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/sqlserver.md -->
# SQL Server

## SQL Server

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` (optional) -> The username to access the instance
* `database` (optional) -> The database name of the instance
* `instance` (optional)  -> The SQL Server instance to use
* `schema` (optional)  -> The default schema to use
* `password` (optional) -> The password to access the instance
* `port` (optional) -> The port of the instance. Default is `1433`.
* `authenticator` (optional) -> Can be used to specify use of a registered authentication provider. (e.g. `ntlm`, `winsspi` (on windows) or `krb5` (on linux))
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.
* `fedauth` (optional)  -> The Azure Active Directory authentication string (e.g. for **Fabric** connection). See [here](https://github.com/microsoft/go-mssqldb?tab=readme-ov-file#azure-active-directory-authentication) for more details. Accepted values:  `ActiveDirectoryDefault`, `ActiveDirectoryIntegrated`, `ActiveDirectoryPassword`, `ActiveDirectoryInteractive`, `ActiveDirectoryMSI`, `ActiveDirectoryManagedIdentity`, `ActiveDirectoryApplication`, `ActiveDirectoryServicePrincipal`, `ActiveDirectoryServicePrincipalAccessToken`, `ActiveDirectoryDeviceCode`, `ActiveDirectoryAzCli`.
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

### Additional Parameters

Sling uses the `go-mssqldb` library and thus will accept any paremters listed [here](https://github.com/denisenkom/go-mssqldb#connection-parameters-and-dsn). Some parameters that may be of interest:

- `encrypt` -> `strict`, `disable`, `false` or `true` - whether data between client and server is encrypted.
- `log` -> logging level (accepts `1`, `2`, `4`, `8`, `16`, `32`).
- `trusted_connection` -> `true` or `false` - whether to connects to SQL Server with a trusted connection using integrated security (also will use the `-T` flag when using bulk loading with `bcp`)
- `trust_server_certificate` -> `true` or `false` - whether the server certificate is checked
- `certificate` -> The file that contains the public key certificate of the CA that signed the SQL Server certificate. The specified certificate overrides the go platform specific CA certificates.
- `hostname_in_certificate` -> Specifies the Common Name (CN) in the server certificate. Default value is the server host.
- `server_spn` -> The kerberos SPN (Service Principal Name) for the server. Default is MSSQLSvc/host:port.
- `driver` -> A way to override the sql driver to connect with. Default is `sqlserver`. Other option is `azuresql`
- `bcp_auth_string` -> A way to override the way `bcp` [authenticates](https://learn.microsoft.com/en-us/sql/tools/bcp-utility?view=sql-server-ver16). Accepts an array of strings, for example `['U', 'username', '-T', '-G']` will append flags `-U username -T -G` to the `bcp` command.
- `bcp_extra_args` -> A way to add additional args to the `bcp` command. Accepts an array of strings, for example `['b', '5000']` will append flags `-b 5000` to the `bcp` command.
- `bcp_entra_auth` -> Appends the `-G` flag to the `bcp` command for MS Entra Auth.
- `bcp_path`-> The path to the `bcp` binary. This is useful if you need to specify a custom path for `bcp`.
- `bcp_azure_token_resource` (*v1.4.24+*) -> where using `fed_auth=ActiveDirectoryAzCli`, this indicates the resource to use to obtain an access token for BCP to use. Default is `https://database.windows.net`
- `az_path` -> The path to the `az` binary. This is useful if you need to specify a custom path for `az` to obtain an access token.

### Kerberos Parameters
* `authenticator` - set this to `krb5` to enable kerberos authentication. If this is not present, the default provider would be `ntlm` for unix and `winsspi` for windows.
* `krb5_config_file` (optional) - path to kerberos configuration file. Defaults to `/etc/krb5.conf`. Can also be set using `KRB5_CONFIG` environment variable.
* `krb5_realm` (required with keytab and raw credentials) - Domain name for kerberos authentication. Omit this parameter if the realm is part of the user name like `username@REALM`.
* `krb5_keytab_file` - path to Keytab file. Can also be set using environment variable `KRB5_KTNAME`. If no parameter or environment variable is set, the `DefaultClientKeytabName` value from the krb5 config file is used.
* `krb5_cred_cache_file` - path to Credential cache. Can also be set using environment variable `KRB5CCNAME`.
* `krb5_dns_lookup_kdc` - Optional parameter in all contexts. Set to lookup KDCs in DNS. Boolean. Default is true. 
* `krb5_udp_preference_limit` - Optional parameter in all contexts. 1 means to always use tcp. MIT krb5 has a default value of 1465, and it prevents user setting more than 32700. Integer. Default is 1.

Sling supports authentication via 3 methods. See [here](https://github.com/microsoft/go-mssqldb?tab=readme-ov-file#kerberos-parameters) for more details.

* Keytabs - Specify the username, keytab file, the krb5.conf file, and realm.
      
      authenticator: krb5
      krb5_realm: domain.com
      krb5_config_file: /etc/krb5.conf
      krb5_keytab_file: ~/user.keytab

* Credential Cache - Specify the krb5.conf file path and credential cache file path.

      authenticator=krb5
      krb5_config_file=/etc/krb5.conf
      krb5_cred_cache_file=~/user_cached_creds 

* Raw credentials - Specity krb5.confg, Username, Password and Realm.

      authenticator=krb5
      krb5_realm=comani.com
      krb5_config_file=/etc/krb5.conf

---

### Using `sling conns`

Here are examples of setting a connection named `MSSQL`. We must provide the `type=sqlserver` property:

{% code overflow="wrap" %}
```bash
$ sling conns set MSSQL type=sqlserver host=<host> user=<user> database=<database> password=<password> port=<port>

# Or use url
$ sling conns set MSSQL url="sqlserver://myuser:mypass@host.ip:1433?database=mydatabase"
$ sling conns set MSSQL url="sqlserver://myuser:mypass@host.ip:1433/my_instance?database=master&encrypt=true&TrustServerCertificate=true"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export MSSQL='sqlserver://myuser:mypass@host.ip:1433?database=mydatabase'
export MSSQL='{ type: sqlserver, user: "myuser", password: "mypass", host: "host.ip", port: 1433, database: "mydatabase" }'
export MSSQL='sqlserver://myuser:mypass@host.ip:1433/my_instance?database=master&encrypt=true&TrustServerCertificate=true'
export MSSQL_FABRIC='{ type: sqlserver, host: "xxx.datawarehouse.fabric.microsoft.com", database: "warehouse1", "trust_server_certificate": false, "encrypt": true, fedauth: ActiveDirectoryAzCli, bcp_entra_auth: true }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  MSSQL:
    type: sqlserver
    host: <host>
    user: <user>
    port: <port>
    instance: <instance>
    database: <database>
    schema: <schema>
    password: <password>
    encrypt: 'true'
    trust_server_certificate: 'true'

  MSSQL_URL:
    url: "sqlserver://myuser:mypass@host.ip:1433?database=mydatabase"

  MSSQL_KERBEROS:
    type: sqlserver
    host: <host>
    user: <user>
    port: <port>
    instance: <instance>
    database: <database>
    schema: <schema>
    password: <password>
    authenticator: 'krb5'
    krb5_config_file: /etc/krb5.conf
    krb5_realm: domain.com
    krb5_keytab_file: ~/MyUserName.keytab
    krb5_credcache_file: ~/MyUserNameCachedCreds
  
  MSSQL_FABRIC:
    type: sqlserver
    database: warehouse1
    host: xxx.datawarehouse.fabric.microsoft.com
    encrypt: true
    trust_server_certificate: false
    fedauth: ActiveDirectoryAzCli  # after running `az login` using Azure CLI Tool
    bcp_entra_auth: true  # in order to use BCP with fabric

```

## ADBC (Arrow Database Connectivity)

Sling supports SQL Server through both the standard TDS path (`go-mssqldb`) and the ADBC path. ADBC keeps data in Apache Arrow columnar format end-to-end and uses SQL Server's bulk-load API for writes, which is significantly faster than row-by-row TDS for large loads — and avoids shelling out to `bcp`. (*v1.5.2+*)

Flip a regular `type: sqlserver` connection over to ADBC by adding `use_adbc: true`. The rest of the connection properties stay the same. See the [ADBC documentation](adbc.md) for driver-manager install (`conda install -c conda-forge libadbc-driver-manager`) and driver install (`dbc install mssql`).

### Enable on an existing connection

```bash
sling conns set MSSQL type=sqlserver \
  host=localhost port=1433 user=sa password='AdbcDemo123!' \
  database=master encrypt=disable use_adbc=true
```

```yaml
connections:
  MSSQL_ADBC:
    type: sqlserver
    host: localhost
    port: 1433
    user: sa
    password: 'AdbcDemo123!'
    database: master
    encrypt: disable
    use_adbc: true
```

### ADBC-specific properties

* `use_adbc` (optional) -> Enable ADBC for this connection (`true` or `false`). Default is `false`.
* `adbc_uri` (optional) -> Override the automatically constructed ADBC URI. Sling builds it from `host`, `port`, `user`, `password`, `database`, but you can supply your own (e.g. for non-default driver flags). Format: `mssql://user:password@host:port/database`.
* `driver` (optional) -> Explicit path to the ADBC SQL Server driver library file (e.g. `~/.dbc/drivers/mssql/lib/libadbc_driver_mssql.dylib`). If not set, Sling auto-discovers the driver from `dbc`'s standard install locations.

### When to use it

ADBC is the better path in three situations:

- **Loading from Arrow-native sources.** When the source is DuckDB, Parquet, or any other Arrow-producing system, ADBC carries Arrow batches end-to-end without re-serializing through SQL strings.
- **Large writes that would otherwise need `bcp`.** ADBC's bulk insert is faster than row-by-row TDS and stays inside one process and one transaction boundary.
- **Read-heavy analytics.** ADBC reads return Arrow record batches; downstream Polars / DuckDB / pandas conversions are free copies. From Python, use `Sling(...).stream_arrow()` instead of `stream()` to receive Arrow batches directly.

The default TDS path is the right answer for short interactive queries, smaller writes, anywhere you want to avoid the `dbc` driver install, and anywhere you need features the ADBC driver doesn't expose yet (Kerberos, certain Azure auth modes).

### ADBC example

A full end-to-end example is available [here](https://slingdata.io/articles/sling-adbc-sqlserver).

## Troubleshooting

### TLS Handshake Failed: cannot read handshake packet: EOF

Starting with Sling v1.5.x, you may encounter this error when connecting to SQL Server instances that use TLS certificates signed with the **SHA-1** algorithm. This is due to Go 1.25 disabling SHA-1 signature algorithms in TLS 1.2 handshakes (per [RFC 9155](https://datatracker.ietf.org/doc/html/rfc9155)).

**Recommended fix:** Replace the SQL Server's TLS certificate with one using **SHA-256** or stronger.

**Workaround:** Set the `GODEBUG` environment variable before running Sling:

{% tabs %}
{% tab title="Linux / macOS" %}
```bash
export GODEBUG=tlssha1=1
sling run -r my_replication.yaml
```
{% endtab %}

{% tab title="Windows" %}
```powershell
set GODEBUG=tlssha1=1
sling run -r my_replication.yaml
```
{% endtab %}
{% endtabs %}

{% hint style="warning" %}
The `tlssha1=1` workaround is temporary. Future versions of Go will remove this option entirely. Upgrading your SQL Server's TLS certificate to SHA-256 is strongly recommended.
{% endhint %}

**Alternative workaround:** If your network environment allows it, you can disable TLS encryption entirely by adding `encrypt=disable` to your connection:

```yaml
connections:
  MSSQL:
    type: sqlserver
    host: <host>
    user: <user>
    password: <password>
    database: <database>
    encrypt: 'disable'
```

{% hint style="danger" %}
Disabling encryption means data is transmitted in plaintext. Only use this on trusted networks where security is not a concern.
{% endhint %}

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/redshift.md -->
# Redshift

## Redshift

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` **(required)** -> The username to access the instance
* `database` **(required)** -> The database name of the instance
* `aws_bucket` **(required)** -> The name of the S3 Bucket for Bulk Loading / Unloading
* `aws_access_key_id` (optional) -> The AWS Access Key ID to access the bucket for Bulk Loading / Unloading
* `aws_secret_access_key` (optional) -> The AWS Secret Key to access the bucket for Bulk Loading / Unloading
* `aws_session_token` (optional) -> The AWS Session token to access the bucket for Bulk Loading / Unloading
* `aws_role_arn` (optional since *v1.4.21*) -> The AWS Role to use to access the bucket for Bulk Loading / Unloading
* `schema` (optional) -> The default schema to use when loading
* `password` (optional) -> The password to access the instance
* `port` (optional) -> The port of the instance. Default is `5439`.
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.

### Using `sling conns`

Here are examples of setting a connection named `REDSHIFT`. We must provide the `type=redshift` property:

{% code overflow="wrap" %}
```bash
$ sling conns set REDSHIFT type=redshift host=<host> user=<user> database=<database> password=<password> port=<port> aws_bucket=<aws_bucket> aws_access_key_id=<aws_access_key_id> aws_secret_access_key=<aws_secret_access_key>

# OR use url
$ sling conns set REDSHIFT url="redshift://myuser:mypass@host.ip:5439/mydatabase" aws_bucket=<aws_bucket> aws_access_key_id=<aws_access_key_id> aws_secret_access_key=<aws_secret_access_key>
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export REDSHIFT='redshift://myuser:mypass@host.ip:5439/mydatabase'
export AWS_BUCKET='<aws_bucket>'
export AWS_ACCESS_KEY_ID='<aws_access_key_id>'
export AWS_SECRET_ACCESS_KEY='<aws_secret_access_key>'

export REDSHIFT='{ type: redshift, url: "redshift://myuser:mypass@host.ip:5439/mydatabase", aws_bucket: "<aws_bucket>", aws_access_key_id: "<aws_access_key_id>", aws_secret_access_key: "<aws_secret_access_key>" }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  REDSHIFT:
    type: redshift
    host: <host>
    user: <user>
    password: <password>
    port: <port>
    database: <database>
    schema: <schema>
    aws_bucket: <aws_bucket>
    aws_access_key_id: <aws_access_key_id>
    aws_secret_access_key: <aws_secret_access_key>

  REDSHIFT_URL:
    url: "redshift://myuser:mypass@host.ip:5439/mydatabase"
    aws_bucket: <aws_bucket>
    aws_access_key_id: <aws_access_key_id>
    aws_secret_access_key: <aws_secret_access_key>
```

## Database user creation

To allow Sling to access your database, we need to create a user with the proper privileges. Please follow the steps below:

1.  Create a user `sling` (or whatever you prefer) by running:

    ```sql
    CREATE USER sling WITH PASSWORD '<password>';
    ```
2.  If you are planning to load data into this connection, grant the following privileges to that user:

    ```sql
    GRANT CREATE ON DATABASE <database_name> TO sling;
    ```
3.  If you are planning to extract data from this connection, you need to give permission to read the tables you'd like Sling to extract.

    ```sql
    -- Need this to read table & column names 
    GRANT SELECT ON ALL TABLES IN SCHEMA information_schema TO sling;
    GRANT SELECT ON ALL TABLES IN SCHEMA pg_catalog TO sling;

    -- run this to grant SELECT permission to all tables in schema `marketing` to user sling
    GRANT SELECT ON ALL TABLES IN SCHEMA marketing TO sling;
    ```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/clickhouse.md -->
# Clickhouse

## Clickhouse

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `database` **(required)** -> The database name of the instance
* `user` (optional) -> The username to access the instance
* `schema` (optional)  -> The default schema to use
* `password` (optional) -> The password to access the instance
* `port` (optional) -> The port of the instance. Default is `9000`.
* `secure` (optional) -> Whether to use TLS for connecting. Default is `false`. **Note: This is required (`secure=true`) when connecting to ClickHouse Cloud.**
* `export_stream_format` (optional) -> Whether to specify the `FORMAT` when exporting (e.g. [`CSVWithNames`](https://clickhouse.com/docs/interfaces/formats/CSVWithNames)). Can help achieve low-memory exports.
* `skip_verify` (optional) -> Whether to skip verification for TLS. Default is `false`.
* `http_url` (optional) -> The HTTP url to override the connection string (see docs at [github.com/ClickHouse/clickhouse-go](https://github.com/ClickHouse/clickhouse-go?tab=readme-ov-file#http-support-experimental)). When specifying `http_url`, sling will use the HTTP clickhouse interface instead of the native interface. Native is recommended for optimal performance. **HTTP still has some limitations to be aware of for things like batch flushing and session context, so be cautious when switching over code to this protocol.**
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.
* `tls` -> TLS configuration name (`true`, `false`, `skip-verify`, or `custom` when providing `cert_*` keys below)

Custom TLS Certificates (v1.4.18+):
* `cert_file` (optional) -> the client certificate to use to access the instance via TLS (file path or raw)
* `cert_key_file` (optional) -> the client key to use to access the instance via TLS (file path or raw)
* `cert_ca_file` (optional) -> the client CA certificate to use to access the instance via TLS (file path or raw)

### Using `sling conns`

Here are examples of setting a connection named `CLICKHOUSE`. We must provide the `type=clickhouse` property:

{% code overflow="wrap" %}
```bash
$ sling conns set CLICKHOUSE type=clickhouse host=<host> user=<user> database=<database> password=<password> port=<port> 

# OR use url
$ sling conns set CLICKHOUSE url="clickhouse://myuser:mypass@host.ip:9000/mydatabase"

# connecting via http
$ sling conns set CLICKHOUSE type=clickhouse http_url="http://myuser:mypass@host.ip:8123/default"

# connecting to ClickHouse Cloud (secure=true is required)
$ sling conns set CLICKHOUSE_CLOUD url="clickhouse://myuser:mypass@host.clickhouse.cloud:9440/mydatabase?secure=true"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export CLICKHOUSE='clickhouse://myuser:mypass@host.ip:9000/mydatabase'
export CLICKHOUSE='{ type: clickhouse, user: "myuser", password: "mypass", host: "host.ip", port: 9000, database: "mydatabase", export_stream_format: "CSVWithNames" }'

# ClickHouse Cloud (secure=true is required)
export CLICKHOUSE_CLOUD='clickhouse://myuser:mypass@host.clickhouse.cloud:9440/mydatabase?secure=true'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  CLICKHOUSE:
    type: clickhouse
    host: <host>
    user: <user>
    port: <port>
    database: <database>
    schema: <schema>
    password: <password>
    export_stream_format: CSVWithNames

  # connecting via http
  CLICKHOUSE_HTTP:
    type: clickhouse
    http_url: http://myuser:mypass@host.ip:8123/default

  # connecting via https
  CLICKHOUSE_HTTP:
    type: clickhouse
    http_url: https://myuser:mypass@host.ip:8123/default?secure=true

  # ClickHouse Cloud (secure=true is required)
  CLICKHOUSE_CLOUD:
    type: clickhouse
    host: host.clickhouse.cloud
    user: <user>
    port: 9440
    database: <database>
    password: <password>
    secure: true
    export_stream_format: CSVWithNames

  CLICKHOUSE_URL:
    url: "clickhouse://myuser:mypass@host.ip:9000/mydatabase"
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/duckdb.md -->
# DuckDB

## DuckDB

## Setup

The following credentials keys are accepted:

* `instance` **(required)** -> The local file path of the database file
* `schema` (optional) -> The default schema to use.
* `duckdb_version` (optional) -> The CLI version of DuckDB to use. You can also specify the env. variable `DUCKDB_VERSION`.
* `read_only` (optional) -> Whether to open the connection in `readonly` mode. Accepts `true` or `false`. Default is `false`.
* `interactive` (optional) -> Whether to communicate to the DuckDB CLI via interactive mode instead of reopening the connection each time. Accepts `true` or `false`. Default is `false`.
* `copy_method` (optional) -> the method data is copied from sling into DuckDB. Acceptable values:  `csv_files`, `csv_http`, `arrow_http`, `named_pipes`. Default is `csv_http`.
* `max_buffer_size` (optional *v1.5*) -> the max buffer size to use when piping data from DuckDB CLI. Specify this if you have extremely large one-line text values in your dataset. Default is `10485760` (10MB).
* `use_adbc` (optional) -> Enable Arrow Database Connectivity (ADBC) driver for high-performance data transfer. See [ADBC](adbc.md) for setup and details. (*v1.5.2+*)
* `adbc_uri` (optional) -> Override the automatically constructed ADBC connection URI when using `use_adbc=true`.

### Using `sling conns`

Here are examples of setting a connection named `DUCKDB`. We must provide the `type=duckdb` property:

{% code overflow="wrap" %}
```bash
# for local files 
$ sling conns set DUCKDB type=duckdb instance=/path/to/file.db

# for local files (Windows), don't use backslash (\)
$ sling conns set DUCKDB type=duckdb instance=C:/path/to/file.db

# Or use url (only for local files)
$ sling conns set DUCKDB url="duckdb:///path/to/file.db"

# url for Windows, don't use backslash (\)
$ sling conns set DUCKDB url="duckdb://C:/path/to/file.db"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export DUCKDB='duckdb:///path/to/file.db'
export DUCKDB='{ type: duckdb, instance: "/path/to/file.db" }'

$env:DUCKDB="duckdb://C:/path/to/file.db" # for Windows PowerShell
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  DUCKDB:
    type: duckdb
    instance: <instance>
    schema: <schema>
    duckdb_version: '<duckdb_version>'
```

### Specifying a DuckDB version

By default, Sling will download the DuckDB binary automatically. If you would like to use a specific DuckDB version, you can specify this way, and sling will download that version:

```bash
export DUCKDB_VERSION='0.7.0'
```

### Specifying a DuckDB Binary Path & Extensions

If you already have DuckDB on your machine and would like to use it (instead of having Sling download it), you can specify the binary path this way:

```bash
export DUCKDB_PATH='/opt/homebrew/bin/duckdb'
export DUCKDB_USE_INSTALLED_EXTENSIONS='true'  # tells sling not to download extensions
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

### Potential Issue

If you have a `.duckdbrc` file, which runs commands whenever the DuckDB CLI is invoked, this may interfere with normal Sling operation. If you are facing weird issues and have this file, try again after deleting it.

<!-- source: sling-docs/connections/database-connections/mongodb.md -->
# MongoDB

## MongoDB

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` **(required)** -> The username to access the instance
* `password` (optional) -> The password to access the instance
* `port` (optional) -> The port of the instance. Default is `27017`.
* `tls` (optional) -> whether to use TLS for connecting (`true`/`false`).
* `cert_file` (optional) -> the client certificate to use to access the instance via TLS (file path or raw)
* `cert_key_file` (optional) -> the client key to use to access the instance via TLS (file path or raw)
* `cert_ca_file` (optional) -> the client CA certificate to use to access the instance via TLS (file path or raw)

### Using `sling conns`

Here are examples of setting a connection named `MONGODB`. We must provide the `type=mongodb` property:

{% code overflow="wrap" %}
```bash
$ sling conns set MONGODB type=mongodb host=<host> user=<user> password=<password> port=<port> 
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
# see https://www.mongodb.com/docs/manual/reference/connection-string-options/#std-label-connections-connection-options for URL options
export MONGODB='mongodb://root:password@host.ip:27017?authSource=<database>'
export MONGODB='{ type: mongodb, user: "root", password: "password", host: "host.ip", port: 27017, authSource: "<database>" }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  MONGODB:
    type: mongodb
    host: host.ip
    port: 27017
    user: <user>
    password: <password>
    tls: true # if TLS connection needed

  # see https://www.mongodb.com/docs/manual/reference/connection-string-options/#std-label-connections-connection-options for URL options
  MONGODB_URL:
    type: mongodb
    url: mongodb://root:password@host.ip:27017?authSource=<database>

  MONGODB_TLS:
    type: mongodb
    url: mongodb://root:password@host.ip:27017?tls=true
```

## Examples

```yaml
source: mongodb
target: oracle

defaults:
  mode: full-refresh
  object: user1.{stream_schema}_{stream_table}

streams:
  db1.collection1:
  db1.collection2:
    select: [col1, col2]
    where: '{"field": "value", "field2": {"$gt": 100}}'  # use "where" starting v1.3.6
  db2.collection3:
    where: '[{"$or": [{"field1": "value1"}, {"field2": "value2"}]}]'  # use "where" starting v1.3.6
  db2.collection4:
    mode: incremental
    primary_key: [id]
    update_key: last_mod_at
  db2.collection5:
    mode: incremental
    primary_key: [id]
    where: '{"_id": {"$gte": "67859d8ee682ab32317abc6f", "$lte": "67859d8ee682ab32317abcc8"}}' # filter on ID (v1.4.21+)
    update_key: last_mod_at
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/database-connections/oracle.md -->
# Oracle

## Oracle


## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the instance
* `user` **(required)** -> The username to access the instance
* `password` **(required)** -> The password to access the instance
* `schema` (optional) -> This is the default schema
* `sid` (optional) -> The Oracle System ID of the instance
* `service_name` (optional) -> The Oracle Service Name of the instance
* `tns` (optional) -> The Oracle TNS string of the instance (example: `(DESCRIPTION=(ADDRESS=(PROTOCOL=TCPS)(HOST=my-oracle-database.provider.com)(PORT=1521))(CONNECT_DATA=(SID=my_service)))`).
* `port` (optional) -> The port of the instance. Default is `1521`.
* `ssh_tunnel` (optional) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional) -> The private key to use to access a SSH server (raw string or path to file).
* `ssh_passphrase` (optional) -> The passphrase to use to access a SSH server.
* `jdbc_str` (optional) -> The JDBC connection string to use. No need to provide `user`, `password` or `host`
* `sqlldr_path` *(v1.3.0)*-> The path to the `sqlldr` binary. This is useful if you need to specify a custom path for `sqlldr`.

### Additional Parameters
* `auth_type` (optional) -> The auth type to use. Options include: `OS`, `TCPS`
* `auth_server` (optional) -> The Windows auth server to use. Options include: `NTS`
* `os_user` (optional) -> The Windows operating system user (with `auth_type=OS`)
* `os_password` (optional) -> The Windows operating system password (with `auth_type=OS`)
* `domain` (optional) -> Windows system domain name (with `auth_type=OS`)
* `proxy_client_name` (optional) -> to use proxy authentication

### Using `sling conns`

Here are examples of setting a connection named `ORACLE`. We must provide the `type=oracle` property:

{% code overflow="wrap" %}
```bash
$ sling conns set ORACLE type=oracle host=<host> port=<port> sid=<sid> user=<user> password=<password> schema=<schema>

# OR use JDBC connection string
$ sling conns set ORACLE type=oracle jdbc_str=<jdbc_str>

# OR use url
$ sling conns set ORACLE url="oracle://myuser:mypass@host.ip:1521/<sid>"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
export ORACLE='oracle://myuser:mypass@host.ip:1521/<sid>'
export ORACLE='{ type: oracle, user: "myuser", password: "mypass", host: "host.ip", port: 1521, sid: "<sid>" }'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  ORACLE:
    type: oracle
    host: <host>
    user: <user>
    port: <port>
    sid: <sid>                   # sid or service_name
    service_name: <service_name> # sid or service_name
    tns: <tns>
    schema: <schema>
    password: <password>

  ORACLE_JDBC:
    type: oracle
    jdbc_str: <jdbc_str>

  ORACLE_URL:
    url: "oracle://myuser:mypass@host.ip:1521/<sid>"
```

## LOB, XML and Binary Types

{% hint style="info" %}

Byte-exact round-trip for large Oracle binary types (`BLOB`, `LONG RAW`) up to the target's per-value limit (64 MB on Snowflake) is available in **v1.5.19+**.

{% endhint %}

Sling supports every Oracle large-object, binary and structured type as a source. Each maps to a generic Sling type, which then maps to the target's native type:

| Oracle source type | Sling generic type | Notes |
|---|---|---|
| `CLOB` | `text` | Character LOB. Streamed without loading the full value into memory. |
| `NCLOB` | `text` | National-character LOB. Converted from `AL16UTF16` to UTF-8 on most targets. |
| `BLOB` | `binary` | Binary LOB. Bytes preserved exactly. |
| `RAW` | `binary` | Fixed-length raw bytes up to `RAW(2000)`. |
| `LONG RAW` | `binary` | Legacy large binary (up to 2 GB). Treated like `BLOB`. |
| `LONG` | `string` | Legacy large character (up to 2 GB). Only one `LONG` *or* `LONG RAW` column per table. |
| `BFILE` | `binary` | Pointer to an external OS file. Sling reads the locator bytes, not the file contents. |
| `XMLTYPE` | `text` | Auto-cast to CLOB via `.getclobval()` so it streams as text. |

Practical per-value limits are set by the **target**:

| Target | Text | Binary |
|---|---|---|
| Snowflake | 128 MB (`VARCHAR`) | 64 MB (`BINARY`) |
| PostgreSQL | 1 GB (`text`) | 1 GB (`bytea`) |
| BigQuery | 10 MiB (`STRING`) | 10 MiB (`BYTES`) per row total |
| File targets | bounded by file format and disk | bounded by file format and disk |

For Snowflake, Sling generates `BINARY(67108864)` DDL and stages via CSV (`TO_BINARY('HEX')`) by default. For binary-heavy workloads, opt into Parquet staging (`target_options.format = parquet`).

## Oracle Client Dependency

Until version 1.1.13, there was a dependency on the Oracle Client for Sling to work. This is because sling used to use a [3rd party driver](https://github.com/godror/godror) which needs it. You can install it by following directions:

- **MacOS**: [https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#scripted-installation](https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#scripted-installation)
- **Linux**: [https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#linux](https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#linux)
- **Windows**: [https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#windows](https://odpi-c.readthedocs.io/en/latest/user_guide/installation.html#windows)

Starting in v1.1.14, Sling uses another [library](https://github.com/sijms/go-ora) for Oracle, which does not need the Oracle client for connection. However, there is an advantage of having the Oracle client installed, as it contains the `sqlldr` tool, which sling can use to load data (if present in PATH). Loading data with `sqlldr` can be much faster.

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/file-connections/s3.md -->
# AWS S3

## AWS S3
## Setup

The following credentials keys are accepted:

* `bucket` **(required)**
* `use_environment` (optional) -> whether to use the AWS environment variables to access. Accepts `true` or `false`.
* `profile` (optional) (or provide environment variable `AWS_PROFILE`)
* `access_key_id` (optional) (or provide environment variable `AWS_ACCESS_KEY_ID`)
* `secret_access_key` (optional) (or provide environment variable `AWS_SECRET_ACCESS_KEY`)
* `session_token` (optional) (or provide environment variable `AWS_SESSION_TOKEN`)
* `region` (optional) (or provide environment variable `AWS_REGION`)
* `endpoint` (optional) (or provide environment variable `AWS_ENDPOINT`)
* `role_arn` (optional) (or provide environment variable `AWS_ROLE_ARN`)
* `anonymous` (optional) Tells to access the bucket as anonymous (set to `true`)
* `encryption_algorithm` (optional since *v1.2.15*) The server-side encryption algorithm to use. Accepts `AES256`, `aws:kms` or `aws:kms:arn`.
* `encryption_kms_key` (optional since *v1.2.15*) If `encryption_algorithm` is `aws:kms` or `aws:kms:arn`, this is the KMS key ID or ARN to use for encryption.
* `http_timeout` (optional) The HTTP client timeout for S3 requests. Accepts Go duration format (e.g., `30s`, `5m`, `10m30s`). Default is `30s`. Increase this value if you're experiencing timeout errors with large files or slow network connections.

### Using `sling conns`

Here are examples of setting a connection named `AWS_S3`. We must provide the `type=s3` property:

{% code overflow="wrap" %}
```bash
$ sling conns set AWS_S3 type=s3 bucket=sling-bucket profile=my-profile

$ sling conns set AWS_S3 type=s3 bucket=sling-bucket access_key_id=ACCESS_KEY_ID secret_access_key="SECRET_ACCESS_KEY"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

In JSON/YAML format:

{% code overflow="wrap" %}
```bash
export AWS_S3='{type: s3, bucket: sling-bucket, access_key_id: ACCESS_KEY_ID, secret_access_key: "SECRET_ACCESS_KEY"}'

export AWS_S3='{type: s3, bucket: sling-bucket, profile: my-profile}'

# or via AWS classic environment variables
export AWS_ACCESS_KEY_ID='<aws_access_key_id>'
export AWS_SECRET_ACCESS_KEY='<aws_secret_access_key>'
export AWS_SESSION_TOKEN='<aws_session_token>'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  AWS_S3:
    type: s3
    bucket: <bucket>
    profile: <my-profile>

  AWS_S3_OTHER:
    type: s3
    bucket: <bucket>
    access_key_id: <access_key_id>
    secret_access_key: '<secret_access_key>'
    http_timeout: 5m  # Increase timeout for large files or slow connections
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/file-connections/gs.md -->
# Google Cloud Storage

## Google Cloud Storage

## Setup

The following credentials keys are accepted:

* `bucket` **(required)**
* `key_file` **(required)** -> This is the account credential JSON file.

### Using `sling conns`

Here are examples of setting a connection named `GOOGLE_STORAGE`. We must provide the `type=gs` property:

{% code overflow="wrap" %}
```bash
$ sling conns set GOOGLE_STORAGE type=gs bucket=<bucket> key_file=/path/to/keyfile.json
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

In JSON/YAML format:

{% code overflow="wrap" %}
```bash
export GOOGLE_STORAGE='{type: gs, bucket: sling-bucket, key_file: /path/to/keyfile.json}'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  GOOGLE_STORAGE:
    type: gs
    bucket: <bucket>
    key_file: /path/to/keyfile.json
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/file-connections/azure.md -->
# Azure Storage

## Azure Storage

## Setup

The following credentials keys are accepted:

* `account` **(required)** -> This is the Azure account string.
* `container` **(required)** -> This is the storage container.
* `conn_str` (optional) -> This is the Connection String from an [Account Access Keys](https://learn.microsoft.com/en-us/azure/storage/common/storage-account-keys-manage?tabs=azure-portal). Looks something like `DefaultEndpointsProtocol=https;AccountName=my_account;AccountKey=thisIsAFakeKeyxxxxxxxxxxxxx/MRHCrBw4BLmypbaJyVbvUWn4YZ1Nw==;EndpointSuffix=core.windows.net`.
* `sas_svc_url` (optional) -> This is the [Shared Access Signature (SAS)](https://learn.microsoft.com/en-us/azure/storage/common/storage-sas-overview) URL.
* `client_id` (optional) -> Application ID of an Azure service principal. Can also set via env var `AZURE_CLIENT_ID`
* `tenant_id` (optional) -> ID of the application's Microsoft Entra tenant. Can also set via env var `AZURE_TENANT_ID`
* `client_secret` (optional) -> A client secret generated for the App Registration. Can also set via env var `AZURE_CLIENT_SECRET`
* `client_certificate_path` (optional) -> Path to a PEM or PKCS12 certificate file including private key. Can also set via env var `AZURE_CLIENT_CERTIFICATE_PATH`
* `client_certificate_password` (optional) -> The password protecting the certificate file (PFX/PKCS12 only). Can also set via env var `AZURE_CLIENT_CERTIFICATE_PASSWORD`

### Using `sling conns`

Here are examples of setting a connection named `AZURE_STORAGE`. We must provide the `type=azure` property:

{% code overflow="wrap" %}
```bash
$ sling conns set AZURE_STORAGE type=azure account=<account> container=<container> sas_svc_url=<sas_svc_url>
$ sling conns set AZURE_STORAGE type=azure account=<account> container=<container> conn_str="<conn_str>"
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

In JSON/YAML format:

{% code overflow="wrap" %}
```bash
export AZURE_STORAGE='{"type": "azure", "account": "<account>", "container": "<container>", "sas_svc_url": "<sas_svc_url>"}'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  AZURE_STORAGE:
    type: azure
    account: <account>
    container: <container>
    sas_svc_url: '<sas_svc_url>'

  AZURE_STORAGE_2:
    type: azure
    account: <account>
    container: <container>
    conn_str: '<conn_str>'

  # Service principal with client secret
  AZURE_STORAGE_SP:
    type: azure
    account: <account>
    container: <container>
    client_id: your-client-id
    tenant_id: your-tenant-id
    client_secret: your-client-secret
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/file-connections/sftp.md -->
# SFTP

## SFTP

## Setup

The following credentials keys are accepted:

* `host` **(required)** -> The hostname / ip of the machine
* `user` **(required)** -> The username to access the machine
* `port` (optional. Default is `22`)
* `password` (optional) -> The password to access the machine
* `path` (optional, *v1.4.20*) -> The root path to use when connecting
* `private_key` (optional) -> This is the raw key string or the path to your SSH private key.
* `passphrase` (optional) -> The passphrase to use to access a SSH server.
* `ssh_tunnel` (optional since *v1.2.15*) -> The URL of the SSH server you would like to use as a tunnel (example `ssh://user:password@db.host:22`)
* `ssh_private_key` (optional since *v1.2.15*) -> The private key to use to access a SSH tunnel server (raw string or path to file).
* `ssh_passphrase` (optional since *v1.2.15*) -> The passphrase to use to access a SSH tunnel server.

### Using `sling conns`

Here are examples of setting a connection named `MY_SFTP`. We must provide the `type=sftp` property:

{% code overflow="wrap" %}
```bash
$ sling conns set MY_SFTP type=sftp host=<host> user=<user> password=<password> port=<port> private_key=<private_key>

# Or use url
$ sling conns set MY_SFTP url=sftp://myuser:mypass@host.ip:22 private_key=<private_key>
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

In JSON/YAML format:

{% code overflow="wrap" %}
```bash
export MY_SFTP='{type: sftp, url: "sftp://myuser:mypass@host.ip:22", private_key: <private_key>}'
export MY_SFTP_WITH_PATH='{type: sftp, url: "sftp://user:password@host.ip:22//absolute/path/to/dir"}'
```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  MY_SFTP:
    type: sftp
    host: <host>
    user: <user>
    port: <port>
    password: <password>
    private_key: <private_key>

  MY_SFTP_URL:
    url: "sftp://myuser:mypass@host.ip:22"
    private_key: <private_key>

  MY_SFTP_URL_PATH:
    url: "sftp://user:password@host.ip:22//absolute/path/to/dir"
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-docs/connections/datalake-connections/iceberg.md -->
# Iceberg

## Iceberg

Apache Iceberg is an open table format for huge analytic datasets. Iceberg adds tables to compute engines including Spark, Trino, PrestoDB, Flink, Hive and Impala using a high-performance table format that works just like a SQL table. See [https://iceberg.apache.org/](https://iceberg.apache.org/) for more details.

Sling supports connecting to Iceberg tables through catalog backends including REST catalogs, AWS Glue, and SQL catalogs.

## Setup

The following credentials keys are accepted:

### Common Properties
* `catalog_type` **(required)** -> The catalog type: `rest`, `glue`, or `sql`. Default is `rest`.
* `schema` (optional) -> The default schema to use to read/write data. Default is `main`.

### REST Catalog Configuration
* `rest_uri` **(required for REST)** -> The REST catalog endpoint URI (e.g., `https://s3tables.us-east-1.amazonaws.com/iceberg`, `https://catalog.cloudflarestorage.com/xxxxxxxxx/warehouse`,  `http://localhost:8181`).
* `rest_warehouse` (optional) -> Warehouse location for the catalog (e.g., `s3://bucket/warehouse`, `arn:aws:s3tables:region:account-id:bucket/namespace`).
* `rest_token` (optional) -> OAuth token for authentication.
* `rest_oauth_client_id` (optional) -> OAuth client ID for authentication.
* `rest_oauth_client_secret` (optional) -> OAuth client secret for authentication.
* `rest_oauth_scope` (optional) -> OAuth scope for authentication.
* `rest_oauth_server_uri` (optional) -> OAuth server URI for token requests.
* `rest_prefix` (optional) -> API prefix for the REST catalog.
* `rest_metadata_location` (optional) -> Custom metadata location.
* `rest_extra_props` (optional) -> Additional properties as JSON string.
* `rest_sigv4_enable` (optional) -> Enable AWS SigV4 authentication (true/false).
* `rest_sigv4_region` (optional) -> AWS region for SigV4 authentication (with `rest_sigv4_enable=true`).
* `rest_sigv4_service` (optional) -> AWS service name for SigV4 authentication (with `rest_sigv4_enable=true`).

### Glue Catalog Configuration
* `glue_warehouse` **(required for Glue)** -> Warehouse location in S3. e.g. `s3://my-bucket/warehouse`
* `glue_account_id` (optional) -> AWS account ID for Glue catalog.
* `glue_namespace` (optional) -> Namespace in the Glue catalog.
* `glue_extra_props` (optional) -> Extra Glue Properties (object).

### SQL Catalog Configuration
* `sql_catalog_name` (optional) -> Name of the SQL catalog. Default is `sql`.
* `sql_catalog_conn` **(required for SQL)** -> Name of a Sling database connection to use as catalog backend.
* `sql_catalog_init` (optional) -> Whether to initialize catalog tables if they don't exist. Default is `true`.

### Storage Configuration for Glue Catalog or S3Tables or SQL Catalog

For **S3/S3-compatible storage**:
* `s3_access_key_id` (optional) -> AWS access key ID
* `s3_secret_access_key` (optional) -> AWS secret access key
* `s3_session_token` (optional) -> AWS session token
* `s3_region` (optional) -> AWS region
* `s3_profile` (optional) -> AWS profile to use
* `s3_endpoint` (optional) -> S3-compatible endpoint URL (e.g. `http://localhost:9000` for MinIO)

### Using `sling conns`

Here are examples of setting a connection named `ICEBERG`. We must provide the `type=iceberg` property:

{% code overflow="wrap" %}
```bash
# REST catalog with local warehouse
$ sling conns set ICEBERG type=iceberg catalog_type=rest rest_uri=http://localhost:8181

# AWS S3 Tables via REST
$ sling conns set ICEBERG type=iceberg catalog_type=rest rest_warehouse="arn:aws:s3tables:us-east-1:123456789012:bucket/my-namespace" s3_profile=my-profile

# Cloudflare R2 Data Catalog via REST
$ sling conns set ICEBERG type=iceberg catalog_type=rest rest_uri="https://catalog.cloudflarestorage.com/xxxxxxxxx/warehouse" rest_warehouse="3fff6d86c73fcxxxxxxx4cb125f34927_warehouse" rest_token="<rest_token>"

# Glue catalog
$ sling conns set ICEBERG type=iceberg catalog_type=glue glue_warehouse=s3://my-bucket/glue-warehouse s3_access_key_id=AKIAIOSFODNN7EXAMPLE s3_secret_access_key=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY s3_region=us-east-1
```
{% endcode %}

### Environment Variable

See [here](../../sling-cli/environment.md#dot-env-file-.env.sling) to learn more about the `.env.sling` file.

{% code overflow="wrap" %}
```bash
# REST catalog local
export ICEBERG='{ 
  type: iceberg, 
  catalog_type: rest,
  rest_uri: "http://localhost:8181",
  rest_token: "<rest_token>",
  rest_warehouse: "<rest_warehouse>"
}'

# REST catalog with OAuth
export ICEBERG='{ 
  type: iceberg, 
  catalog_type: rest,
  rest_uri: "http://localhost:8181",
  rest_oauth_client_id: "my-client-id",
  rest_oauth_client_secret: "my-client-secret",
  rest_oauth_scope: "catalog:write",
  rest_oauth_server_uri: "https://auth.example.com/oauth2/token"
}'

# AWS S3 Tables
export ICEBERG_S3='{ 
  type: iceberg, 
  catalog_type: rest,
  rest_warehouse: "arn:aws:s3tables:us-east-1:123456789012:bucket/my-namespace",
  s3_access_key_id: "AKIAIOSFODNN7EXAMPLE",
  s3_secret_access_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
}'

# Iceberg with GCP backend (Apache Gravitino)
export ICEBERG_GCP='{ 
  type: iceberg
  catalog_name: iceberg_catalog
  catalog_type: rest
  rest_uri: http://endpoint.com:9001/iceberg
  rest_warehouse: s3://my-bucket/gravitino/gcs_as_s3
  s3_access_key_id: <s3_access_key_id>
  s3_secret_access_key: <s3_secret_access_key>
  s3_endpoint: https://myhost.storage.googleapis.com
  s3_region: auto
  schema: iceberg_schema
}'

# Glue catalog
export ICEBERG_GLUE='{ 
  type: iceberg, 
  catalog_type: glue,
  glue_warehouse: "s3://my-bucket/glue-warehouse",
  s3_region: "us-east-1",
  s3_profile: "default"
}'

# SQL catalog with PostgreSQL
export ICEBERG='{ 
  type: iceberg
  catalog_type: sql
  sql_catalog_conn: "POSTGRES_DB"
  sql_catalog_name: "iceberg_catalog"
  sql_catalog_init: true
  sql_warehouse: s3://my-bucket/iceberg/warehouse
  s3_access_key_id: AKIAIOSFODNN7EXAMPLE
  s3_secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
  s3_region: us-east-1
}'

```
{% endcode %}

### Sling Env File YAML

See [here](../../sling-cli/environment.md#sling-env-file-env.yaml) to learn more about the sling `env.yaml` file.

```yaml
connections:
  ICEBERG:
    type: iceberg
    catalog_type: rest  # or glue, or sql
    catalog_name: iceberg
    schema: main
    
    # REST Catalog Configuration
    rest_uri: http://localhost:8181
    rest_warehouse: s3://my-bucket/warehouse
    rest_token: my-bearer-token
    rest_oauth_client_id: my-client-id
    rest_oauth_client_secret: my-client-secret
    rest_oauth_scope: catalog:write
    rest_oauth_server_uri: https://auth.example.com/oauth2/token
    
    # Glue Catalog Configuration
    catalog_type: glue
    glue_warehouse: s3://my-bucket/glue-warehouse
    s3_access_key_id: AKIAIOSFODNN7EXAMPLE
    s3_secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
    s3_region: us-east-1
    
    # SQL Catalog Configuration (alternative to REST/Glue)
    # sql_catalog_conn: POSTGRES_DB
    # sql_catalog_name: iceberg_catalog
    # sql_catalog_init: true
    
    # S3 Configuration (if using S3 storage)
    s3_access_key_id: AKIAIOSFODNN7EXAMPLE
    s3_secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
    s3_region: us-east-1
    s3_endpoint: http://localhost:9000  # for MinIO or other S3-compatible

  # Example SQL catalog setup with PostgreSQL backend
  ICEBERG_SQL:
    type: iceberg
    catalog_type: sql
    sql_catalog_conn: POSTGRES_CATALOG  # defined below
    sql_catalog_name: iceberg_catalog
    sql_catalog_init: true
    schema: main

    # S3 Configuration for storage
    sql_warehouse: s3://my-bucket/iceberg/warehouse
    s3_access_key_id: AKIAIOSFODNN7EXAMPLE
    s3_secret_access_key: wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
    s3_region: us-east-1

  # PostgreSQL connection for SQL catalog backend
  POSTGRES_CATALOG:
    type: postgres
    host: localhost
    port: 5432
    user: postgres
    password: mypassword
    database: iceberg_catalog
```

## Common Usage Examples

### Basic Operations
```bash
# List namespaces (schemas)
sling conns discover ICEBERG

# List tables in a namespace
sling conns discover ICEBERG --schema my_namespace

# Query data
sling run --src-conn ICEBERG --src-stream "SELECT * FROM my_namespace.orders LIMIT 10" --stdout

# Export to CSV
sling run --src-conn ICEBERG --src-stream my_namespace.orders --tgt-object file://./orders.csv
```

### Data Import/Export
```bash
# Import CSV to Iceberg
sling run --src-stream file://./data.csv --tgt-conn ICEBERG --tgt-object my_namespace.new_data

# Import from PostgreSQL
sling run --src-conn POSTGRES_DB --src-stream public.customers --tgt-conn ICEBERG --tgt-object sales.customers

# Export to Parquet files
sling run --src-conn ICEBERG --src-stream sales.orders --tgt-conn AWS_S3 --tgt-object s3://bucket/exports/orders.parquet

# Incremental sync with timestamp
sling run --src-conn POSTGRES_DB --src-stream public.events --tgt-conn ICEBERG --tgt-object events.raw --mode incremental --primary-key id --update-key updated_at
```

### Advanced Queries with DuckDB Integration
```bash
# For complex SQL queries, Sling uses DuckDB with Iceberg extension
# Tables must be qualified with "iceberg_catalog" prefix
# Note: Custom SQL queries via DuckDB are only supported with REST and Glue catalog types
sling run --src-conn ICEBERG --src-stream "SELECT count(*) FROM iceberg_catalog.my_namespace.my_table" --stdout

# Join multiple Iceberg tables
sling run --src-conn ICEBERG --src-stream "
  SELECT o.*, c.customer_name 
  FROM iceberg_catalog.sales.orders o 
  JOIN iceberg_catalog.sales.customers c ON o.customer_id = c.id
" --stdout

# Export with custom SQL
sling run --src-conn ICEBERG --src-stream "
  SELECT date_trunc('month', order_date) as month, sum(amount) as total
  FROM iceberg_catalog.sales.orders 
  GROUP BY 1 ORDER BY 1
" --tgt-object file://./monthly_sales.csv
```

If you are facing issues connecting, please reach out to us at [support@slingdata.io](mailto:support@slingdata.io), on [discord](https://discord.gg/q5xtaSNDvp) or open a Github Issue [here](https://github.com/slingdata-io/sling-cli/issues).

<!-- source: sling-website/src/content/articles/en/bigquery-to-postgres-sling/index.mdx -->
# Export from BigQuery to PostgreSQL with Sling

> Last updated: June 2026

## Introduction

Moving data between Google BigQuery and PostgreSQL traditionally involves complex ETL processes, custom scripts, and significant engineering effort. Organizations often face challenges such as:

- Setting up and maintaining data extraction processes from BigQuery
- Managing authentication and permissions across platforms
- Handling schema compatibility and data type conversions
- Implementing efficient data loading into PostgreSQL
- Monitoring and maintaining the data pipeline
- Dealing with incremental updates and schema changes

According to industry research, setting up a traditional data pipeline between BigQuery and PostgreSQL can take weeks or even months, requiring specialized knowledge of both platforms and custom code development. Common approaches include:

1. Writing custom Python scripts using libraries like `pandas` and `sqlalchemy`
2. Using ETL tools that require extensive configuration and maintenance
3. Implementing Apache Airflow DAGs with custom operators
4. Developing and maintaining complex data transformation logic

These approaches often lead to:
- Increased development and maintenance costs
- Complex error handling and retry mechanisms
- Difficulty in handling schema changes
- Performance bottlenecks
- Limited monitoring and observability

Sling simplifies this entire process by providing a streamlined, configuration-based approach that eliminates the need for custom code and complex infrastructure setup. With Sling, you can:

- Configure connections with simple environment variables or CLI commands
- Automatically handle schema mapping and data type conversions
- Optimize performance with built-in batch processing and parallel execution
- Monitor and manage replications through both CLI and web interface
- Implement incremental updates with minimal configuration

In this guide, we'll walk through the process of setting up a BigQuery to PostgreSQL replication using Sling, demonstrating how to overcome common challenges and implement an efficient data pipeline in minutes rather than days or weeks. If you need the opposite direction, our [MySQL to BigQuery guide](/articles/mysql-bigquery-sling-data-transfer) and [Postgres to BigQuery guide](/articles/export-postgres-to-bigquery-database) cover loading into BigQuery.

# Installation

Getting started with Sling is straightforward. You can install it using various package managers depending on your operating system:

```bash
# macOS / Linux
curl -fsSL https://slingdata.io/install.sh | bash

# Windows
irm https://slingdata.io/install.ps1 | iex

# Python
pip install sling
```

After installation, verify that Sling is properly installed by running:

```bash
# Check sling version
sling --version
```

For more detailed installation instructions and options, visit the [installation guide](https://docs.slingdata.io/sling-cli/getting-started#installation).

# Setting Up Connections

Before we can start replicating data, we need to configure our source (BigQuery) and target (PostgreSQL) connections. Sling provides multiple ways to manage connections, including environment variables, the `sling conns` command, and a YAML configuration file.

## BigQuery Connection Setup

The BigQuery source connection here is identical to the one used when BigQuery is the source in other pipelines, such as [BigQuery to Snowflake](/articles/bigquery-to-snowflake-sling).

For BigQuery, you'll need:
- Google Cloud project ID
- Service account credentials with appropriate permissions
- Dataset information
- Google Cloud Storage bucket (for data transfer)

You can set up the BigQuery connection in several ways:

1. **Using the `sling conns set` Command**

```bash
# Set up BigQuery connection using CLI
sling conns set bigquery_source type=bigquery \
  project=<project> \
  dataset=<dataset> \
  gc_bucket=<gc_bucket> \
  key_file=/path/to/service.account.json \
  location=<location>
```

2. **Using Environment Variables**

```bash
# Set up using service account JSON content
export GC_KEY_BODY='{"type": "service_account", ...}'
export BIGQUERY_SOURCE='{type: bigquery, project: <project>, dataset: <dataset>, gc_bucket: <gc_bucket>}'
```

3. **Using Sling Environment YAML**

Create or edit `~/.sling/env.yaml`:

```yaml
connections:
  bigquery_source:
    type: bigquery
    project: your-project
    dataset: your_dataset
    gc_bucket: your-bucket
    key_file: /path/to/service.account.json
    location: US  # optional
```

## PostgreSQL Connection Setup

For PostgreSQL, you'll need:
- Host address
- Port number (default: 5432)
- Database name
- Username and password
- Schema (optional)
- SSL mode (if required)

Here's how to set up the PostgreSQL connection:

1. **Using the `sling conns set` Command**

```bash
# Set up PostgreSQL connection using CLI
sling conns set postgres_target type=postgres \
  host=<host> \
  user=<user> \
  database=<database> \
  password=<password> \
  port=<port> \
  schema=<schema>

# Or use connection URL format
sling conns set postgres_target url="postgresql://user:password@host:5432/database?sslmode=require"
```

2. **Using Environment Variables**

```bash
# Set up using connection URL format
export POSTGRES_TARGET='postgresql://user:password@host:5432/database?sslmode=require'
```

3. **Using Sling Environment YAML**

Add to your `~/.sling/env.yaml`:

```yaml
connections:
  postgres_target:
    type: postgres
    host: your-host
    user: your-username
    password: your-password
    database: your-database
    port: 5432
    schema: public
    sslmode: require  # optional
```

## Testing Connections

After setting up your connections, it's important to verify they work correctly:

```bash
# Test BigQuery connection
sling conns test bigquery_source

# Test PostgreSQL connection
sling conns test postgres_target

# List available tables in BigQuery
sling conns discover bigquery_source
```

You can also manage your connections through the Sling Platform's web interface:

![Sling Connections Management](../../../../assets/images/screenshots/ui.connections.light.png)

For more details about connection configuration, visit the [environment documentation](https://docs.slingdata.io/sling-cli/environment).

# Data Replication Methods

Sling provides multiple ways to replicate data from BigQuery to PostgreSQL. Let's explore both CLI-based and YAML-based approaches, starting from simple configurations to more advanced use cases.

## Using CLI Flags

The quickest way to start a replication is using CLI flags. Here are two examples:

### Basic CLI Example

This example shows how to replicate a single table with default settings:

```bash
# Replicate a single table from BigQuery to PostgreSQL
sling run \
  --src-conn bigquery_source \
  --src-stream "analytics.daily_sales" \
  --tgt-conn postgres_target \
  --tgt-object "analytics.daily_sales" \
  --tgt-options '{ "column_casing": "snake" }'
```

### Advanced CLI Example

This example demonstrates more advanced options including column selection and incremental updates:

```bash
# Replicate with advanced options
sling run \
  --src-conn bigquery_source \
  --src-stream "analytics.customer_orders" \
  --select "order_id, customer_id, order_date, total_amount" \
  --tgt-conn postgres_target \
  --tgt-object "analytics.customer_orders" \
  --mode incremental \
  --primary-key order_id \
  --update-key order_date \
  --tgt-options '{ "column_casing": "snake", "add_new_columns": true, "table_keys": { "unique": ["order_id"] } }'
```

For more CLI flag options, visit the [CLI flags documentation](https://docs.slingdata.io/sling-cli/run#cli-flags-overview).

## Using YAML Configuration

For more complex replication scenarios, YAML configuration files provide better maintainability and reusability. Let's look at two examples:

### Basic YAML Example

Create a file named `bigquery_to_postgres.yaml`:

```yaml
# Define source and target connections
source: bigquery_source
target: postgres_target

# Default settings for all streams
defaults:
  mode: full-refresh
  target_options:
    column_casing: snake
    add_new_columns: true

# Define streams to replicate
streams:
  analytics.daily_sales:
    object: analytics.daily_sales
    primary_key: [date, product_id]

  analytics.customer_orders:
    object: analytics.{stream_table}
    primary_key: order_id
```

Run the replication:

```bash
# Run the replication using YAML config
sling run -r bigquery_to_postgres.yaml
```

### Advanced YAML Example

Here's a more complex example that demonstrates various features including runtime variables, custom SQL, and multiple streams:

```yaml
source: bigquery_source
target: postgres_target

env:
  DATE: ${DATE}  # from env var

defaults:
  mode: incremental
  target_options:
    column_casing: snake
    add_new_columns: true

streams:
  # Stream with custom SQL and runtime variables
  analytics.orders_{DATE}:
    object: analytics.orders
    sql: |
      SELECT *
      FROM analytics.orders
      WHERE DATE(created_at) = '{DATE}'
    primary_key: order_id
    update_key: updated_at

  # Stream with column selection and transforms
  analytics.customers:
    object: analytics.customers
    select:
      - customer_id
      - first_name
      - last_name
      - email
      - -internal_notes  # exclude this column
    transforms:
      email: [lower, trim]
    primary_key: customer_id
    target_options:
      table_keys:
        primary: [customer_id]
        unique: [email]

  # Stream with wildcard pattern
  analytics.events_*:
    object: analytics.events
    mode: full-refresh
    primary_key: event_id
```

Run the replication with runtime variables:

```bash
# Run the replication with a specific date
export DATE=2024-02-10
sling run -r bigquery_to_postgres.yaml
```

For more details about replication configuration, visit:
- [Replication Structure](https://docs.slingdata.io/concepts/replication/structure)
- [Runtime Variables](https://docs.slingdata.io/concepts/replication/runtime-variables)
- [Source Options](https://docs.slingdata.io/concepts/replication/source-options)
- [Target Options](https://docs.slingdata.io/concepts/replication/target-options)

# Sling Platform UI

While the CLI provides powerful functionality for local development and automation, the Sling Platform offers a comprehensive web interface for managing and monitoring your data replications at scale.

## Platform Overview

The Sling Platform provides:
- Visual interface for creating and managing data workflows
- Team collaboration features
- Monitoring and alerting
- Centralized connection management
- Job scheduling and orchestration
- Agent-based architecture for secure execution

## Managing Connections

The platform provides an intuitive interface for managing your connections:

![Sling Connections Management](../../../../assets/images/screenshots/ui.connections.light.png)

You can:
- Create and edit connections with a visual form
- Test connections directly from the UI
- View and manage connection permissions
- Share connections with team members

## Visual Replication Editor

The platform includes a powerful visual editor for creating and managing replications:

![Sling Editor](../../../../assets/images/screenshots/ui.editor.light.png)

Features include:
- Visual stream configuration
- Syntax highlighting for SQL and YAML
- Real-time validation
- Version control integration

## Monitoring and Execution

Track your replication jobs with detailed execution information:

![Sling Execution](../../../../assets/images/screenshots/ui.execution.light.png)

The platform provides:
- Real-time execution monitoring
- Detailed logs and error messages
- Performance metrics and statistics
- Historical execution data

For more information about the Sling Platform, visit the [platform documentation](https://docs.slingdata.io/sling-platform/getting-started).

# Getting Started

Now that we've covered the various aspects of using Sling for BigQuery to PostgreSQL data replication, here's a quick guide to get you started:

1. **Install Sling**
   - Choose the appropriate installation method for your system
   - Verify the installation with `sling --version`

2. **Set Up Connections**
   - Configure BigQuery source connection
   - Configure PostgreSQL target connection
   - Test both connections using `sling conns test`

3. **Start Simple**
   - Begin with a basic CLI command to replicate a single table
   - Monitor the replication process
   - Verify the data in PostgreSQL

4. **Scale Up**
   - Create a YAML configuration file for multiple streams
   - Add incremental updates and transformations
   - Implement runtime variables for flexibility

5. **Consider Platform Features**
   - Sign up for the Sling Platform for advanced features
   - Set up team access and permissions
   - Configure monitoring and alerts

## Best Practices

1. **Connection Management**
   - Use environment variables or YAML files for connection configuration
   - Keep credentials secure and never commit them to version control
   - Use separate connections for development and production

2. **Replication Configuration**
   - Start with simple configurations and gradually add complexity
   - Use YAML files for better maintainability
   - Document your configurations with clear descriptions

3. **Performance Optimization**
   - Use appropriate batch sizes for your data volume
   - Implement incremental updates when possible
   - Monitor and adjust configurations based on performance metrics

4. **Monitoring and Maintenance**
   - Regularly check replication logs
   - Set up alerts for failed replications
   - Keep Sling updated to the latest version

## Next Steps

Once your data lands in Postgres, you can keep moving it downstream; see [Postgres to DuckDB](/articles/postgres-to-duckdb-with-sling) for fast local analytics or [loading JSON data into Postgres](/articles/exporting-loading-json-data-sling-postgres) for the file-to-Postgres direction.

To learn more about Sling's capabilities, explore these resources:
- [Sling Documentation](https://docs.slingdata.io/)
- [Example Configurations](https://docs.slingdata.io/examples/database-to-database)
- [CLI Reference](https://docs.slingdata.io/sling-cli/run)
- [Platform Features](https://docs.slingdata.io/sling-platform/getting-started)

For additional examples and community support:
- Join the [Sling Discord Community](https://discord.gg/q5xtaSNDvp)
- Follow Sling on [GitHub](https://github.com/slingdata-io/sling-cli)
- Contact [support@slingdata.io](mailto:support@slingdata.io) for assistance

# FAQ

## How do I export a BigQuery table to PostgreSQL with Sling?

Set `bigquery` as the source connection and `postgres` as the target, then point a stream's `object` key at the destination schema and table. A single `sling run` command moves the data, creating the target table if it does not exist.

## Can I run incremental loads from BigQuery into Postgres?

Yes. Set `mode` to `incremental` on the stream and supply a primary key plus an `update_key`. Sling compares the `update_key` against the existing target rows and loads only newer records.

## How do I filter BigQuery rows before loading into Postgres?

Add a `sql` key to the stream with a SELECT statement. The query runs against BigQuery and only the result set is loaded, which is handy for date-partitioned or filtered extracts.

## Does Sling automatically convert BigQuery data types to Postgres types?

Yes. Sling maps each BigQuery column to a compatible Postgres type during the load. Enable `adjust_column_type` under `target_options` if you want it to widen existing target columns when needed.

## How do I set a primary key or unique constraint on the Postgres target?

Use a `table_keys` block under `target_options` with `primary` and `unique` lists, for example `primary: [customer_id]` and `unique: [email]`. Sling applies these keys when it creates or maintains the target table.

## Can I use runtime variables like a date in the replication config?

Yes. Reference variables with single braces such as `{DATE}` in object names, stream names, or SQL, and pass the value through the `env` block or an environment variable at run time.

## What permissions does the BigQuery service account need?

It needs read access to the source dataset and read/write access to the GCS bucket set as `gc_bucket`, since Sling stages data through Cloud Storage during extraction.

<!-- source: sling-website/src/content/articles/en/bigquery-to-snowflake-sling/index.mdx -->
# Moving Data from BigQuery to Snowflake Using Sling

## Introduction

> Last updated: June 2026

Moving data between cloud data warehouses like Google BigQuery and Snowflake traditionally involves complex ETL processes, custom scripts, and significant engineering effort. Organizations often face challenges such as:

- Setting up and maintaining data extraction processes from BigQuery
- Managing authentication and permissions across platforms
- Handling schema compatibility and data type conversions
- Implementing efficient data loading into Snowflake
- Monitoring and maintaining the data pipeline
- Dealing with incremental updates and schema changes

According to industry research, setting up a traditional data pipeline between BigQuery and Snowflake can take weeks or even months, requiring specialized knowledge of both platforms and custom code development. This complexity often leads to increased costs, maintenance overhead, and potential reliability issues.

Sling simplifies this entire process by providing a streamlined, configuration-based approach that eliminates the need for custom code and complex infrastructure setup. With Sling, you can:

- Configure connections with simple environment variables or CLI commands
- Automatically handle schema mapping and data type conversions
- Optimize performance with built-in batch processing and parallel execution
- Monitor and manage replications through both CLI and web interface
- Implement incremental updates with minimal configuration

In this guide, we'll walk through the process of setting up a BigQuery to Snowflake replication using Sling, demonstrating how to overcome common challenges and implement an efficient data pipeline in minutes rather than days or weeks. If you need to move data the other direction, see the guide on [exporting Snowflake to BigQuery](/articles/exporting-snowflake-to-bigquery-using-sling).

# Installation

Getting started with Sling is straightforward. You can install it using various package managers depending on your operating system:

```bash
# macOS / Linux
curl -fsSL https://slingdata.io/install.sh | bash

# Windows
irm https://slingdata.io/install.ps1 | iex

# Python
pip install sling
```

After installation, verify that Sling is properly installed by running:

```bash
# Check sling version
sling --version
```

For more detailed installation instructions and options, visit the [installation guide](https://docs.slingdata.io/sling-cli/getting-started#installation).

# Setting Up Connections

Before we can start replicating data, we need to configure our source (BigQuery) and target (Snowflake) connections. Sling provides multiple ways to manage connections, including environment variables, the `sling conns` command, and a YAML configuration file.

## BigQuery Connection Setup

For BigQuery, you'll need:
- Google Cloud project ID
- Service account credentials with appropriate permissions
- Dataset information
- Google Cloud Storage bucket (for data transfer)

You can set up the BigQuery connection in several ways:

1. **Using the `sling conns set` Command**

```bash
# Set up BigQuery connection using CLI
sling conns set bigquery_source type=bigquery \
  project=<project> \
  dataset=<dataset> \
  gc_bucket=<gc_bucket> \
  key_file=/path/to/service.account.json \
  location=<location>
```

2. **Using Environment Variables**

```bash
# Set up using service account JSON content
export GC_KEY_BODY='{"type": "service_account", ...}'
export BIGQUERY_SOURCE='{type: bigquery, project: <project>, dataset: <dataset>, gc_bucket: <gc_bucket>}'
```

3. **Using Sling Environment YAML**

Create or edit `~/.sling/env.yaml`:

```yaml
connections:
  bigquery_source:
    type: bigquery
    project: your-project
    dataset: your_dataset
    gc_bucket: your-bucket
    key_file: /path/to/service.account.json
    location: US  # optional
```

## Snowflake Connection Setup

For Snowflake, you'll need:
- Account identifier (e.g., `xy12345.us-east-1`)
- Username and password
- Database name
- Warehouse name
- Role (optional)
- Schema (optional)

Here's how to set up the Snowflake connection:

1. **Using the `sling conns set` Command**

```bash
# Set up Snowflake connection using CLI
sling conns set snowflake_target type=snowflake \
  account=<account> \
  user=<user> \
  password=<password> \
  database=<database> \
  warehouse=<warehouse> \
  role=<role>
```

2. **Using Environment Variables**

```bash
# Set up using connection URL format
export SNOWFLAKE_TARGET='snowflake://user:password@account/database?warehouse=compute_wh&role=sling_role'
```

3. **Using Sling Environment YAML**

Add to your `~/.sling/env.yaml`:

```yaml
connections:
  snowflake_target:
    type: snowflake
    account: xy12345.us-east-1
    user: your_username
    password: your_password
    database: your_database
    warehouse: compute_wh
    role: sling_role  # optional
    schema: public    # optional
```

## Testing Connections

After setting up your connections, it's important to verify they work correctly:

```bash
# Test BigQuery connection
sling conns test bigquery_source

# Test Snowflake connection
sling conns test snowflake_target

# List available tables in BigQuery
sling conns discover bigquery_source
```

You can also manage your connections through the Sling Platform's web interface:

![Sling Connections Management](../../../../assets/images/screenshots/ui.connections.light.png)

For more details about connection configuration, visit the [environment documentation](https://docs.slingdata.io/sling-cli/environment).

# Data Replication Methods

Sling provides multiple ways to replicate data from BigQuery to Snowflake. Let's explore both CLI-based and YAML-based approaches, starting from simple configurations to more advanced use cases.

## Using CLI Flags

The quickest way to start a replication is using CLI flags. Here are two examples:

### Basic CLI Example

This example shows how to replicate a single table with default settings:

```bash
# Replicate a single table from BigQuery to Snowflake
sling run \
  --src-conn bigquery_source \
  --src-stream "analytics.daily_sales" \
  --tgt-conn snowflake_target \
  --tgt-object "ANALYTICS.DAILY_SALES" \
  --tgt-options '{ "column_casing": "upper" }'
```

### Advanced CLI Example

This example demonstrates more advanced options including column selection and data type handling:

```bash
# Replicate with advanced options
sling run \
  --src-conn bigquery_source \
  --src-stream "analytics.customer_orders" \
  --select "order_id, customer_id, order_date, total_amount" \
  --tgt-conn snowflake_target \
  --tgt-object "ANALYTICS.CUSTOMER_ORDERS" \
  --mode incremental \
  --primary-key order_id \
  --update-key order_date \
  --tgt-options '{ "column_casing": "upper", "add_new_columns": true }'
```

For more CLI flag options, visit the [CLI flags documentation](https://docs.slingdata.io/sling-cli/run#cli-flags-overview).

## Using YAML Configuration

For more complex replication scenarios, YAML configuration files provide better maintainability and reusability. Let's look at two examples:

### Basic YAML Example

Create a file named `bigquery_to_snowflake.yaml`:

```yaml
# Define source and target connections
source: bigquery_source
target: snowflake_target

# Default settings for all streams
defaults:
  mode: full-refresh
  target_options:
    column_casing: upper
    add_new_columns: true

# Define the streams to replicate
streams:
  # Replicate multiple tables using wildcards
  "analytics.*":
    object: "ANALYTICS.{stream_table}"
    
  # Replicate a specific table with custom settings
  "sales.transactions":
    object: "SALES.TRANSACTIONS"
    mode: incremental
    primary_key: ["transaction_id"]
    update_key: "transaction_date"
```

### Advanced YAML Example

Here's a more complex configuration with multiple streams and custom options:

```yaml
source: bigquery_source
target: snowflake_target

env:
  run_date: ${RUN_DATE}

defaults:
  mode: incremental
  target_options:
    column_casing: upper
    add_new_columns: true

streams:
  # Replicate with custom SQL and column selection
  "custom_sales_report":
    object: "ANALYTICS.SALES_REPORT"
    sql: |
      SELECT 
        o.order_id,
        c.customer_name,
        p.product_name,
        o.quantity,
        o.total_amount,
        o.order_date
      FROM `dataset.orders` o
      JOIN `dataset.customers` c ON o.customer_id = c.customer_id
      JOIN `dataset.products` p ON o.product_id = p.product_id
      WHERE o.order_date >= '2023-01-01'
    
  # Replicate with runtime variables and transformations
  "daily_metrics":
    object: "ANALYTICS.DAILY_METRICS_{run_timestamp}"
    sql: |
      SELECT 
        date,
        product_category,
        SUM(revenue) as total_revenue,
        COUNT(DISTINCT customer_id) as unique_customers
      FROM `dataset.sales`
      WHERE date = '{run_date}'
      GROUP BY date, product_category
    
  # Replicate with advanced options
  "customer_segments":
    mode: truncate
    object: "ANALYTICS.CUSTOMER_SEGMENTS"
    select: ["segment_id", "segment_name", "created_at", "updated_at"]
```

To run a replication using a YAML configuration:

```bash
# Run the entire replication
sling run -r bigquery_to_snowflake.yaml

# Run specific streams
sling run -r bigquery_to_snowflake.yaml --stream analytics.daily_sales
```

For more information about runtime variables and configuration options, visit:
- [Runtime Variables Documentation](https://docs.slingdata.io/concepts/replication/runtime-variables)
- [Replication Configuration](https://docs.slingdata.io/concepts/replication)
- [Source Options](https://docs.slingdata.io/concepts/replication/source-options)
- [Target Options](https://docs.slingdata.io/concepts/replication/target-options)

# Sling Platform Features

While the CLI provides powerful command-line capabilities, the Sling Platform offers a comprehensive web-based solution for managing your data pipelines at scale. Let's explore some key features that make it easier to manage BigQuery to Snowflake replications.

## Visual Configuration Editor

The Sling Platform includes a sophisticated configuration editor that makes it easy to create and modify replication configurations through a user-friendly interface.

![Sling Configuration Editor](../../../../assets/images/screenshots/ui.editor.light.png)

The editor provides:
- Syntax highlighting for YAML configurations
- Auto-completion for connection names and options
- Real-time validation of your configuration
- Easy access to documentation and examples
- Version control for configuration changes

## Execution Monitoring

Monitor your replications in real-time with detailed execution statistics and logs.

![Sling Execution Monitoring](../../../../assets/images/screenshots/ui.execution.light.png)

Key monitoring features include:
- Real-time progress tracking
- Detailed execution logs
- Performance metrics and statistics
- Error reporting and diagnostics
- Historical execution data

## Team Collaboration

The platform facilitates team collaboration with features such as:
- Role-based access control
- Shared connection management
- Configuration version history
- Team activity monitoring
- Collaborative troubleshooting

## Additional Platform Benefits

The Sling Platform offers several advantages for enterprise users:
- Scheduled executions
- Automated retries and error handling
- Integration with notification systems
- Audit logging
- Resource usage monitoring

For more information about the Sling Platform and its features, visit the [platform documentation](https://docs.slingdata.io/sling-platform/getting-started).

# Getting Started

Ready to start using Sling for your BigQuery to Snowflake data pipeline? Here's how to get started:

1. **Set Up Your Environment**
   - [Install Sling CLI](https://docs.slingdata.io/sling-cli/getting-started#installation)
   - Configure your connections
   - Test connectivity to both platforms

2. **Create Your First Replication**
   - Start with a simple table replication
   - Test the replication process
   - Monitor the results

3. **Scale Your Implementation**
   - Add more tables and transformations
   - Implement incremental updates
   - Set up scheduling and monitoring

4. **Explore Advanced Features**
   - Try the Sling Platform
   - Implement complex transformations
   - Set up team collaboration

## Additional Resources

- [Database to Database Examples](https://docs.slingdata.io/examples/database-to-database)
- [Replication Modes Documentation](https://docs.slingdata.io/concepts/replication/modes)
- [Runtime Variables Guide](https://docs.slingdata.io/concepts/replication/runtime-variables)
- [Sling Platform Documentation](https://docs.slingdata.io/sling-platform/getting-started)

## Related Guides

- [Export Snowflake to BigQuery](/articles/exporting-snowflake-to-bigquery-using-sling)
- [Move data from BigQuery to Postgres](/articles/bigquery-to-postgres-sling)
- [Export SQL Server to BigQuery](/articles/exporting-sqlserver-to-bigquery-using-sling)
- [Load Postgres into Snowflake](/articles/export-postgres-to-snowflake-database)

For more examples and detailed documentation, visit the [Sling Documentation](https://docs.slingdata.io/).

# FAQ

**Why does the BigQuery connection need a Google Cloud Storage bucket?**

Sling exports BigQuery data through GCS for efficient bulk extraction. It unloads query results to the bucket you set with `gc_bucket`, then streams them into Snowflake, so the bucket works as a staging area.

**How do I match Snowflake's uppercase identifier convention?**

Set `column_casing` to `upper` in `target_options`. Snowflake stores unquoted identifiers in uppercase, so this keeps column and table names lined up with what Snowflake expects.

**Can I replicate many BigQuery tables at once?**

Yes. Use a wildcard stream such as `analytics.*` to match every table in a dataset, and reference the `{stream_table}` runtime variable in the target object so each Snowflake table is named after its source.

**How do I run incremental loads from BigQuery to Snowflake?**

Set `mode` to `incremental` and define `table_keys` with a primary key plus an `update_key`. Sling pulls only the rows that changed on each run and merges them into the Snowflake target.

**Does Sling create the Snowflake tables, and can it add new columns automatically?**

Sling creates target tables that do not exist and infers types from BigQuery. Turn on `add_new_columns` in `target_options` so new source columns are added to the Snowflake table instead of breaking the run.

**Which Snowflake role and warehouse does Sling use for loading?**

Sling uses the warehouse and optional role you set on the Snowflake connection. The role needs permission to create or write to the target schema, and the warehouse has to be able to run the load.

<!-- source: sling-website/src/content/articles/en/export-d1-duckdb-sling/index.mdx -->
# Migrating Data from Cloudflare D1 to DuckDB with Sling

## Introduction

> Last updated: June 2026

Cloudflare D1 and DuckDB represent two innovative approaches to database management - D1 as a serverless SQL database integrated with Cloudflare's edge network, and DuckDB as a high-performance analytical database engine. While both serve distinct purposes, organizations often need to move data between these systems for analytics, reporting, and data processing workflows. This is where Sling comes in, offering a streamlined solution for data migration between these platforms.

## What is Cloudflare D1?

Cloudflare D1 is a serverless SQL database that runs at the edge, built on SQLite and integrated seamlessly with Cloudflare Workers. It offers:
- Zero configuration database management
- Automatic replication across Cloudflare's global network
- SQLite compatibility
- Seamless integration with Cloudflare Workers
- Built-in security and performance optimizations

## Understanding DuckDB

DuckDB is an in-process SQL OLAP database management system that excels at analytical queries. Key features include:
- Columnar-vectorized query execution
- Complex query optimization
- Efficient data compression
- Seamless integration with Python and R
- Support for structured and semi-structured data

## Introduction to Sling

As an open-source data movement tool, Sling simplifies the process of transferring data between different database systems. It offers:
- Simple installation and configuration
- Multiple replication modes
- Efficient handling of large datasets
- Built-in data type mapping
- Automated schema management
- Real-time monitoring 

# Getting Started with Sling

Before we dive into replicating data from D1 to DuckDB, let's set up Sling on your system. Sling offers multiple installation methods to suit different operating systems and preferences.

## Installation

Choose the installation method that matches your operating system:

```bash
# macOS / Linux
curl -fsSL https://slingdata.io/install.sh | bash

# Windows
irm https://slingdata.io/install.ps1 | iex

# Python
pip install sling
```

After installation, verify that Sling is properly installed by checking its version:

```bash
# Check Sling version
sling --version
```

## Initial Configuration

Sling uses a configuration directory to store connection details and other settings. The configuration directory is typically located at:
- Linux/macOS: `~/.sling/`
- Windows: `C:\Users\<username>\.sling\`

The first time you run Sling, it will automatically create this directory and a default configuration file. You can also specify a custom location using the `SLING_HOME_DIR` environment variable.

## Understanding Sling's Architecture

Sling's architecture is designed for efficient data movement:

1. **Connection Management**
   - Secure credential storage
   - Multiple connection methods support
   - Connection pooling and reuse

2. **Data Processing**
   - Streaming data transfer
   - Automatic type conversion
   - Configurable batch sizes

3. **Monitoring and Control**
   - Real-time progress tracking
   - Detailed logging
   - Error handling and recovery

For more detailed installation instructions and configuration options, visit the [Sling CLI Getting Started Guide](https://docs.slingdata.io/sling-cli/getting-started). 

# Setting Up Database Connections

Before we can replicate data, we need to configure our source (D1) and target (DuckDB) connections. Sling provides multiple ways to set up and manage connections securely.

## Configuring D1 Connection

Cloudflare D1 connections in Sling require your Cloudflare account credentials and database information. Here's how to set up a D1 connection:

### Using Environment Variables

The simplest way is to use environment variables:

```bash
# Set D1 connection using environment variable
export D1_SOURCE='d1://account_id:api_token@database_name'
```

### Using the Sling CLI

Alternatively, use the `sling conns set` command:

```bash
# Set up D1 connection with individual parameters
sling conns set D1_SOURCE type=d1 account_id=your_account_id api_token=your_api_token database=your_database_name

# Or use a connection URL
sling conns set D1_SOURCE url="d1://account_id:api_token@database_name"
```

### Using the Sling Environment File

You can also add the connection details to your `~/.sling/env.yaml` file:

```yaml
connections:
  D1_SOURCE:
    type: d1
    account_id: your_account_id
    api_token: your_api_token
    database: your_database_name
```

## Setting Up DuckDB Connection

DuckDB connections in Sling are straightforward as they primarily involve specifying a file path. Here's how to set up a DuckDB connection:

### Using the Sling CLI

```bash
# Set up DuckDB connection
sling conns set DUCKDB type=duckdb path=/path/to/database.duckdb
```

### Using Environment Variables

```bash
# Set DuckDB connection using environment variable
export DUCKDB='duckdb:///path/to/database.duckdb'
```

### Using the Sling Environment File

Add to your `~/.sling/env.yaml`:

```yaml
connections:
  DUCKDB:
    type: duckdb
    path: /path/to/database.duckdb
```

## Testing Connections

After setting up your connections, verify them using the `sling conns` commands:

```bash
# List all configured connections
sling conns list

# Test D1 connection
sling conns test D1_SOURCE

# Test DuckDB connection
sling conns test DUCKDB

# Discover available tables in D1
sling conns discover D1_SOURCE
```

For more details about connection configuration and options, refer to:
- [D1 Connection Guide](https://docs.slingdata.io/connections/database-connections/d1)
- [DuckDB Connection Guide](https://docs.slingdata.io/connections/database-connections/duckdb)
- [Sling Environment Documentation](https://docs.slingdata.io/sling-cli/environment) 

# Basic Data Replication with CLI

Once you have your connections set up, you can start replicating data from D1 to DuckDB using Sling's CLI flags. Let's look at some common usage patterns.

## Simple Table Replication

The most basic way to replicate data is using the `sling run` command with source and target specifications:

```bash
# Replicate a single table from D1 to DuckDB
sling run \
  --src-conn D1_SOURCE \
  --src-stream "users" \
  --tgt-conn DUCKDB \
  --tgt-object "main.users"
```

## Using Custom SQL Queries

You can use custom SQL queries to transform or filter data during replication:

```bash
# Replicate with a custom SQL query
sling run \
  --src-conn D1_SOURCE \
  --src-stream "SELECT id, name, email, created_at FROM users WHERE created_at > '2024-01-01'" \
  --tgt-conn DUCKDB \
  --tgt-object "main.recent_users"
```

## Advanced CLI Options

For more complex scenarios, you can use additional flags to customize the replication:

```bash
# Replicate with specific columns and options
sling run \
  --src-conn D1_SOURCE \
  --src-stream "orders" \
  --tgt-conn DUCKDB \
  --tgt-object "main.orders" \
  --select "id,customer_id,total_amount,status" \
  --mode incremental \
  --update-key "updated_at" \
  --tgt-options '{ "add_new_columns": true, "table_keys": { "primary": ["id"] } }'
```

For a complete list of available CLI flags and options, refer to the [Sling CLI Flags documentation](https://docs.slingdata.io/sling-cli/run#cli-flags-overview). 

# Advanced Replication with YAML

While CLI flags are great for simple operations, YAML configurations provide more flexibility and maintainability for complex replication scenarios. Let's explore how to use YAML configurations for D1 to DuckDB data replication.

## Understanding YAML Configuration Structure

A typical replication YAML file includes source and target connections, along with stream definitions. Here's a basic example:

```yaml
# Basic replication configuration
source: D1_SOURCE
target: DUCKDB

streams:
  main.users:
    object: main.users
    mode: full-refresh
```

## Multiple Stream Replication

You can configure multiple streams in a single YAML file:

```yaml
# Multiple stream replication
source: D1_SOURCE
target: DUCKDB

defaults:
  mode: incremental
  target_options:
    add_new_columns: true
    column_casing: snake
    table_keys:
      primary: [id]

streams:
  users:
    object: main.users
    update_key: updated_at
    source_options:
      datetime_format: YYYY-MM-DD HH:mm:ss

  orders:
    object: main.orders
    update_key: order_date
    select: [id, customer_id, order_date, total_amount, status]

  customers:
    object: main.customers
    mode: full-refresh
    source_options:
      empty_as_null: true
```

## Using Runtime Variables

Sling supports runtime variables that can be used in your YAML configurations:

```yaml
source: D1_SOURCE
target: DUCKDB

env:
  SCHEMA: main
  TABLE_PREFIX: analytics_

streams:
  # Use runtime variables for dynamic table names, load all tables in schema
  main.*:
    object: '{SCHEMA}.{TABLE_PREFIX}{stream_table}'
    mode: incremental
    primary_key: id
    update_key: updated_at
```

## Configuring Source and Target Options

You can specify various options for both source and target connections:

```yaml
source: D1_SOURCE
target: DUCKDB

streams:
  analytics_data:
    object: main.analytics
    mode: incremental
    update_key: event_date
    
    # Source-specific options
    source_options:
      datetime_format: YYYY-MM-DD HH:mm:ss
      empty_as_null: true
      
    # Target-specific options
    target_options:
      add_new_columns: true
      column_casing: snake
      table_keys:
        primary: [id]
        unique: [event_id]
```


To run a replication using a YAML configuration:

```bash
# Run replication using YAML config
sling run -r d1_to_duckdb_basic.yaml
```

For more information about YAML configurations, refer to:
- [Replication Structure](https://docs.slingdata.io/concepts/replication/structure)
- [Runtime Variables](https://docs.slingdata.io/concepts/replication/runtime-variables)
- [Source Options](https://docs.slingdata.io/concepts/replication/source-options)
- [Target Options](https://docs.slingdata.io/concepts/replication/target-options) 

# Sling Platform Overview

While the CLI is powerful for local development and simple workflows, the Sling Platform provides a comprehensive web-based interface for managing your data operations at scale. Let's explore the key components and features of the platform.

## Web Interface

The Sling Platform offers an intuitive web interface for managing your data operations:

![Sling Editor Interface](../../../../assets/images/screenshots/ui.editor.light.png)

The web interface provides:
- Visual replication editor
- Real-time validation
- Syntax highlighting
- Auto-completion
- Version control integration

## Connection Management

Manage all your connections in one place:

![Sling Connections](../../../../assets/images/screenshots/ui.connections.light.png)

The platform offers:
- Centralized credential management
- Connection health monitoring
- Easy testing and validation
- Team access controls

## Monitoring and Logging

Track your data operations in real-time:

![Sling Execution Details](../../../../assets/images/screenshots/ui.execution.light.png)

Features include:
- Real-time progress tracking
- Detailed execution logs
- Performance metrics
- Error reporting and alerts

For more information about the Sling Platform, visit:
- [Getting Started with Sling Platform](https://docs.slingdata.io/sling-platform/getting-started)
- [Platform Architecture](https://docs.slingdata.io/sling-platform/architecture)
- [Agent Management](https://docs.slingdata.io/sling-platform/agents) 

# Best Practices and Next Steps

Now that we've covered the various aspects of using Sling for D1 to DuckDB data replication, here are some recommended practices and next steps.

## Best Practices

1. **Start Small**
   - Begin with a simple table replication
   - Test with a subset of your data
   - Validate the results thoroughly

2. **Optimize Performance**
   - Use appropriate batch sizes
   - Consider incremental updates for large tables
   - Monitor system resources
   - Use efficient SQL queries

3. **Maintain Security**
   - Use environment variables for credentials
   - Implement proper access controls
   - Regularly rotate API tokens
   - Follow least privilege principle

4. **Version Control**
   - Keep YAML configurations in version control
   - Document your replication setup
   - Track changes and updates
   - Use descriptive commit messages

## Additional Resources

To learn more about Sling and its capabilities:

1. **Documentation**
   - [Sling CLI Documentation](https://docs.slingdata.io/sling-cli/getting-started)
   - [Sling Platform Documentation](https://docs.slingdata.io/sling-platform/getting-started)
   - [Replication Concepts](https://docs.slingdata.io/concepts/replication)

2. **Examples**
   - [File to Database Examples](https://docs.slingdata.io/examples/file-to-database)
   - [Database to Database Examples](https://docs.slingdata.io/examples/database-to-database)
   - [Database to File Examples](https://docs.slingdata.io/examples/database-to-file)

3. **Community and Support**
   - Join our [Discord community](https://discord.gg/q5xtaSNDvp)
   - Report issues on [GitHub](https://github.com/slingdata-io/sling-cli/issues)
   - Contact support at support@slingdata.io

## Next Steps

To continue your journey with Sling:

1. **Explore Advanced Features**
   - Try different replication modes
   - Experiment with transformations
   - Test various source and target options

2. **Scale Your Operations**
   - Move to YAML configurations for complex workflows
   - Set up monitoring and alerting
   - Implement proper error handling

3. **Consider Platform**
   - Evaluate the Sling Platform for enterprise needs
   - Set up agents for distributed processing
   - Implement team collaboration workflows

Remember that Sling is continuously evolving with new features and improvements. Stay connected with the community to learn about updates and best practices as they emerge.

## Related Guides

If you are working with D1 or DuckDB-style local analytics, these walkthroughs cover adjacent workflows:

- [Migrating Data from Cloudflare D1 to MySQL with Sling](/articles/export-d1-load-mysql-sling) for loading edge data into a relational target.
- [Export Data from Cloudflare D1 to Local Files with Sling](/articles/export-d1-to-local-csv-json-parquet) when you want CSV, JSON, or Parquet output instead of a database.
- [Loading Local Parquet to Postgres with Sling](/articles/loading-local-parquet-to-postgres-with-sling) for moving columnar files into a database.

# FAQ

**Does Sling read directly from Cloudflare D1 or through the HTTP API?**

Sling talks to D1 over Cloudflare's HTTP query API using your account ID and an API token. There is no local socket, so make sure the token has D1 read permission for the database you target.

**Why move D1 data into DuckDB instead of querying D1 directly?**

D1 is built for transactional edge workloads, while DuckDB is a columnar engine tuned for analytical scans and joins. Copying the data into DuckDB lets you run heavy aggregations locally without adding load to your edge database.

**Can I run incremental syncs from D1 to DuckDB?**

Yes. Set mode to `incremental` and provide an `update_key` such as `updated_at`, and Sling will only pull rows newer than the last run. Define a primary key under `table_keys` so updates merge correctly.

**Does the target DuckDB file need to exist before the first run?**

No. If the path you point the DuckDB connection at does not exist yet, Sling creates the database file on the first replication and adds the target tables automatically.

**How does Sling handle SQLite data types coming from D1?**

Sling infers column types from the source data and maps them to DuckDB equivalents during loading. You can override casing and column behavior with `target_options` like `column_casing` and `add_new_columns`.

**Can I select only a few columns or filter rows during the export?**

Yes. Use the `select` option to limit columns, or pass a full SQL query as the source stream to filter rows with a WHERE clause before they reach DuckDB.

<!-- source: sling-website/src/content/articles/en/export-load-data-mysql-sqlserver-sling/index.mdx -->
# Export and Load Data from MySQL to SQL Server Using Sling

## Introduction

> Last updated: June 2026

Data migration between different database systems is a common yet challenging task in modern data operations. When it comes to moving data from MySQL to SQL Server, organizations often face numerous hurdles: differences in data types, handling large datasets efficiently, maintaining data integrity, and ensuring minimal downtime. Traditional approaches might involve writing custom scripts, using ETL tools that require significant setup, or relying on expensive enterprise solutions.

Enter Sling: a modern data movement and transformation platform that simplifies these complex migrations. In this article, we'll explore how Sling streamlines the process of migrating data from MySQL to SQL Server, making it accessible, efficient, and reliable.

# Understanding Sling's Capabilities

Sling stands out in the data migration landscape by offering a powerful combination of simplicity and flexibility. Unlike traditional ETL tools that often require extensive setup and configuration, Sling provides a straightforward approach to database migrations while maintaining the robustness needed for production environments.

Key features that make Sling particularly effective for MySQL to SQL Server migrations include:
- Automatic schema mapping and creation
- Efficient bulk data transfer
- Support for incremental updates
- Built-in data type handling
- Real-time monitoring and error handling

Let's dive into how you can leverage these capabilities to simplify your database migration process. 

# Installing Sling

Before we begin migrating data, let's get Sling installed on your system. Sling provides multiple installation methods to suit different operating systems and preferences.

## System Requirements

Sling is designed to be lightweight and efficient, with minimal system requirements:
- Operating System: Windows, macOS, or Linux
- Memory: 512MB minimum (2GB recommended for large datasets)
- Storage: 100MB for installation
- Network: Access to both source (MySQL) and target (SQL Server) databases

## Installation Methods

Choose the installation method that best suits your environment:

```bash
# macOS / Linux
curl -fsSL https://slingdata.io/install.sh | bash

# Windows
irm https://slingdata.io/install.ps1 | iex

# Python
pip install sling
```

For more detailed installation instructions, visit the [official documentation](https://docs.slingdata.io/sling-cli/getting-started#installation).

## Verifying Installation

After installation, verify that Sling is properly installed by checking its version:

```bash
# Check Sling version
sling --version
```

If you see the version number displayed, you're ready to proceed with configuring your database connections. 

# Setting Up Database Connections

A crucial step in the migration process is properly configuring connections to both your source MySQL database and target SQL Server database. Sling provides multiple ways to manage these connections, including environment variables and a YAML configuration file.

## MySQL Connection Setup

For MySQL, you'll need to provide the following connection details:
- Host address
- Port (default: 3306)
- Database name
- Username and password
- Optional SSL/TLS configuration

Here's how to set up a MySQL connection using the `sling conns` command:

```bash
# Set up MySQL connection using command line
sling conns set MYSQL type=mysql host=your-mysql-host user=your-user database=your-database password=your-password port=3306
```

Alternatively, you can use a connection URL:

```bash
# Set up MySQL connection using URL
sling conns set MYSQL url="mysql://your-user:your-password@your-mysql-host:3306/your-database"
```

## SQL Server Connection Setup

For SQL Server, you'll need:
- Host address
- Port (default: 1433)
- Database name
- Username and password
- Optional instance name
- Optional encryption settings

Here's how to set up a SQL Server connection:

```bash
# Set up SQL Server connection using command line
sling conns set MSSQL type=sqlserver host=your-sqlserver-host user=your-user database=your-database password=your-password port=1433
```

Or using a connection URL:

```bash
# Set up SQL Server connection using URL
sling conns set MSSQL url="sqlserver://your-user:your-password@your-sqlserver-host:1433?database=your-database&encrypt=true&TrustServerCertificate=true"
```

## Using Environment Variables

You can also set up connections using environment variables:

```bash
# MySQL connection string
export MYSQL='mysql://your-user:your-password@your-mysql-host:3306/your-database'

# SQL Server connection string
export MSSQL='sqlserver://your-user:your-password@your-sqlserver-host:1433?database=your-database'
```

## YAML Configuration

For a more maintainable approach, especially in production environments, you can use Sling's `env.yaml` file:

```yaml
connections:
  MYSQL:
    type: mysql
    host: your-mysql-host
    user: your-user
    port: 3306
    database: your-database
    password: your-password

  MSSQL:
    type: sqlserver
    host: your-sqlserver-host
    user: your-user
    port: 1433
    database: your-database
    password: your-password
    encrypt: 'true'
    trust_server_certificate: 'true'
```

Save this configuration in `~/.sling/env.yaml`. For more details about environment configuration, refer to the [environment documentation](https://docs.slingdata.io/sling-cli/environment).

## Testing Connections

After setting up your connections, it's important to verify they work correctly:

```bash
# Test MySQL connection
sling conns test MYSQL

# Test SQL Server connection
sling conns test MSSQL
```

You can also list available tables and views in your databases:

```bash
# List available streams in MySQL
sling conns discover MYSQL

# List available streams in SQL Server
sling conns discover MSSQL
```

![Sling Connections UI](../../../../assets/images/screenshots/ui.connections.light.png)

With both connections properly configured and tested, we're ready to start migrating data between the databases. 

# Simple Data Migration with CLI

Sling's command-line interface provides a straightforward way to migrate data between databases. Let's explore how to use CLI flags for both simple and more complex migration scenarios.

## Basic Table Migration

The simplest form of migration involves copying a single table from MySQL to SQL Server. Here's a basic example:

```bash
# Migrate a single table with default options
sling run --src-conn MYSQL --src-stream customers \
          --tgt-conn MSSQL --tgt-object dbo.customers
```

This command will:
1. Read the `customers` table from MySQL
2. Automatically create the target table in SQL Server if it doesn't exist
3. Copy all data maintaining the original schema structure

## Advanced CLI Options

For more complex scenarios, Sling provides various CLI flags to customize the migration process. Here's an example with additional options:

```bash
# Migrate data with custom options
sling run --src-conn MYSQL --src-stream orders \
          --tgt-conn MSSQL --tgt-object dbo.orders \
          --select 'order_id,customer_id,order_date,total_amount'
          --tgt-options '{ "column_casing": "snake", "add_new_columns": true, "table_keys": { "primary_key": ["order_id"] } }' \
          --mode incremental \
          --update-key order_date
```

This advanced example:
- Selects specific columns from the source table
- Configures primary keys
- Uses snake case for column names in the target
- Enables automatic addition of new columns
- Performs an incremental update based on the `order_date` column

For a complete list of available CLI flags and options, refer to the [CLI flags documentation](https://docs.slingdata.io/sling-cli/run#cli-flags-overview).


# Advanced Data Migration with YAML

While CLI flags are great for quick migrations, YAML configurations provide a more maintainable and powerful way to define complex migration tasks. Let's explore how to use YAML configurations for MySQL to SQL Server migrations.

## Basic Multi-Stream Configuration

Here's a basic example that migrates multiple tables with default settings:

```yaml
# mysql_to_sqlserver.yaml
source: mysql
target: mssql

streams:
  customers:
    object: dbo.customers
    mode: full-refresh
    
  orders:
    object: dbo.orders
    mode: incremental
    update_key: order_date
    primary_key: [order_id]
```

To run this configuration:

```bash
# Run the replication using YAML config
sling run -r mysql_to_sqlserver.yaml
```

## Advanced Configuration Example

Here's a more complex example that showcases various features:

```yaml
# mysql_to_sqlserver_advanced.yaml
source: mysql
target: mssql

defaults:
  mode: incremental
  target_options:
    column_casing: snake
    add_new_columns: true
    batch_limit: 10000

streams:
  customers:
    object: dbo.customers
    primary_key: [customer_id]
    update_key: last_modified
    select: [customer_id, first_name, last_name, email, status, last_modified]
    source_options:
      table_keys: [customer_id]
    target_options:
      table_keys:
        unique_key: [email]
      pre_sql: |
        IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = 'dbo')
        BEGIN
          EXEC('CREATE SCHEMA dbo')
        END
  
  orders:
    object: dbo.orders
    primary_key: [order_id]
    update_key: order_date
    columns:
      total_amount: decimal(10, 2)
    source_options:
      select: |
        SELECT 
          order_id,
          customer_id,
          order_date,
          COALESCE(total_amount, 0) as total_amount,
          status
        FROM orders
        WHERE status != 'DELETED'
    target_options:
      table_keys:
        primary: [order_id]

  order_items:
    object: dbo.order_items
    primary_key: [order_id, item_id]
    update_key: last_modified
```

This advanced configuration demonstrates:
- Default settings for all streams
- Custom SQL queries for data selection
- Column transformations
- Primary and foreign key configurations
- Schema creation with pre-SQL
- Batch size control for performance optimization

## Using Runtime Variables

Sling supports runtime variables that can make your configurations more dynamic:

```yaml
# mysql_to_sqlserver_dynamic.yaml
source: mysql
target: mssql

env:
  tgt_schema: dbo
  batch_size: 5000

streams:
  main.*:
    object: {tgt_schema}.{stream_table}
    mode: full-refresh
    target_options:
      batch_limit: {batch_size}
```

You can use this configuration with runtime variables:

```bash
# Run with specific table and mode
sling run -r mysql_to_sqlserver_dynamic.yaml --var stream_table=customers --var mode=incremental
```

For more information about runtime variables, see the [runtime variables documentation](https://docs.slingdata.io/concepts/replication/runtime-variables).

## Monitoring Replications

When running YAML-based replications, Sling provides detailed execution information through both the CLI and the web interface:

![Sling Execution UI](../../../../assets/images/screenshots/ui.execution.light.png)

The web interface provides additional features like:
- Detailed progress tracking for each stream
- Historical execution statistics
- Error logs and debugging information
- Performance metrics and optimization suggestions

# Sling Platform Overview

While the CLI is powerful for individual migrations, the Sling Platform provides a comprehensive web interface for managing and monitoring data operations at scale. Let's explore the key components of the platform that make database migrations even more manageable.

## Web Interface Features

The Sling Platform offers a modern, intuitive interface for:
- Visual configuration of database connections
- Interactive replication builder
- Real-time monitoring of data transfers
- Team collaboration tools
- Detailed logging and analytics

## Monitoring and Management

The platform provides extensive monitoring capabilities:
- Real-time progress tracking
- Historical performance analytics
- Error detection and alerting
- Resource utilization metrics
- Automated retry mechanisms

## Team Collaboration

Enterprise features include:
- Role-based access control
- Shared connection management
- Replication templates
- Audit logging
- Team-wide notifications

For more information about the Sling Platform, visit the [platform documentation](https://docs.slingdata.io/sling-platform/getting-started).

# Next Steps

Now that you understand how to use Sling for MySQL to SQL Server migrations, here are some resources to help you get started and expand your usage:

## Additional Resources

- [Database to Database Examples](https://docs.slingdata.io/examples/database-to-database)
- [Replication Concepts](https://docs.slingdata.io/concepts/replication)
- [Source Options Documentation](https://docs.slingdata.io/concepts/replication/source-options)
- [Target Options Documentation](https://docs.slingdata.io/concepts/replication/target-options)

## Community and Support

- Join our [Discord community](https://discord.gg/q5xtaSNDvp) for real-time help
- Visit our [GitHub repository](https://github.com/slingdata-io/sling-cli) for the latest updates
- Contact [support@slingdata.io](mailto:support@slingdata.io) for enterprise support

## Further Use Cases

Beyond MySQL to SQL Server migrations, Sling supports various other scenarios:
- Real-time data synchronization
- Database backup and archival
- Cross-platform data replication
- Schema migration and validation
- Data warehouse loading

Whether you're performing a one-time migration or setting up ongoing data synchronization, Sling provides the tools and flexibility to handle your database migration needs efficiently and reliably.

## Related Guides

If you are working across MySQL and SQL Server, these guides cover adjacent scenarios:

- [Loading Local JSON Data into MySQL Using Sling](/articles/export-load-data-local-json-mysql-sling)
- [Loading Local JSON into SQL Server with Sling](/articles/export-load-data-local-json-sqlserver-sling)
- [Exporting and Loading MySQL to Snowflake with Sling](/articles/export-load-mysql-snowflake-sling)
- [Loading Local CSV Data into SQL Server with Sling](/articles/sling-export-load-local-csv-to-sqlserver)
- [Syncing MySQL to Postgres with Sling](/articles/sync-mysql-to-postgres-database)

# FAQ

**Does Sling create the SQL Server target tables automatically during a MySQL migration?**

Yes. When a target table does not exist, Sling creates it based on the source schema and maps MySQL types to their SQL Server equivalents. You can override the inferred types with a `columns` block per stream.

**How do I run an incremental sync from MySQL to SQL Server instead of a full reload?**

Set `mode: incremental`, give the stream a primary key, and choose an `update_key` column such as a last-modified timestamp. Sling then loads only rows newer than the last run.

**Can I migrate many MySQL tables to SQL Server in one configuration?**

Yes. List each table as a stream in a replication YAML, or use a wildcard like `main.*` to match all tables in a schema. Shared settings go under the `defaults` block.

**How do I select only specific columns when migrating from MySQL?**

Use the `--select` flag on the CLI or a `select` list in the stream definition. You can also write a custom SQL query in `source_options` to filter rows and compute derived columns.

**How do I set a primary key for a stream in Sling?**

Use `primary_key: [col]` on the stream, or `table_keys` with a `primary` entry under `target_options`. The primary key drives upserts in incremental mode and deduplication on reload.

**Does data flow through the Sling Platform servers during a migration?**

No. Sling agents execute the data movement inside your own infrastructure, so data goes directly from MySQL to SQL Server. The control plane only handles orchestration and monitoring.

**How can I improve performance for large MySQL to SQL Server loads?**

Tune `batch_limit` in target options, run multiple streams in parallel, and select only the columns you need. For very large tables, an incremental `update_key` avoids reloading unchanged rows.

<!-- source: sling-website/src/content/articles/en/export-load-data-sftp-json-snowflake-sling/index.mdx -->
# Export and Load JSON Data from SFTP to Snowflake using Sling

> Last updated: May 2026

## Introduction

In today's data-driven world, organizations frequently need to move data between different systems efficiently and reliably. A common scenario is transferring JSON data from SFTP servers to Snowflake data warehouses. While this seems straightforward, it often involves complex ETL processes, custom scripts, and ongoing maintenance.

Enter Sling, an open-source data integration tool designed to simplify these data movement challenges. Sling provides a streamlined approach to data replication, making it easy to transfer data between various sources and destinations, including SFTP servers and Snowflake.

In this comprehensive guide, we'll walk through how to use Sling to efficiently move JSON data from an SFTP server to Snowflake. You'll learn about:

- Setting up Sling in your environment
- Configuring SFTP and Snowflake connections
- Creating and running data replication tasks
- Using both the CLI and Platform interfaces
- Best practices for optimal performance

Whether you're a data engineer, developer, or analyst, this guide will help you implement a robust data pipeline using Sling's powerful features.

# Understanding the Data Pipeline Challenge

Moving data from SFTP to Snowflake traditionally involves several complex steps and considerations:

1. **Authentication and Access**: Managing SFTP credentials, SSH keys, and Snowflake authentication
2. **Data Parsing**: Handling JSON files with varying structures and nested data
3. **Type Mapping**: Converting JSON data types to appropriate Snowflake column types
4. **Error Handling**: Dealing with malformed JSON, network issues, and failed transfers
5. **Scalability**: Processing large files and multiple directories efficiently
6. **Monitoring**: Tracking successful transfers and troubleshooting failures

Many organizations resort to writing custom scripts or using expensive ETL tools to handle these challenges. This approach often leads to:

- High development and maintenance costs
- Complex error handling and retry logic
- Limited scalability and performance
- Difficulty in monitoring and troubleshooting
- Technical debt from custom solutions

Sling addresses these challenges by providing:

- Built-in support for SFTP and Snowflake connections
- Automatic JSON parsing and type inference
- Efficient bulk loading capabilities
- Robust error handling and retry mechanisms
- Comprehensive monitoring and logging
- Simple configuration through YAML or CLI flags

Let's explore how to implement this solution using Sling.

# Getting Started with Sling

Installing Sling is straightforward and supports multiple platforms. Choose the installation method that best suits your environment:

```bash
# macOS / Linux
curl -fsSL https://slingdata.io/install.sh | bash

# Windows
irm https://slingdata.io/install.ps1 | iex

# Python
pip install sling
```

After installation, verify that Sling is properly installed by checking its version:

```bash
# Check Sling version
sling --version
```

# Setting Up Connections

## SFTP Connection Setup

Sling provides multiple ways to configure your SFTP connection. Here are the available options:

### Using `sling conns set` Command

```bash
# Set up SFTP connection with password authentication
sling conns set MY_SFTP type=sftp host=sftp.example.com user=myuser password=mypassword port=22

# Set up SFTP connection with SSH key authentication
sling conns set MY_SFTP type=sftp host=sftp.example.com user=myuser private_key=/path/to/private_key port=22
```

### Using Environment Variables

```bash
# Using JSON format
export MY_SFTP='{
  "type": "sftp",
  "host": "sftp.example.com",
  "user": "myuser",
  "password": "mypassword",
  "port": 22
}'
```

### Using YAML Configuration

Create or edit `~/.sling/env.yaml`:

```yaml
connections:
  MY_SFTP:
    type: sftp
    host: sftp.example.com
    user: myuser
    port: 22
    password: mypassword  # Or use private_key
    private_key: /path/to/private_key  # Optional, alternative to password
```

## Snowflake Connection Setup

Similarly, you can configure your Snowflake connection using multiple methods:

### Using `sling conns set` Command

```bash
# Set up Snowflake connection
sling conns set MY_SNOWFLAKE type=snowflake account=myaccount user=myuser password=mypassword database=mydatabase warehouse=mywarehouse role=myrole
```

### Using Environment Variables

```bash
# Using connection URL
export MY_SNOWFLAKE='snowflake://myuser:mypassword@myaccount/mydatabase?warehouse=mywarehouse&role=myrole'
```

### Using YAML Configuration

Add to your `~/.sling/env.yaml`:

```yaml
connections:
  MY_SNOWFLAKE:
    type: snowflake
    account: myaccount
    user: myuser
    password: mypassword
    database: mydatabase
    warehouse: mywarehouse
    role: myrole
    schema: myschema  # Optional
```

# Data Synchronization Methods

Sling offers two primary methods for configuring and running data synchronization: CLI flags and replication YAML files. Let's explore both approaches.

## Using CLI Flags

For quick, one-off transfers or simple configurations, you can use CLI flags:

### Basic Example

```bash
# Basic JSON file transfer
sling run \
  --src-conn MY_SFTP \
  --src-stream '/path/to/data.json' \
  --src-options '{ "format": "json", "flatten": true }' \
  --tgt-conn MY_SNOWFLAKE \
  --tgt-object 'mydatabase.myschema.mytable' \
  --mode full-refresh
```

### Advanced Example

```bash
# Advanced JSON transfer with options, create a new table for each file
sling run \
  --src-conn MY_SFTP \
  --src-stream '/path/to/data/*.json' \
  --src-options '{
    "format": "json",
    "flatten": true,
    "jmespath": "records[*]",
    "empty_as_null": true
  }' \
  --transforms '[remove_diacritics, trim_space]' \
  --tgt-conn MY_SNOWFLAKE \
  --tgt-object 'mydatabase.myschema.{stream_file_name}' \
  --tgt-options '{
    "column_casing": "snake",
    "table_keys": {"primary": ["id"]},
    "add_new_columns": true
  }'
```

## Using Replication YAML

For more complex scenarios or when managing multiple streams, use a replication YAML file:

### Basic Multi-stream Example

```yaml
source: MY_SFTP
target: MY_SNOWFLAKE

defaults:
  mode: full-refresh
  source_options:
    format: json
    flatten: true
    empty_as_null: true

streams:
  '/data/customers/*.json':
    object: mydatabase.myschema.customers
    transforms: [remove_diacritics, trim_space]
    
  '/data/orders/*.json':
    object: mydatabase.myschema.orders
    transforms: [remove_diacritics, trim_space]
```

### Complex Multi-stream Example

```yaml
source: MY_SFTP
target: MY_SNOWFLAKE

env:
  SLING_SAMPLE_SIZE: 2000
  SLING_STREAM_URL_COLUMN: true

defaults:
  mode: incremental
  source_options:
    format: json
    flatten: true
    empty_as_null: true
    jmespath: "records[*]"
  target_options:
    column_casing: snake
    add_new_columns: true

streams:
  '/data/customers/${stream_date}/*.json':
    object: mydatabase.myschema.customers
    transforms:
      - remove_diacritics
      - trim_space
    primary_key: [customer_id]
    update_key: last_modified
    source_options:
      datetime_format: "YYYY-MM-DD HH:mm:ss"
    target_options:
      table_keys:
        primary: [customer_id]
        unique: [email]
  
  '/data/orders/${stream_date}/*.json':
    object: mydatabase.myschema.orders
    transforms:
      customer_name: [remove_diacritics, trim_space]
      email: [lower, trim_space]
    primary_key: [order_id]
    update_key: order_date
    source_options:
      datetime_format: "YYYY-MM-DD"
    target_options:
      table_keys:
        primary: [order_id]
```

To run a replication YAML configuration:

```bash
# Run the replication
sling run -r path/to/replication.yaml
```

# The Sling Platform

While the CLI is powerful for local development and automation, Sling also offers a comprehensive platform for managing data operations at scale. The platform provides a user-friendly interface for:

- Managing connections
- Creating and editing replications
- Monitoring executions
- Scheduling jobs
- Team collaboration

![Sling Platform Connections](../../../../assets/images/screenshots/ui.connections.light.png)

The Connections page allows you to manage your data sources and destinations securely.

![Sling Platform Editor](../../../../assets/images/screenshots/ui.editor.light.png)

The Editor provides a visual interface for creating and modifying replication configurations.

![Sling Platform Execution](../../../../assets/images/screenshots/ui.execution.light.png)

The Execution view shows detailed information about your data transfers, including progress, logs, and statistics.

## Platform Components

The Sling Platform consists of several key components:

1. **Control Plane**
   - Manages authentication and authorization
   - Handles job scheduling and orchestration
   - Provides the web interface

2. **Agents**
   - Execute data operations
   - Run in your infrastructure
   - Maintain secure access to your data sources

3. **Editor**
   - Visual replication configuration
   - Syntax highlighting and validation
   - Template management

4. **Monitoring**
   - Real-time execution tracking
   - Error reporting and alerting
   - Performance metrics

To get started with the platform:

1. Sign up at [https://app.slingdata.io/signup](https://app.slingdata.io/signup)
2. Create your first project
3. Install and configure a Sling agent
4. Set up your connections
5. Create and run your first replication

# Best Practices and Tips

To ensure optimal performance and reliability when transferring JSON data from SFTP to Snowflake:

1. **JSON Handling**
   - Use `flatten: true` for nested JSON structures
   - Specify `jmespath` expressions for complex JSON parsing
   - Set `empty_as_null: true` to handle empty strings consistently

2. **Performance Optimization**
   - Use wildcards for processing multiple files
   - Implement incremental loading for large datasets
   - Configure appropriate batch sizes

3. **Error Handling**
   - Set up proper logging and monitoring
   - Use primary keys to prevent duplicates
   - Implement retry mechanisms for network issues

4. **Security**
   - Use SSH keys instead of passwords for SFTP
   - Rotate credentials regularly
   - Follow the principle of least privilege

5. **Maintenance**
   - Regular testing of connections
   - Monitoring of job execution times
   - Cleanup of processed files

For more detailed information and examples, visit:
- [Sling Documentation](https://docs.slingdata.io/)
- [Replication Concepts](https://docs.slingdata.io/concepts/replication)
- [CLI Documentation](https://docs.slingdata.io/sling-cli/getting-started)
- [Platform Documentation](https://docs.slingdata.io/sling-platform/getting-started)

# Related Guides

These walkthroughs cover related SFTP and Snowflake loading scenarios:

- [Transferring CSV data from SFTP to Snowflake](/articles/sling-sftp-csv-to-snowflake) for delimited files instead of JSON.
- [Loading SFTP Parquet files into Snowflake](/articles/sling-sftp-parquet-to-snowflake) when your source data is columnar Parquet.
- [Importing files from SFTP into PostgreSQL](/articles/import-sftp-files-to-postgres-database) for the same source with a PostgreSQL target.
- [Loading local CSV data into Snowflake](/articles/export-load-local-csv-snowflake-sling) for file-based loads that originate on your own machine.
- [Migrating SQL Server to Snowflake](/articles/sling-sqlserver-to-snowflake) for a database-to-warehouse pipeline into the same target.

# FAQ

### How does Sling parse nested JSON files from an SFTP server?

Set `flatten` to `true` under `source_options` and Sling expands nested objects into individual columns using dotted names. For deeply nested or array-heavy payloads you can also supply a `jmespath` expression to select the exact records to load.

### What does the jmespath option do when loading JSON?

The `jmespath` option lets you pick a sub-path inside each JSON file before Sling processes it, such as `records[*]` to load only the items in a records array. It is useful when the file wraps the real data in an outer envelope.

### Can Sling load multiple JSON files from an SFTP directory in one run?

Yes. Use a wildcard path like `/data/orders/*.json` as the stream and Sling processes every matching file. The `{stream_file_name}` runtime variable lets you route each file to its own target table when needed.

### How do I authenticate to an SFTP server with an SSH key instead of a password?

Provide a `private_key` path instead of a password when defining the connection, either through `sling conns set` or in `env.yaml`. Key-based authentication is recommended over passwords for production pipelines.

### Does Sling support incremental loading of JSON files from SFTP?

Yes. Set `mode` to `incremental` and define a `primary_key` and `update_key` on the stream so Sling loads only new or changed records. This avoids reprocessing files that were already ingested.

### How can I handle empty strings in JSON source data?

Enable `empty_as_null` under `source_options` so Sling treats empty strings as NULL values. This keeps the loaded data consistent and avoids mixing empty strings with genuine nulls in Snowflake.

### Can I apply transformations to JSON data before it reaches Snowflake?

Yes. Use the `transforms` key on a stream with built-in functions such as `remove_diacritics` or `trim_space` to clean values during the load. Transforms run inline so no separate processing step is required.

# Conclusion

Sling provides a powerful and flexible solution for transferring JSON data from SFTP to Snowflake. Its combination of simple configuration, robust features, and comprehensive platform makes it an excellent choice for organizations of all sizes.

Key benefits include:
- Easy setup and configuration
- Support for complex JSON structures
- Robust error handling
- Scalable performance
- Comprehensive monitoring
- Both CLI and Platform interfaces

To take your data integration to the next level:
1. Install Sling and set up your connections
2. Start with simple transfers using CLI flags
3. Progress to replication YAML for complex scenarios
4. Consider the Platform for enterprise needs
5. Join the Sling community for support and updates

For more examples and detailed documentation, visit [https://docs.slingdata.io/](https://docs.slingdata.io/).