Hi, we use pipeline with a loop from a
# sling-helpi
itamar s
07/16/2026, 3:24 PMHi, we use pipeline with a loop from a database query. each table gets several variables from the table row .
for each row from the table we run its own stream on the replication
how can we run specific tables manually ?
We dont want to change the main is_active columns because it serves the scheduled process.
1. best option is if there is a --tag or --stream name ?
2. fails: when one of the streams fail with abort the all pipeline stops . I prefer the behavior of replication
that it will fail the specific stream and at the end the status of the pipeline will be failed.
for now we put warn. the process is success even if one of the stream failed and then we
need to query the log and find the error to alert the developers.
thanks
steps:
# All active medium tables
- type: query
id: cfg
connection: target_stg
query: |
SELECT table_name, cycle_limit, hours_back, cutoff
FROM zstg.vw_z_refresh
WHERE is_active = 1 and size_group = 'medium'
ORDER BY table_name
- type: group
id: refresh
loop: state.cfg.result
steps:
- type: replication
path: elts/fin/fin_inc.yaml
streams:
- "{loop.value.table_name}"
env:
CUTOFF: "{loop.value.cutoff}"
HOURS_BACK: "{loop.value.hours_back}"
CYCLE_LIMIT: "{loop.value.cycle_limit}"
on_failure: warn
f
flarco
07/18/2026, 8:28 PMhey @itamar s ,
#1 see https://docs.slingdata.io/concepts/replication/tags-wildcards
and https://docs.slingdata.io/concepts/hooks/replication
you can pass in
streamstag:my_tagon_failure: deferi
itamar s
07/19/2026, 12:00 PM@flarco Hi, in my example i am running a pipeline (not replication) each loop i get another stream from database. I put a tag on one of the streams in the replication and tried to run it. but it run all.
uv run sling run -p elts/fin/fin_inc_pipeline.yaml --streams tag:mttransactions
itamar s
07/19/2026, 12:59 PMAlso 2: the on_failure: defer fail in the middle of a group.
- type: group
id: load_calls
loop: state.brands.result
steps:
- type: log
message: "Loading n.calls for brand_id={loop.value.brand_id}"
- type: query
id: brand_watermark
connection: target_stg
query: |
SELECT COALESCE(CAST(MAX(watermark) AS text), '1900-01-01 00:00:00') AS watermark
FROM zstg.z_wm_n_calls
WHERE brand_id = '{loop.value.brand_id}'
on_failure: warn
- type: replication
path: elts/n/n_inc.yaml
streams: [n.calls]
env:
SOURCE_CONN: "{loop.value.conn_str}"
BRAND_ID: "{loop.value.brand_id}"
CYCLE_LIMIT: "{loop.value.cycle_limit}"
JOIN_ADD: "{loop.value.join_add}"
WHERE_ADD: "{loop.value.where_add}"
WATERMARK: "{state.brand_watermark.result[0].watermark}"
CUTOFF: "{loop.value.cutoff}"
on_failure: defer # one brand failing must not abort the rest
f
flarco
07/21/2026, 12:04 PM@itamar s you're correct, defer is broken and now fixed for next release (releasing by end of this week)
flarco
07/21/2026, 12:05 PM> --streams tag:mttransactions
Oh i see, this flag is not being passed in pipelines, will add 👍
flarco
07/21/2026, 12:05 PMit only works for replicaitons
flarco
07/27/2026, 1:55 AM