JDBC Task

"type" : "JDBC"

The JDBC task (JDBC) executes SQL statements against relational databases. It supports SELECT queries, UPDATE/INSERT/DELETE statements, parameterized queries, and transaction management with automatic rollback on failure.

Multiple named database connections can be configured, allowing workflows to interact with different databases (MySQL, PostgreSQL, Oracle, etc.) within the same workflow.

Task parameters

Parameter	Type	Description	Required / Optional
integrationName	String	The name of the configured Relational Database integration.	Required.
type	String	The SQL operation type. Supported: `SELECT`, `UPDATE`.	Required.
statement	String	The SQL statement to execute. Use `?` for parameterized queries.	Required.
parameters	List[String]	Ordered list of parameter values for `?` placeholders in the statement.	Optional.
expectedUpdateCount	Integer	For `UPDATE` type only. The number of rows the statement is expected to change. If the actual count is different, the transaction is rolled back and the task fails. Must be greater than 0. If not set, or set to 0, the update count is not checked.	Optional.
schemaName	String	Database schema name (reserved for future use).	Optional.

Configuration JSON

SELECT query

{
  "name": "query_users",
  "taskReferenceName": "query_users_ref",
  "type": "JDBC",
  "inputParameters": {
    "integrationName": "postgres-prod",
    "type": "SELECT",
    "statement": "SELECT id, name, email FROM users WHERE status = ?",
    "parameters": ["active"]
  }
}

UPDATE with expected count

{
  "name": "update_order_status",
  "taskReferenceName": "update_order_ref",
  "type": "JDBC",
  "inputParameters": {
    "integrationName": "postgres-prod",
    "type": "UPDATE",
    "statement": "UPDATE orders SET status = ? WHERE order_id = ?",
    "parameters": [
      "shipped",
      "${workflow.input.orderId}"
    ],
    "expectedUpdateCount": 1
  }
}

Output

SELECT output

Name	Type	Description
result	List[Map[String, Any]]	List of rows, where each row is a map of column names to values.

Example output:

{
  "result": [
    {"id": 1, "name": "Alice", "email": "alice@example.com"},
    {"id": 2, "name": "Bob", "email": "bob@example.com"}
  ]
}

UPDATE output

Name	Type	Description
update_count	Integer	The number of rows affected by the statement.

Example output:

{
  "update_count": 1
}

Transaction behavior

SELECT statements run with auto-commit enabled (default JDBC behavior).
UPDATE statements run with auto-commit disabled. The transaction is committed on success.
If expectedUpdateCount is set and the actual count doesn't match, the transaction is automatically rolled back and the task fails.
If a SQL exception occurs during an UPDATE, the transaction is automatically rolled back.

Execution

The JDBC task completes as follows:

COMPLETED: The SQL statement executed successfully. For SELECT, results are in output.result. For UPDATE, the count is in output.update_count.
FAILED: The task fails if:
- The integrationName doesn't match any configured integration.
- A SQL exception occurs (syntax error, constraint violation, connection timeout).
- The expectedUpdateCount doesn't match the actual update count (UPDATE only, triggers rollback).

Examples

Parameterized SELECT

{
  "name": "find_active_orders",
  "taskReferenceName": "find_orders_ref",
  "type": "JDBC",
  "inputParameters": {
    "integrationName": "postgres-analytics",
    "type": "SELECT",
    "statement": "SELECT order_id, total, created_at FROM orders WHERE customer_id = ? AND status = ? ORDER BY created_at DESC",
    "parameters": [
      "${workflow.input.customerId}",
      "active"
    ]
  }
}

INSERT with expected count

{
  "name": "create_audit_record",
  "taskReferenceName": "audit_ref",
  "type": "JDBC",
  "inputParameters": {
    "integrationName": "postgres-prod",
    "type": "UPDATE",
    "statement": "INSERT INTO audit_log (action, user_id, details, created_at) VALUES (?, ?, ?, NOW())",
    "parameters": [
      "${workflow.input.action}",
      "${workflow.input.userId}",
      "${workflow.input.details}"
    ],
    "expectedUpdateCount": 1
  }
}

Chaining SELECT and UPDATE

Use the output of a SELECT task as input to an UPDATE task:

[
  {
    "name": "get_order",
    "taskReferenceName": "get_order_ref",
    "type": "JDBC",
    "inputParameters": {
      "integrationName": "postgres-prod",
      "type": "SELECT",
      "statement": "SELECT id, total FROM orders WHERE order_id = ?",
      "parameters": ["${workflow.input.orderId}"]
    }
  },
  {
    "name": "apply_discount",
    "taskReferenceName": "apply_discount_ref",
    "type": "JDBC",
    "inputParameters": {
      "integrationName": "postgres-prod",
      "type": "UPDATE",
      "statement": "UPDATE orders SET total = total * 0.9 WHERE order_id = ? AND total > 0",
      "parameters": ["${workflow.input.orderId}"],
      "expectedUpdateCount": 1
    }
  }
]

Using with different databases in the same workflow

[
  {
    "name": "read_from_mysql",
    "taskReferenceName": "mysql_read_ref",
    "type": "JDBC",
    "inputParameters": {
      "integrationName": "postgres-prod",
      "type": "SELECT",
      "statement": "SELECT user_id, email FROM users WHERE user_id = ?",
      "parameters": ["${workflow.input.userId}"]
    }
  },
  {
    "name": "write_to_postgres",
    "taskReferenceName": "pg_write_ref",
    "type": "JDBC",
    "inputParameters": {
      "integrationName": "postgres-analytics",
      "type": "UPDATE",
      "statement": "INSERT INTO user_activity (user_id, email, event_type, event_time) VALUES (?, ?, ?, NOW())",
      "parameters": [
        "${workflow.input.userId}",
        "${mysql_read_ref.output.result[0].email}",
        "workflow_triggered"
      ],
      "expectedUpdateCount": 1
    }
  }
]

SQL injection

Always use parameterized queries (? placeholders with the parameters list). Never concatenate user input directly into SQL statements.