Sorted Dynamic Table Sink in YTsaurus Flow

Write modes
- Synchronous sink parameters
- Asynchronous sink parameters

Use this connector to write data to sorted dynamic tables in YTsaurus.

Attention

In most cases, use StateManager to work with sorted dynamic tables, not this connector.

Key differences:

StateManager supports read-modify-write operations, row deletion based on non-trivial rules, and guarantees a single modification point for each row. The group_by_schema computation must match (or be a prefix of) the state’s key columns.
This connector is a low-level primitive that supports only write-only operations (inserting or deleting rows). There are no restrictions on group_by_schema, which lets you write to the table from multiple computations with different keys. However, the order of modifications to a single row isn’t controlled.

You can find the connector code here.

Write modes

The sink accepts messages and writes their rows to a sorted dynamic table. The message schema must be compatible with the target table’s schema.

The synchronous sink (NYT::NFlow::NSortedDynamicTable::TSyncSink) writes rows within the main epoch transaction. The write is atomic together with the other epoch changes, but the table must reside on the main processing cluster — the same cluster as the pipeline.

The asynchronous sink (NYT::NFlow::NSortedDynamicTable::TAsyncSink) writes each batch in a separate tablet transaction. The target table may reside on another cluster specified in table_path. This write is not atomic with the main epoch transaction: the sink acknowledges messages after the separate transaction commits successfully.

The asynchronous sink reports the latest YT write error in /async_write and retries every error except Canceled until the write succeeds or the job is cancelled. Retry delays use exponential backoff with jitter. The delay before jitter starts at backoff_duration and is capped at one minute (or at backoff_duration when it is greater). Pending messages remain accounted for in the output buffers, so the regular output backpressure applies while the sink is retrying.

Within one sink instance, batches are written sequentially. This ordering does not extend across job restarts or partition handoffs: a tablet transaction already started by the old job may commit after the replacement job writes the replayed batch and subsequent batches. As a result, an older modification may overwrite a newer modification of the same key. Use the asynchronous sink only when both replay and this reordering are safe, for example when every key is always written with the same value or the logical version is part of the key. Aggregate writes are unsupported.

The number of receiver channels automatically adjusts to match the number of tablets in the target table. For a chaos replicated table, the controller reads the tablet count from the enabled data replica with the lexicographically smallest replica id; its sync/async mode does not affect the choice. Resharding that replica may change the number of receiver channels.

Synchronous sink parameters

Sink class: NYT::NFlow::NSortedDynamicTable::TSyncSink.

Static spec:

Source: yt/yt/flow/library/cpp/common/registry-inl.h

Parameter	Description
`table_path`	Type: `NYT::NYPath::TRichYPath` Required parameter Path to the target sorted dynamic table with the cluster specified.
`column_filter`	Type: `std::optional<THashSet<std::string>>` Which columns from the message to write to the table (default — all). When using the `delete_rows` parameter, you must specify the key columns of the target table, otherwise deletions will fail.
`aggregate_columns`	Type: `std::optional<THashSet<std::string>>` List of columns for which the aggregation function is considered during writing. By default, the value in the column is overwritten.
`delete_rows`	Type: `bool` Default value: `false` If set to `true`, rows will be deleted instead of inserted. In this mode, it is important that the key columns of the target table are selected in `column_filter`. Note A Sink can work in either insert or delete mode, but not simultaneously. If you need both insert and delete functionality, use StateManager.
`require_sync_replica`	Type: `bool` Default value: `true` If set to `false`, the presence of a synchronous replica is not checked when writing rows to the target table. The default value is `true` — the check is performed.

Additional parameters

update_partition_count_period

Type: TDuration
Default value: 1m
How often the controller should update the number of tablets in the table (the controller changes the number of receiver channels accordingly). For a chaos replicated table, the controller uses the enabled data replica with the lexicographically smallest replica id; its current sync/async mode does not affect the choice. Resharding that replica may change the number of receiver channels.

Dynamic spec:

Source: yt/yt/flow/library/cpp/common/registry-inl.h

This structure has no main parameters.

Asynchronous sink parameters

Sink class: NYT::NFlow::NSortedDynamicTable::TAsyncSink.

Static spec:

Source: yt/yt/flow/library/cpp/common/registry-inl.h

Parameter	Description
`table_path`	Type: `NYT::NYPath::TRichYPath` Required parameter Path to the target sorted dynamic table with the cluster specified.
`column_filter`	Type: `std::optional<THashSet<std::string>>` Which columns from the message to write to the table (default — all). When using the `delete_rows` parameter, you must specify the key columns of the target table, otherwise deletions will fail.
`aggregate_columns`	Type: `std::optional<THashSet<std::string>>` Aggregate writes are unsupported: omit this parameter or provide an empty list. An asynchronous write may replay an already committed batch after recovery, which makes non-idempotent aggregation unsafe.
`delete_rows`	Type: `bool` Default value: `false` If set to `true`, rows will be deleted instead of inserted. In this mode, make sure that `column_filter` selects the key columns of the target table.
`require_sync_replica`	Type: `bool` Default value: `true` If set to `false`, the presence of a synchronous replica is not checked when writing rows to the target table. The default value is `true`.

Additional parameters

update_partition_count_period

Dynamic spec:

Source: yt/yt/flow/library/cpp/common/registry-inl.h

Parameter	Description
`max_rows_per_batch`	Type: `long` Default value: `1000` Maximum number of Flow messages in a batch: the batching sink flushes the accumulated batch when it reaches this threshold. The valid range is 1 through 2000.
`max_bytes_per_batch`	Type: `long` Default value: `5242880` Maximum total size of Flow messages in a batch: the batching sink flushes the accumulated batch when it reaches this threshold. The valid range is 1 byte through 10 MiB.

Additional parameters

backoff_duration

Type: TDuration
Default value: 3s
Initial retry delay before jitter. Retry delays use exponential backoff with jitter; the delay before jitter is capped at one minute (or at this value when it is greater). The sink keeps retrying until the write succeeds or the job is cancelled, and reports the latest error in its status.

Static table

ServiceLog