> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streambuild.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Sources and Pipelines

> Reusable streaming sources and the pipelines that consume them.

Sources are project-wide identities under `sources/*.yml`. Pipelines reference those identities
from `pipelines/<name>/pipeline.yml`.

```text theme={null}
streambuild_project.toml
sources/
  order_events.yml
pipelines/
  order_events/
    pipeline.yml
    orders.sql
    order_items.sql
```

Only direct `sources/*.yml` files are discovered. Source names must be unique across the project.

## Managed Kafka source

```yaml theme={null}
sources:
  - name: order_events
    kind: kafka
    broker_list: redpanda:9092
    topic: source.order_events.live
    consumer_group: optional_group
    format: JSONAsString
    ttl: _replay_landed_at + INTERVAL 30 DAY
    settings:
      kafka_num_consumers: "1"
    replay_boundary:
      mode: offsets
```

StreamBuild creates:

```text theme={null}
kafka__order_events   Kafka engine table
raw__order_events     MergeTree landing table
mv__order_events      landing materialized view
```

Managed sources support `offsets`, `timestamp`, and `landed_at`. `JSONAsString` is currently the
only supported Kafka format. StreamBuild owns these source objects and blocks incompatible source
drift rather than silently recreating them.

`ttl` is an optional ClickHouse expression applied to the managed `raw__order_events` landing
table. It overrides `[defaults].managed_source_ttl` from `streambuild_project.toml`. Use
`_replay_landed_at` for ingestion-age retention, for example
`_replay_landed_at + INTERVAL 30 DAY`. If the source and project default both omit TTL,
StreamBuild creates the landing table without a TTL clause.

`format` defaults to `JSONAsString`. When `consumer_group` is omitted, the effective ClickHouse
group is `streambuild_<source>_<source>_<database>`. Hyphens in the database suffix are normalized
to underscores so each target database receives an isolated consumer group.

## Adopted stream table

Adopt an existing table without transferring ownership:

```yaml theme={null}
sources:
  - name: orders
    kind: stream_table
    table_name: orders_existing
    replay_boundary:
      mode: offsets
      columns:
        _replay_partition: kafka_partition
        _replay_offset: kafka_offset
        _replay_timestamp: kafka_timestamp
```

The table must already exist as a bare relation name in the selected target database. StreamBuild
validates its replay columns but never creates, drops, or replaces the adopted table.
Because StreamBuild does not own adopted tables, `ttl` is rejected on `stream_table` sources.

| Mode        | Required mappings                                          |
| ----------- | ---------------------------------------------------------- |
| `offsets`   | `_replay_partition`, `_replay_offset`, `_replay_timestamp` |
| `timestamp` | `_replay_timestamp`                                        |
| `cursor`    | `_replay_cursor`, `_replay_timestamp`                      |

Integer roles must use ClickHouse integer types. Timestamp roles must use `DateTime` or
`DateTime64`. Adopted sources do not support `landed_at`.

## Pipeline declaration

`pipelines/order_events/pipeline.yml` contains the source identity:

```yaml theme={null}
source: order_events
```

The directory name is the pipeline name. Model files are discovered recursively, and each model
name is its SQL filename stem. Every pipeline needs at least one model. Nested `pipeline.yml` files,
duplicate model stems, and project-wide duplicate model names are rejected.

Virtual-environment replay policies may also be set at pipeline level:

```yaml theme={null}
source: order_events
replay_on_change:
  breaking: full
  non_breaking: bounded-7d
bounded_replay_fallback: full
```

These policy fields are invalid in direct mode because direct builds always rebuild their selected
closure.