SQL Query Node

Node Configuration

{
  "nodeType": "SqlQuery",
  "name": "queryPendingInvoices",
  "config": {
    "datasourceId": "payroll-db",
    "sql": "SELECT invoice_id, invoice_number, vendor_id, total, currency, submitted_at FROM invoices WHERE status = @status AND vendor_id = @vendorId AND total >= @minAmount ORDER BY submitted_at ASC LIMIT @pageSize OFFSET @offset",
    "parameters": {
      "status": "pending-approval",
      "vendorId": "$output.fetchVendor.entityId",
      "minAmount": "$json.minimumAmount ?? 0",
      "pageSize": 50,
      "offset": 0
    },
    "rowLimit": 500,
    "timeoutSeconds": 30
  }
}

Configuration Fields

Node Output

{
  "rows": [
    {
      "invoice_id": "inv-001",
      "invoice_number": "INV-2026-001",
      "vendor_id": "vnd-acme",
      "total": 4750.00,
      "currency": "GBP",
      "submitted_at": "2026-05-20T09:00:00Z"
    }
  ],
  "rowCount": 1
}

SqlQueryExecutor

protected override async Task<NodeExecutionResult> ExecuteAsync(
    SqlQueryConfig config,
    NodeDataContext ctx,
    CancellationToken ct)
{
    var datasource = _registry.GetDatasource(config.DatasourceId);
    var parameters = _evaluator.EvaluateParameters(config.Parameters, ctx);

    // Validate: only SELECT statements are allowed
    if (!_sqlValidator.IsSelectOnly(config.Sql))
        return NodeExecutionResult.Fail(new InvalidOperationException(
            "SqlQueryNode only allows SELECT statements. Use EntityCreateNode/EntityUpdateNode for writes."));

    using var connection = await _connectionFactory.OpenAsync(datasource, ct);
    using var cmd = connection.CreateCommand();
    cmd.CommandText = config.Sql;
    cmd.CommandTimeout = config.TimeoutSeconds;
    foreach (var (key, value) in parameters)
        cmd.Parameters.AddWithValue("@" + key, value ?? DBNull.Value);

    var rows = new List<Dictionary<string, object?>>();
    using var reader = await cmd.ExecuteReaderAsync(ct);
    while (await reader.ReadAsync(ct) && rows.Count < config.RowLimit)
    {
        var row = new Dictionary<string, object?>();
        for (int i = 0; i < reader.FieldCount; i++)
            row[reader.GetName(i)] = reader.IsDBNull(i) ? null : reader.GetValue(i);
        rows.Add(row);
    }

    return NodeExecutionResult.Success(new { rows, rowCount = rows.Count });
}