ingestr ingest

The ingest command is a core feature of the ingestr tool, allowing users to transfer data from a source to a destination with optional support for incremental updates.

Example

The following example demonstrates how to use the ingest command to transfer data from a source to a destination.

ingestr ingest \
   --source-uri '<your-source-uri-here>' \
   --source-table '<your-schema>.<your-table>' \
   --dest-uri '<your-destination-uri-here>'

Required flags

• --source-uri TEXT: Required. Specifies the URI of the data source.
• --dest-uri TEXT: Required. Specifies the URI of the destination where data will be ingested.
• --source-table TEXT: Defines the source table to fetch data from. Required except for CDC sources, which replicate every table in their capture set when it is omitted. CDC sources also accept a comma-separated list (public.users,public.orders) to replicate just those tables; see Multi-table CDC.

Optional flags

• --dest-table TEXT: Designates the destination table to save the data. If not specified, defaults to the value of --source-table.
• --incremental-key TEXT: Identifies the column used for incremental reads or replacement. For append and merge, ingestr passes it to the source read for interval filtering when the source supports it; append does not compare it with the destination table. For delete+insert, this is the interval column used to decide which destination rows to replace, and should normally be a date, timestamp, partition column, or numeric batch column rather than a row primary key. Numeric delete+insert keys work when bounds are inferred from staged rows; --interval-start and --interval-end are parsed as datetime values, not numeric values. Defaults to None.
• --incremental-predicate TEXT: Appends a destination-specific SQL predicate to the target match condition for supported SQL destinations with single-table, non-full-refresh merges. Use t as the destination alias for BigQuery and Trino, and target for other supported destinations. For example: t.event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY) on BigQuery, or target.event_date >= CURRENT_DATE - INTERVAL '7 days' on PostgreSQL. Use only when every matching destination row is guaranteed to satisfy the predicate; an incorrect predicate can insert duplicates or fail a uniqueness constraint depending on the destination's merge mechanism.
• --incremental-strategy TEXT: Defines the strategy for incremental updates. Options include replace, append, delete+insert, merge, or scd2. The default strategy is replace. Not every source and destination supports every strategy; unsupported combinations fail at runtime. The former truncate+insert strategy has been removed; existing configurations should use replace, which selects a destination-specific replacement implementation and preserves the table object for single-table PostgreSQL loads.
• --interval-start: Sets the inclusive start of the interval for the incremental key and passes that start bound to the source read when the source supports interval filtering. For delete+insert, this becomes the lower delete bound. If omitted, ingestr can infer the lower bound from staged rows; if it cannot infer a required bound, delete+insert skips the delete and insert. Defaults to None.
• --interval-end: Sets the inclusive end of the interval for the incremental key and passes that end bound to the source read when the source supports interval filtering. For delete+insert, this becomes the upper delete bound. If omitted, ingestr can infer the upper bound from staged rows; if it cannot infer a required bound, delete+insert skips the delete and insert. Defaults to None.
• --extract-parallelism INTEGER: The number of parallel jobs used to extract data from the source. With --extract-partition-by, this is the number of windows read at once. When set explicitly, it also sets the destination write parallelism. Defaults to 5. Environment variable: INGESTR_EXTRACT_PARALLELISM.
• --extract-partition-by TEXT: Splits extraction from a SQL source into windows over this date, timestamp, or integer column, read in parallel. See Parallel Extract Partitioning. Environment variable: INGESTR_EXTRACT_PARTITION_BY.
• --extract-partition-interval TEXT: The width of each extract window: a duration (1h, 7d), an integer step (10000), or auto. Defaults to auto when --extract-partition-by is set. Environment variable: INGESTR_EXTRACT_PARTITION_INTERVAL.
• --primary-key TEXT: Specifies a column used to identify one logical row for merge and scd2. For delete+insert, some destinations can use it to deduplicate staged rows during the insert or overwrite step, but this is destination-specific. Use the flag multiple times for composite keys. Primary key values should be non-null: some destinations match null keys as equal during merge, while others reject or duplicate them. This is ingestr strategy configuration; do not rely only on a primary key constraint already existing in the destination database. Defaults to None.
• --partition-by TEXT: Partitions the destination table by a column or, on BigQuery and Iceberg, a partition expression. See Partitioning.
• --columns <name>:<type>:<source>: Specifies the columns to be ingested. Use name:type to override a column's type, name:type:source to rename source to name with a type, or name::source to rename only. Multiple entries are comma-separated. Defaults to None.
• --no-inference: Skips schema inference for schema-less sources and uses --columns as the source schema. Requires --columns.
• --mask <column_name>:<algorithm>[:param]: Applies data masking to specified columns. Can be used multiple times for different columns. See the Data Masking documentation for available algorithms and usage examples. Defaults to None.
• --trim-whitespace: Trims leading and trailing whitespace from all string column values before writing to the destination. This applies to regular batch ingestions and CDC ingestions, preserves nulls and column types, and leaves non-string columns unchanged. Defaults to false. Can also be set with TRIM_WHITESPACE=true or INGESTR_TRIM_WHITESPACE=true.
• --schema-naming Specifies what naming convention to use for table and column names on the destination. Can be default or direct.default is snake_case. `direct is case sensitive and doesn't contract underscores.
• --cdc-table-naming: In multi-table (CDC) mode with dest_schema set, controls how a schema-qualified source table maps to its destination table. schema_table (default) flattens the source schema into the table name (foo.A → foo_cdc.foo_A); table keeps only the table name (foo.A → foo_cdc.A). With table, two source schemas containing a table of the same name fail the run with a collision error instead of sharing a destination table. Switching an existing pipeline to table creates a fresh CDC connector (new replication slot, state and snapshot) and leaves the old flattened tables untouched. Can also be set with CDC_TABLE_NAMING or INGESTR_CDC_TABLE_NAMING.
• --cdc-snapshot-mode auto|never: PostgreSQL CDC only. auto resumes when possible and permits initial and recovery snapshots and backfills as needed. never prohibits all snapshots and backfills; it requires explicit table selection, an existing slot=, and keyed merge. The policy is saved with connector state and inherited when omitted; new connectors default to auto. Cannot combine never with full refresh. Environment variable: INGESTR_CDC_SNAPSHOT_MODE.
• --cdc-start-lsn TEXT: Bootstrap boundary for preloaded PostgreSQL CDC destinations. Requires --cdc-snapshot-mode never, a transaction-consistent external baseline at this LSN, and retained WAL. An interrupted bootstrap can retry the same boundary. After adoption completes, the same boundary may remain configured: subsequent runs resume from managed checkpoints without replaying the original baseline. A different boundary or existing state without an external baseline is rejected; the flag cannot reset progress. See CDC without snapshots. Environment variable: INGESTR_CDC_START_LSN.
• --cdc-snapshot-chunked: Opt into PostgreSQL CDC snapshots that read bounded chunks in separate short transactions. Disabled by default. Uses keyed merge; incompatible with explicit non-merge strategies, --cdc-snapshot-mode never, --cdc-start-lsn, and SQL/incremental extraction filters. See Large initial snapshots. Environment variable: INGESTR_CDC_SNAPSHOT_CHUNKED.
• --cdc-snapshot-execution parallel|serial: Execution order when chunked snapshots are enabled. serial (default) copies all selected tables before decoding and applying retained WAL; PostgreSQL must retain the required WAL throughout the copy. Serial requires --cdc-wal-staging inmemory. Set parallel to decode changes while copying chunks and catch up after each chunk. Environment variable: INGESTR_CDC_SNAPSHOT_EXECUTION.
• --cdc-snapshot-chunk-size INTEGER: Maximum rows per chunk when --cdc-snapshot-chunked is enabled. Must be positive; defaults to 10000. Environment variable: INGESTR_CDC_SNAPSHOT_CHUNK_SIZE.
• --cdc-snapshot-chunk-timeout DURATION: Timeout for an individual chunk read when --cdc-snapshot-chunked is enabled. Must be positive; defaults to 30s. Environment variable: INGESTR_CDC_SNAPSHOT_CHUNK_TIMEOUT.
• --cdc-wal-staging inmemory|destination: Storage for pending logical changes during parallel chunked PostgreSQL CDC snapshots. inmemory (default) uses memory and retains source WAL for recovery; destination uses a managed staging table in the destination and requires --cdc-snapshot-execution parallel. Default serial execution requires inmemory and retains pending changes in source WAL until copying finishes. Environment variable: INGESTR_CDC_WAL_STAGING.
• --cdc-wal-buffer-bytes INTEGER: Memory budget in bytes for pending changes during chunked snapshots. Must be positive; defaults to 67108864 (64 MiB). Environment variable: INGESTR_CDC_WAL_BUFFER_BYTES.
• --cdc-delete-behavior: How a CDC merge applies source deletes. soft (default) keeps the deleted row and sets _cdc_deleted = true; hard deletes the row from the destination. hard is supported on PostgreSQL, MySQL/MariaDB, SQL Server, Snowflake, BigQuery, DuckDB, and StarRocks destinations, with the merge strategy or a non-streaming --full-refresh, and for sources that keep CDC state in the destination (PostgreSQL, MySQL/MariaDB, and SQL Server CDC, and SQL Server Change Tracking started with --stream --full-refresh). See Deletes. Can also be set with CDC_DELETE_BEHAVIOR or INGESTR_CDC_DELETE_BEHAVIOR.
• --stream: Runs continuous (streaming) ingestion instead of a one-shot load. Supported by CDC sources (postgres+cdc, mysql+cdc, mariadb+cdc, vitess+cdc, ps_mysql+cdc, mssql+cdc, mongodb+cdc) and message brokers. The process runs until interrupted (SIGINT/SIGTERM), flushing buffered records to the destination on an interval or record-count trigger. See Streaming ingestion below.
• --full-refresh: Rebuild destination tables from fresh source snapshots. With --stream, continue streaming in the same process after the snapshots. Streaming refresh requires a PostgreSQL CDC, MySQL CDC, or SQL Server CDC/Change Tracking source and a destination supporting managed CDC state. With --cdc-snapshot-chunked, finite runs also rebuild through checkpointed merges rather than the replacement table swap path.
• --refresh-id TEXT: An opaque, globally unique operation ID for --full-refresh --stream. Reuse this ID across process restarts; use a new ID for each requested refresh. Can also be set with INGESTR_REFRESH_ID.
• --flush-interval: In streaming mode, flush buffered records to the destination at least this often. Defaults to 30s. Only valid with --stream.
• --flush-records: In streaming mode, flush when this many records have been buffered. Defaults to 50000. Only valid with --stream.
• --metrics-addr: In streaming mode, serve replication lag and throughput metrics over HTTP on this address (e.g. 127.0.0.1:6060). Disabled unless set. Only valid with --stream. See Monitoring a stream below.
• --debug: Enables debug logging. Some destinations print generated SQL in debug logs; parameterized queries may show placeholders such as $1, ?, @p1, or @p2 for values bound separately by the database driver.

The interval-start and interval-end options support various datetime formats. When both are provided, interval-start must be earlier than interval-end. Here are some examples:

• %Y-%m-%d: 2023-01-31
• %Y-%m-%dT%H:%M:%S: 2023-01-31T15:00:00
• %Y-%m-%dT%H:%M:%S%z: 2023-01-31T15:00:00+00:00
• %Y-%m-%dT%H:%M:%S.%f: 2023-01-31T15:00:00.000123
• %Y-%m-%dT%H:%M:%S.%f%z: 2023-01-31T15:00:00.000123+00:00

INFO

For the details around the incremental key and the various strategies, please refer to the Incremental Loading section.

Partitioning

--partition-by takes a column name on every destination that supports partitioning. BigQuery and Iceberg also accept their native partition expressions, written the same way as in their own SQL:

Destination	Column only	Expressions	Details
BigQuery	Daily partitions	DATE(col), TIMESTAMP_TRUNC(col, HOUR), DATE_TRUNC(col, MONTH), RANGE_BUCKET(col, GENERATE_ARRAY(0, 1000, 10))	BigQuery partitioning
Iceberg	One partition per value	day(col), hour(col), month(col), year(col), bucket(16, col), truncate(10, col)	Iceberg partitioning
DuckLake	Daily partitions for timestamps	Not supported	DuckLake table layout

Other destinations do not partition tables, and reject partition expressions.

Streaming ingestion

The --stream flag turns ingest into a long-running process that continuously pulls changes from the source and flushes them to the destination, rather than running once and exiting. It is supported by:

• CDC sources (postgres+cdc, mysql+cdc, mariadb+cdc, vitess+cdc, ps_mysql+cdc, mssql+cdc, mongodb+cdc): captures every insert, update, and delete across all selected tables and applies them with the merge strategy.
• Message brokers (kafka, amqp): consumes messages into a fixed envelope schema — a msg_id primary key, a JSON data column holding the decoded body and metadata, and an _ingestr_order column (source offset / delivery tag) — and applies them with merge keyed on msg_id, keeping the latest record per key within each flush window. Schema inference is skipped (a never-ending stream has no end to infer from).

A flush happens whenever either --flush-interval (default 30s) or --flush-records (default 50000) is reached, whichever comes first. --flush-records is the memory bound: records are buffered until a flush.

Each flush writes the buffered records, merges them into the destination, and only then confirms the source position as durable. This gives at-least-once delivery: a crash before a flush completes re-delivers the un-flushed changes on restart, and the merge (by primary key / msg_id) makes replays idempotent. The stream resumes automatically — CDC from the destination's last recorded LSN, brokers from their committed offset / unacknowledged messages.

Stop a stream with Ctrl+C (SIGINT) or SIGTERM; ingestr performs a final flush of buffered data and exits cleanly.

# Stream all changes from a Postgres publication into BigQuery, flushing
# every 15 seconds or 100k changes, whichever comes first.
ingestr ingest \
   --source-uri 'postgres+cdc://user:pass@localhost:5432/mydb?publication=my_pub' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/sa.json' \
   --stream \
   --flush-interval 15s \
   --flush-records 100000

INFO

Postgres publications. Pass publication=<name> to use a publication you manage yourself. If you omit it, ingestr creates and maintains a publication named ingestr_publication, refreshing it on every run to include every logged table that has a replica identity (a primary key, REPLICA IDENTITY FULL, or a replica-identity index). Tables that are unlogged, or that lack a replica identity, are skipped with a warning — their changes either never reach the WAL or would make UPDATE/DELETE on the source fail.

INFO

New tables. Postgres CDC picks up tables created after ingestion started. A batch run detects them at startup; a stream additionally re-checks the source on an interval (discover_interval URI parameter, default 30s). When a running stream finds a new eligible table, it exits before changing destination data and asks its supervisor to restart it. The restarted process snapshots the table and then streams its retained WAL. User-managed publications are respected, so add the table to the publication yourself. See the Postgres CDC documentation for details.

INFO

Column-level schema changes are picked up at startup. If a table's columns change while a stream is running, restart the stream to apply the new schema. Run streaming ingestion under a supervisor (systemd, Kubernetes, etc.) so it restarts after transient source/destination outages.

Refreshing a stream

Stop the existing process before starting its replacement:

ingestr ingest \
   --source-uri "$SOURCE_URI" \
   --dest-uri "$DEST_URI" \
   --source-table public.orders \
   --stream --full-refresh --refresh-id "orders:550e8400-e29b-41d4-a716-446655440000"

The stream truncates each selected destination table at its snapshot boundary, loads fresh source data, and continues applying changes with merge or append. Multi-table refresh covers the currently selected or discovered source tables; destinations for tables that are no longer selected or present in the source are left in place. Snapshots are not an atomic replacement across tables.

The refresh ID is the sole lookup key in destination metadata. Its record links the operation to its CDC generation and per-table snapshot certificates. Restarting with the same ID resumes completed snapshots and retries interrupted ones. A completed refresh resumes ordinary streaming; a new ID requests another refresh. Reusing an ID for different connection or selection settings is rejected. The ID restores its recorded checkpoint and replication slot associations before source slots and leases are acquired, even if a recomputed connector identity changes. BigQuery credential rotation preserves these associations. Keep the destination metadata alongside the destination data.

Without --refresh-id, every process invocation with --full-refresh requests a new reset. Supervisors should persist one ID per refresh request and reuse it on retries. Unsupported sources and destinations fail before resetting destination data.

Monitoring a stream

Passing --metrics-addr starts a small HTTP server for the lifetime of the stream that exposes Prometheus metrics at /metrics. It is off unless the flag is set, and the address is bound before ingestion starts, so a port conflict fails immediately rather than half-way through a run. Only ingestr's own metrics are served — the Go runtime and process collectors are deliberately excluded.

ingestr ingest \
   --source-uri 'postgres+cdc://user:pass@localhost:5432/mydb' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/sa.json' \
   --stream \
   --metrics-addr 127.0.0.1:6060

curl -s localhost:6060/metrics | grep -E '^ingestr_replication|^ingestr_stream_table'

ingestr publishes:

Metric	Meaning
ingestr_replication_*{source}	Replication lag for the current source (see below). Absent when the source cannot report lag.
ingestr_stream_rows_synced_total	Cumulative rows written and confirmed durable since the process started.
ingestr_stream_bytes_synced_total	Cumulative logical Arrow payload bytes written and confirmed durable since the process started.
ingestr_stream_flush_cycles_total	Number of completed flush cycles.
ingestr_stream_last_synced_timestamp_seconds	Unix time of the last successful commit.
ingestr_stream_table_rows_synced_total{table}	Cumulative rows written and confirmed durable, broken out per destination table.
ingestr_stream_table_bytes_synced_total{table}	Cumulative logical Arrow payload bytes written and confirmed durable, broken out per destination table.
ingestr_stream_table_last_flush_rows{table}	Rows written for a table in its most recent successful flush cycle.
ingestr_stream_table_last_flush_bytes{table}	Logical Arrow payload bytes written for a table in its most recent successful flush cycle.
ingestr_stream_table_last_synced_timestamp_seconds{table}	Unix time of the most recent successful flush cycle that included a table.

The row counters advance only after a flush's destination write and its source-position commit have both succeeded, so they count durable rows rather than merely written ones. ingestr_stream_last_synced_timestamp_seconds also advances on cycles that commit a position without writing rows, which is what makes it usable as a staleness alarm: if it stops moving, the stream is stuck.

The byte counters use the Arrow record-batch buffers ingestr hands to the destination after stream transformations such as _ingestr_loaded_at. They are useful for stream throughput and workload monitoring, but they are not source log bytes, network bytes, compressed staging-file bytes, warehouse billing bytes, or destination table growth after merge/deduplication.

The ingestr_replication_* series carry a source label and depend on the engine, because "lag" is not the same quantity everywhere:

• Postgres (postgres+cdc) reports ingestr_replication_bytes_behind: the distance between the server's WAL head and the position ingestr has confirmed durable. This is the same number as pg_current_wal_lsn() - confirmed_flush_lsn for the replication slot, so it is what predicts unbounded WAL growth on the source. It is the value to alert on.
• MongoDB (mongodb+cdc) reports ingestr_replication_seconds_behind: the gap between the server's operationTime and the cluster time of the last processed change event. Both clocks are server-side, so an idle collection converges to zero instead of drifting upward.
• SQL Server (mssql+cdc) reports ingestr_replication_seconds_behind: the change time between the processed LSN and the capture watermark, via sys.fn_cdc_map_lsn_to_time. SQL Server's binary(10) LSNs are ordered but their difference is not a log distance, so no bytes_behind is published.

Fields that an engine cannot express are omitted rather than reported as a misleading zero. Postgres, for instance, has no per-LSN timestamp, so it publishes no ingestr_replication_seconds_behind. Message-broker sources report no replication series at all.

General flags

• --help: Displays the help message and exits the command.

Examples

Ingesting a CSV file to DuckDB

ingestr ingest \
   --source-uri 'csv://input.csv' \
   --source-table 'sample' \
   --dest-uri 'duckdb://output.duckdb'

Copy a table from Postgres to DuckDB

ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --source-table 'public.input_table' \
   --dest-uri 'duckdb://output.duckdb' \
   --dest-table 'public.output_table'

Replace a staged date slice from Postgres to BigQuery

ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --source-table "query:SELECT * FROM public.users WHERE dt = '2023-01-01'" \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/service/account.json&location=EU' \
   --dest-table 'raw.users' \
   --incremental-key 'dt' \
   --incremental-strategy 'delete+insert' \
   --columns 'dt:date'

Load an interval of data from Postgres to BigQuery using a date column

ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --source-table 'public.users' \
   --dest-uri 'bigquery://my_project?credentials_path=/path/to/service/account.json&location=EU' \
   --dest-table 'raw.users' \
   --incremental-key 'dt' \
   --incremental-strategy 'delete+insert' \
   --interval-start '2023-01-01' \
   --interval-end '2023-01-31' \
   --columns 'dt:date'

Load a specific query from Postgres to Snowflake

ingestr ingest \
   --source-uri 'postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=disable' \
   --dest-uri 'snowflake://user:password@account/dbname?warehouse=COMPUTE_WH&role=my_role' \
   --source-table 'query:SELECT * FROM public.users as pu JOIN public.orders as o ON pu.id = o.user_id WHERE pu.dt BETWEEN :interval_start AND :interval_end' \
   --dest-table 'raw.users' \
   --incremental-key 'dt' \
   --incremental-strategy 'delete+insert' \
   --interval-start '2023-01-01' \
   --interval-end '2023-01-31' \
   --columns 'dt:date'

Ingesting with Data Masking

ingestr ingest \
   --source-uri 'postgresql://user:pass@localhost/customers' \
   --source-table 'customer_data' \
   --dest-uri 'duckdb:///masked_customers.db' \
   --dest-table 'masked_customers' \
   --mask 'email:hash' \
   --mask 'phone:partial:3' \
   --mask 'ssn:redact' \
   --mask 'salary:round:5000'

This example demonstrates masking sensitive customer data:

• Email addresses are hashed for consistent anonymization
• Phone numbers show only first and last 3 digits
• SSNs are completely redacted
• Salaries are rounded to nearest $5000

Trimming whitespace from string values

ingestr ingest \
   --source-uri 'postgresql://user:pass@localhost/app?sslmode=disable' \
   --source-table 'public.customers' \
   --dest-uri 'duckdb:///warehouse.duckdb' \
   --dest-table 'raw.customers' \
   --trim-whitespace

This trims leading and trailing whitespace from string values as data streams through ingestr. For example, " Alice " becomes "Alice" and "\tA-123\n" becomes "A-123". Interior whitespace, such as "ACME Inc", is preserved.

Overriding column types

Use --columns to set the type of one or more columns on the destination. Each entry is name:type, and multiple entries are comma-separated:

ingestr ingest \
   --source-uri 'postgresql://user:pass@localhost/app?sslmode=disable' \
   --source-table 'public.customers' \
   --dest-uri 'snowflake://user:password@account/dbname?warehouse=COMPUTE_WH' \
   --dest-table 'raw.customers' \
   --columns 'id:bigint,signup_date:date,balance:decimal(18,2)'

Supported types include bigint, int, smallint, tinyint, float, double, decimal(p,s), string, text, varchar(n), boolean, date, timestamp, timestamp_ntz, json, uuid, and binary.

Sized string types

String types accept an optional length, so you can create a bounded column such as varchar(50) instead of an unbounded text column:

ingestr ingest \
   --source-uri 'postgresql://user:pass@localhost/app?sslmode=disable' \
   --source-table 'public.customers' \
   --dest-uri 'postgresql://user:pass@localhost/warehouse?sslmode=disable' \
   --dest-table 'raw.customers' \
   --columns 'name:varchar(100),email:varchar(255)'

The types that accept a length are varchar(n), string(n), and text(n) — all equivalent. A string type given without a length (varchar, string, or text) creates an unbounded column.

INFO

For more examples, please refer to the specific platforms' documentation on the sidebar.