---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/state.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/state.md
  - href: https://ytsaurus.tech/docs/ru/flow/cpp/state.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: ru/_includes/flow/cpp/state.md -->
# Работа со стейтами в YTsaurus Flow (C++)

{% note info %}

На этой странице описаны особенности работы со стейтами на C++. Языконезависимое описание концепции см. в разделе [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md).

{% endnote %}

Все примеры пользовательской логики на этой странице используют [process function](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md). Не создавайте для работы со стейтом собственный класс-наследник `Computation`.

## Internal State {#internal-state}

Простейший способ хранить стейт внутри `Computation`. Данные автоматически подгружаются в начале [эпохи](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#epoch) и записываются при коммите. Не требует самостоятельного создания таблиц — Flow управляет ими автоматически.

### Использование

Для работы с Internal State необходимо:

В качестве типа стейта можно использовать **любой тип**, для которого определены функции сериализации и десериализации. По умолчанию поддерживаются все типы, для которых определены стандартные функции `Serialize`/`Deserialize` в YT (в том числе `TYsonStruct`-наследники); для произвольного типа достаточно предоставить их перегрузки.

По умолчанию стейт считается пустым, если он равен значению по умолчанию (`TMyState{}`), а очистка пересоздаёт его. Чтобы переопределить это поведение, тип стейта может унаследовать опциональные миксины из `NYT::NFlow`:

- `ICustomStateOps` — собственная логика `Clear()` и `IsEmpty()` (задаются парой);
- `ICustomYsonView` — собственное представление `ToYsonView()` для read-state-интроспекции.

Необходимо:

1. Завести поле `TMutableStateKeyClient<TMyState> MyStateClient_` в своей process function.
2. Переопределить `Init(const IRuntimeInitContextPtr& initContext)` и вызвать в нём `initContext->InitClient<TMyState>(MyStateClient_, "my_state")`. В качестве имени стоит взять уникальную в рамках компьютейшена строку.
3. Для получения стейта по ключу использовать `MyStateClient_.GetState(message->Key)`. Возвращаемый аксессор `TStateAccessor<TMyState>` ведёт себя как умный указатель на `TMyState` (`state->...`, `*state`) и действителен только в пределах текущей эпохи — сохранять его в полях нельзя.
4. Для очистки стейта (удаления строки из таблицы) вызвать `state.Clear()` на аксессоре.

Если стейт пуст, соответствующая строчка будет удалена.

Пример:

```cpp
struct TMyState
    : public NYTree::TYsonStruct
{
    std::optional<ui64> SomeValue;

    REGISTER_YSON_STRUCT(TMyState);

    static void Register(TRegistrar registrar)
    {
        registrar.Parameter("some_value", &TThis::SomeValue)
            .Default();
    }
};

class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient(MyStateClient_, "my_state");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = MyStateClient_.GetState(message->Key);
        state->SomeValue = 42;
        // ...
    }

    void ProcessTimer(
        const TInputTimerConstPtr& timer,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = MyStateClient_.GetState(timer->Key);
        // Очистить стейт (удалить строку из таблицы):
        state.Clear();
    }

private:
    TMutableStateKeyClient<TMyState> MyStateClient_;
};
```

### Сжатие

Internal State поддерживает сжатие данных. Оно настраивается в [DynamicSpec](https://ytsaurus.tech/docs/ru/flow/concepts/spec.md) для каждого стейта по отдельности — в секции `state_manager` компьютейшена, в `overrides/<имя_стейта>/format`:

- `compress` — включить сжатие (по умолчанию `false`);
- `recode_probability` — вероятность перекодировать стейт в заданный формат при очередной обработке; обеспечивает постепенную миграцию стейтов после смены формата (по умолчанию `0.1`).

## Yson State Reader

{% note warning %}

Функциональность ещё не реализована.

{% endnote %}

## External State {#external-state}

External State — стейт, хранимый в пользовательской динамической таблице YTsaurus. В отличие от Internal State, таблицы создаются и управляются пользователем.

External state manager объявляется в спеке `Computation` под уникальным именем в секции `external_state_managers` и подключается к `Computation` через типизированный клиент `TMutableStateKeyClient<TState>`. Имена в спеке должны начинаться с `/` (например, `"/state"`) и совпадают с именем, передаваемым в `InitExternalStateClient`. Реализация ищется в реестре по `external_state_manager_class_name`.

### Использование

Для работы с External State необходимо:

1. Завести поле `TMutableStateKeyClient<TState> StateClient_` в своей process function, где `TState` — тип стейта, который возвращает соответствующий external state manager (см. ниже про конкретные реализации).
2. Переопределить `Init(const IRuntimeInitContextPtr& initContext)` и вызвать в нём `initContext->InitExternalStateClient(StateClient_, "/state")`. В качестве имени стоит взять уникальную в рамках компьютейшена строку — это же имя должно фигурировать в спеке.
3. Для получения стейта по ключу использовать `StateClient_.GetState(message->Key)`. Возвращаемый аксессор `TStateAccessor<TState>` ведёт себя как умный указатель на `TState` и действителен только в пределах текущей эпохи.

Пример:

```cpp
class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitExternalStateClient(StateClient_, "/state");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = StateClient_.GetState(message->Key);
        i64 count = state->GetColumnValue<std::optional<i64>>("count").value_or(0);
        TPayloadBuilder builder(state->Schema);
        builder.Set(count + 1, "count");
        state->Payload = builder.Finish();
    }

private:
    TMutableStateKeyClient<TSimpleExternalState> StateClient_;
};
```

Спека `Computation` с подключённым `TSimpleExternalStateManager`:

```yson
"external_state_managers" = {
    "/state" = {
        "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
        "parameters" = {
            "path" = "//path/to/state/table";
        };
    };
};
```

### Предзагрузка и `auto_preload` {#external-state-preload}

По умолчанию (`auto_preload = %true`) фреймворк перед обработкой эпохи подгружает в каждый external state manager все ключи сообщений, таймеров и визитов — по одному lookup'у на ключ. При `"auto_preload" = %false;` process function подгружает только нужные ключи через клиент: `PreloadKeyStates(THashSet<TKey>)`, `PreloadKeyStates(IInputContextPtr)` или `PreloadKeyStates(IInputContextPtr, TExtractKeysOptions)`; опции выбирают, ключи каких сущностей брать, например `{.Visits = false}` для стейта, который обработчик визитов не читает. Предзагрузка инкрементальна: её можно вызывать несколько раз за эпоху, подгружаются только ещё не загруженные ключи, а уже загруженный (возможно, изменённый) стейт остаётся как есть. `GetState` по ключу, не подгруженному в эпохе, бросает исключение. `auto_preload = %false` запрещён для companion-вычислений.

В `IBatchProcessFunction::Process` сущности можно раздать через хелперы из раздела [Process functions](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md):

```cpp
void Process(const IInputContextPtr& input, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context) override
{
    if (!context->GetSpec()->ExternalStateManagers.at("/state")->AutoPreload) {
        WaitFor(StateClient_.PreloadKeyStates(input, {.Visits = false})).ThrowOnError();
    }
    ProcessMessages(input, output, context, BIND(&TMyFunction::ProcessMessage, MakeStrong(this)));
    ProcessVisits(input, output, context, BIND(&TMyFunction::ProcessVisit, MakeStrong(this)));
}
```

### Удаление без чтения: `EraseState` {#external-state-erase}

`StateClient_.EraseState(key)` ставит строку на удаление, не подгружая и не читая её — key visitor, вычищающий строки по колонке ключа (например, timestamp), не платит за них ни одного lookup'а. Erase терминален для эпохи: после него `GetState` бросает исключение, а `PreloadKeyStates` не возвращает ключ обратно — независимо от того, был ли он загружен. `TSimpleExternalStateManager`, `TProfileStateManager` и внутренний per-key стейт удаляют только по ключу; пользовательский менеджер без переопределения `EraseKeyState` отвергает вызов.

```cpp
for (const auto& visit : input->GetVisits()) {
    if (GetColumnValue<ui64>(visit->Key, "ts") + Week < now) {
        StateClient_.EraseState(visit->Key);   // Без lookup'а.
    }
}
```

### TSimpleExternalStateManager

`TSimpleExternalStateManager` — стандартная реализация external state manager'а. Работает с одной динамической таблицей, ключи которой совпадают с `group_by_schema`. `GetState` отдаёт аксессор поверх `TSimpleExternalState` с полями `Payload` и `Schema`; колонки достаются и записываются через `GetColumn[Value]<T>` / `TPayloadBuilder` по имени или индексу. Кэширование стейтов происходит автоматически через общий [StateCache](#state-cache).

Регистрировать `TSimpleExternalStateManager` в `register.cpp` не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TSimpleExternalStateManagerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/simple_external_state_manager.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/simple_external_state_manager.h)

#|
|| **Параметр** | **Описание** ||
|| `path` | **Тип**: `NYT::NYPath::TRichYPath`
**Обязательный параметр**
Путь до динтаблицы со стейтами (с указанием кластера в rich path). ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TSimpleExternalStateManagerSpec.md -->

Динамическая спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TDynamicSimpleExternalStateManagerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/simple_external_state_manager.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/simple_external_state_manager.h)

**В структуре нет основных параметров.**
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TDynamicSimpleExternalStateManagerSpec.md -->

Полный пример использования `TSimpleExternalStateManager` см. в разборе примера [word_count](https://ytsaurus.tech/docs/ru/flow/cpp/examples/word_count.md).


## External State Joiner {#external-state-joiner}

External State Joiner предоставляет доступ только на чтение к внешним стейтам через join по ключу. Загруженные стейты кэшируются в общем StateCache с TTL: повторные обращения к тому же ключу до истечения TTL обслуживаются из кэша без обращения к YT.

External state joiner объявляется в спеке `Computation` под уникальным именем в секции `external_state_joiners` и подключается к `Computation` через типизированный клиент `TJoinedStateKeyClient<TState>`. Имена в спеке должны начинаться с `/` (например, `"/reference"`) и совпадать с именем, передаваемым в `InitExternalStateClient`. Реализация ищется в реестре по `external_state_joiner_class_name`.

### Использование

Аналогично external state manager'у:

```cpp
class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitExternalStateClient(StateReaderClient_, "/reference");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = StateReaderClient_.GetState(message->Key);
        // state — read-only аксессор TConstStateAccessor<TSimpleExternalState> (действителен в пределах эпохи).
    }

private:
    TJoinedStateKeyClient<TSimpleExternalState> StateReaderClient_;
};
```

Спека `Computation` с подключённым `TSimpleExternalStateJoiner`:

```yson
"external_state_joiners" = {
    "/reference" = {
        "external_state_joiner_class_name" = "NYT::NFlow::TSimpleExternalStateJoiner";
        "parameters" = {
            "path" = "//path/to/reference/table";
        };
    };
};
```

### TSimpleExternalStateJoiner {#simple-external-state-joiner}

`TSimpleExternalStateJoiner` — стандартная реализация external state joiner'а. По интерфейсу аналогична `TSimpleExternalStateManager`: читает строки одной динамической таблицы, ключи которой совпадают с `group_by_schema`. `GetState` отдаёт read-only аксессор поверх `TSimpleExternalState`, который содержит `TPayload` и схему value-колонок таблицы; колонки достаются через `GetColumn[Value]<T>` по имени или индексу.

Регистрировать `TSimpleExternalStateJoiner` в `register.cpp` не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TSimpleExternalStateJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/simple_external_state_manager.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/simple_external_state_manager.h)

#|
|| **Параметр** | **Описание** ||
|| `path` | **Тип**: `NYT::NYPath::TRichYPath`
**Обязательный параметр**
Путь до динтаблицы со стейтами (с указанием кластера в rich path). ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TSimpleExternalStateJoinerSpec.md -->

Динамическая спека (TTL кэша задаётся через секцию `cache`):

<!-- source: ru/flow/generated_docs/NYT_NFlow_TDynamicSimpleExternalStateJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/simple_external_state_manager.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/simple_external_state_manager.h)

#|
|| **Параметр** | **Описание** ||
|| `cache` | **Тип**: `NYT::TIntrusivePtr<`[NYT::NFlow::TDynamicExpiringJobNamedStateCacheSpec](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TDynamicExpiringJobNamedStateCacheSpec)`>`
**Значение по умолчанию**: `{}`
Настройки TTL-кэша загруженных стейтов. ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TDynamicSimpleExternalStateJoinerSpec.md -->

### TStaticTableKeyVisitorJoiner {#static-table-key-visitor-joiner}

`TStaticTableKeyVisitorJoiner` — external state joiner поверх **статической** сортированной таблицы. Работает только в паре с [key-visitor-стримом](https://ytsaurus.tech/docs/ru/flow/concepts/key_visitor.md#static-table-joiner): в отличие от `TSimpleExternalStateJoiner`, он не делает случайных чтений — фоновый обход визитора последовательно читает таблицу теми же диапазонами ключей, что и внутренний стейт, и строка таблицы становится доступна в `ProcessVisit` как read-only стейт ключа визита. Ключ, которого в таблице нет, отдаёт пустой стейт (`IsEmpty() == true`).

Джойнер привязывается к visit-стриму перечислением его имени в `external_names` секции `key_visitor_streams` (см. [параметры статической спеки](https://ytsaurus.tech/docs/ru/flow/concepts/key_visitor.md#static-params)); один джойнер можно привязать не более чем к одному visit-стриму (проверяется при submit'е спеки). `join_on/key_schema_override` не поддерживается — ключом всегда служит собственная `group_by_schema`.

Требования к таблице-источнику:

- статическая — динамическая таблица отвергается при чтении схемы;
- отсортирована так, что префикс её ключевых колонок совпадает с `group_by_schema` `Computation`'а по именам и типам. Таблица может лежать на другом кластере (`<cluster=...>` в rich path).

{% note warning %}

Выражения вычисляемых колонок не сверяются: партиционная колонка источника должна быть **материализована** значениями, совпадающими с выражением из `group_by_schema` (обычно `farm_hash(key)`). Таблицу, материализованную иначе, фреймворк не отличит от корректной — каждый ключ визита молча резолвится как отсутствующий.

{% endnote %}

Канонический паттерн — зеркалирование (reconciliation): ключи таблицы участвуют в обходе наравне с ключами собственного стейта, поэтому визит приходит и для ключа, которого в стейте ещё нет (его надо создать), и для ключа, которого уже нет в таблице (его надо удалить). Периодический обход таким образом сводит стейт `Computation`'а к внешней таблице:

```cpp
class TMirrorFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient<TMirrorState>(MirroredState_, "/mirror");
        initContext->InitExternalStateClient(SourceState_, "/source");
    }

    void ProcessVisit(
        const TInputVisitConstPtr& visit,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto src = SourceState_.GetState(visit->Key);
        if (!src.IsInitialized()) {
            // Диапазон источника не прочитан (mark_unreadable): состояние ключа неизвестно.
            return;
        }
        auto mirror = MirroredState_.GetState(visit->Key);
        if (src.IsEmpty()) {
            mirror.Clear();
        } else {
            mirror->Payload = src->GetColumnValue<std::optional<std::string>>("payload").value_or(std::string{});
        }
    }

private:
    TMutableStateKeyClient<TMirrorState> MirroredState_;
    TJoinedStateKeyClient<TSimpleExternalState> SourceState_;
};
```

Спека `Computation` для этого примера:

```yson
"external_state_joiners" = {
    "/source" = {
        "external_state_joiner_class_name" = "NYT::NFlow::TStaticTableKeyVisitorJoiner";
        "parameters" = {
            "path" = "//path/to/source";
        };
    };
};
"key_visitor_streams" = {
    "visit_iter" = {
        "names" = ["/mirror"];
        "external_names" = ["/source"];
    };
};
```

Поведение при недоступности источника задаёт `unavailable_source_policy`. Неуспешным считается чтение, исчерпавшее `read_attempts` попыток:

- `retry` (по умолчанию) — ошибка роняет итерацию обхода, чтение повторяется; обход не продвигается за непрочитанный диапазон.
- `mark_unreadable` — ошибка проглатывается, обход идёт дальше, а ключи непрочитанного диапазона резолвятся **неинициализированным** аксессором: `IsInitialized() == false`, разыменование бросает исключение. `Computation` сам решает, что делать с ключом, состояние которого в источнике неизвестно (обычно — пропустить визит, как в примере выше).

После неуспешного чтения источник помечается недоступным на `unavailable_source_backoff`: пока пометка действует, обращений к источнику нет — чтения немедленно резолвятся согласно полиси; первое чтение после истечения окна снова пробует источник. Состояние джойнера видно в сенсорах воркера `static_table_key_visitor_joiner/{source_unavailable,failed_reads,reader_opens,listed_size}` с тегом `external_state_joiner=<имя>`.

Регистрировать `TStaticTableKeyVisitorJoiner` в `register.cpp` не нужно: он уже зарегистрирован в самой библиотеке Flow.

Спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TStaticTableKeyVisitorJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/static_table_key_visitor_joiner.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/static_table_key_visitor_joiner.h)

#|
|| **Параметр** | **Описание** ||
|| `path` | **Тип**: `NYT::NYPath::TRichYPath`
**Обязательный параметр**
Путь до статической сортированной таблицы-источника (с указанием кластера в rich path). Префикс её ключевых колонок должен совпадать с `group_by_schema` computation'а по именам и типам; вычисляемая (партиционная) колонка должна быть материализована значениями своего выражения. ||
|| `unavailable_source_policy` | **Тип**: [NYT::NFlow::EUnavailableSourcePolicy](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NFlow_EUnavailableSourcePolicy)
**Значение по умолчанию**: `retry`
Поведение при неуспешном чтении источника (исчерпавшем `read_attempts` попыток).
`retry` (по умолчанию) — ошибка роняет итерацию обхода и чтение повторяется; обход не продвигается за непрочитанный диапазон.
`mark_unreadable` — ошибка проглатывается, обход идёт дальше, а ключи непрочитанного диапазона резолвятся неинициализированным аксессором (`IsInitialized() == false`). ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TStaticTableKeyVisitorJoinerSpec.md -->

Динамическая спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TDynamicStaticTableKeyVisitorJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/computation/static_table_key_visitor_joiner.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/computation/static_table_key_visitor_joiner.h)

#|
|| **Параметр** | **Описание** ||
|| `read_attempts` | **Тип**: `int`
**Значение по умолчанию**: `3`
Бюджет попыток одного чтения источника; чтение, исчерпавшее бюджет, считается неуспешным и обрабатывается согласно `unavailable_source_policy`. ||
|| `unavailable_source_backoff` | **Тип**: [TDuration](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#TDuration)
**Значение по умолчанию**: `5m`
На какое время неуспешное чтение помечает источник недоступным.
Пока пометка действует, обращений к источнику нет: каждое чтение немедленно резолвится согласно `unavailable_source_policy`;
первое чтение после истечения окна снова пробует источник. ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TDynamicStaticTableKeyVisitorJoinerSpec.md -->


## State Joiner {#state-joiner}

State Joiner предоставляет доступ только на чтение к **внутреннему** стейту другого `Computation`'а: один `Computation` может обогащаться стейтом, который накапливает другой `Computation`, в том числе когда ключ join'а не совпадает с собственной `group_by_schema`.

Joiner объявляется в спеке `Computation` под уникальным именем в секции `state_joiners` и подключается через типизированный клиент `TJoinedStateKeyClient<TState>`. Имена в спеке должны начинаться с `/` и совпадать с именем, передаваемым в `InitClient`. Отдельной реализации или регистрации не требуется — это встроенная возможность Flow.

В спеке указывается:

- `computation_id` — `Computation`, чей внутренний стейт читается;
- `state_name` — имя стейт-клиента целевого `Computation`'а (тот префикс, который он передал в `InitClient`, начинается с `/`);
- `join_on/key_schema_override` — какими колонками текущей строки задаётся ключ. Если не задана, используется собственная `group_by_schema` (join по тому же ключу);
- `join_on/key_provider_streams` — стримы, из сообщений и таймеров которых берутся ключи (`nullopt` — все входные стримы);
- `auto_preload` — если `true` (по умолчанию), фреймворк сам подгружает ключи перед каждым `DoProcess`; иначе `Computation` вызывает `PreloadKeyStates` сам.

{% note warning %}

Типы ключа проверяются на старте: `key_schema_override` (или собственная `group_by_schema`, если override не задан) должна по числу колонок и типам совпадать с `group_by_schema` целевого `Computation`'а — joiner вычисляет полный ключ целевого стейта. Имена колонок и выражения при этом могут отличаться.

На стороне пользователя остаётся (фреймворк это не проверяет):

- *соответствие значений* — какая колонка во что мапится, чтобы вычисленный ключ реально указывал на нужную строку чужого стейта;
- `TState` должен совпадать с типом стейта целевого `Computation`'а: joiner десериализует его строки так же, как `TMutableStateKeyClient<TState>` на стороне-владельце.

{% endnote %}

### Использование

```cpp
class TMyFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitClient(UpstreamClient_, "/upstream");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = UpstreamClient_.GetState(message);
        // state — read-only аксессор TConstStateAccessor<TUpstreamState> (действителен в пределах эпохи).
    }

private:
    TJoinedStateKeyClient<TUpstreamState> UpstreamClient_;
};
```

Спека `Computation` с подключённым joiner'ом по ключу, отличному от собственной `group_by_schema`:

```yson
"state_joiners" = {
    "/upstream" = {
        "computation_id" = "accumulator";
        "state_name" = "/total";
        "join_on" = {
            "key_schema_override" = [
                {name = "Hash"; expression = "farm_hash(UserId)"; type = "uint64"; required = %true;};
                {name = "UserId"; type = "string"; required = %true;};
            ];
        };
    };
};
```

По умолчанию кэш выключен (`ttl = 0`): joiner заново читает стейт целевого `Computation`'а на `SyncLastCommittedTimestamp` каждую эпоху, поэтому всегда видит свежие коммиты. Кэш с TTL включается в dynamic spec через секцию `cache` (загруженные стейты живут до истечения `ttl`, повторные чтения того же ключа обслуживаются без обращения к YT). Для `Swift`-вычислений с требованием строгой детерминированности кэш и чтение на `SyncLastCommittedTimestamp` нужно учитывать.

Спека:

<!-- source: ru/flow/generated_docs/NYT_NFlow_TStateJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/common/spec.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/spec.h)

#|
|| **Параметр** | **Описание** ||
|| `computation_id` | **Тип**: `NYT::NFlow::TStrongIdentifierTypedef<NYT::NFlow::TComputationIdTag>`
**Обязательный параметр**
`Computation`, чей внутренний стейт читается этим joiner'ом. ||
|| `state_name` | **Тип**: `std::string`
**Обязательный параметр**
Имя стейт-клиента целевого `Computation`'а — тот префикс, который он передал в `InitClient` (начинается с `/`). ||
|| `join_on` | **Тип**: `NYT::TIntrusivePtr<`[NYT::NFlow::TStateJoinSpec](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TStateJoinSpec)`>`
**Значение по умолчанию**: `{}`
Описание соответствия между ключами входного потока computation'а и ключом стейта целевого `Computation`'а (см. `TStateJoinSpec`). `key_schema_override` должна совпадать с `group_by_schema` целевого `Computation`'а. ||
|| `auto_preload` | **Тип**: `bool`
**Значение по умолчанию**: `true`
Если `true` (по умолчанию), фреймворк сам вызывает `PreloadKeyStates` у этого joiner'а перед каждым `DoProcess`, формируя ключи по `join_on`. Если `false`, computation сам отвечает за вызов `PreloadKeyStates` до `GetState`. ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TStateJoinerSpec.md -->

Динамическая спека (TTL кэша задаётся через секцию `cache`):

<!-- source: ru/flow/generated_docs/NYT_NFlow_TDynamicStateJoinerSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/common/spec.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/spec.h)

#|
|| **Параметр** | **Описание** ||
|| `cache` | **Тип**: `NYT::TIntrusivePtr<`[NYT::NFlow::TDynamicExpiringJobNamedStateCacheSpec](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NFlow_TDynamicExpiringJobNamedStateCacheSpec)`>`
**Значение по умолчанию**: `{}`
Настройки TTL-кэша загруженных стейтов. По умолчанию выключен (`ttl = 0`). ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TDynamicStateJoinerSpec.md -->

## StateCache {#state-cache}

Flow предоставляет общий двухуровневый (uncompressed + compressed) LRU кэш для стейтов. Конфигурирование осуществляется через `/dynamic_spec/job_tracker/state_cache`. Он используется в первую очередь для стейтов менеджеров; у joiner'ов кэш по умолчанию выключен, а данные грузятся на `SyncLastCommittedTimestamp`.

<!-- source: ru/flow/generated_docs/NYT_NFlow_TDynamicStateCacheSpec.md -->
<!-- This file is generated by yt/yt/flow/yandex/tools/generate_yson_struct_doc/generate.sh script -->
<!-- Before using doc generation tool check readme: yt/yt/flow/yandex/tools/generate_yson_struct_doc/README.md -->
Источник: [yt/yt/flow/library/cpp/common/spec.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/spec.h)

#|
|| **Параметр** | **Описание** ||
|| `compressed_cache_weight` | **Тип**: [NYT::NYTree::TSize](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NYTree_TSize)
**Значение по умолчанию**: `1Gi`
Ограничение на `compressed`. ||
|| `uncompressed_cache_weight` | **Тип**: [NYT::NYTree::TSize](https://ytsaurus.tech/docs/ru/flow/generated_docs/all_yson_structs.md#NYT_NYTree_TSize)
**Значение по умолчанию**: `100Mi`
Ограничение на `uncompressed`. ||
|#
<!-- endsource: ru/flow/generated_docs/NYT_NFlow_TDynamicStateCacheSpec.md -->

Подробнее о согласованности `group_by_schema` и общих принципах работы со стейтами см. [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md).

## См. также

- [Stateful processing (концепция)](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md)
- [Computation (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/computation.md)
- [Быстрый старт (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.md)
<!-- endsource: ru/_includes/flow/cpp/state.md -->