Ingest a Snowflake Table Programmatically

Use this quickstart when your data lives in a Snowflake table and your application needs its own copy of the rows. Copying the rows is ingestion. The C3 Agentic AI Platform reads each row once and runs it through a Transform. The platform stores the result in an entity Type that your application owns. Virtualization is the alternative, where the platform reads the table live on every query and stores nothing. For that route, see Query a Snowflake Table Programmatically with Virtualization.

You connect to Snowflake, inspect the table, model the data, create a source collection, run the pipeline, and verify the stored entities. A later section shows how to make the pipeline part of your package, and "Fix common problems" covers common errors. The worked example copies a sample table of person records, PUBLIC.DEMO_TABLE. The example names (the snowflakeDemo source system, the demoTableColl collection, and the PersonSqlSource and DemoPerson Types) stand in for your own; use your own table, source, and Type names.

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

A SQL ingest pipeline links these Types:

SqlSourceSystem and JdbcCredentials: the source system and its credentials, which together are the connection to the Snowflake database.
SqlSourceCollection: the source collection, which binds one table to the Source Type.
A Source Type: the Type that describes one row of the table.
The transform, which maps each source row to the target.
A target entity Type: the Type that stores the data in the application.

Data passes through these five Types in the order listed. The source system reaches the database, and the source collection reads a table as Source Type rows. The transform writes each row to the target entity Type.

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, source systems, 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.
Snowflake connection details: the account endpoint, database, warehouse, and a Snowflake user whose public key is registered in Snowflake.
The Snowflake user's private key in PEM format, and the key's passphrase if the key is encrypted.
A Snowflake table to copy. The example uses PUBLIC.DEMO_TABLE.
The package source for your application, open in your IDE, such as VS Code with the C3 AI extension.

Step 1: Connect to Snowflake

In the C3 AI Console, build the key-pair credentials, create the source system, and set the credentials on it. Every later step reads the table through this connection. If you completed the virtualization quickstart in the same application, the snowflakeDemo source system already exists and you can skip this step. To check, run SqlSourceSystem.forName('snowflakeDemo').

JavaScript

var key = PrivateKeyAuth.createFromPem('<private-key-pem>', '<passphrase>', 'PKCS8', 'RSA');

var creds = JdbcCredentials.make({
  serverEndpoint: '<account>.snowflakecomputing.com',
  port: 443,
  datastore: 'snowflake',
  database: '<database>',
  username: '<user>',
  privateKeyAuth: key,
  properties: { warehouse: '<warehouse>' }
});

SqlSourceSystem.make({ name: 'snowflakeDemo' }).upsert();
SqlSourceSystem.forName('snowflakeDemo').setCredentials(creds, false);

SqlSourceSystem.forName('snowflakeDemo').ping();

Replace the following placeholders with your values:

<private-key-pem>: The private key in PEM format, including the BEGIN and END lines. Use \n for line breaks.
<passphrase>: The passphrase for the private key, or null if the key is unencrypted.
<account>.snowflakecomputing.com: The Snowflake account endpoint.
<database>, <warehouse>, <user>: The Snowflake database, warehouse, and user.

Keep the other values as shown. 'PKCS8' and 'RSA' name the key format and algorithm of a standard Snowflake key pair, 443 is the Snowflake port, and 'snowflake' sets the datastore kind. The name snowflakeDemo is your choice; later steps use it again.

Each call does one job:

setCredentials stores the credentials for the source system in your application. The second argument, false, replaces any credentials stored earlier instead of merging with them.
ping tests the connection. The platform stores credentials without validating them, so run ping after every credential change. A working connection returns {reachable: true}. A failed connection returns {reachable: false} with an errorMsg that describes the error.

Step 2: Inspect the source data

In the Console, preview the table before you model it. The Source Type in Step 3 maps its fields to these column names. Note the row count too: processing copies every row of the table into the application.

JavaScript

SqlSourceSystem.forName('snowflakeDemo').previewTable('<table>', 3);

Replace <table> with the table name, for example PUBLIC.DEMO_TABLE. PUBLIC is the Snowflake schema. A name qualified with its schema works whatever the connection's default schema is. The second argument is the number of rows to return. The result lists the column names, then the first three rows.

The example table DEMO_TABLE has three columns: ID (a number), NAME (a string), and LAST_MODIFIED (a timestamp that changes whenever a row changes).

Step 3: Model the data

Define what a source row looks like, what the application stores, and how one maps to the other. In your IDE, add three files to the package source, then sync the package to the application.

Define the Source Type in src/PersonSqlSource.c3typ. The Source Type mixes in three Types:

External: the Type reads rows from a table outside the platform, named by schema name on the Type.
NoSystemCols: the table has none of the system columns that the platform adds to its own tables.
Source: the Type can feed a transform.

The schema name on each field names the source column it reads:

Type

entity type PersonSqlSource mixes External, NoSystemCols, Source schema name 'PUBLIC.DEMO_TABLE' {
  id: string schema name 'ID'
  personName: string schema name 'NAME'
  lastModified: datetime schema name 'LAST_MODIFIED'
}

Replace PUBLIC.DEMO_TABLE with your schema and table, and each column name with yours.

Define the target entity Type in src/DemoPerson.c3typ. The application stores and queries this Type, and the pipeline fills it:

Type

entity type DemoPerson {
  personName: string
}

Define the transform that maps source rows to the target. Save the transform as metadata/Transform/PersonSqlSource-DemoPerson.js:

JavaScript

data = {
  name: 'PersonSqlSource-DemoPerson',
  source: 'PersonSqlSource',
  target: 'DemoPerson',
  projection: {
    id: id,
    personName: personName,
  },
};

The projection lists each target field and the expression that fills it. Each expression reads fields of the Source Type, so id: id copies the source row's id into the entity ID. The platform upserts each row: a row whose ID already exists updates that entity instead of creating a duplicate, so processing a row again is safe.

Save the three files and wait for the platform to finish syncing the package source. After the platform deploys the Types, DemoPerson.fetchCount() runs in the Console and returns 0 before the first load.

Step 4: Create the source collection

In the Console, create the source collection. The source collection links the connection from Step 1 to the Source Type from Step 3, and names the field that orders the rows:

JavaScript

SqlSourceCollection.make({
  name: 'demoTableColl',
  source: PersonSqlSource,
  sourceSystem: SqlSourceSystem.forName('snowflakeDemo'),
  cdcOrder: 'lastModified'
}).upsert();

cdcOrder names the Source Type field that the platform uses for change data capture (CDC). After each run, the platform records the highest value of that field in a checkpoint, and the next run reads only rows past it. Without cdcOrder, process fails in Step 5. You can also set up CDC in Data Fusion. For those steps, see Enable Change Data Capture for a SQL Source via UI.

Use a field whose value only grows, such as a last-updated timestamp. You can name several fields, separated by commas, for example 'lastUpdated, id'. The platform then reads every row in which any one of those fields is past its checkpoint value. As a result, a row with an older timestamp but a higher key loads again.

Declare the cdcOrder field with a type that sorts the way its values grow, such as a number or a date and time. Do not use a numeric ID that the Source Type declares as string, such as id in this example. The platform compares string values as text, so "10" sorts before "9". The checkpoint then stops advancing, and every later run reads the same rows again.

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 5: Run the pipeline

In the Console, process the source collection. The platform reads the table, runs each row through the transform, and writes DemoPerson entities:

JavaScript

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

An empty DataIntegSpec runs with the settings in the collection's configuration. The call reads the table before it returns. When the platform processes the rows as one unit, the entities exist as soon as the call returns. When the platform splits the rows into chunks, the chunks finish in the background, and the status record in Step 6 shows completed when they are done.

Each run advances the CDC checkpoint. A later process call that finds no rows past the checkpoint copies nothing and writes no new status record. To see the checkpoint, run SqlSourceCollection.forName('demoTableColl').cdcCheckpoint(); it returns nothing before the first run. To reload every row, reset the checkpoint as described in the "Reset the checkpoint to reload" section of Load Data Incrementally.

Step 6: Verify the data

Check the status records to find any stage that failed, then fetch the entities to confirm that the data loaded.

JavaScript

SourceStatus.fetch({
  filter: "sourceCollectionName=='demoTableColl'",
  order: 'descending(meta.updated)'
});

Each SourceStatus record reports the state of a run and its failed count. In the newest record, a completed state with failed at 0 indicates a clean run. When a run fails, firstError holds the first error.

Read the persisted entities. c3Grid displays the result as a table in the Console:

JavaScript

c3Grid(DemoPerson.fetch());

The grid shows the copied rows as DemoPerson entities. Your application can now query DemoPerson like any other entity Type, independent of Snowflake.

Make the pipeline part of your package

Declare the source system and source collection as package metadata when the pipeline must deploy with your package. Objects you create in the Console are runtime updates, and deploying a new version of the package overrides them.

The folder under metadata/ must match the Type name, and each file carries the same fields you passed to make:

Create metadata/SqlSourceSystem/snowflakeDemo.json:
JSON
```
{ "name": "snowflakeDemo" }
```

Create metadata/SqlSourceCollection/demoTableColl.json:

JSON

{
  "name": "demoTableColl",
  "source": "PersonSqlSource",
  "sourceSystem": { "type": "SqlSourceSystem", "name": "snowflakeDemo" },
  "cdcOrder": "lastModified"
}

Sync the package to the application.

Keep credentials out of the package. Credentials are secrets: set them at runtime with setCredentials, as in Step 1.

Fix common problems

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

Symptom	Cause	Fix
`createFromPem` fails with `Invalid private key format: unable to read encrypted data` (wrong passphrase) or `Invalid private key format: Passphrase required for encrypted key` (no passphrase)	The key is encrypted and the passphrase is wrong or missing.	Pass the key's passphrase as the second argument, then rerun Step 1.
`ping` returns `{reachable: false}` and `errorMsg` mentions a missing password	The credentials carry no private key, so the connection has no secret to sign in with.	Build the key with `createFromPem`, pass it as `privateKeyAuth`, and rerun `setCredentials` and `ping`.
`ping` returns `{reachable: false}` with another error	Wrong endpoint, user, key, or passphrase; the platform does not validate credentials when you set them.	Correct the values, rerun `setCredentials` from Step 1, and run `ping` again.
`process` fails with an error that names `cdcOrder`	The source collection has no `cdcOrder`.	Set `cdcOrder` on the source collection, as in Step 4.
A repeat `process` copies nothing and writes no status record	The CDC checkpoint already covers every row.	Inspect the checkpoint with `cdcCheckpoint()` and reset it for a full reload, as described in Load Data Incrementally.
Rows missing from one target but present in another	Each target persists independently; a row that one target rejects can still load into another.	Check `firstError` in the status record, fix the data or the Type, and process again.

Copy link to this sectionPrerequisites

Copy link to this sectionStep 1: Connect to Snowflake

Copy link to this sectionStep 2: Inspect the source data

Copy link to this sectionStep 3: Model the data

Copy link to this sectionStep 4: Create the source collection

Copy link to this sectionStep 5: Run the pipeline

Copy link to this sectionStep 6: Verify the data

Copy link to this sectionMake the pipeline part of your package

Copy link to this sectionFix common problems

Copy link to this sectionSee also

Prerequisites

Step 1: Connect to Snowflake

Step 2: Inspect the source data

Step 3: Model the data

Step 4: Create the source collection

Step 5: Run the pipeline

Step 6: Verify the data

Make the pipeline part of your package

Fix common problems

See also