Articles in this section

Query files in Celigo Storage

Kathyana Queeman
Updated

Run a SQL statement against files in Celigo Storage and read the results back as JSON rows. You write a SELECT statement that references a file or a folder by its storage path, submit it as a query job, poll the job until it completes, and fetch the result one page at a time. Every query, whatever its size, follows the same three calls.

For an introduction, including when to use a query and how folder references and authorization work, see Querying Celigo Storage files.

Prerequisites:

  • Querying Celigo Storage is enabled for your account. It's enabled per account on request — if it isn't, every request fails with STORAGE_QUERY_NOT_ENTITLED. Contact your Celigo account team to turn it on.
  • A token whose scope covers the files you want to query: a Full access API token, a Custom API token whose File storage paths include the folders the files are in, or a personal access token whose user can open the files. See Create an API token or Personal Access Token.
  • The files are in Celigo Storage, in a supported format: CSV, JSON, NDJSON, or Parquet. Their uploads are complete.
  • If you're querying a folder, every file directly inside it has the same format, and the folder holds at most 100 files.

Write the statement

  1. Go to Resources resources-icon.png, and then select File storage. Copy the path of each file or folder you want to query exactly, including its case.
  2. Write one SELECT statement and put each path in single quotes where a table name would go. Reference a file by its full path, for example '/staging/orders.csv'. Reference a folder with a trailing slash, for example '/staging/orders/', to read every file directly inside it.
  3. If the engine's defaults don't parse a file correctly, wrap the path in the reader function for its format and pass options, for example read_csv('/staging/orders/', header = false). See Read options and format behavior.
  4. To see a file's columns and inferred types before you write the real query, submit DESCRIBE SELECT * FROM '/staging/orders.csv' as a query of its own.

Note

Celigo checks a statement before it runs. At submit, it rejects a second statement, a statement outside the SELECT family, a function outside the allowlist, a schema-qualified table name, and a wildcard in a path, each with its own error code. See Celigo SQL reference.

Submit the query job

  1. Send a POST request to /v1/storage/query on your integrator.io API host with the token in the Authorization header and the statement in the sql field.

    POST /v1/storage/query
Authorization: Bearer <your API token>
Content-Type: application/json
{
  "sql": "SELECT region, COUNT(*) AS orders, SUM(total) AS revenue FROM '/staging/orders/' GROUP BY region ORDER BY revenue DESC"
}

  2. Check the response status. 202 means Celigo accepted the job.

    {
  "jobId": "66e1f2a3b4c5d6e7f8a9b0c1",
  "status": "queued"
}
  3. Save the jobId. You need it to poll the job and fetch its results.

A 4xx response means Celigo rejected the statement, a path, or your token's access, and no job exists. The body carries an errors array with a code and message.

{
  "errors": [
    { "code": "STORAGE_QUERY_SOURCE_NOT_FOUND", "message": "No storage file found at path \"/staging/orders/\"" }
  ]
}

Fix the cause and submit again. For each code, see Troubleshoot a Celigo Storage query job.

Poll the job

  1. Send a GET request to /v1/storage/query/{jobId}/status with the same token.

    GET /v1/storage/query/66e1f2a3b4c5d6e7f8a9b0c1/status
Authorization: Bearer <your API token>
  2. Check the status field. It is one of the following values.
    • queued — The job is waiting to run. Time spent here doesn't count against the timeout.
    • running — The query is executing. The 300-second timeout counts from startedAt.
    • completed — The result is ready to fetch.
    • failed — The query didn't finish. The error field carries the code and message.
  3. While the status is queued or running, wait a few seconds and poll again. Most queries finish within seconds; a sort or join over several gigabytes can take minutes. If a job is still running well past 300 seconds after startedAt, it was interrupted, and Celigo marks it failed within about 7 minutes. Submit the query again.

A completed job reports the result's shape alongside its timing.

{
  "jobId": "66e1f2a3b4c5d6e7f8a9b0c1",
  "status": "completed",
  "createdAt": "2026-09-12T18:02:11.418Z",
  "startedAt": "2026-09-12T18:02:12.907Z",
  "completedAt": "2026-09-12T18:02:14.115Z",
  "executionMs": 1208,
  "rowCount": 4680,
  "truncated": false,
  "pageCount": 3,
  "columns": [
    { "name": "region", "type": "VARCHAR" },
    { "name": "orders", "type": "BIGINT" },
    { "name": "revenue", "type": "DOUBLE" }
  ]
}
  • columns — The result's column names and the types the engine inferred or computed, in result order.
  • rowCount — The total number of rows across all pages.
  • truncated — true when the result was cut off at the 1,000,000-row cap. Otherwise false.
  • pageCount — The number of result pages to fetch. A query that matched no rows completes with rowCount: 0 and pageCount: 0.
  • executionMs — How long the query ran.

A failed job reports the error instead.

{
  "jobId": "66e1f2a3b4c5d6e7f8a9b0c1",
  "status": "failed",
  "createdAt": "2026-09-12T18:02:11.418Z",
  "startedAt": "2026-09-12T18:02:12.907Z",
  "completedAt": "2026-09-12T18:07:13.002Z",
  "error": {
    "code": "STORAGE_QUERY_TIMEOUT",
    "message": "Query exceeded the maximum execution time of 300 seconds"
  }
}

Fetch the results

  1. Send a GET request to /v1/storage/query/{jobId}/results?page=1.

    GET /v1/storage/query/66e1f2a3b4c5d6e7f8a9b0c1/results?page=1
Authorization: Bearer <your API token>

  2. Read the rows from the response. Each row is a JSON object keyed by column name.

    {
  "page": 1,
  "pageCount": 3,
  "rowCount": 2000,
  "hasMore": true,
  "rows": [
    { "region": "NA", "orders": "1840", "revenue": 91230.5 },
    { "region": "EU", "orders": "1211", "revenue": 60117.25 }
  ]
}
  3. While hasMore is true, request the next page: ?page=2, ?page=3, and so on, up to pageCount. Pages are roughly 5 MB each, and a row is never split across pages.
  • rowCount here is the number of rows on this page. The total is on the status response.
  • A page number above pageCount, or anything other than a positive integer, returns 400 STORAGE_QUERY_PAGE_OUT_OF_RANGE.
  • Requesting results before the job completes returns 409 STORAGE_QUERY_JOB_NOT_READY. Poll the status until it is completed.
  • A completed job with no rows returns page: 1, pageCount: 0, hasMore: false, and an empty rows array for page 1.

Note

In the example, orders is a string and revenue is a number. Celigo returns 64-bit integers and decimals as strings so that large values keep every digit, and COUNT is a 64-bit integer. Whole-number columns read from CSV and JSON files follow the same rule. SUM over a DOUBLE column and AVG return numbers. To receive a number, declare the column's type in the reader, for example read_csv('/staging/orders/', types = {'qty': 'INTEGER'}), or cast the expression, for example CAST(COUNT(*) AS INTEGER). See Read options and format behavior.

The job record and its result pages are available for about 24 hours after the job finishes. After that, both endpoints return 404 STORAGE_QUERY_JOB_NOT_FOUND. There is no cancel request. A job runs until it completes, fails, or reaches the timeout.

Deduplicate a folder of daily files

A flow writes one CSV of invoices to /inbound/invoices/ each day, and the same invoice can appear in more than one day's file. This query reads the whole folder and keeps one row per invoice.

  1. Submit the job.

    {
  "sql": "SELECT DISTINCT invoice_number, invoice_date, amount FROM '/inbound/invoices/' ORDER BY invoice_date"
}

    The folder reference reads every CSV directly inside /inbound/invoices/. Celigo checks that the folder holds one format and at most 100 files, and returns 202 with the jobId.

  2. Poll /v1/storage/query/{jobId}/status until status is completed. The response reports columns for invoice_number, invoice_date, and amount, the deduplicated rowCount, and pageCount.
  3. Fetch /v1/storage/query/{jobId}/results?page=1 and continue while hasMore is true.

To keep only the latest copy of an invoice when the files disagree, use a window function instead of DISTINCT.

SELECT invoice_number, invoice_date, amount
FROM (
  SELECT *, row_number() OVER (PARTITION BY invoice_number ORDER BY invoice_date DESC) AS rn
  FROM '/inbound/invoices/'
)
WHERE rn = 1

To join the deduplicated invoices to a customer file, reference the second path in the same statement. Each distinct path counts as one source, and a query can reference up to 10.

SELECT i.invoice_number, i.amount, c.tier
FROM '/inbound/invoices/' AS i
LEFT JOIN '/reference/customers.csv' AS c ON i.customer_id = c.customer_id

Learn more