Implement Write-Back for a Custom Connector
A custom connector can write data from the C3 Agentic AI Platform to an external system. It can also read from that system.
The platform supports two write paths:
- Live writes: An application writes a connector-backed entity type with
create,update,upsert,merge, orremove. - Data Integration write-back: A Data Integration pipeline writes transformed records to an external system through a
VirtualTargetCollection.
A connector exposes writes in one of two styles. Both styles are optional. A writable connector implements at least one. A read-only connector implements neither.
- Batch style: The connector mixes
VirtualSourceSystem.BatchWriter. The platform cuts the records into batches, then runs a commit protocol.writeBatchorremoveBatchruns one time for each batch. - Stream style: The connector implements
writeStreamorremoveStream. The platform passes the whole stream in one call and pulls it lazily. That call is the only task of the same commit protocol.
A whole-array write prefers the batch style. create, update, upsert, merge, and remove, including their batch forms, go through VirtualPersistable#saveBatch or VirtualPersistable#deleteBatch. A write uses writeStream when writeBatch is missing. A delete uses removeStream when removeBatch is missing.
A VirtualTargetCollection prefers the stream style. If the connector does not implement writeStream, the platform uses the batch style.
Every writable connector also implements toWriteSpec. That hook builds the connector's write request before either style runs.
Implement the hooks in the sections below. writeObjs, deleteObjs, and removeObjs on VirtualPersistable are caller-facing pass-throughs, not connector hooks. VirtualTargetCollection#writeObjs and VirtualTargetCollection#writeArrow are the Data Integration entry points. They route into the style the connector implements.
Build the write request
Implement toWriteSpec on every connector that writes.
toWriteSpec turns the caller's UpsertSpec into the connector's own write-request type (WSPEC). merge and mergeBatch pass a MergeSpec, which is an UpsertSpec. Return a WSPEC whose table is the type's table (type.table()).
Carry only the parts of the write your external system can honor. Reject a filter or option you cannot apply. A write must not do more than the caller asked.
A VirtualPersistable write fails as unsupported when toWriteSpec is missing. A VirtualTargetCollection calls toWriteSpec when the target mixes VirtualPersistable and the connector implements the hook. Otherwise it builds an empty write request with VirtualSourceSystem.Write.Spec#emptyWriteSpec for the target table.
Example
function toWriteSpec(type, spec) {
return VirtualSourceSystem.Write.Spec.make({ table: type.table() });
}Bind a subtype of VirtualSourceSystem.Write.Spec as WSPEC when the request must carry more than the table. beginWrite can add state to that subtype for the rest of the write.
Implement the batch write style
Use the batch style when you want each write split into tasks under a commit protocol.
The platform runs the steps in this order. A step you do not implement is skipped.
| Step | Hook | When it runs |
|---|---|---|
| 1 | writeDistribution | Before any batch. Sets how many records each batch holds. Without it, the write is one batch. |
| 2 | beginWrite | A single call, before the first batch. Use it to open a transaction or staging area. Return the request that later steps receive. |
| 3 | writeBatch or removeBatch | A single call for each batch. writeBatch creates or updates. removeBatch deletes. |
| 4 | commitWrite | A single call, after every batch succeeds. Use it to commit or publish. |
| 5 | abortWrite | A single call, when a batch throws, reports an error, or commitWrite fails. Use it to roll back. |
writeBatch and removeBatch take an array of the connector's element type R, not a stream. R is Obj or Arrow.
A batch that reports an error ends the write. Later batches are not sent. commitWrite does not run. abortWrite runs when you implement it. Batches that already ran stay written unless abortWrite undoes them. The caller retries the whole input, not only the failed records.
Use upsert semantics
Implement writeBatch and writeStream with upsert semantics.
The platform does not tell the connector whether the caller used create, update, upsert, or merge. The connector receives the records to write. merge and mergeBatch ignore mergeInclude. The connector writes the records it is given.
Your implementation should:
- Create the record when it does not exist.
- Update the record when it already exists.
If the external system needs true merge behavior, do that inside the write hook. Read the current record, apply the incoming changes, and write the result back.
Make writes idempotent
Each batch request carries a deterministic batchId of the form <batchId>/<task>:0. The task index starts at 0. The platform does not retry a task, so the attempt is always 0.
When the incoming write has no batchId, the platform assigns a new identifier. That call is a new write. A retry, such as a Data Integration chunk, stays idempotent only when the caller sends the same batchId on every attempt.
Treat the same batchId as the same write. Repeating it must not create duplicate records. If the connector sends notifications or starts external workflows, make those actions safe to repeat.
Report per-record errors
Return a VirtualSourceSystem.Write.Result. Put successful records and per-record failures on its objs field, which is an ObjList.
Inside one batch, record the error and continue with the remaining records in that batch when you can. Add an ObjError with targetId and message.
createBatch, updateBatch, upsertBatch, and mergeBatch return that ObjList to the caller. A single-record create, update, upsert, or merge has no ObjList to return. If the result contains an error for that record, the platform throws. A single-record remove returns false instead, because remove returns a boolean.
You can also set stats.failedObjCount when the external system returns a failure count and not a record ID. The platform treats that count as a failed write.
Example
function writeBatch(type, objs, spec) {
var written = [];
var errors = [];
objs.each(function(obj) {
try {
writeToExternalSystem(type.name(), obj);
written.push(obj);
} catch (e) {
errors.push(ObjError.make({
targetId: obj.id,
message: e.message
}));
}
});
return VirtualSourceSystem.Write.Result.make({
objs: ObjList.make({
objs: written,
errors: errors
})
});
}
function removeBatch(type, objs, spec) {
var removed = [];
var errors = [];
objs.each(function(obj) {
try {
deleteFromExternalSystem(type.name(), obj.id);
removed.push(obj);
} catch (e) {
errors.push(ObjError.make({
targetId: obj.id,
message: e.message
}));
}
});
return VirtualSourceSystem.Write.Result.make({
objs: ObjList.make({
objs: removed,
errors: errors
})
});
}removeBatch deletes a specific set of records that the platform has already resolved. The records should contain the identity fields the external system needs. Make the delete idempotent. A second delete of a record that is already gone should leave the external system unchanged.
To cap batch size, implement writeDistribution. maxRows of -1, the default, sends every record in one batch.
function writeDistribution(type, spec) {
return VirtualSourceSystem.Write.Distribution.make({ maxRows: 500 });
}Implement the stream write style
Use the stream style when the write should stay a stream and be pulled lazily.
Implement writeStream to create or update, and removeStream to delete. Each hook receives the whole stream and the write request from toWriteSpec. The same upsert, idempotency, and per-record error rules apply as for writeBatch and removeBatch.
The platform runs beginWrite, then one writeStream or removeStream call, then commitWrite. abortWrite runs when that call throws or reports an error. Read the entire stream before you return. If you return early, the platform aborts the write so unread records are not dropped.
writeStream and removeStream are the style a VirtualTargetCollection tries first.
Example
function writeStream(type, data, spec) {
return writeBatch(type, data.collect(), spec);
}Write Arrow batches
Bind the connector's element type R to Arrow when the external system accepts columnar data. writeStream and writeBatch then receive Arrow batches. Bind R to Obj to receive records.
The platform converts between Obj and Arrow. A connector that binds R to Obj still accepts an Arrow write. There is no connector method named writeArrow.
VirtualTargetCollection#writeArrow is the Data Integration method that starts an Arrow write. It converts the Arrow stream into R, then uses the stream style, or the batch style when writeStream is absent.
Example
type MyConnector mixes VirtualSourceSystem<
MyConnector.Config,
MyConnector.Credentials,
MyReadSpec,
MyReadBatch,
VirtualSourceSystem.Write.Spec,
Arrow> {
writeStream: ~ js-server
toWriteSpec: ~ js-server
}Delete records by filter
Implement deleteAll to remove every record of a type that matches a filter, without the platform listing those records first.
deleteAll backs VirtualPersistable#removeAll. It is separate from removeBatch and removeStream, which delete a set of records the platform already resolved. Implement deleteAll when the external system can delete by filter, for example with a SQL DELETE ... WHERE statement. Return the number of records removed.
The platform builds req from the caller's RemoveAllSpec the same way it builds a read request. req is the connector's read-request type, not a FetchSpec and not the write request. The filter on req scopes the delete. An empty filter removes every record of the type.
Callers must pass confirm: true. The platform rejects the call before it reaches deleteAll otherwise.
If you do not implement deleteAll, removeAll fails and names the missing capability.
A connector without a native delete-by-filter can still implement deleteAll. Resolve the matches with readStream, pass them to removeBatch or removeStream, and return the count removed.
Example
function deleteAll(type, req) {
var result = removeBatch(type, readStream(type, req).collect(),
VirtualSourceSystem.Write.Spec.make({ table: type.table() }));
var removed = result.objs != null && result.objs.objs != null ? result.objs.objs.length : 0;
return removed;
}Enable live writes
A connector-backed entity becomes writable the same way it becomes queryable. Mix VirtualPersistable into the entity type. Bind it to a registered connector with @connector(sourceSystem=...).
No other write-specific configuration is required.
Example
@connector(sourceSystem="MyConnector")
entity type WidgetWritable mixes VirtualPersistable {
id: ~
name: string
category: string
price: double
}Applications call standard persistence methods, such as upsert() and remove(). Every write, including a single record, calls toWriteSpec first. The platform then routes the call to the hooks below.
Route each operation
| Operation | Batch operation | Preferred hook | Fallback hook |
|---|---|---|---|
create, update, upsert, merge | createBatch, updateBatch, upsertBatch, mergeBatch | writeBatch | writeStream |
remove | removeBatch | removeBatch | removeStream |
removeAll | None | deleteAll | None |
Data Integration write-back uses the opposite preference. It calls writeStream first, then writeBatch. It does not call removeBatch, removeStream, or deleteAll.
Enable Data Integration write-back
A VirtualTargetCollection is the destination of a Data Integration pipeline. The pipeline sends transformed records to the connector instead of loading them into a C3-managed entity.
writeObjs and writeArrow on VirtualTargetCollection are that entry point. They resolve the connector from sourceSystem and write the target type. The platform uses writeStream when the connector implements it, and the batch protocol otherwise. Implement writeStream or writeBatch on the connector. The same names on VirtualPersistable (writeObjs, removeObjs, deleteObjs) are a separate caller-facing layer.
If the result reports a per-record error or a failedObjCount, the platform fails the Data Integration write. The job can retry. It does not advance past records that were not written.
Create a VirtualTargetCollection
Create a VirtualTargetCollection with a collection name, the target type, and the registered connector.
Example
var connector = MyConnector.forName("MyConnector");
VirtualTargetCollection.builder()
.name("MyWidgetTargetCollection")
.target(WidgetTarget)
.sourceSystem(connector)
.build()
.upsert();The target collection must reference a connector that implements writeStream or writeBatch, and toWriteSpec when the target mixes VirtualPersistable.
Write to the same or a different connector
A Data Integration pipeline can:
- Read from and write back to the same connector instance.
- Read from one connector instance and write to a different connector instance.
The connector on the VirtualTargetCollection is the write destination.
Finish asynchronous writes
If your connector buffers writes, or flushes them to the external system asynchronously, returning from writeBatch or writeStream does not mean the record is durable yet.
When a later step depends on those records, confirm that they have reached the external system before you continue.
Write-back behavior and limitations
- Per-record
ObjListerrors are returned from the batch surfaces (createBatch,updateBatch,upsertBatch,mergeBatch, andremoveBatch). A single-recordcreate,update,upsert, ormergethrows. A single-recordremovereturnsfalse. - A batch that reports an error stops the write. Later batches are not sent. Implement
abortWritewhen earlier batches must be undone. - Data Integration write-back does not call
removeBatch,removeStream, ordeleteAll. Delete operations are live writes only. - A Data Integration write fails when the connector reports any per-record error or a
failedObjCount. - A connector that implements neither
writeBatchnorwriteStreamfails a write as unsupported. A connector that implements neitherremoveBatchnorremoveStreamfails a delete the same way.removeAllfails whendeleteAllis missing. - Records passed to
removeBatchorremoveStreamneed the identity fields the external system uses. removeAllrequiresconfirm: true. The platform rejects the call beforedeleteAllruns otherwise.writeBatchandwriteStreamuse upsert semantics. The platform does not distinguishcreate,update,upsert, andmerge.