> For the complete documentation index, see [llms.txt](https://docs.adiom.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.adiom.io/reference/connectors/sql-batch.md). # SQL Batch The SQL Batch connector is supported as a source only. It is built into dsync/dsynct and reads data from SQL databases using configurable queries. It supports SQL Server, PostgreSQL, and Oracle. ## Configuration Specify `sqlbatch` as the source with a `--config` flag pointing to a YAML configuration file: ``` sqlbatch --config=config.yml ``` ## Data Type The SQL Batch connector uses **JSON**. ## Config File The configuration file is a YAML file with the following top-level fields: ```yaml id: my-source driver: postgres connectionstring: "postgres://user:password@host:5432/dbname" mappings: - namespace: db.users query: "SELECT id, name, email FROM users" partitionquery: "SELECT id FROM users WHERE id % 4 = 0 ORDER BY id" cols: [id] limit: 1000 ``` ### Top-Level Fields | Field | Required | Description | | ------------------ | -------- | ----------------------------------------------------------------------- | | `id` | No | Identifier for the connector. Defaults to `"sql"` if not set. | | `driver` | Yes | SQL driver to use. Supported values: `sqlserver`, `postgres`, `oracle`. | | `connectionstring` | Yes | Connection string for the database. | | `mappings` | Yes | List of mapping definitions (see below). | ### Mapping Fields Each entry in `mappings` defines a namespace (analogous to a collection or table) that the connector will read. | Field | Required | Description | | ---------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `namespace` | Yes | Name used to identify this data set (e.g. `db.users`). This is the namespace that dsync will use. | | `query` | Yes | SQL query that returns the data. Must include all key columns from `cols`. | | `partitionquery` | Yes | SQL query that returns partition boundary values. The columns must match `cols`. Results must be sorted in ascending order and contain no duplicates. Used to split the initial sync into parallel partitions. | | `cols` | Yes | List of key column names. Used for partitioning, ordering, and identifying rows. | | `limit` | Yes | Batch size for reading data. Must be at least 1. | | `countquery` | No | SQL query that returns a single count value. If omitted, defaults to `WITH QUERY AS () SELECT COUNT(*) from QUERY`. | | `nocount` | No | If `true`, skips the count query entirely. | | `decodejson` | No | List of column names whose values should be decoded from JSON strings into structured data. | | `fetchers` | No | Number of parallel fetchers for change streaming. Defaults to 1. | | `changes` | No | List of change tracking configurations for streaming updates (see below). | ### Change Tracking If you want the connector to stream ongoing changes (not just do an initial sync), configure the `changes` field. Each entry polls for changes using a query. The changes `query` receives the current cursor value as a single positional parameter. The placeholder syntax depends on the driver: | Driver | Placeholder | | ----------- | ----------- | | `sqlserver` | `@p1` | | `postgres` | `$1` | | `oracle` | `:1` | ```yaml mappings: - namespace: db.users query: "SELECT id, name, email FROM users" partitionquery: "SELECT id FROM users WHERE id % 4 = 0 ORDER BY id" cols: [id] limit: 1000 changes: - initialcursorquery: "SELECT MAX(updated_at) FROM users" query: "SELECT id, 'U', updated_at FROM users WHERE updated_at > $1 ORDER BY updated_at LIMIT 1000" interval: 5s ``` | Field | Required | Description | | -------------------- | -------- | ----------------------------------------------------------------------------------------------------------------- | | `initialcursorquery` | Yes | Query that returns a single value to use as the starting cursor. | | `query` | Yes | Query that returns changed rows (see format below). Use the driver-specific placeholder for the cursor parameter. | | `interval` | No | How long to wait before polling again when no changes are found. Defaults to `5s`. | #### Changes Query Format The changes `query` is critical to get right. It must return columns in this exact order: 1. **Key columns** -- the same columns listed in `cols`, used to identify which row changed. 2. **Update type** -- a string column: `"D"` for deletes, any other value (e.g. `"U"`, `"I"`) for upserts. 3. **Cursor value** -- a monotonically increasing value (e.g. a version number or timestamp) that becomes the input for the next poll cycle. For example, with `cols: [id]`, the query must return exactly 3 columns: `id`, update type, cursor value. The connector polls in a loop: it passes the current cursor as the placeholder parameter, reads all returned rows, advances the cursor to the last row's cursor value, and repeats until no more rows are returned. It then waits for `interval` before polling again. **Important considerations:** * The query should filter using `> cursor` (not `>=`) to avoid reprocessing the same row, unless the cursor value is set up to guarantee no overlap. * Results should be ordered by the cursor column in ascending order so that the connector advances through changes sequentially. * The number of rows returned per poll is bounded by `limit` from the mapping. If the query returns `limit` or more rows, the connector immediately polls again without waiting for `interval`. ## Emitted Format This is what the SQL Batch source hands to Dsync, and therefore what a [transformation](/enterprise/running-dsynct/data-transformations.md) receives. **Namespace:** the `namespace` string of the mapping. It is not derived from a table name. **ID:** the columns listed in `cols`, in that order. Each part is named after its column and keeps the column's type (for example, an integer key stays an integer). **Initial sync:** each row of `query` is emitted as one flat JSON object whose fields are the result columns, named as the database driver returns them. No `id` field is added. | Column value | JSON value | | ----------------------------- | ----------------------------------------------------- | | Integer, float | Number | | Boolean | `true` / `false` | | String | String | | `NULL` | `null` | | Date and time | RFC 3339 string, for example `"2024-01-02T03:04:05Z"` | | Binary | Base64 string | | Column listed in `decodejson` | The decoded JSON value (object, array, etc.) | How decimal, UUID and other database-specific types appear depends on the driver. For example, the SQL Server driver returns `DECIMAL` and `NUMERIC` columns as bytes, so they arrive as base64 strings; cast such columns to a float or a string in `query`. Check a sample row before writing a transform that relies on them. **CDC:** each row returned by a `changes` query produces one event. | Update type column | Event type | ID | Payload | | ------------------ | ---------- | ---------------------------------- | ---------------------------------------------- | | `"D"` | `DELETE` | Key columns from the changes query | None. | | Anything else | `UPDATE` | Key columns | The full current row, re-read through `query`. | ### Example With `cols: [id]` and `query: "SELECT id, name, created_at, avatar FROM users"`: | Stage | ID | Payload | | ---------------------------------------------- | ----------------- | ---------------------------------------------------------------------------------- | | Initial sync | (part of the row) | `{"id": 7, "name": "Ann", "created_at": "2024-01-02T03:04:05Z", "avatar": "AQI="}` | | Changes row `(7, 'I', 101)` or `(7, 'U', 102)` | `id` = `7` | `{"id": 7, "name": "Ann", "created_at": "2024-01-02T03:04:05Z", "avatar": "AQI="}` | | Changes row `(7, 'D', 103)` | `id` = `7` | None | ### Things to Know * Inserts are never emitted as `INSERT`. New and changed rows both arrive as `UPDATE` with the full row. * If a row changes several times between polls, only the last change is emitted. If a changed row no longer exists when it is re-read, no event is emitted for that change. * The key columns in the changes query must have the same types as in `query`. A cast in one and not the other produces delete IDs that do not match the IDs written earlier. * The destination ID is not set for you. Alias a column to `id` in `query`, or map the ID in a transform. See [IDs, Data Types and CDC Events](/enterprise/running-dsynct/data-transformations/data-types.md). * The connector does not support fetching rows by ID, so `testsync`, Verify IDs and the Sync Tester's Test Sync cannot use it as a source. Test transforms in Transform Studio with sample rows, then with a sync into a test destination. ## Full Example ```yaml id: my-sql-source driver: sqlserver connectionstring: "sqlserver://user:password@host:1433?database=mydb" mappings: - namespace: dbo.orders query: "SELECT order_id, customer_id, total, details FROM orders" partitionquery: "SELECT order_id FROM orders WHERE order_id % 8 = 0 ORDER BY order_id" cols: [order_id] limit: 5000 decodejson: [details] changes: - initialcursorquery: "SELECT CHANGE_TRACKING_CURRENT_VERSION()" query: "SELECT CT.order_id, CT.SYS_CHANGE_OPERATION, CT.SYS_CHANGE_VERSION FROM CHANGETABLE(CHANGES dbo.orders, @p1) AS CT ORDER BY CT.SYS_CHANGE_VERSION" interval: 10s - namespace: dbo.customers query: "SELECT customer_id, name, email FROM customers" partitionquery: "SELECT customer_id FROM customers WHERE customer_id % 4 = 0 ORDER BY customer_id" cols: [customer_id] limit: 2000 nocount: true ```