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:

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 (<query>) 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

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.

Full Example

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