Delta Lake output connector
Delta Lake is a popular open table format based on Parquet files. It is typically used with the Apache Spark runtime. Data in a Delta Lake is organized in tables, stored in a file system or an object stores like AWS S3, Google GCS, or Azure Blob Storage.
The Delta Lake output connector does not yet support fault tolerance.
Update modes
The connector has two ways of applying a view's changes to a Delta table, chosen with
update_mode.
| Mode | What the table holds | Who folds the changes in |
|---|---|---|
cdc (default) | A change log: every insert and delete, tagged with __feldera_op and __feldera_ts | You, with a periodic Spark MERGE INTO job |
merge (experimental) | Exactly the view's rows, one per key, no extra columns | The connector |
cdc is described in CDC mode, merge in Merge mode.
CDC mode
The Delta Lake format does not support efficient real-time deletes and updates. To delete a record from a Delta table, one must first locate the record, which often requires an expensive table scan. This limitation makes it inefficient to directly write the output of a Feldera pipeline, which consists of both inserts and deletes, to a Delta table.
To address this issue, the Delta Lake connector transforms both inserts and deletes into table records with additional metadata columns that describe the type and order of operations. Specifically, the connector adds the following columns to the output Delta table:
| Column | Type | Description |
|---|---|---|
__feldera_op | VARCHAR | Operation that this record represents: i for "insert", d for "delete", or u for "update". |
__feldera_ts | BIGINT | Timestamp of the update, used to establish the order of updates. Updates with smaller timestamps are applied before those with larger timestamps. |
Effectively, we treat the table as a change log, where every record corresponds to
either an insert or delete operation. The user can run a periodic Spark job to
incorporate this change log into another Delta table, using the SQL MERGE INTO operation. An example of the code is below:
MERGE INTO {target_table} AS target
USING (
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY {merge_key}
ORDER BY __feldera_ts DESC
) AS rn
FROM {source_table}
-- Only consider new updates since the last merge.
WHERE __feldera_ts >= (
SELECT COALESCE(MAX(__feldera_ts), 0)
FROM {target_table}
)
)
-- Only apply the last update for each key.
WHERE rn = 1
) AS source
ON target.{merge_key} = source.{merge_key}
WHEN MATCHED AND source.__feldera_op = 'd' THEN
DELETE
WHEN MATCHED AND source.__feldera_op = 'u' THEN
UPDATE SET *
WHEN NOT MATCHED AND source.__feldera_op = 'i' THEN
INSERT *
WHEN NOT MATCHED AND source.__feldera_op = 'u' THEN
INSERT *
Views with unique keys
If the SQL view contains a unique key—a set of columns that uniquely identify each record—the Delta Lake connector can optimize updates by combining a delete and insert with the same key into a single atomic update. In such cases, the connector emits a record with the __feldera_op field set to 'u' (for update).
To enable this optimization:
- Use the
CREATE INDEXstatement to define the unique key on the view. - Set the connector's
indexproperty to reference this index.
For more information, see the documentation on views with unique keys.
Delta Lake output connector configuration
| Parameter | Description |
|---|---|
uri* | Table URI, e.g., "s3://feldera-fraud-detection-data/feature_train". |
mode* | Determines how the Delta table connector handles an existing table at the target location. Options: |
- append: New updates will be appended to the existing table at the target location. If the table doesn't exist, it will be created. | |
- truncate: Existing table at the specified location will be truncated on the first pipeline start. When the pipeline resumes from a checkpoint the table is kept as-is so that data written before the restart is preserved. | |
- error_if_exists: If a table exists at the specified location, the operation will fail. When the pipeline resumes from a checkpoint the existing table is opened without error. | |
checkpoint_interval | Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables. The option is only available when creating the Delta table ( 0 means no checkpoints are created. If writing a checkpoint fails, the connector keeps the commit, which is already in the table, and logs a warning. The connector reports itself unhealthy and tries to write the missed checkpoint again after each commit, until it succeeds. Default: 10. |
log_retention_duration | Log retention duration for newly created Delta tables. Configures the The option is only available when creating the Delta table ( The value follows the Delta Lake interval syntax Default: |
enable_expired_log_cleanup | Whether to clean up expired log entries when a checkpoint is written. Configures the The option is only available when creating the Delta table ( Default: |
max_retries | Maximum number of retries for failed Delta Lake operations like writing Parquet files and committing transactions. The connector performs retries on several levels: individual S3 operations, Delta Lake transaction commits, and overall operation retries. This setting controls the overall operation retries. When a write to the table fails, because of an S3 timeout or any other reason that was not resolved by lower-level retries, the connector will retry the entire operation. When not specified, the connector performs infinite retries. When set to 0, the connector doesn't retry failed operations. |
threads | Number of parallel threads used by the connector. Increasing this value can improve Delta Lake write throughput by enabling concurrent writes. Values above 1 require the view to have a unique key, so that the connector can order inserts and deletes correctly. Define the key with In Default: |
variant_encoding | Encoding of - - See VARIANT. Default: |
update_mode | How the connector applies the view's changes to the table. Orthogonal to - - Default: |
lookup_chunk_size | Ceiling on the encoded keys the connector holds while locating rows to supersede. A flush whose key set exceeds this budget is split into successive lookup passes, which bounds memory at the cost of re-scanning candidate files. Like A size such as |
max_concurrent_probes | Number of data files read concurrently while locating rows to supersede. The lookup reads each file in its own task, so this sets how much of the decoding runs in parallel as well as how many requests are in flight. Each concurrent read holds one file's projected key column, so it bounds memory too. Like |
auto_compact | Compact the target table from the connector, between flushes. Merge mode marks an old row version deleted but leaves it in place. Without compaction the table keeps growing and reads keep slowing down, however few live rows it holds. See Compaction is required. Off by default, because maintenance is normally the table administrator's job. Set it for tables where Feldera is the only writer and nothing else will maintain them. Default: |
[*]: Required fields
The connector cannot write to a Delta table that sets delta.columnMapping.mode to name or id.
The pipeline fails to start with an error that names the table property.
Storage parameters
Additional configuration options are defined for specific storage backends. Refer to backend-specific documentation for details:
On S3 the connector commits with a conditional put, so one writer cannot
overwrite another writer's commit. The S3-compatible store must support
conditional put. Amazon S3 does; on a store that does not, two writers can
overwrite each other's commit. DynamoDB-based locking
(AWS_S3_LOCKING_PROVIDER = 'dynamodb') is no longer supported; the connector
logs an error and commits without it.
Merge mode
Merge mode is an experimental feature of Feldera. Read Limits and Flush cost before you use it on a production table.
With "update_mode": "merge" the connector keeps the target table equal to the current
contents of the view: one row per key, no metadata columns, no change history. There is no
Spark merge job to run.
A flush never rewrites a data file. A change to a row appends the new version and marks the old one deleted in a deletion vector attached to the file that holds it. To find the old row, the flush reads the key columns of the table; see Flush cost. One flush produces one commit, so a reader never sees half of one.
Requirements
The connector checks all of these when the pipeline starts, so a misconfiguration fails
before any data moves. It checks the table requirements again each time it reloads the
table, so an ALTER TABLE that breaks one while the pipeline runs stops the connector with
an error.
| Requirement | Why |
|---|---|
The view has a unique key: the index property | The key identifies the row to supersede |
delta.enableDeletionVectors is true on the table | Without it the connector cannot mark a row deleted |
| The table's schema matches the view's | The table holds the view's rows, so its columns are the view's columns |
Key columns are scalars, or ROW of scalars | See Key types |
| At least one key column is stored in the data files | A key made entirely of partition columns leaves nothing in the file to read the key from, and would mean one partition directory per row |
delta.enableChangeDataFeed is off | A change data feed needs _change_data files that this connector does not write |
delta.appendOnly is off | Superseding a row means removing the old one |
| The table uses only table features that delta-rs can write | Tables with row tracking, liquid clustering, UniForm (icebergCompat*), identity columns, type widening, in-commit timestamps or column mapping are refused. Databricks turns some of these on by default. Create the table without them, or remove them with ALTER TABLE ... DROP FEATURE |
A key column that is a partition column is not BINARY | delta-rs stores binary partition values as escaped text, so the key would not match |
Merge mode also assumes the pipeline is the only process writing rows to the table.
Maintenance is fine: when an OPTIMIZE or VACUUM conflicts with a flush, the connector
redoes the flush. Each redo uses one of the max_retries attempts, so leave max_retries
unset in merge mode. A second writer that inserts rows is not fine. The connector finds the
row to supersede in the snapshot it read, so an insert of the same key committed in between
leaves two live rows for one key.
On a table the connector creates, it enables deletion vectors itself. On a table that already exists it does not, and reports the statement to run:
ALTER TABLE <table> SET TBLPROPERTIES ('delta.enableDeletionVectors' = 'true')
That is deliberate. Enabling deletion vectors raises the table to reader version 3, and any consumer that cannot read deletion vectors then loses access to the table entirely. Whether to accept that is the table owner's decision, not the pipeline's.
Key types
A type may form part of the key only when a value survives the round trip into Parquet and back still comparing equal to itself, because that comparison is how the connector finds the row to supersede.
| Type | Usable as a key |
|---|---|
Scalars, except INTERVAL | Yes |
ROW of scalars | Yes |
TINYINT UNSIGNED to BIGINT UNSIGNED | No: Delta has no unsigned types. Cast to a signed type that holds the values |
ARRAY | No |
MAP, VARIANT | No: two values Feldera considers equal can be stored differently, so the connector could fail to find the row |
For an unsupported type, project a scalar surrogate into the view and index on that.
What it reads, and what it skips
The connector finds the row to supersede by reading the key columns of the target table. It narrows that read two ways, both of which it will keep a file for whenever it cannot be certain:
| Test | Effect |
|---|---|
| Key range, from the statistics in the Delta log | A file whose key range cannot hold any changed key is skipped without being opened |
| Key range, from each Parquet file's footer | Within a file that is opened, only the row groups whose key range can hold a changed key are read |
| Partition, when partition columns are key columns | A file in a partition none of the changed keys belongs to is skipped without being opened |
A few things make the read wider than it needs to be, all worth knowing when a flush costs more than expected:
- A key that is not part of the table's clustering. If changed keys are spread across the
whole table, every file's range contains some of them and nothing is skipped. Partitioning
or
ZORDERing the table on the key is what makes pruning effective. - The files the connector itself appends. A flush writes a file holding the new versions
of whatever keys changed, so if those keys are spread across the key space, that one file's
key range spans the key space too, and no later flush can prune it. Clustering the table
therefore only helps for as long as it lasts: every file appended since the last
OPTIMIZEis read by every flush. Measured on a table grown one flush at a time with scattered updates, pruning skipped no files at all and each flush read every file in the table. How oftenOPTIMIZEruns is what bounds that, which is a second reason to run it on a schedule. - A
FLOATorDOUBLEkey column. Parquet and Delta leaveNaNout of min/max statistics, so a range test could skip a file that holds the key. Range pruning is therefore switched off for the whole key, and every file is read. Floating point makes a poor key in any case; use an exact type. - A null key value. Nulls are left out of min/max statistics too, so a null key sits below the recorded range of every file. Range pruning is therefore switched off for any flush that changes a null-keyed row, and every file is read. Only that flush is affected; a nullable key column that holds no nulls costs nothing.
Statistics only exist for the columns the table asks to index, and the connector collects
exactly the set the table declares: delta.dataSkippingStatsColumns when it names one, and
otherwise the first delta.dataSkippingNumIndexedCols columns, 32 by default. The count is
over leaf columns, so a nested column early in the schema uses up several positions.
A key column outside that set has no statistics, so its files are always read. The connector says so at startup, naming the property to set:
ALTER TABLE <table> SET TBLPROPERTIES ('delta.dataSkippingStatsColumns' = '<key columns>')
It does not set the property itself. What a table collects statistics for is the table owner's decision, and a connector that widened it silently would write files that disagree with the table's own declaration.
Metrics
The connector exports these alongside the standard connector metrics.
| Metric | Meaning |
|---|---|
output_connector_delta_merge_tombstone_ratio_permille | Superseded rows per thousand rows in the table. The signal for when to compact |
output_connector_delta_merge_rows_appended_total, output_connector_delta_merge_rows_superseded_total | Change volume, split by side |
output_connector_delta_merge_keys_probed_total | Changed keys whose row had to be located. Below the changed-key count by whatever the insert shortcut saved |
output_connector_delta_merge_keys_not_found_total | Changed keys the lookup did not find. This includes inserts of new keys, which the connector looks up whenever the table held data when it started, so a nonzero rate is normal |
output_connector_delta_merge_probe_files_scanned_total, output_connector_delta_merge_probe_files_pruned_total | Whether file pruning is doing anything |
output_connector_delta_merge_probe_row_groups_scanned_total, output_connector_delta_merge_probe_row_groups_pruned_total | The same, within the files that are opened |
output_connector_delta_merge_files_appended_total, output_connector_delta_merge_files_dropped_total | Small-file growth, and the files reclaimed because every row in them was superseded |
output_connector_delta_merge_lookup_passes_total | Above one per flush only when a key set exceeded its share of lookup_chunk_size |
output_connector_delta_merge_bytes_written_total | Bytes written: new data files plus deletion vectors |
output_connector_delta_merge_ranges_walked_total | Key ranges walked, summed over flushes. Above the flush count when threads split a batch into ranges written in parallel. |
output_connector_delta_merge_probe_key_bytes_read_total | Key-column bytes the lookup read. Against bytes_written_total, the flush's read amplification |
output_connector_delta_merge_compactions_total, output_connector_delta_merge_compaction_failures_total | Maintenance steps, and how many failed. Only when auto_compact is set |
output_connector_delta_merge_compaction_wait_microseconds_total | Time flushes waited for a maintenance step to finish before they started. Not part of the flush latency. Only when auto_compact is set |
output_connector_delta_merge_flush_latency_microseconds | Histogram of whole flushes. Where object-store latency shows up; no combination of the counters above reveals it |
output_connector_delta_merge_probe_microseconds_total, ..._append_..., ..._deletion_vector_..., ..._commit_..., ..._log_scan_..., ..._batch_walk_... | Where flush time goes. Only successful flushes report them. At threads 1 they partition a flush and sum to what the latency histogram records for it; above 1 the phases a key range performs (batch_walk, probe and append) are summed across ranges that ran at once, so they measure work rather than elapsed time and can exceed the flush's latency |
output_connector_delta_merge_reclaimed_rows_total | Superseded rows that maintenance removed from storage |
output_connector_delta_merge_reclaim_incomplete | 1 while files a step can rewrite are still mostly superseded after the last step. A step rewrites one, so sustained, reclamation is falling behind the writes |
These counters start from zero each time the pipeline starts.
Compaction is required
Every update adds a row and marks one deleted. Without compaction the table grows without bound, and read cost grows with the number of updates rather than the number of live rows.
The connector warns, at most once an hour, when at least 20% of the table's rows are superseded versions.
If Feldera is the table's only writer and nothing else will compact it, set
auto_compact. After each commit the connector then starts one small compaction step in
the background; a flush that starts before the step ends waits for it. A step does one of:
- Pack a partition's small files, each under half the target file size (the table's
delta.targetFileSize, or 100 MiB), into one, once they fill one target file or 50 of them fall in the same size band (between two consecutive powers of 50). At most one target file and 100 files. - Otherwise, or after a pack, rewrite one file whose rows are at least half superseded.
Files over two target files and over 200 MiB are left to
REORG TABLE ... APPLY (PURGE), which the warning above then asks for, because rewriting one would hold up the next flush. The connector itself writes no file that big, so such files come from other writers, or from a larger target file size set earlier.
A step that fails, or runs over 10 minutes and is stopped, pauses compaction for 10
minutes. A step replaces files rather than
deleting them, so VACUUM still reclaims the space, as described below. A table that gains
small files faster than one pack per commit, or a pipeline whose output arrives
continuously and needs every flush equally fast, should leave auto_compact off and
compact the table from Spark or Databricks instead.
Do not set auto_compact when anything else compacts the table, for example
Databricks auto compaction or predictive optimization. The two would rewrite the same
files, and an engine whose conflict check misses that (delta-rs, issue #4851) can leave
the rows of such a file in the table twice.
Otherwise, maintain the table yourself. OPTIMIZE alone is not enough: it plans on file
size, bin-packing files below the target size, so a file that has reached that size is never
rewritten however many of its rows are dead. Run REORG TABLE <table> APPLY (PURGE)
alongside it, which is the Delta command that rewrites files to drop the rows their deletion
vectors mark deleted.
A file whose every row has been superseded is dropped outright rather than kept behind a full
deletion vector, so a delete-heavy workload reclaims some space without OPTIMIZE, but an
update-heavy one does not. Writing to the table never rewrites a data file, changes the
schema, changes the partitioning, alters a table property, or upgrades the protocol, so it
does not conflict with that maintenance: if a compaction replaces a file mid-flush, the
connector notices and redoes the flush against the new files.
Every flush that supersedes rows writes one deletion_vector_<uuid>.bin at the table root,
and leaves whichever vector file the data files it touched pointed at before unreferenced. So
VACUUM is required maintenance here, alongside OPTIMIZE. It is also what reclaims a data
file the connector drops whole, so a merge-mode table needs it either way.
Use Delta Spark's VACUUM if you can. delta-rs, and so the deltalake Python package, does
not account for vector files: its default mode never deletes them, so they accumulate one per
flush, and its full mode deletes ones that are still in use, which brings back every row they
marked deleted. Never pass full=True to it.
Flush cost, and how to size it
A flush reads the table's file list. Then, for every file whose key range can hold a changed
key, it reads the file's footer and the key column of the row groups that can hold one. When
the changed keys are spread over the whole key range, every file can hold one, so a flush of
ten keys costs about as much as a flush of ten thousand: about two object-store requests per
data file. threads above 1 splits the lookup into key ranges, and each range checks every
file, so on such a table it multiplies those requests. Two things keep the cost down, and
both matter more than the pipeline's row rate:
| Lever | Why it moves the cost |
|---|---|
| Output buffering | Merge mode writes at least one data file per flush, and the pass over the file list is linear in the file count. Buffering cuts the number of flushes, which cuts both terms at once |
OPTIMIZE | Fewer, larger files. The lookup reads one row group of one column, so a file 100x larger costs about 3x more to search, far less than searching 100 files |
Output buffering is a requirement of merge mode rather than a tuning knob. Measured on a local table, an unbuffered flush of 16 keys cost 4.5 ms per record where a buffered flush of 1000 keys cost 0.13 ms per record, and the buffered pipeline also grew the file count 60x more slowly. See output buffer.
max_concurrent_probes defaults to 16, which is the setting that matters most on a table of
many files: measured against 800 of them, raising it from 1 to 4 cut the lookup to a third,
and to 16 cut it by a further half. Raise it further if the pipeline has more cores than that
to spend, or if the object store's latency leaves them idle; each concurrent read holds one
file's projected key column, so it costs memory in proportion.
Limits
Merge mode is not exactly-once, and the connector is still not fault tolerant. After a restart the connector replays whatever the pipeline replays, and because a Feldera change states the final value or a deletion per key, re-applying one converges: it marks the row it previously wrote as deleted and appends an identical one. The cost is one extra superseded row per replayed row, which the next compaction reclaims. While the replay runs, readers can see older values.
The table stops matching the view in these cases:
| Case | What happens | What to do |
|---|---|---|
A batch still fails after max_retries | The batch is dropped and the connector reports an error | Leave max_retries unset, or rebuild the table (see below) |
| The connector is paused | The changes made while it is paused are dropped | Rebuild the table after you resume it |
The index columns are not unique | A key with more than one value in a step is skipped and reported as a delta_uniqueness_violation error. The table keeps that key's previous row | Index on a key that is unique |
| You change the view's query | The pipeline sends the view's new contents but no deletes for rows that left it, so those rows stay | Rebuild the table |
To rebuild the table, start the connector with mode: truncate and send_snapshot: true.
With mode: truncate, every start that is not a resume of the same configuration re-creates
the table. That includes a resume after you change any of this connector's settings. Re-creating
the table resets its partitioning and properties and turns on deletion vectors, and without
send_snapshot: true the table then holds only the keys that change after the start.
Data type mapping
See source connector documentation for DeltaLake to Feldera SQL type mapping.
VARIANT
By default, the connector writes a VARIANT column as the Delta variant type,
holding the Parquet variant binary encoding. Alternatively, the user can configure
the connector to write VARIANT columns as JSON-encoded string columns by
setting "variant_encoding": "json_string":
{
"transport": {
"name": "delta_table_output",
"config": {
"uri": "s3://feldera-fraud-detection-demo/feature_train",
"variant_encoding": "json_string"
}
}
}
The variant encoding enables the variantType table feature on tables the
connector creates. Under the variant encoding, three SQL types have no counterpart
and are rejected: long and short INTERVAL, and GEOMETRY. Unsigned
integers widen to the smallest signed type that holds them, and a BIGINT UNSIGNED too large for BIGINT becomes a decimal. SQL NULL and a VARIANT
null share one encoding.
Appending to a table whose VARIANT column is already a string requires
variant_encoding: json_string. The connector compares the two when it opens
the table and refuses to start on a mismatch, naming the column and the setting
to change.
The small file problem and output buffer configuration
By default a Feldera pipeline sends a batch of changes to the output transport for each batch of input updates it processes. This can result in a stream of small updates, which is normal and even preferable for output transports like Kafka; however it can cause problems for the Delta Lake format by creating a large number of small files.
The output buffer mechanism is designed to solve this problem by decoupling the rate at which the pipeline pushes changes to the output transport from the rate of input changes. It works by accumulating updates inside the pipeline for up to a user-defined period of time or until accumulating a user-defined number of updates and writing them to the Delta Table as a small number of large files.
See output buffer for details on configuring the output buffer mechanism.
Example usage
Streaming incremental updates
Create a Delta Lake output connector that writes a stream of updates to a table stored in an S3 bucket, truncating any existing contents of the table.
CREATE VIEW v
WITH (
'connectors' = '[{
"transport": {
"name": "delta_table_output",
"config": {
"uri": "s3://feldera-fraud-detection-demo/feature_train",
"mode": "truncate",
"aws_access_key_id": <AWS_ACCESS_KEY_ID>,
"aws_secret_access_key": <AWS_SECRET_ACCESS_KEY>,
"aws_region": "us-east-1"
}
},
"enable_output_buffer": true,
"max_output_buffer_time": "10s"
}]'
)
AS SELECT * FROM my_table;
Sending a snapshot at startup
Set send_snapshot: true to have the connector emit a full snapshot of a
materialized view to the Delta table before streaming incremental updates. This
is useful when downstream consumers need the complete current state of the
view, not just changes since the connector was created.
The snapshot is sent exactly once per connector lifetime. Resuming the
pipeline from a checkpoint does not re-send it. Modifying the connector
triggers a fresh snapshot on the next start, and is delivered even if the
pipeline is started or resumed in Paused state.
CREATE MATERIALIZED VIEW v
WITH (
'connectors' = '[{
"name": "delta_sink",
"send_snapshot": true,
"transport": {
"name": "delta_table_output",
"config": {
"uri": "s3://my-bucket/my-table",
"mode": "truncate",
"aws_access_key_id": <AWS_ACCESS_KEY_ID>,
"aws_secret_access_key": <AWS_SECRET_ACCESS_KEY>,
"aws_region": "us-east-1"
}
},
"enable_output_buffer": true,
"max_output_buffer_time": "10s"
}]'
)
AS SELECT * FROM my_table;
Writing in parallel
Set threads above 1 to have the connector write Parquet files concurrently. Parallel writes require the view to have a
unique key, so that the connector can order inserts and deletes correctly; without one the pipeline fails to start.
Define the key with CREATE INDEX and set the connector's index property to that index. See
views with unique keys for additional details.
Each thread writes its own files, so both the memory a write needs and the uploads it has in flight scale with
threads: a thread holds the Parquet file it is building, and the buffers that feed it, until the file is
uploaded. On a twenty-million-row backfill of fifty columns, threads 8 peaked between 2.5 GB and 3.8 GB above
threads 1 across runs. The spread is wide because peak memory also depends on what the pipeline itself is
holding, so treat this as a few hundred megabytes per added thread and measure your own workload rather than
taking the figure. Raise the pipeline's memory limit along with threads, or a thread count that writes faster
will run out of memory instead.
On a constrained network path a high thread count can saturate the link before it saturates the writers, which shows up as a slower write rather than a faster one.
Threads speed a backfill up sublinearly, because a thread spends most of its time serializing rows and the
threads compete for the same cores. On a twelve-core machine, a ten-million-row backfill of fifty columns flushed
in 65.4 s at threads 1, 24.2 s at 4 (2.7x) and 18.1 s at 8 (3.6x), each the mean of three runs. The CPU the
flush spends rises with it (15% above threads 1 at 4, and 52% at 8), so the gain costs cores and memory that
the rest of the pipeline could use. Raise threads a step at a time and keep the setting that stops improving
throughput; the point where it stops depends on how many cores the pipeline itself needs.
In merge mode a thread also needs enough rows to be worth a file of its own, so the connector splits a batch only
when each thread would get a substantial share of it. A steady stream of small transactions therefore keeps writing
one file per flush whatever threads says, and only a large change, typically a backfill, is split.
A backfill is the one flush the connector cannot size that way, because no flush has yet measured what a row costs,
so it splits into as many ranges as threads allows. Each range ends on a partially filled file, so a high setting
leaves the table with more files than it needs: the 2.2 GB backfill above fills 21 files, and wrote 24 at threads
8 but 32 at 16. Those files cost every later lookup, so raise threads for the speed a backfill needs and not
beyond it. A rough ceiling is the backfill's size divided by 300 MB, which gives each range three full files.
max_concurrent_probes and lookup_chunk_size are budgets for a flush rather than for a thread: the threads of
one flush divide them between them, so raising threads multiplies neither the requests the lookup has in flight
nor the keys it holds.
CREATE VIEW v
WITH (
'connectors' = '[{
"index": "v_idx",
"transport": {
"name": "delta_table_output",
"config": {
"uri": "s3://feldera-fraud-detection-demo/feature_train",
"threads": 4,
"mode": "append",
"aws_access_key_id": <AWS_ACCESS_KEY_ID>,
"aws_secret_access_key": <AWS_SECRET_ACCESS_KEY>,
"aws_region": "us-east-1"
}
},
"enable_output_buffer": true,
"max_output_buffer_time": "10s"
}]'
)
AS SELECT * FROM my_table;
-- Unique key required by "threads": 4.
CREATE INDEX v_idx ON v(id);
Note where each option goes: threads is a transport option, while index is a connector option.