Configure CDC and Parallel Processing

Change Data Capture (CDC) enables a VirtualSourceCollection to retrieve only records that have changed since the previous Data Integration run.

The extraction strategy, configured on the collection, determines whether the platform manages the checkpoint automatically or the collection implements its own checkpointing logic.

This topic assumes that you have already configured a VirtualSourceCollection and chosen an extraction strategy, as described in Expose a Connector to Data Integration. It focuses on the checkpoint mechanics for each extraction strategy, parallel batch processing, retry behavior, and file staging.

Copy link to this section

Implement WATERMARK extraction

With the WATERMARK extraction strategy, the platform creates and persists the checkpoint.

Set cdcOrder on the VirtualSourceCollection to name the watermark field. On each Data Integration run, the platform builds an incremental predicate from cdcOrder and the previous checkpoint. It applies that predicate to the connector request before it calls batchesForWatermark. The collection reads that request through the connector's readStream or readTupleStream hook. The connector has no CDC-specific read hook.

For WATERMARK and CUSTOM, bind R to Obj and implement readStream, or implement readTupleStream. The checkpoint is accumulated per row. Only STATELESS extraction uses the Arrow read path.

cdcOrder can name more than one field. The platform treats a record as changed when it has advanced past the checkpoint in any configured field. It compares each field using that field's sort direction.

By default, batchesForWatermark returns one batch for the whole dataset. To partition a WATERMARK read, override batchesForWatermark on the VirtualSourceCollection subtype. The request already includes the watermark predicate. Combine that predicate with the partition filter on each batch.

A WATERMARK collection cannot run unless cdcOrder is set.

Copy link to this section

Implement CUSTOM extraction

Use the CUSTOM extraction strategy when the external system tracks changes with its own cursor or delta token.

With CUSTOM extraction, the VirtualSourceCollection subtype owns checkpoint generation, updates, and application. The connector binds R to Obj and implements readStream, or implements readTupleStream. Either hook reads the rows that match the filter on each batch.

Implement these members on the VirtualSourceCollection subtype:

These four members form a pipeline. generateCdcCheckpointFields seeds an accumulator from the previously persisted checkpoint. updateCheckpointFields folds each processed record into that accumulator while a batch is read. mergeCheckpointAccumulators combines those accumulators after every batch completes successfully. The platform persists the merged accumulator's snd map directly as the next checkpoint value.

The accumulator is a Pair<any, map<string, any>>. fst is the winning row so far, or null when no row has won. snd is the checkpoint field map derived from that row. The row lets mergeCheckpointAccumulators compare checkpoints that use more than one field, such as a composite key or an opaque continuation token.

Implement the accumulator so it initializes checkpoint state and handles a batch that reads no rows. It must also merge accumulators from parallel batches. mergeCheckpointAccumulators must be associative and commutative. The checkpoint is then the same regardless of the order in which batches complete.

Example

// VirtualSourceCollection subtype: builds the batch whose filter encodes the cursor.
// The batch keeps the connector's own request type.
function batchesForCustom(req, checkpoint) {
    var cursor = cursorFromCheckpoint(checkpoint);
    return [req.withField("filter", Filter.gt("updatedAt", cursor).toString(), false)];
}

// VirtualSourceCollection subtype: keeps the row with the max updatedAt seen so far.
function updateCheckpointFields(currentRow, previousAccumulator) {
    var winner = previousAccumulator != null ? previousAccumulator.fst : null;
    if (rowCursor(currentRow) > rowCursor(winner)) {
        winner = currentRow;
    }
    return rowAccumulator(winner);
}

// VirtualSourceCollection subtype: associative, commutative merge across parallel batches.
function mergeCheckpointAccumulators(a, b) {
    var ra = a != null ? a.fst : null;
    var rb = b != null ? b.fst : null;
    return rowCursor(ra) >= rowCursor(rb) ? rowAccumulator(ra) : rowAccumulator(rb);
}

// VirtualSourceCollection subtype: seeds the accumulator from an existing checkpoint.
function generateCdcCheckpointFields(checkpoint) {
    return Pair.of(null, C3.Map.ofStrToAny("cursor", cursorFromCheckpoint(checkpoint)));
}

function cursorFromCheckpoint(checkpoint) {
    if (checkpoint == null || checkpoint.value == null) {
        return 0;
    }
    var cursor = checkpoint.value.get("cursor");
    return cursor == null ? 0 : Number(cursor);
}

function rowCursor(row) {
    return row != null && row.updatedAt != null ? Number(row.updatedAt) : 0;
}

function rowAccumulator(row) {
    return Pair.of(row, C3.Map.ofStrToAny("cursor", rowCursor(row)));
}

// Connector: readStream reads only rows past the cursor encoded in the batch filter.
function readStream(type, batch) {
    var cursor = cursorFromBatch(batch);
    var objs = queryExternalSystem(type, cursor).map(function (record) {
        return type.make(record);
    });
    return C3.Stream.of(type.name(), objs);
}

Copy link to this section

Checkpoint behavior

The platform manages checkpoint persistence consistently across incremental Data Integration runs:

• The checkpoint is updated only after a Data Integration run completes successfully.
• If a run fails, the previously persisted checkpoint is retained.
• If no new records are retrieved, the existing checkpoint is retained.
• After every scheduled batch succeeds, the platform folds checkpoint state from every batch and persists the next checkpoint.
• For WATERMARK, the folded state is the per-field extrema.
• For CUSTOM, the folded state is the Pair accumulators.
• Delivery is at least once, so a record can arrive more than once. When a limit is set, a run re-reads the records tied at the checkpoint value. Make your target loads idempotent.

These guarantees apply regardless of the extraction strategy.

Copy link to this section

Configure parallel processing

A VirtualSourceCollection mixes VirtualSourceSystem.Batchable. The mixin lets the collection split a read into multiple VirtualSourceSystem.Read.Batch instances. The platform reads those batches concurrently.

By default, each extraction strategy reads as a single batch. To partition a read, override the batch method for that extraction strategy:

Each method returns one VirtualSourceSystem.Read.Batch per partition. Give each batch a filter that selects only that batch's rows. For WATERMARK and CUSTOM, include the incremental predicate in every batch filter. Each batch then reads only its own share of the delta. For WATERMARK, req already selects rows past the checkpoint. Keep that predicate when you add a partition filter.

Example

// Partitions a CUSTOM read into two batches by category. Each filter still carries the cursor.
function batchesForCustom(req, checkpoint) {
    var cursor = cursorFromCheckpoint(checkpoint);
    return ["tools", "gadgets"].map(function (category) {
        var filter = Filter.eq("category", category).and(Filter.gt("updatedAt", cursor));
        return req.withField("filter", filter.toString(), false);
    });
}

The platform runs the batches in a VirtualSourceCollection.ReadBatchJob. The job is a BatchJob that reads batches across the cluster. For WATERMARK and CUSTOM, each batch stores its checkpoint state in a VirtualSourceCollection.ReadBatchResult row. The row id joins jobId and batchId with an underscore. A retry of that batch overwrites the same row and keeps a single checkpoint value. STATELESS extraction stores no per-batch checkpoint row.

The platform allows one read per collection at a time. It holds a lease for that collection. A second run on the same collection is rejected while the lease is held. If the run that holds the lease fails, the next run claims the lease.

The platform sweeps staged files after every scheduled batch has completed. For WATERMARK and CUSTOM, it persists the merged checkpoint only after every batch has stored a result.

Set maxConcurrency and maxConcurrencyPerNode on readJobOptions to limit how many batches run at once. The platform resolves readJobOptions in the order listed under Configure retry behavior.

Without an override of the batch method, the platform reads the workload as a single batch.

Partition only work that each batch can process on its own. Keep sequential cursor pagination and continuation-token pagination in one batch. The platform collapses batches with the same serialized form into one batch. Make each partition's serialized form unique, usually with its own filter.

Copy link to this section

Configure retry behavior

The platform retries a failed batch during Data Integration processing.

Set numAutoRecoveryAttempts on readJobOptions. The platform clamps the value to the range 1 through 5. When the value is unset, the platform uses 1.

The platform takes the first readJobOptions value that is set, in this order:

1. The run's VirtualSourceCollection.DISpec#readJobOptions
2. The collection's VirtualSourceCollection.Config#readJobOptions
3. The connector's VirtualSourceSystem.Batchable#batchOptions

A set value replaces the options below it entirely. When none of these is set, the platform uses its default options.

If a retried batch succeeds, the Data Integration run continues. If every retry fails, the run fails and the checkpoint stays unchanged. The next run resumes from the previously persisted checkpoint.

Copy link to this section

Configure file staging

File staging can be combined with parallel processing. When staging is enabled, each batch writes its retrieved records to staged files before the Data Integration pipeline loads the data. When staging is disabled, retrieved records are streamed directly into the pipeline.

Configure file staging by setting stage and stageFileSourceCollectionName on the VirtualSourceCollection configuration. No additional connector implementation is required. For more information, see Expose a Connector to Data Integration.

Copy link to this section

Limitations and considerations

• During the first WATERMARK or CUSTOM execution, no checkpoint exists. The initial run therefore retrieves all records that match the configured collection scope.
• Partition batches only for independently processable workloads. Keep sequential cursor pagination and continuation-token pagination in a single batch. The platform collapses batches with the same serialized form into one batch. Make each partition's serialized form unique.
• When implementing CUSTOM extraction, ensure that mergeCheckpointAccumulators is associative and commutative so that checkpoint generation remains deterministic during parallel execution.
• A collection configured with the CUSTOM extraction strategy cannot run unless it implements batchesForCustom and all three checkpoint-accumulator hooks.
• For WATERMARK and CUSTOM extraction, bind R to Obj and implement readStream, or implement readTupleStream. The checkpoint is accumulated per row, so only STATELESS extraction uses the Arrow read path.

Copy link to this section

See also

• Expose a Connector to Data Integration: configure the extraction strategy and file staging when creating the VirtualSourceCollection.
• Custom Connector Framework: architecture and components of the custom connector framework.
• Implement a Custom Connector: implement the connector's readStream and other live-query hooks.