Load Data Incrementally

Incremental loading makes each pipeline run in the C3 Agentic AI Platform read only the data that changed since the previous run. Runs stay fast, and the load on your source system stays low as the data grows. A source system holds the connection to an external store, and a source collection names the files, table, or topic that a pipeline reads from it.

Each source family (file, SQL, API, or streaming) tracks change its own way. File and streaming collections load incrementally without setup. SQL and API collections need change data capture (CDC): a stored checkpoint that records the last position loaded. The next run then reads only what comes after that position. You can set up CDC for a SQL collection, reset its checkpoint to reload its data, and reload a small table in full on every run. You can also speed up file listings with a native change stream and replace the default read with your own logic.

In Data Fusion, you turn on CDC for a SQL source collection on the canvas. For those steps, see Enable Change Data Capture for a SQL Source via UI.

Copy link to this section

Prerequisites

Make sure you have:

• A working SQL pipeline: a source system, a source collection, and a transform to a target entity Type. The pipeline also needs a source Type that describes one row of the table. Ingest a Snowflake Table Programmatically builds such a pipeline.
• Access to the C3 AI Console of your application. You run every snippet on this page in the Console.

The examples use a SQL source system named ordersDb, a source Type named OrderSqlSource, and a collection named ordersDbColl. The file source example uses a Google Drive collection named driveContactsColl. Replace these names with your own.

Copy link to this section

How each source family tracks change

Each source family tracks change differently, and only SQL and API collections need configuration:

The checkpoint is a SourceCollection.Cdc.Checkpoint record, one per collection. Every source collection has the cdcOrder, fullSync, and cdcSourcesOverride fields that this page describes, but only SQL collections act on fullSync.

Copy link to this section

Configure change data capture for a SQL source

To set up CDC, name the field to track, run the collection, and confirm the checkpoint:

1. Name the field that defines "newer than last time" when you create the collection. The field's values must only ever increase (the field must be monotonic), such as an auto-incrementing key, a modification timestamp, or a load batch number:

   JavaScript

   Copy code to clipboard

   SqlSourceCollection.make({
     name: 'ordersDbColl',
     source: OrderSqlSource,
     sourceSystem: SqlSourceSystem.forName('ordersDb'),
     cdcOrder: 'id'
   }).upsert();

   cdcOrder names the field on the source Type, not the database column; the source Type maps each field to its column. cdcOrder takes a sort order. A field name alone, such as 'id', means ascending order, which is the default: each run reads the rows whose value is greater than the checkpoint.

2. Run the collection. The first process() call reads every row and records the highest cdcOrder value it loaded:

   JavaScript

   Copy code to clipboard

   SqlSourceCollection.forName('ordersDbColl').process(DataIntegSpec.make({}));

3. After the run completes, read the checkpoint:

   JavaScript

   Copy code to clipboard

   SqlSourceCollection.forName('ordersDbColl').cdcCheckpoint().value;
   // { "id": "2" }

   The value is a map keyed by the cdcOrder field name. In this example, the table holds rows with id values 1 and 2, so the checkpoint stores 2. Before the first run completes, cdcCheckpoint() returns null.

4. Run process() again on an unchanged table to confirm that CDC works. The run loads no rows, creates no new SourceStatus records (SourceStatus records the outcome of each run), and leaves the checkpoint at the same value.

From then on, each run reads only rows past the checkpoint and advances it. An empty run means the source did not change. If you expected new rows, confirm that they have a higher cdcOrder value than the checkpoint.

If you change cdcOrder on a collection that does not stage its data, the platform keeps the old checkpoint. Reset the checkpoint after you change cdcOrder, as described in Reset the checkpoint to reload.

Copy link to this section

Reset the checkpoint to reload

Reset the checkpoint when you need to re-read data it already passed. For example, reset the checkpoint after you fix a transform bug, when you backfill a target you truncated, or when a chunk fails after the checkpoint advanced.

You cannot rewind the checkpoint to a value you choose. The supported reset is Reset Run State in Data Fusion. Reset Run State clears the checkpoint, so the next run reads every row in the table.

1. In Data Fusion, select Execute pipeline on the SQL collection node.

2. In the Execute Pipeline dialog, turn on Reset Run State, then select Execute.

The run re-reads every row in the table, and the checkpoint advances again. By default, the target merges each reloaded row by its ID: a matching row receives the new non-empty field values. Reloaded rows update their existing copies only when the transform sets a stable ID on each target row. A transform that generates a new ID on every run creates new rows instead.

Copy link to this section

Reload everything on every run with fullSync

Set fullSync on a small table whose rows change in place, such as a reference or dimension table. Checkpoint tracking misses in-place updates: an updated row keeps its old cdcOrder value, so later runs skip it. With fullSync on, every run re-reads the entire table, and the collection ignores cdcOrder and the checkpoint:

SqlSourceCollection.forName('ordersDbColl')
  .withFullSync(true)
  .upsert();

Keep cdcOrder set when you turn on fullSync, as in this example, unless the collection stages its data into files first.

Only SQL collections act on fullSync, and a limit applies. The platform refuses to fully sync a table larger than maxFullSyncRowCount, a field on the collection with a default of 100,000 rows. The platform counts the source rows each time the collection processes. A collection whose table exceeds the limit therefore saves without an error, and then every run fails with an error like this one:

SqlSourceCollection ordersDbColl cannot enable fullSync: source type OrderSqlSource
has 250000 rows, exceeding the configured maxFullSyncRowCount of 100000.

To fully sync a larger table, raise maxFullSyncRowCount on the collection. A full re-read of a large table on every scheduled run costs time and load on the source, so the limit blocks full re-reads of large tables by default.

Copy link to this section

Use native change streams for file sources

Set cdcOrder to $native on a file collection so that each sync lists only the files that changed. Without $native, file collections still skip files they already processed, but every sync lists the full inbox. On a large drive, the full listing takes most of the sync time.

The $native value works with file systems that expose a native change stream: Google Drive (changes.list) and OneDrive or SharePoint (the Microsoft Graph delta API):

FileSourceCollection.forName('driveContactsColl')
  .withCdcOrder('$native')
  .upsert();

With $native set, each sync stores the provider's delta token, a marker of the last change the provider reported, in the collection's checkpoint. For these collections, the checkpoint holds that token instead of a field value. The next sync asks the provider for only the files that changed since that token.

You can set $native on any file collection. File systems without a change stream, including Amazon S3, Google Cloud Storage, and Azure Blob Storage, ignore it and return a full listing. For those file systems, the platform creates no checkpoint. The pipeline behaves as before, without the faster listing. A file written right before a sync can appear one sync later; it loads on the next run.

If the provider rejects the stored token as invalid, the sync does not fail. The sync falls back to a full listing, the same listing a collection without $native makes.

The token advances only after every batch of the sync succeeds. If a batch fails, the token stays at its old value, and the next sync replays the same changes. A sync that finds no changes advances the token at once. To confirm that the stream is in use, read the token after each sync and check that it changes:

FileSourceCollection.forName('driveContactsColl').cdcCheckpoint().value.deltaToken;

Change streams are eventually consistent. A sync that runs right after a write can find no changes, and the file then loads on a later sync. For Google Drive, schedule syncs at least a minute or two apart instead of back to back.

The change stream also reports files that were deleted at the source. The platform does not remove the records that earlier runs loaded from those files. To act on deletions, run your own cleanup logic after the pipeline. For the pattern, see Run BatchJobs Before and After a Pipeline.

Copy link to this section

Write custom checkpoint logic with cdcSourcesOverride

Write your own read logic when the source has no field whose values only increase. For example, the source is an API that pages by cursor, or a table keyed on business identifiers. The cdcSourcesOverride field on the collection holds a Lambda that receives the current checkpoint and returns a stream of source records to load. When the field is set, the collection uses it instead of the default read on every process() run, manual or scheduled.

Your lambda decides what to read from the checkpoint. After the run, the platform advances the checkpoint to the highest cdcOrder value among the records your lambda returned. Set cdcOrder on the collection and populate that field on every record. Records with no value in the field do not move the checkpoint, and the checkpoint never moves backward.

An ApiSourceCollection uses cdcSourcesOverride for CDC: its lambda field reads the data, and the override reads only the changed data. An API collection that sets the override stages the records it reads, so it also needs a stage file source collection.

Copy link to this section

Fix common problems

Each row starts from what you see:

Copy link to this section

See also

• Integrate Data Programmatically
• Data Integration End-to-End Workflow
• Ingest a Snowflake Table Programmatically
• Select a Pipeline Execution Type
• Run BatchJobs Before and After a Pipeline