# Incremental Data Loading
Source: https://docs.file.ai/docs-api/api-incremental-data-loading

> Sync files incrementally to data warehouses using Change Data Capture patterns with the GET /files endpoint.

## On this page

*   [Overview](#overview)
*   [Timestamp Filters](#timestamp-filters)
    *   [Timestamp Contract](#timestamp-contract)
*   [Watermark Semantics](#watermark-semantics)
    *   [Inclusive Comparison (>=)](#inclusive-comparison-%3E%3D)
    *   [Deduplication Requirement](#deduplication-requirement)
*   [Ordering & Pagination](#ordering-%26-pagination)
    *   [Default Sort Order](#default-sort-order)
    *   [Recommended Sort for CDC](#recommended-sort-for-cdc)
    *   [Pagination Parameters](#pagination-parameters)
    *   [Pagination Stability Warning](#pagination-stability-warning)
*   [What Triggers updatedAt?](#what-triggers-updatedat)
*   [Delete Handling](#delete-handling)
    *   [Tracking Deletions](#tracking-deletions)
*   [Best Practices](#best-practices)
*   [Snowflake Integration Example](#snowflake-integration-example)
*   [Response Schema](#response-schema)
*   [Quick Reference](#quick-reference)
*   [Need Help?](#need-help)

Key Functions

# Incremental Data Loading

Copy pageCopy page

Sync files incrementally to data warehouses using Change Data Capture patterns with the GET /files endpoint.

Copy pageCopy page

## 

[​

](#overview)

Overview

The `GET /files` endpoint supports **incremental data loading** through timestamp-based filters. This enables efficient Change Data Capture (CDC) workflows for syncing to Snowflake, BigQuery, or other data warehouses.

Instead of fetching all files on every sync, use `updatedAfter` or `createdAfter` to retrieve only what’s changed since your last sync.

* * *

## 

[​

](#timestamp-filters)

Timestamp Filters

## updatedAfter

Filters files updated **on or after** the specified timestamp. `GET /files?updatedAfter=2025-01-01T00:00:00.000Z`

## createdAfter

Filters files created **on or after** the specified timestamp. `GET /files?createdAfter=2025-01-01T00:00:00.000Z`

### 

[​

](#timestamp-contract)

Timestamp Contract

Property

Value

**Format**

ISO 8601

**Timezone**

UTC

**Precision**

Milliseconds

**Example**

`2025-01-01T00:00:00.000Z`

All timestamps in the response (`createdAt`, `updatedAt`) are returned in UTC with millisecond precision.

* * *

## 

[​

](#watermark-semantics)

Watermark Semantics

### 

[​

](#inclusive-comparison-\>=)

Inclusive Comparison (`>=`)

The `updatedAfter` and `createdAfter` filters use **inclusive** comparison. This means:

*   A file with `updatedAt = 2025-01-01T12:00:00.000Z` **will be returned** when querying with `updatedAfter=2025-01-01T12:00:00.000Z`

### 

[​

](#deduplication-requirement)

Deduplication Requirement

Because of inclusive comparison, **clients may receive duplicate records** across paginated requests or subsequent syncs. You must deduplicate using `fileId + updatedAt`.

**Example deduplication in Snowflake:**

```
MERGE INTO target_table t
USING staging_table s
ON t.file_id = s.file_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET ...
WHEN NOT MATCHED THEN
  INSERT ...
```

* * *

## 

[​

](#ordering-&-pagination)

Ordering & Pagination

### 

[​

](#default-sort-order)

Default Sort Order

Parameter

Default Value

`sortBy`

`createdAt`

`sortOrder`

`ASC`

### 

[​

](#recommended-sort-for-cdc)

Recommended Sort for CDC

When using `updatedAfter` for incremental loading, always sort by `updatedAt` in ascending order for reliable watermark tracking.

```
GET /files?updatedAfter=2025-01-01T00:00:00.000Z&sortBy=updatedAt&sortOrder=ASC
```

### 

[​

](#pagination-parameters)

Pagination Parameters

The API uses **offset-based pagination** with `page` and `limit` parameters.

Parameter

Default

Maximum

`page`

1

—

`limit`

100

100

**Example request:**

```
GET /files?updatedAfter=2025-01-01T00:00:00.000Z&sortBy=updatedAt&sortOrder=ASC&page=1&limit=100
```

### 

[​

](#pagination-stability-warning)

Pagination Stability Warning

**Offset-based pagination may produce inconsistent results** if records are updated during pagination. You may miss records or see duplicates.

**Mitigation strategies:**

1.  Use smaller time windows for `updatedAfter`
2.  Always deduplicate by `fileId + updatedAt`
3.  Re-sync periodically with a larger time window to catch missed records

* * *

## 

[​

](#what-triggers-updatedat)

What Triggers `updatedAt`?

The `updatedAt` timestamp changes when any of these events occur:

Event

Updates `updatedAt`

File metadata changes

✅ Yes

Schema/form field updates

✅ Yes

Document status changes

✅ Yes

Form filling/reprocessing

✅ Yes

Tag modifications

✅ Yes

Contact/vendor updates

✅ Yes

Workflow step changes

✅ Yes

Classification changes

✅ Yes

OCR reprocessing

✅ Yes

* * *

## 

[​

](#delete-handling)

Delete Handling

**Important**: Deleted files are **excluded** from the `GET /files` response and do **not** appear in `updatedAfter` queries. This endpoint provides **insert/update only** — not full CDC.

### 

[​

](#tracking-deletions)

Tracking Deletions

If you need to detect deleted files:

1

Option A: Periodic Full Sync

Do a full sync periodically and compare with your existing data to detect missing records.

2

Option B: Deletion Events Endpoint

Contact the fileAI team about a dedicated deletion events endpoint (roadmap item).

* * *

## 

[​

](#best-practices)

Best Practices

1\. Store Your Watermark

After each successful sync, store the **maximum `updatedAt`** value from the batch:

```
# Pseudocode
last_sync_watermark = max(record['updatedAt'] for record in batch)
save_watermark(last_sync_watermark)
```
2\. Use Recommended Query Parameters

Always include sorting parameters for reliable CDC:

```
GET /files?updatedAfter={watermark}&sortBy=updatedAt&sortOrder=ASC&limit=100
```
3\. Handle Pagination Completely

Paginate through **all pages** before updating your watermark:

```
# Pseudocode
page = 1
all_records = []

while True:
    response = get_files(updatedAfter=watermark, page=page, limit=100)
    all_records.extend(response['files'])

    if len(response['files']) < 100:
        break
    page += 1

# Only update watermark after all pages are processed
if all_records:
    new_watermark = max(r['updatedAt'] for r in all_records)
    save_watermark(new_watermark)
```

4\. Deduplicate Before Loading

Always deduplicate records by `fileId + updatedAt` before inserting into your data warehouse.

5\. Schedule Periodic Full Syncs

Run full syncs (e.g., weekly) to:

*   Catch any records missed due to pagination issues
*   Detect deleted records by comparing with existing data

* * *

## 

[​

](#snowflake-integration-example)

Snowflake Integration Example

*   Initial Load
    
*   Incremental Load
    

```
-- First sync: load all files
COPY INTO raw_files
FROM (
  SELECT $1:fileId, $1:fileName, $1:updatedAt, ...
  FROM @fileai_stage/files.json
);
```

```
-- Create staging table for incremental data
CREATE TEMPORARY TABLE staging_files AS
SELECT * FROM raw_files WHERE 1=0;

-- Load incremental data
COPY INTO staging_files
FROM (
  SELECT $1:fileId, $1:fileName, $1:updatedAt, ...
  FROM @fileai_stage/incremental_files.json
);

-- Merge with deduplication
MERGE INTO raw_files t
USING (
  SELECT * FROM staging_files
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY file_id
    ORDER BY updated_at DESC
  ) = 1
) s
ON t.file_id = s.file_id
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET
    file_name = s.file_name,
    updated_at = s.updated_at,
    ...
WHEN NOT MATCHED THEN
  INSERT (file_id, file_name, updated_at, ...)
  VALUES (s.file_id, s.file_name, s.updated_at, ...);
```

* * *

## 

[​

](#response-schema)

Response Schema

```
{
  "files": [
    {
      "fileId": "53d6a0b1-2a8d-4ed9-9e6a-ceaef7ca3908",
      "fileName": "invoice.pdf",
      "fileType": "application/pdf",
      "fileSize": 261928,
      "fileStoragePath": "path/to/file.pdf",
      "fileHash": "be3ef5fbb21e31c1281300b23b1c918a8ba54427c799aea21865f68d5efd01b7",
      "uploadId": "f2538513-f0b9-4aa8-9c57-bc0a85c77de6",
      "status": "processed",
      "currency": "USD",
      "summary": "Invoice from Example Corp",
      "duplicateToFileId": null,
      "referenceId": "1e70a5e860",
      "isDuplicate": false,
      "isEmbedded": false,
      "schemaId": "6835aca6030a79ffaabca742",
      "fileClass": "Invoice",
      "fileContactId": "6835aca2281d9ed1bab90b11",
      "fileContactName": "Example Company",
      "createdAt": "2025-05-27T12:14:24.258Z",
      "updatedAt": "2025-05-27T14:30:00.123Z"
    }
  ],
  "count": 1,
  "currentPage": 1
}
```

* * *

## 

[​

](#quick-reference)

Quick Reference

Feature

Behavior

Filter semantics

`>=` (inclusive)

Timestamp format

ISO 8601, UTC, milliseconds

Pagination

Offset-based (`page`, `limit`)

Default sort

`createdAt ASC`

Recommended sort for CDC

`updatedAt ASC`

Delete visibility

❌ Not visible (insert/update only)

Deduplication required

✅ Yes, by `fileId + updatedAt`

* * *

## 

[​

](#need-help)

Need Help?

## Contact Support

Reach out to the fileAI engineering team for questions about CDC implementation or to request new features.

[

Processing Callback

](/docs-api/api-processing-callback)[

Pagination

](/docs-api/api-pagination)

[linkedin](https://www.linkedin.com/company/file-ai)[x](https://x.com/fileAI_)[youtube](https://www.youtube.com/@file_AI)

[Powered byThis documentation is built and hosted on Mintlify, a developer documentation platform](https://www.mintlify.com?utm_campaign=poweredBy&utm_medium=referral&utm_source=undefined)