sqlload.example.toml

# sqlload configuration. `${ENV:NAME}` is replaced from the environment; `$${` is a literal `${`.

[source]
kind = "kafka"
brokers = "${ENV:KAFKA_BROKERS}"
include = ["shop.*"]
exclude = ["shop.*.dlq"]
# `{stream}` is the lower-case table name; `{table}` and `{table_flat}` also work.
consumer_group = "sqlload-{stream}"
# Where a group with no committed offset starts.
start_from = "latest"
discovery_interval = "5m"
# Kafka prefetch shared by all consumers (at most 64MiB and at least 1MiB each).
prefetch_budget = "1GiB"
# Streams start this many at a time, `startup_interval` apart; 0 starts them all at once.
startup_batch = 20
startup_interval = "500ms"

# Any librdkafka consumer property. `queued.max.messages.kbytes` replaces the
# prefetch share; `fetch.wait.max.ms` defaults to 2000.
[source.properties]
# "security.protocol" = "SASL_SSL"
# "sasl.mechanisms" = "SCRAM-SHA-512"

[destination]
kind = "snowflake"
account = "myorg-myaccount"
user = "SQLLOAD"
auth = { method = "pat", token = "${ENV:SNOWFLAKE_TOKEN}" }
# auth = { method = "key_pair", private_key_file = "/etc/sqlload/key.p8" }
role = "SQLLOAD_ROLE"
# Needed for `COPY` loads and `plan`/`apply`; streaming does not use it.
warehouse = "LOAD_WH"
database = "RAW"
schema = "KAFKA"

[staging]
# none: Snowpipe Streaming. internal: files on a table stage. s3: files in your bucket.
kind = "none"
# internal and s3: how files are loaded. auto picks Snowpipe with a key pair, otherwise COPY.
load = "auto"
# internal: how long dated stage folders are kept.
retain = "7d"
# Uploads, and separately loads, in flight at once across all streams.
max_in_flight = 32
# Snowpipe loads: Kafka offsets are committed only once Snowpipe reports the files loaded
# (insertReport). The loading role needs OPERATE on each pipe for insertFiles and MONITOR for
# insertReport and loadHistoryScan; pipes created by `sqlload apply` are owned by it.
# The most files per stream sent but not committed yet; a full stream pauses its partitions
# (time paused: sqlload_pending_wait_milliseconds_total).
max_pending_files = 32
# A file Snowpipe reports as failed or partially loaded, or that is gone from the stage:
# halt stops the stream without committing it; skip logs and counts it, then commits past it.
# To recover from halt, fix the cause and restart with halt; replays usually get new file names
# that load normally. Only if the same file is reported failed again, load it with
# COPY INTO ... FILES = (...) and restart once with skip.
on_load_error = "halt"
# Files with no report are checked (load history, then the object) after this long, then
# again every pending_recheck_every. Slow loads are never failed: alert on
# sqlload_oldest_pending_seconds instead.
pending_check_after = "5m"
pending_recheck_every = "10m"

# [staging.s3]
# bucket = "my-bucket"
# prefix = "sqlload"
# region = "eu-west-2"
# stage = "SQLLOAD_S3_STAGE"
# auto_ingest = true

[batching]
max_bytes = "32MB"
max_rows = 100000
max_age = "60s"
# Across every stream; any stream flushes early when this is reached.
max_total_bytes = "2GB"
# Streaming: compressed bytes per append request (service limit 4MiB).
max_request_bytes = "3.5MB"
# Streaming: what to do with a row larger than one request: fail or skip.
oversized = "fail"

# Later matching entries win.
[[topics]]
match = "shop.*.events"
start_from = "earliest"
max_age = "10s"

[naming]
# Default rule: strip this prefix, upper-case, `.` becomes `__`, other symbols `_`.
strip_prefix = "shop."
# What happens to topics no rule or map entry names: default or fail.
unmatched = "default"
stage = "{table}_SQLLOAD_STAGE"
pipe = "{table}_SQLLOAD_PIPE"

# Rules run in order; `{name}` captures text without dots, `*` matches anything.
[[naming.rules]]
match = "shop.{service}_{variant}.{kind}"
table = "{service}__{kind}"

# Exact names win over rules.
[naming.map]
"shop.legacy-orders" = "ORDERS_V1"

[metrics]
listen = "0.0.0.0:9898"