Combine Multiple Sources with a Canonical

Use this quickstart when two systems send records about the same things in different shapes, and you want them in one entity Type. To build this pipeline on the Data Fusion canvas, the visual pipeline editor in C3 AI Studio, see Combine Multiple Sources with a Canonical via UI.

One transform cannot read two different schemas, so the pipeline adds an intermediate step. Each source gets its own transform into a shared canonical Type, an intermediate Type that holds the common shape of both sources. A single second transform then maps the canonical Type to the target. The C3 Agentic AI Platform chains the two hops, the steps from one Type to the next. One run call per source therefore executes the whole path.

You model the data, create the source collections, stage sample files, inspect the transform chain, run the pipeline, and verify the stored entities. An optional section adds more transforms after the canonical, and a later section shows how to make the pipeline part of your package. A troubleshooting table covers the errors you are most likely to see, and a closing section explains how the pattern applies to other sources. The worked example ingests contact records from two CSV files on the application's own file system, so it needs no external credentials.

You work in two places. You define Types and transforms in your IDE as package source, which the platform syncs to the application. You run runtime commands in the C3 AI Console of your application.

The pipeline uses these Types:

Two Source Types: the Source Types, which describe the two incoming row formats.
A canonical Type: the shared intermediate model that both sources map into.
FileSourceCollection: one source collection per source, each binding an inbox folder to a Source Type. The inbox is the folder the source collection reads files from.
Three Transform definitions: one per source into the canonical Type, and one from the canonical Type to the target.
A target entity Type: the Type that stores the data in the application.

Prerequisites

Before you start, make sure you have the following:

A C3 Agentic AI Platform application in dev mode, with access to the C3 AI Console. The platform accepts new Types, transforms, and source collections only while the application is in dev mode.
A role that allows data integration from code. To choose one, see Data Integration Roles and Permissions.
The package source for your application, open in your IDE, such as VS Code with the C3 AI extension.

Step 1: Model the data

Define what each source row looks like, the shared canonical shape, what the application stores, and how each maps to the next. In your IDE, add the Types and transforms to the package source, then sync the package to the application.

Define one Source Type per incoming schema. In the example, a US system exports split names and an EU system exports full names. The two Types therefore differ in both field names and shape. Save the two Source Types as src/ContactSourceUS.c3typ and src/ContactSourceEU.c3typ:

Type

type ContactSourceUS mixes Source {
  firstName: string
  lastName: string
  email: string
  phone: string
}

Type

type ContactSourceEU mixes Source {
  fullName: string
  emailAddress: string
  telephone: string
}

Define the canonical Type in src/CanonicalContact.c3typ. Mixing Canonical makes the chain work. A canonical Type is itself a Source, so the platform passes the rows that land in it to the next transform. The canonical Type stores no records; only the target entity Type persists data.

Type

type CanonicalContact mixes Canonical {
  contactId: string
  fullName: string
  email: string
  phone: string
  region: string
}

Define the target entity Type in src/Contact.c3typ:

Type

entity type Contact {
  fullName: string
  email: string
  phone: string
  region: string
}

Define the three transforms. Each transform's projection lists the target fields and the expression that fills each one from the source row. The platform finds the transforms for a Type by matching each transform's source field to that Type. The chain forms where one transform's target Type is the next transform's source Type; the transform names are labels.

The first two transforms reshape each source into the canonical Type. The US transform builds fullName with an expression, and the EU transform passes it through. Save the two transforms as metadata/Transform/ContactSourceUS-CanonicalContact.js and metadata/Transform/ContactSourceEU-CanonicalContact.js:

JavaScript

data = {
  name: 'ContactSourceUS-CanonicalContact',
  source: 'ContactSourceUS',
  target: 'CanonicalContact',
  projection: {
    contactId: email,
    fullName: concat(firstName, ' ', lastName),
    email: email,
    phone: phone,
    region: 'US',
  },
};

JavaScript

data = {
  name: 'ContactSourceEU-CanonicalContact',
  source: 'ContactSourceEU',
  target: 'CanonicalContact',
  projection: {
    contactId: emailAddress,
    fullName: fullName,
    email: emailAddress,
    phone: telephone,
    region: 'EU',
  },
};

The third transform maps the canonical Type to the entity. You write the third transform once, however many sources feed the canonical Type. To add a source later, add one Source Type and one transform into the canonical Type; nothing downstream changes. Save the third transform as metadata/Transform/CanonicalContact-Contact.js:

JavaScript

data = {
  name: 'CanonicalContact-Contact',
  source: 'CanonicalContact',
  target: 'Contact',
  projection: {
    id: contactId,
    fullName: fullName,
    email: email,
    phone: phone,
    region: region,
  },
};

The id projection is the merge key across every source. The platform upserts rows that produce the same id, from one source or from two, onto one entity. The later row's non-empty values overwrite the earlier row's. The overwrite is the intended result when both sources describe the same real-world record. When two different records only happen to share an id, you lose data without an error. Build the id from a value that identifies the record the same way in every source. When the sources must never merge, prefix the id with the source name.

Save the files and wait for the platform to finish syncing the package source before you continue.

Step 2: Create the source collections

In the Console, create one source collection per source. Both source collections attach to the platform's built-in file source system, which is named Canonical; the name is unrelated to the canonical Type. The platform creates this source system in every application, and it reads the application's own file system, so it needs no credentials. Each source collection gets an inbox folder derived from its name:

JavaScript

var fss = FileSourceSystem.forName('Canonical');

FileSourceCollection.make({ name: 'contactsUSColl', source: ContactSourceUS, sourceSystem: fss }).upsert();
FileSourceCollection.make({ name: 'contactsEUColl', source: ContactSourceEU, sourceSystem: fss }).upsert();

var usInbox = FileSourceCollection.forName('contactsUSColl').inboxUrl();
var euInbox = FileSourceCollection.forName('contactsEUColl').inboxUrl();

Each inboxUrl value ends with a slash. Steps 3 and 5 reuse the usInbox and euInbox variables, so run them in the same Console session, or run the two inboxUrl lines again first.

The Console route suits iteration in a dev application. To deploy the pipeline with your package instead, follow "Make the pipeline part of your package".

Step 3: Stage the source files

In the Console, write one CSV file into each inbox. The column headers match each Source Type's field names exactly. Both files contain bob@example.com on purpose: Step 6 shows how the shared merge key combines the two rows.

JavaScript

var usCsv = 'firstName,lastName,email,phone\n' +
  'Alice,Anderson,alice@example.com,555-0101\n' +
  'Bob,Brown,bob@example.com,555-0102\n' +
  'Carol,Clark,carol@example.com,555-0103\n';

var euCsv = 'fullName,emailAddress,telephone\n' +
  'Derek Dupont,derek@example.eu,+33-555-0201\n' +
  'Eva Engel,eva@example.eu,+49-555-0202\n' +
  'Bob Brown,bob@example.com,+44-555-0203\n';

FileSystem.makeFile(usInbox + 'contacts_us.csv').writeString(usCsv);
FileSystem.makeFile(euInbox + 'contacts_eu.csv').writeString(euCsv);

In a real pipeline, the source systems deliver the files. This staging step stands in for an S3 upload, an SFTP delivery, or an export job.

Step 4: Inspect the transform chain

Before you run anything, confirm how the platform routes each source. allReachableTransforms follows the chain from a Source Type and returns every transform the platform runs for it:

JavaScript

ContactSourceUS.allReachableTransforms();

The result lists two transforms, ContactSourceUS-CanonicalContact and CanonicalContact-Contact, which proves that the chain resolves before any data moves. If only the first transform appears, the canonical Type or the third transform is missing from the application. Sync the package again, then repeat this step before you run the pipeline.

Step 5: Run the pipeline

In the Console, sync and process each inbox. Syncing registers each file in the inbox; the process: true setting on the DataIntegSpec also runs the rows through the transforms. One call per source collection is enough. The platform runs each row through every transform in the chain, including the hop from the canonical Type to the target.

JavaScript

SourceFile.syncAll(usInbox, DataIntegSpec.make({ process: true }));
SourceFile.syncAll(euInbox, DataIntegSpec.make({ process: true }));

Both calls return immediately, each with a reference to the batch job that does the work, and process in the background. Processing of a file has finished when its SourceStatus record shows completed, failed, rejected, or stopped. Step 6 shows how to fetch these records.

Step 6: Verify the data

Check the target entity Type and the status records. The canonical Type stores no records, so the results appear only in Contact and in SourceStatus.

JavaScript

c3Grid(Contact.fetch({ order: 'ascending(id)' }));

SourceStatus.fetch({
  filter: "sourceCollectionName=='contactsUSColl' || sourceCollectionName=='contactsEUColl'",
  order: 'descending(meta.created)'
});

c3Grid shows the result as a table in the Console. The six source rows produce five Contact entities:

Alice and Carol carry region US and the names the US transform built from first and last name.
Derek and Eva carry region EU.
Bob appears in both files and becomes one entity. Bob's entity holds the values from the file that the platform processed last, so its phone is either 555-0102 or +44-555-0203. The merged Bob entity is the overwrite that the warning in Step 1 describes.

The status records for each file show the completed state.

Add more transforms after the canonical

A canonical Type can feed more than one transform, and a transform can target a second canonical Type instead of an entity. Use more transforms to load a second entity from the same rows, or to refine the model in a second canonical before the entity. The run call stays the same: one syncAll per source collection runs every hop.

Define the new Types, each in its own file. This example adds an audit entity fed by CanonicalContact, and a second canonical with its own entity:

Type

entity type ContactAudit {
  contactId: string
  region:    string
  auditNote: string
}

Type

type CanonicalPerson mixes Canonical {
  personKey:   string
  displayName: string
  region:      string
}

Type

entity type Person {
  displayName: string
  region:      string
}

Add one transform file per hop. Each new transform from CanonicalContact names it as its source:

JavaScript

data = {
  name: 'CanonicalContact-ContactAudit',
  source: 'CanonicalContact',
  target: 'ContactAudit',
  projection: {
    id: concat('audit_', contactId),
    contactId: contactId,
    region: region,
    auditNote: concat('loaded from ', region),
  },
};

JavaScript

data = {
  name: 'CanonicalContact-CanonicalPerson',
  source: 'CanonicalContact',
  target: 'CanonicalPerson',
  projection: {
    personKey: contactId,
    displayName: upperCase(fullName),
    region: region,
  },
};

JavaScript

data = {
  name: 'CanonicalPerson-Person',
  source: 'CanonicalPerson',
  target: 'Person',
  projection: {
    id: personKey,
    displayName: displayName,
    region: region,
  },
};

After the package syncs, list every hop the platform runs for a source:
JavaScript
```
ContactSourceUS.allReachableTransforms().map(t => t.name);
```
The result lists all five hops. The list follows each branch to its end before it starts the next, so its order does not group the hops by layer:
Text
```
ContactSourceUS-CanonicalContact
CanonicalContact-CanonicalPerson
CanonicalPerson-Person
CanonicalContact-Contact
CanonicalContact-ContactAudit
```
Run the pipeline as in Step 5. Each transform from CanonicalContact processes every row one time, and CanonicalPerson-Person stores the rows that passed through the second canonical.

To run only some hops in one run, name them in transformOverrides on the DataIntegSpec. This run loads Contact and skips ContactAudit and the second canonical:

JavaScript

SourceFile.syncAll(usInbox, DataIntegSpec.make({
  process: true,
  transformOverrides: ['ContactSourceUS-CanonicalContact', 'CanonicalContact-Contact']
}));

The run writes transform status records only for the named hops.

To preview a hop without storing anything, call transformSource on the hop's source Type. transformSource runs only the direct hops from that Type, not the whole chain, so preview a chain one Type at a time:

JavaScript

ContactSourceUS.transformSource([ContactSourceUS.make({
  firstName: 'Ann', lastName: 'Lee', email: 'ann@example.com', phone: '555-0104'
})]);
CanonicalContact.transformSource([CanonicalContact.make({
  contactId: 'ann@example.com', fullName: 'Ann Lee', email: 'ann@example.com', phone: '555-0104', region: 'US'
})]);

The first call returns one CanonicalContact object. The second returns one object for each target of CanonicalContact: CanonicalPerson, Contact, and ContactAudit.

Make the pipeline part of your package

Declare the source collections as package metadata when the pipeline must deploy with your package. Package metadata deploys with the package and survives re-provisioning: the platform provisions it again on every deploy.

Create metadata/FileSourceCollection/contactsUSColl.json:

JSON

{
  "name": "contactsUSColl",
  "source": "ContactSourceUS",
  "sourceSystem": { "type": "FileSourceSystem", "name": "Canonical" }
}

Create metadata/FileSourceCollection/contactsEUColl.json:

JSON

{
  "name": "contactsEUColl",
  "source": "ContactSourceEU",
  "sourceSystem": { "type": "FileSourceSystem", "name": "Canonical" }
}

Sync the package to the application.

The folder name under metadata/ is the Type name, FileSourceCollection.

Fix common problems

Find the symptom you see, then apply the fix in the same row.

Symptom	Cause	Fix
Sync reports `Invalid value 'ContactSourceUS' for !Type` for a transform file	The platform read the transform before the Source Type it names was deployed.	After the Types are deployed, save and sync the transform files again.
Rows stop at the canonical Type: the run completes but the entity is empty	The third transform's `source` does not name the canonical Type.	Set the transform's `source` to your canonical Type (`CanonicalContact` in the example), then confirm that `allReachableTransforms` lists both transforms.
Rows stop at the canonical Type, and the transform `source` is correct	The canonical Type does not mix `Canonical`.	Add `mixes Canonical` to the canonical Type declaration, then confirm that `allReachableTransforms` lists both transforms.
Fetching the canonical Type fails with `Cannot find function fetch`	A canonical Type passes rows through and stores nothing.	Verify at the target entity Type and the `SourceStatus` records instead.
One entity where two were expected	Both sources produced the same `id`, and the rows merged.	Build the `id` from a value that is unique across sources, or prefix it with the source name.
Two entities where one was expected	The sources compute different `id` values for the same record, so nothing merged.	Align the `id` expression across the per-source transforms.
A new source's rows land nowhere	The new Source Type has no transform into the canonical Type.	Add one transform from the new Source Type into the canonical Type; the canonical-to-target transform needs no change.

Use the pattern with other sources

The canonical pattern works the same for every kind of source. The same canonical Type works downstream of an Amazon S3 file source, a Snowflake table, or a Kafka topic. The chain begins at whatever Source Type the source collection binds. Virtualization is the exception: a virtualized table is queried in place and has no transforms, so there is nothing to chain. For virtualization, see Query a Snowflake Table Programmatically with Virtualization.

Copy link to this sectionPrerequisites

Copy link to this sectionStep 1: Model the data

Copy link to this sectionStep 2: Create the source collections

Copy link to this sectionStep 3: Stage the source files

Copy link to this sectionStep 4: Inspect the transform chain

Copy link to this sectionStep 5: Run the pipeline

Copy link to this sectionStep 6: Verify the data

Copy link to this sectionAdd more transforms after the canonical

Copy link to this sectionMake the pipeline part of your package

Copy link to this sectionFix common problems

Copy link to this sectionUse the pattern with other sources

Copy link to this sectionSee also

Prerequisites

Step 1: Model the data

Step 2: Create the source collections

Step 3: Stage the source files

Step 4: Inspect the transform chain

Step 5: Run the pipeline

Step 6: Verify the data

Add more transforms after the canonical

Make the pipeline part of your package

Fix common problems

Use the pattern with other sources

See also