| name | add-pipeline |
| description | Registers cds-data-pipeline jobs via addPipeline with correct source and target shapes. Use when creating replicate, materialize, or move-to-service pipelines, configuring delta sync, schedules, or REST/OData sources. |
addPipeline
Entity-shape replicate (most common)
await pipelines.addPipeline({
name: 'LocalProducts',
source: { service: 'northwind', entity: 'northwind.Products' },
target: { entity: 'db.LocalProducts' },
mode: 'delta',
delta: { field: 'modifiedAt', mode: 'timestamp' },
schedule: 600_000,
});
Model the target as a consumption view with @cds.persistence.table:
@cds.persistence.table
entity LocalProducts as projection on northwind.Products {
ProductID, ProductName, UnitPrice
} where Discontinued = false;
Query-shape materialize
await pipelines.addPipeline({
name: 'DailyCustomerRevenue',
source: {
service: 'db',
query: SELECT.from('sales.Orders')
.columns('customerId', { func: 'sum', args: [{ ref: ['amount'] }], as: 'totalAmount' })
.groupBy('customerId'),
},
target: { entity: 'db.DailyCustomerRevenue' },
schedule: '0 2 * * *',
});
Prefer @materialize.snapshot on a CDS projection when the aggregate is declarative — see cds-data-materialization.
Inference rules (summary)
| Config | Read shape | Defaults |
|---|
source.query only | Query-shape | mode: 'full', refresh: 'full' |
source.entity or rest.path | Entity-shape | mode: 'full'; add mode: 'delta' + delta for incremental sync |
| Both query and entity | Error | — |
| Neither | Error | — |
Do not pass kind — intent is derived from shape.
Run retention (optional)
When global housekeeping is enabled in cds.requires['data-pipeline'].housekeeping, override counts per pipeline:
await pipelines.addPipeline({
name: 'HighVolumeOrders',
source: { service: 'OrdersService', entity: 'Orders' },
target: { entity: 'db.ArchivedOrders' },
retention: { retentionDays: 30, maxRuns: 200 },
});
See Run housekeeping.
Initial load on startup (optional)
preload runs one sync right after registration so the target is not empty
between boot and the first scheduled tick. Independent of schedule.
await pipelines.addPipeline({
name: 'LocalProducts',
source: { service: 'northwind', entity: 'northwind.Products' },
target: { entity: 'db.LocalProducts' },
schedule: 600_000,
preload: true,
});
await pipelines.addPipeline({
name: 'LocalProducts',
source: { service: 'northwind', entity: 'northwind.Products' },
target: { entity: 'db.LocalProducts' },
preload: { mode: 'full', wait: true },
});
preload | Behavior |
|---|
false / omitted | No startup run (default). |
true | Background run (async), pipeline's effective mode, boot not blocked. |
{ mode, wait } | mode overrides the run mode; wait: true blocks addPipeline until the run finishes (failure propagates). |
Preload runs record trigger: 'preload' in PipelineRuns. A disabled pipeline skips its preload.
REST source
source: {
service: 'RestProvider',
rest: {
path: '/api/items',
pagination: { type: 'offset', pageSize: 100 },
deltaParam: 'modifiedSince',
dataPath: 'results',
},
},
Anti-patterns
❌ Wrong:
await pipelines.addPipeline({ kind: 'replicate', name: 'X', ... });
✅ Correct — omit kind; use shape-appropriate source / target.
❌ Wrong — source.query + mode: 'delta'.
✅ Correct — snapshots rebuild fully; use entity-shape for row delta.
❌ Wrong — target.service: 'RemoteOData' without target.adapter.
✅ Correct — implement custom target adapter or write to db.
Docs