# Writing Sql

> Write SQL queries for Celigo RDBMS exports and imports -- SELECT, INSERT, UPDATE, UPSERT, MERGE, delta, once, and bulk operations across Snowflake, Postgres, MySQL, SQL Server, Oracle, BigQuery, and Redshift. Use when editing rdbms.query or troubleshooting SQL errors.

- **Type:** Skill
- **Install:** `agentstack add skill-celigo-ai-writing-sql`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [celigo](https://agentstack.voostack.com/s/celigo)
- **Installs:** 0
- **Category:** [Databases](https://agentstack.voostack.com/c/databases)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [celigo](https://github.com/celigo)
- **Source:** https://github.com/celigo/ai/tree/main/skills/writing-sql

## Install

```sh
agentstack add skill-celigo-ai-writing-sql
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# Writing SQL for RDBMS Integrations

RDBMS exports and imports use SQL queries to read from and write to relational databases. Queries are Handlebars templates -- the platform evaluates expressions like `{{{record.fieldName}}}` at runtime, then executes the resulting SQL against the connected database.

This skill covers **what SQL to write and how to write it**. For which adaptor type to use, connection setup, and queryType selection, see the related skills below.

## Quick Reference

### Context decision matrix

| Where | Field | Handlebars prefix | Braces | Example |
|---|---|---|---|---|
| Export query | `rdbms.query` | `record.` (parameterized) or none (standalone) | `{{{ }}}` | `SELECT * FROM orders WHERE id = {{{record.id}}}` |
| Export delta query | `rdbms.query` | platform-injected tokens | `{{ }}` or `{{{ }}}` | `WHERE updated_at > '{{lastExportDateTime}}'` |
| Export once mark-as-processed | `rdbms.once.query` | `record.` | `{{{ }}}` | `UPDATE orders SET exported = true WHERE id = {{{record.id}}}` |
| Import per_record query | `rdbms.query[]` | `record.` | `{{{ }}}` | `INSERT INTO users (name) VALUES ('{{{record.name}}}')` |
| Import per_page query | `rdbms.query[]` | `batch_of_records` + `record.` | `{{{ }}}` | `{{#each batch_of_records}}...{{{record.field}}}...{{/each}}` |
| Import bulk_load override | `rdbms.bulkLoad.overrideMergeQuery` | staging table ref | `{{ }}` | `MERGE INTO target USING {{import.rdbms.bulkLoad.preMergeTemporaryTable}}` |

### Key syntax rules

1. **Always use `record.` prefix** (AFE 2.0). Never bare `fieldName` or `data.fieldName`.
2. **Prefer triple braces `{{{ }}}`** for values in SQL. Double braces `{{ }}` auto-wrap values in single quotes in RDBMS context -- this silently corrupts numeric values and breaks SQL syntax.
3. **Add your own quotes** for strings: `'{{{record.name}}}'`. Triple braces give you full control.
4. **No quotes** for numbers: `{{{record.quantity}}}`.
5. **Import `query` is an array of strings**, not a single string: `["INSERT INTO ..."]`.
6. **Nested fields** use dot notation: `{{{record.address.city}}}`.

### Related skills

- [configuring-exports > RDBMS](../configuring-exports/SKILL.md) -- adaptorType, delta/once mode, export-level config
- [configuring-imports > RDBMS](../configuring-imports/SKILL.md) -- queryType decision tree, bulkInsert/bulkLoad config
- [writing-handlebars](../writing-handlebars/SKILL.md) -- full Handlebars syntax, helpers, block expressions
- [configuring-connections > RDBMS](../configuring-connections/SKILL.md) -- database connection setup

### Schema reference

- [rdbms-export.yml](references/schemas/rdbms-export.yml) -- export query and once fields
- [rdbms-import.yml](references/schemas/rdbms-import.yml) -- import query, queryType, bulkInsert, bulkLoad fields

## How to Write a SQL Query

### 1. Discover the schema

Query the live database to find tables and columns before writing SQL:

```bash
celigo metadata types                     # List tables
celigo metadata fields         # List columns and types
```

For Snowflake, use fully qualified names: `database.schema.table`.

### 2. Determine the query pattern

**Exports** -- what kind of data fetch?

| Pattern | Export type | Query template |
|---|---|---|
| Full fetch | `type: null` | `SELECT columns FROM table WHERE conditions` |
| Delta/incremental | `type: "delta"` | `SELECT ... WHERE updated_at > '{{lastExportDateTime}}'` |
| Once (mark-as-processed) | `type: "once"` | `rdbms.query` = SELECT unprocessed; `rdbms.once.query` = UPDATE to mark processed |

**Imports** -- what kind of write operation?

| Pattern | queryType | What to configure |
|---|---|---|
| INSERT with duplicate check | `["per_record"]` | `rdbms.query` with INSERT ... ON CONFLICT / ON DUPLICATE KEY |
| UPDATE | `["per_record"]` | `rdbms.query` with UPDATE ... WHERE |
| UPSERT / MERGE | `["per_record"]` or `["bulk_load"]` | per_record: `rdbms.query` with MERGE; bulk_load: `bulkLoad.primaryKeys` |
| Pure INSERT (no checking) | `["bulk_insert"]` | `rdbms.bulkInsert.tableName` and `batchSize` -- no SQL needed |
| Bulk load with upsert | `["bulk_load"]` | `rdbms.bulkLoad.tableName` and `primaryKeys` -- auto-generates MERGE |
| Custom bulk merge | `["bulk_load"]` | `bulkLoad.overrideMergeQuery: true` + custom SQL referencing staging table |

### 3. Write the SQL

Follow the database-specific [dialect patterns](#dialect-patterns) below. Use `{{{record.fieldName}}}` for runtime values.

### 4. Test the query

```bash
# Test an export query -- invoke returns real data
celigo exports invoke 

# Test an import -- submit test records
echo '[{"name":"test","email":"test@example.com"}]' | celigo imports invoke 
```

## Export Query Patterns

### Standard SELECT

```sql
SELECT id, name, email, status
FROM customers
WHERE status = 'ACTIVE'
ORDER BY id
```

### Delta export (incremental)

Use `{{lastExportDateTime}}` (platform-injected, not from a record). The token resolves to the timestamp of the last successful export run.

```sql
SELECT id, name, email, updated_at
FROM customers
WHERE updated_at > '{{lastExportDateTime}}'
ORDER BY updated_at ASC
```

For databases that need specific timestamp formats:

```sql
-- Snowflake
WHERE updated_at > TO_TIMESTAMP('{{lastExportDateTime}}', 'YYYY-MM-DD HH24:MI:SS')

-- MySQL
WHERE updated_at > STR_TO_DATE('{{lastExportDateTime}}', '%Y-%m-%d %H:%i:%s')

-- PostgreSQL
WHERE updated_at > '{{lastExportDateTime}}'::timestamp
```

### Once export (mark-as-processed)

Two queries work together. The export `rdbms.query` fetches unprocessed records; `rdbms.once.query` marks each one after successful export.

```json
{
  "type": "once",
  "rdbms": {
    "query": "SELECT id, name, email FROM orders WHERE exported = false",
    "once": {
      "query": "UPDATE orders SET exported = true WHERE id = {{{record.id}}}"
    }
  }
}
```

### JOINs

```sql
SELECT o.id, o.order_date, c.name AS customer_name, c.email,
       p.product_name, oi.quantity, oi.unit_price
FROM orders o
JOIN customers c ON o.customer_id = c.id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
WHERE o.status = 'SHIPPED'
  AND o.order_date > '{{lastExportDateTime}}'
```

### Aggregations

```sql
SELECT customer_id, COUNT(*) AS order_count, SUM(total) AS lifetime_value
FROM orders
WHERE status IN ('COMPLETED', 'SHIPPED')
GROUP BY customer_id
HAVING SUM(total) > 1000
```

## Import Query Patterns

All import queries use `{{{record.fieldName}}}` to inject values from incoming records. The `query` field is an **array of strings**.

### INSERT (per_record)

```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["INSERT INTO customers (name, email, phone) VALUES ('{{{record.name}}}', '{{{record.email}}}', '{{{record.phone}}}')"]
  }
}
```

### UPDATE (per_record)

```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["UPDATE customers SET name = '{{{record.name}}}', email = '{{{record.email}}}' WHERE id = {{{record.id}}}"]
  }
}
```

### UPSERT -- database-specific

**MySQL** (ON DUPLICATE KEY UPDATE):
```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["INSERT INTO customers (id, name, email) VALUES ({{{record.id}}}, '{{{record.name}}}', '{{{record.email}}}') ON DUPLICATE KEY UPDATE name = '{{{record.name}}}', email = '{{{record.email}}}'"]
  }
}
```

**PostgreSQL** (ON CONFLICT):
```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["INSERT INTO customers (id, name, email) VALUES ({{{record.id}}}, '{{{record.name}}}', '{{{record.email}}}') ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email"]
  }
}
```

**Snowflake** (MERGE):
```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["MERGE INTO customers AS t USING (SELECT {{{record.id}}} AS id, '{{{record.name}}}' AS name, '{{{record.email}}}' AS email) AS s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.name = s.name, t.email = s.email WHEN NOT MATCHED THEN INSERT (id, name, email) VALUES (s.id, s.name, s.email)"]
  }
}
```

**SQL Server** (MERGE):
```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": ["MERGE INTO customers AS t USING (SELECT {{{record.id}}} AS id, '{{{record.name}}}' AS name, '{{{record.email}}}' AS email) AS s ON t.id = s.id WHEN MATCHED THEN UPDATE SET t.name = s.name, t.email = s.email WHEN NOT MATCHED THEN INSERT (id, name, email) VALUES (s.id, s.name, s.email);"]
  }
}
```

### Multiple statements (per_record)

When you need to run multiple SQL statements per record, add them as separate array elements:

```json
{
  "rdbms": {
    "queryType": ["per_record"],
    "query": [
      "INSERT INTO orders (id, customer_id, total) VALUES ({{{record.orderId}}}, {{{record.customerId}}}, {{{record.total}}})",
      "UPDATE customers SET last_order_date = CURRENT_TIMESTAMP WHERE id = {{{record.customerId}}}"
    ]
  }
}
```

### Bulk insert (no SQL needed)

For pure INSERTs with no duplicate checking, use `bulkInsert` -- the platform generates the SQL:

```json
{
  "rdbms": {
    "queryType": ["bulk_insert"],
    "bulkInsert": {
      "tableName": "customers",
      "batchSize": "1000"
    }
  }
}
```

Column mapping comes from `mappings[]` on the import (Mapper 2.0). Each mapping's `generate` value must match a column name in the target table.

### Bulk load with auto-generated MERGE

For high-volume upsert on Snowflake or Azure Synapse, use `bulkLoad` with `primaryKeys`:

```json
{
  "rdbms": {
    "queryType": ["bulk_load"],
    "bulkLoad": {
      "tableName": "WAREHOUSE.PUBLIC.CUSTOMERS",
      "primaryKeys": "id"
    }
  }
}
```

The platform stages data into a temporary table, then auto-generates a MERGE using the primary keys. For composite keys: `"primaryKeys": "order_id,product_id"`.

### Bulk load with custom merge

When the auto-generated MERGE isn't sufficient (conditional updates, ignore-existing, multi-table ops), override it:

```json
{
  "rdbms": {
    "queryType": ["bulk_load"],
    "bulkLoad": {
      "tableName": "WAREHOUSE.PUBLIC.CUSTOMERS",
      "primaryKeys": "id",
      "overrideMergeQuery": true
    }
  }
}
```

The custom SQL goes in the `rdbms.query` field (yes, even with `bulkLoad`). Reference the staging table via `{{import.rdbms.bulkLoad.preMergeTemporaryTable}}`:

```sql
MERGE INTO WAREHOUSE.PUBLIC.CUSTOMERS AS t
USING {{import.rdbms.bulkLoad.preMergeTemporaryTable}} AS s
ON t.id = s.id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET t.name = s.name, t.email = s.email, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (id, name, email, updated_at) VALUES (s.id, s.name, s.email, s.updated_at)
```

### per_page batch operations

`per_page` gives you the entire page of records as `batch_of_records`. Use `{{#each}}` to iterate:

```json
{
  "rdbms": {
    "queryType": ["per_page"],
    "query": ["INSERT INTO customers (name, email) VALUES {{#each batch_of_records}}('{{{record.name}}}', '{{{record.email}}}'){{#unless @last}},{{/unless}}{{/each}}"]
  }
}
```

## Dialect Patterns

### Snowflake

```sql
-- Fully qualified table names (required unless connection sets default schema)
SELECT * FROM MY_DATABASE.MY_SCHEMA.MY_TABLE

-- FLATTEN for semi-structured data (VARIANT columns)
SELECT f.value:name::STRING AS name, f.value:email::STRING AS email
FROM MY_TABLE, LATERAL FLATTEN(input => MY_TABLE.json_column) f

-- QUALIFY for window function filtering
SELECT * FROM orders
QUALIFY ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) = 1

-- Timestamp handling
WHERE updated_at > TO_TIMESTAMP('{{lastExportDateTime}}', 'YYYY-MM-DD"T"HH24:MI:SS')

-- Case: Snowflake uppercases unquoted identifiers. Use double quotes to preserve case
SELECT "camelCaseColumn" FROM "MixedCaseTable"
```

### PostgreSQL

```sql
-- JSONB operators
SELECT data->>'name' AS name, data->'address'->>'city' AS city
FROM customers WHERE data @> '{"active": true}'

-- ILIKE for case-insensitive matching
SELECT * FROM products WHERE name ILIKE '%widget%'

-- ON CONFLICT for upsert
INSERT INTO customers (id, name, email) VALUES ({{{record.id}}}, '{{{record.name}}}', '{{{record.email}}}')
ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, email = EXCLUDED.email

-- Array operations
SELECT * FROM users WHERE 'admin' = ANY(roles)

-- CTEs
WITH recent_orders AS (
  SELECT * FROM orders WHERE order_date > '{{lastExportDateTime}}'
)
SELECT c.name, r.total FROM customers c JOIN recent_orders r ON c.id = r.customer_id
```

### MySQL / MariaDB

```sql
-- ON DUPLICATE KEY UPDATE for upsert
INSERT INTO customers (id, name, email)
VALUES ({{{record.id}}}, '{{{record.name}}}', '{{{record.email}}}')
ON DUPLICATE KEY UPDATE name = VALUES(name), email = VALUES(email)

-- JSON_EXTRACT for JSON columns
SELECT JSON_EXTRACT(data, '$.name') AS name FROM customers

-- GROUP_CONCAT for string aggregation
SELECT customer_id, GROUP_CONCAT(product_name SEPARATOR ', ') AS products
FROM order_items GROUP BY customer_id

-- IFNULL for null handling
SELECT IFNULL(middle_name, '') AS middle_name FROM users

-- Timestamp formatting
WHERE updated_at > STR_TO_DATE('{{lastExportDateTime}}', '%Y-%m-%d %H:%i:%s')
```

### SQL Server / Azure Synapse

```sql
-- MERGE with required semicolon terminator
MERGE INTO customers AS t
USING (SELECT {{{record.id}}} AS id, '{{{record.name}}}' AS name) AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.name = s.name
WHEN NOT MATCHED THEN INSERT (id, name) VALUES (s.id, s.name);

-- TOP instead of LIMIT
SELECT TOP 100 * FROM orders ORDER BY order_date DESC

-- STRING_AGG for string aggregation (SQL Server 2017+)
SELECT customer_id, STRING_AGG(product_name, ', ') AS products
FROM order_items GROUP BY customer_id

-- CROSS APPLY for row-valued functions
SELECT c.name, o.total FROM customers c
CROSS APPLY (SELECT TOP 1 * FROM orders WHERE customer_id = c.id ORDER BY order_date DESC) o

-- Bracket identifiers for reserved words or special characters
SELECT [order], [name] FROM [my-table]
```

### Oracle

```sql
-- NVL for null handling
SELECT NVL(middle_name, '') AS middle_name FROM users

-- ROWNUM for limiting results (pre-12c)
SELECT * FROM (SELECT * FROM orders ORDER BY order_date DESC) WHERE ROWNUM  TIMESTAMP('{{lastExportDateTime}}')
```

### Redshift

```sql
-- PostgreSQL-based syntax
SELECT * FROM orders WHERE status = 'ACTIVE'

-- LISTAGG for string aggregation
SELECT customer_id, LISTAGG(product_name, ', ') WITHIN GROUP (ORDER BY product_name)
FROM order_items GROUP BY customer_id

-- COPY-oriented -- bulk_load uses Redshift COPY under the hood
-- For custom merge logic, use overrideMergeQuery with staging table reference
```

## Pre-Submit Checklist

### Export queries
- [ ] **SELECT query is syntactically valid** for the target database dialect
- [ ] **Delta uses `{{lastExportDateTime}}`** in the query, NOT a `delta` object inside `rdbms`
- [ ] **Once export has both queries** -- `rdbms.query` for SELECT and `rdbms.once.query` for UPDATE
- [ ] **Table and column names exist** -- verify with `celigo metadata types/fields `
- [ ] **Snowflake uses fully qualified names** -- `database.schema.table` unless the connection sets defaults

### Import queries
- [ ] **`query` is an array of strings** -- `["INSERT INTO ..."]`, not `"INSERT INTO ..."`
- [ ] **`queryType` matches the operation** -- `["per_record"]` for UPDATE/UPSERT, `["bulk_insert"]` for pure INSERT, `["bulk_load"]` for high-volume
- [ ] **All `{{{record.fieldName}}}` paths match incoming data** -- invoke the upstream export to verify field names
- [ ] **Strings are quoted, numbers are not** -- `'{{{record.name}}}'` vs `{{{record.id}}}`
- [ ] **Uses `record.` prefix** -- not bare `fieldName` or `data.fieldName`
- [ ] **Uses triple braces `{{{ }}}`** -- double braces auto-wrap in quotes, breaking numeric values and SQL syntax
- [ ] **bulkInsert/bulkLoad not set alongside query** -- these are mutually exclusive with the `query` field (except `overrideMergeQuery`)

### Cross-resource
- [ ] **Connection type is

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [celigo](https://github.com/celigo)
- **Source:** [celigo/ai](https://github.com/celigo/ai)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-celigo-ai-writing-sql
- Seller: https://agentstack.voostack.com/s/celigo
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.