Import libs 1 (#590)

* Import libs 1

* Add new file without extension

* Add file missed in export config

46 files changed, 716 insertions, 270 deletions

diff --git a/yt/yql/CMakeLists.txt b/yt/yql/CMakeLists.txt
deleted file mode 100644
index 8180a508fa..0000000000
--- a/yt/yql/CMakeLists.txt
+++ /dev/null
@@ -1,9 +0,0 @@
-
-# This file was generated by the build system used internally in the Yandex monorepo.
-# Only simple modifications are allowed (adding source-files to targets, adding simple properties
-# like target_include_directories). These modifications will be ported to original
-# ya.make files by maintainers. Any complex modifications which can't be ported back to the
-# original buildsystem will not be accepted.
-
-
-add_subdirectory(plugin)
diff --git a/yt/yt/client/api/bundle_controller_client.cpp b/yt/yt/client/api/bundle_controller_client.cpp
deleted file mode 100644
index 454cbe5b4d..0000000000
--- a/yt/yt/client/api/bundle_controller_client.cpp
+++ /dev/null
@@ -1,9 +0,0 @@
-#include "bundle_controller_client.h"
-
-namespace NYT::NApi {
-
-////////////////////////////////////////////////////////////////////////////////
-
-////////////////////////////////////////////////////////////////////////////////
-
-} // namespace NYT::NApi
diff --git a/yt/yt/client/api/client.h b/yt/yt/client/api/client.h
index 07f5536e5e..2af3ab0b18 100644
--- a/yt/yt/client/api/client.h
+++ b/yt/yt/client/api/client.h
@@ -13,7 +13,8 @@
 #include "table_client.h"
 #include "queue_client.h"
 #include "query_tracker_client.h"
-#include "bundle_controller_client.h"
+
+#include <yt/yt/client/bundle_controller_client/bundle_controller_client.h>
 
 namespace NYT::NApi {
 
diff --git a/yt/yt/client/api/delegating_client.cpp b/yt/yt/client/api/delegating_client.cpp
index 75f4006502..c878d878ad 100644
--- a/yt/yt/client/api/delegating_client.cpp
+++ b/yt/yt/client/api/delegating_client.cpp
@@ -1004,7 +1004,7 @@ TFuture<void> TDelegatingClient::AlterQuery(
     return Underlying_->AlterQuery(queryId, options);
 }
 
-TFuture<TBundleConfigDescriptor> TDelegatingClient::GetBundleConfig(
+TFuture<TBundleConfigDescriptorPtr> TDelegatingClient::GetBundleConfig(
     const TString& bundleName,
     const TGetBundleConfigOptions& options)
 {
diff --git a/yt/yt/client/api/delegating_client.h b/yt/yt/client/api/delegating_client.h
index 07429b103f..15abcbedec 100644
--- a/yt/yt/client/api/delegating_client.h
+++ b/yt/yt/client/api/delegating_client.h
@@ -618,7 +618,7 @@ public:
         NQueryTrackerClient::TQueryId queryId,
         const TAlterQueryOptions& options) override;
 
-    virtual TFuture<TBundleConfigDescriptor> GetBundleConfig(
+    virtual TFuture<TBundleConfigDescriptorPtr> GetBundleConfig(
         const TString& bundleName,
         const TGetBundleConfigOptions& options = {}) override;
 
diff --git a/yt/yt/client/api/rpc_proxy/api_service_proxy.h b/yt/yt/client/api/rpc_proxy/api_service_proxy.h
index d7597992cf..354c883ef5 100644
--- a/yt/yt/client/api/rpc_proxy/api_service_proxy.h
+++ b/yt/yt/client/api/rpc_proxy/api_service_proxy.h
@@ -62,6 +62,8 @@ public:
     DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, AlterTable);
     DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, AlterTableReplica);
     DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, GetTablePivotKeys);
+    DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, CreateTableBackup);
+    DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, RestoreTableBackup);
 
     DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, LookupRows);
     DEFINE_RPC_PROXY_METHOD(NRpcProxy::NProto, VersionedLookupRows);
diff --git a/yt/yt/client/api/rpc_proxy/client_impl.cpp b/yt/yt/client/api/rpc_proxy/client_impl.cpp
index aa76339c5f..8d0a7530e2 100644
--- a/yt/yt/client/api/rpc_proxy/client_impl.cpp
+++ b/yt/yt/client/api/rpc_proxy/client_impl.cpp
@@ -504,17 +504,38 @@ TFuture<TYsonString> TClient::GetTablePivotKeys(
 }
 
 TFuture<void> TClient::CreateTableBackup(
-    const TBackupManifestPtr& /*manifest*/,
-    const TCreateTableBackupOptions& /*options*/)
+    const TBackupManifestPtr& manifest,
+    const TCreateTableBackupOptions& options)
 {
-    ThrowUnimplemented("CreateTableBackup");
+    auto proxy = CreateApiServiceProxy();
+
+    auto req = proxy.CreateTableBackup();
+    ToProto(req->mutable_manifest(), *manifest);
+
+    SetTimeoutOptions(*req, options);
+    req->set_checkpoint_timestamp_delay(ToProto<i64>(options.CheckpointTimestampDelay));
+    req->set_checkpoint_check_period(ToProto<i64>(options.CheckpointCheckPeriod));
+    req->set_checkpoint_check_timeout(ToProto<i64>(options.CheckpointCheckTimeout));
+    req->set_force(options.Force);
+
+    return req->Invoke().As<void>();
 }
 
 TFuture<void> TClient::RestoreTableBackup(
-    const TBackupManifestPtr& /*manifest*/,
-    const TRestoreTableBackupOptions& /*options*/)
+    const TBackupManifestPtr& manifest,
+    const TRestoreTableBackupOptions& options)
 {
-    ThrowUnimplemented("RestoreTableBackup");
+    auto proxy = CreateApiServiceProxy();
+
+    auto req = proxy.RestoreTableBackup();
+    ToProto(req->mutable_manifest(), *manifest);
+
+    SetTimeoutOptions(*req, options);
+    req->set_force(options.Force);
+    req->set_mount(options.Mount);
+    req->set_enable_replicas(options.EnableReplicas);
+
+    return req->Invoke().As<void>();
 }
 
 TFuture<std::vector<TTableReplicaId>> TClient::GetInSyncReplicas(
@@ -2033,7 +2054,7 @@ TFuture<void> TClient::AlterQuery(
     ThrowUnimplemented("AlterQuery");
 }
 
-TFuture<TBundleConfigDescriptor> TClient::GetBundleConfig(
+TFuture<TBundleConfigDescriptorPtr> TClient::GetBundleConfig(
     const TString& /*bundleName*/,
     const TGetBundleConfigOptions& /*options*/)
 {
diff --git a/yt/yt/client/api/rpc_proxy/client_impl.h b/yt/yt/client/api/rpc_proxy/client_impl.h
index e6ab691e43..77c1b429f8 100644
--- a/yt/yt/client/api/rpc_proxy/client_impl.h
+++ b/yt/yt/client/api/rpc_proxy/client_impl.h
@@ -482,7 +482,7 @@ public:
         const TString& passwordSha256,
         const TListUserTokensOptions& options) override;
 
-    TFuture<TBundleConfigDescriptor> GetBundleConfig(
+    TFuture<TBundleConfigDescriptorPtr> GetBundleConfig(
         const TString& bundleName,
         const TGetBundleConfigOptions& options = {}) override;
 
diff --git a/yt/yt/client/api/rpc_proxy/helpers.cpp b/yt/yt/client/api/rpc_proxy/helpers.cpp
index ca0ebfeab4..a118284286 100644
--- a/yt/yt/client/api/rpc_proxy/helpers.cpp
+++ b/yt/yt/client/api/rpc_proxy/helpers.cpp
@@ -1,6 +1,7 @@
 #include "helpers.h"
 
 #include <yt/yt/client/api/rowset.h>
+#include <yt/yt/client/api/table_client.h>
 
 #include <yt/yt/client/table_client/columnar_statistics.h>
 #include <yt/yt/client/table_client/column_sort_schema.h>
@@ -35,6 +36,9 @@ void ThrowUnimplemented(const TString& method)
 
 namespace NProto {
 
+using NYT::ToProto;
+using NYT::FromProto;
+
 ////////////////////////////////////////////////////////////////////////////////
 // OPTIONS
 ////////////////////////////////////////////////////////////////////////////////
@@ -1275,6 +1279,56 @@ void FromProto(
     }
 }
 
+void ToProto(
+    NProto::TTableBackupManifest* protoManifest,
+    const NApi::TTableBackupManifestPtr& manifest)
+{
+    protoManifest->set_source_path(manifest->SourcePath);
+    protoManifest->set_destination_path(manifest->DestinationPath);
+    protoManifest->set_ordered_mode(ToProto<i32>(manifest->OrderedMode));
+}
+
+void FromProto(
+    NApi::TTableBackupManifestPtr* manifest,
+    const NProto::TTableBackupManifest& protoManifest)
+{
+    *manifest = New<NApi::TTableBackupManifest>();
+
+    (*manifest)->SourcePath = protoManifest.source_path();
+    (*manifest)->DestinationPath = protoManifest.destination_path();
+    (*manifest)->OrderedMode = CheckedEnumCast<EOrderedTableBackupMode>(protoManifest.ordered_mode());
+}
+
+void ToProto(
+    NProto::TBackupManifest::TClusterManifest* protoEntry,
+    const std::pair<TString, std::vector<NApi::TTableBackupManifestPtr>>& entry)
+{
+    protoEntry->set_cluster(entry.first);
+    ToProto(protoEntry->mutable_table_manifests(), entry.second);
+}
+
+void FromProto(
+    std::pair<TString, std::vector<NApi::TTableBackupManifestPtr>>* entry,
+    const NProto::TBackupManifest::TClusterManifest& protoEntry)
+{
+    entry->first = protoEntry.cluster();
+    FromProto(&entry->second, protoEntry.table_manifests());
+}
+
+void ToProto(
+    NProto::TBackupManifest* protoManifest,
+    const NApi::TBackupManifest& manifest)
+{
+    ToProto(protoManifest->mutable_clusters(), manifest.Clusters);
+}
+
+void FromProto(
+    NApi::TBackupManifest* manifest,
+    const NProto::TBackupManifest& protoManifest)
+{
+    FromProto(&manifest->Clusters, protoManifest.clusters());
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // ENUMS
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/client/api/rpc_proxy/helpers.h b/yt/yt/client/api/rpc_proxy/helpers.h
index 8c240cac5f..0eb4b3594c 100644
--- a/yt/yt/client/api/rpc_proxy/helpers.h
+++ b/yt/yt/client/api/rpc_proxy/helpers.h
@@ -204,6 +204,30 @@ void FromProto(
     NQueueClient::TQueueRowBatchReadOptions* result,
     const NProto::TRowBatchReadOptions& proto);
 
+void ToProto(
+    NProto::TTableBackupManifest* protoManifest,
+    const NApi::TTableBackupManifestPtr& manifest);
+
+void FromProto(
+    NApi::TTableBackupManifestPtr* manifest,
+    const NProto::TTableBackupManifest& protoManifest);
+
+void ToProto(
+    NProto::TBackupManifest::TClusterManifest* protoEntry,
+    const std::pair<TString, std::vector<NApi::TTableBackupManifestPtr>>& entry);
+
+void FromProto(
+    std::pair<TString, std::vector<NApi::TTableBackupManifestPtr>>* entry,
+    const NProto::TBackupManifest::TClusterManifest& protoEntry);
+
+void ToProto(
+    NProto::TBackupManifest* protoManifest,
+    const NApi::TBackupManifest& manifest);
+
+void FromProto(
+    NApi::TBackupManifest* manifest,
+    const NProto::TBackupManifest& protoManifest);
+
 NProto::EOperationType ConvertOperationTypeToProto(
     NScheduler::EOperationType operationType);
 
diff --git a/yt/yt/client/bundle_controller_client/bundle_controller_client.cpp b/yt/yt/client/bundle_controller_client/bundle_controller_client.cpp
new file mode 100644
index 0000000000..b22ff43d94
--- /dev/null
+++ b/yt/yt/client/bundle_controller_client/bundle_controller_client.cpp
@@ -0,0 +1,29 @@
+#include "bundle_controller_client.h"
+
+namespace NYT::NApi {
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TBundleConfigDescriptor::Register(TRegistrar registrar)
+{
+    registrar.Parameter("bundle_name", &TThis::BundleName)
+        .Default();
+
+    registrar.Parameter("cpu_limits", &TThis::CpuLimits)
+        .DefaultNew();
+    registrar.Parameter("memory_limits", &TThis::MemoryLimits)
+        .DefaultNew();
+
+    registrar.Parameter("rpc_proxy_count", &TThis::RpcProxyCount)
+        .Default(0);
+    registrar.Parameter("rpc_proxy_resource_guarantee", &TThis::RpcProxyResourceGuarantee)
+        .DefaultNew();
+    registrar.Parameter("tablet_node_count", &TThis::TabletNodeCount)
+        .Default(0);
+    registrar.Parameter("tablet_node_resource_guarantee", &TThis::TabletNodeResourceGuarantee)
+        .DefaultNew();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NYT::NApi
diff --git a/yt/yt/client/api/bundle_controller_client.h b/yt/yt/client/bundle_controller_client/bundle_controller_client.h
index 7d61f93562..0341458955 100644
--- a/yt/yt/client/api/bundle_controller_client.h
+++ b/yt/yt/client/bundle_controller_client/bundle_controller_client.h
@@ -1,11 +1,17 @@
 #pragma once
 
-#include "client_common.h"
+#include "bundle_controller_settings.h"
+
+#include <yt/yt/client/api/client_common.h>
 
 namespace NYT::NApi {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+DECLARE_REFCOUNTED_STRUCT(TBundleConfigDescriptor)
+
+////////////////////////////////////////////////////////////////////////////////
+
 struct TGetBundleConfigOptions
     : public TTimeoutOptions
 { };
@@ -13,19 +19,33 @@ struct TGetBundleConfigOptions
 ////////////////////////////////////////////////////////////////////////////////
 
 struct TBundleConfigDescriptor
+    : public NYTree::TYsonStruct
 {
     TString BundleName;
+
+    NBundleControllerClient::TCpuLimitsPtr CpuLimits;
+    NBundleControllerClient::TMemoryLimitsPtr MemoryLimits;
+
     int RpcProxyCount;
+    NBundleControllerClient::TInstanceResourcesPtr RpcProxyResourceGuarantee;
+
     int TabletNodeCount;
+    NBundleControllerClient::TInstanceResourcesPtr TabletNodeResourceGuarantee;
+
+    REGISTER_YSON_STRUCT(TBundleConfigDescriptor);
+
+    static void Register(TRegistrar registrar);
 };
 
+DEFINE_REFCOUNTED_TYPE(TBundleConfigDescriptor)
+
 ////////////////////////////////////////////////////////////////////////////////
 
 struct IBundleControllerClient
 {
     virtual ~IBundleControllerClient() = default;
 
-    virtual TFuture<TBundleConfigDescriptor> GetBundleConfig(
+    virtual TFuture<TBundleConfigDescriptorPtr> GetBundleConfig(
         const TString& bundleName,
         const TGetBundleConfigOptions& options = {}) = 0;
 };
diff --git a/yt/yt/client/bundle_controller_client/bundle_controller_settings.cpp b/yt/yt/client/bundle_controller_client/bundle_controller_settings.cpp
new file mode 100644
index 0000000000..194b020c1d
--- /dev/null
+++ b/yt/yt/client/bundle_controller_client/bundle_controller_settings.cpp
@@ -0,0 +1,108 @@
+#include "bundle_controller_settings.h"
+
+namespace NYT::NBundleControllerClient {
+
+////////////////////////////////////////////////////////////////////////////////
+
+void TCpuLimits::Register(TRegistrar registrar)
+{
+    registrar.Parameter("write_thread_pool_size", &TThis::WriteThreadPoolSize)
+        .GreaterThan(0)
+        .Default(5);
+    registrar.Parameter("lookup_thread_pool_size", &TThis::LookupThreadPoolSize)
+        .GreaterThan(0)
+        .Default(4);
+    registrar.Parameter("query_thread_pool_size", &TThis::QueryThreadPoolSize)
+        .GreaterThan(0)
+        .Default(4);
+}
+
+void TMemoryLimits::Register(TRegistrar registrar)
+{
+    registrar.Parameter("tablet_static", &TThis::TabletStatic)
+        .Optional();
+    registrar.Parameter("tablet_dynamic", &TThis::TabletDynamic)
+        .Optional();
+    registrar.Parameter("compressed_block_cache", &TThis::CompressedBlockCache)
+        .Optional();
+    registrar.Parameter("uncompressed_block_cache", &TThis::UncompressedBlockCache)
+        .Optional();
+    registrar.Parameter("key_filter_block_cache", &TThis::KeyFilterBlockCache)
+        .Optional();
+    registrar.Parameter("versioned_chunk_meta", &TThis::VersionedChunkMeta)
+        .Optional();
+    registrar.Parameter("lookup_row_cache", &TThis::LookupRowCache)
+        .Optional();
+}
+
+void TInstanceResources::Register(TRegistrar registrar)
+{
+    registrar.Parameter("vcpu", &TThis::Vcpu)
+        .GreaterThanOrEqual(0)
+        .Default(18000);
+    registrar.Parameter("memory", &TThis::Memory)
+        .GreaterThanOrEqual(0)
+        .Default(120_GB);
+    registrar.Parameter("net", &TThis::Net)
+        .Optional();
+    registrar.Parameter("type", &TThis::Type)
+        .Default();
+}
+
+void TInstanceResources::Clear()
+{
+    Vcpu = 0;
+    Memory = 0;
+}
+
+bool TInstanceResources::operator==(const TInstanceResources& other) const
+{
+    return std::tie(Vcpu, Memory, Net) == std::tie(other.Vcpu, other.Memory, other.Net);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+namespace NProto {
+
+// TODO(alexmipt): make ToProto for TCpuLimits, TMemoryLimits, TInstanceResources
+
+////////////////////////////////////////////////////////////////////////////////
+
+void FromProto(NBundleControllerClient::TCpuLimitsPtr cpuLimits, const NBundleController::NProto::TCpuLimits* protoCpuLimits)
+{
+    cpuLimits->LookupThreadPoolSize = protoCpuLimits->lookup_thread_pool_size();
+    cpuLimits->QueryThreadPoolSize = protoCpuLimits->query_thread_pool_size();
+    cpuLimits->WriteThreadPoolSize = protoCpuLimits->write_thread_pool_size();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+void FromProto(NBundleControllerClient::TMemoryLimitsPtr memoryLimits, const NBundleController::NProto::TMemoryLimits* protoMemoryLimits)
+{
+    memoryLimits->CompressedBlockCache = protoMemoryLimits->compressed_block_cache();
+    memoryLimits->KeyFilterBlockCache = protoMemoryLimits->key_filter_block_cache();
+    memoryLimits->LookupRowCache = protoMemoryLimits->lookup_row_cache();
+
+    memoryLimits->TabletDynamic = protoMemoryLimits->tablet_dynamic();
+    memoryLimits->TabletStatic = protoMemoryLimits->tablet_static();
+
+    memoryLimits->UncompressedBlockCache = protoMemoryLimits->uncompressed_block_cache();
+
+    memoryLimits->VersionedChunkMeta = protoMemoryLimits->versioned_chunk_meta();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+void FromProto(NBundleControllerClient::TInstanceResourcesPtr instanceResources, const NBundleController::NProto::TInstanceResources* protoInstanceResources)
+{
+    instanceResources->Memory = protoInstanceResources->memory();
+    instanceResources->Net = protoInstanceResources->net();
+    instanceResources->Type = protoInstanceResources->type();
+    instanceResources->Vcpu = protoInstanceResources->vcpu();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NProto
+
+} // namespace NYT::NBundleControllerClient
diff --git a/yt/yt/client/bundle_controller_client/bundle_controller_settings.h b/yt/yt/client/bundle_controller_client/bundle_controller_settings.h
new file mode 100644
index 0000000000..8dcf241ae1
--- /dev/null
+++ b/yt/yt/client/bundle_controller_client/bundle_controller_settings.h
@@ -0,0 +1,97 @@
+#pragma once
+
+#include <optional>
+
+#include <yt/yt/client/tablet_client/public.h>
+
+#include <yt/yt/core/ytree/yson_serializable.h>
+#include <yt/yt/core/ytree/yson_struct.h>
+
+#include <yt/yt_proto/yt/client/bundle_controller/proto/bundle_controller_service.pb.h>
+
+namespace NYT::NBundleControllerClient {
+
+////////////////////////////////////////////////////////////////////////////////
+
+DECLARE_REFCOUNTED_STRUCT(TCpuLimits)
+DECLARE_REFCOUNTED_STRUCT(TMemoryLimits)
+DECLARE_REFCOUNTED_STRUCT(TInstanceResources)
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TCpuLimits
+    : public NYTree::TYsonStruct
+{
+    int LookupThreadPoolSize;
+    int QueryThreadPoolSize;
+    int WriteThreadPoolSize;
+
+    REGISTER_YSON_STRUCT(TCpuLimits);
+
+    static void Register(TRegistrar registrar);
+};
+
+DEFINE_REFCOUNTED_TYPE(TCpuLimits)
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TMemoryLimits
+    : public NYTree::TYsonStruct
+{
+    std::optional<i64> CompressedBlockCache;
+    std::optional<i64> KeyFilterBlockCache;
+    std::optional<i64> LookupRowCache;
+    std::optional<i64> TabletDynamic;
+    std::optional<i64> TabletStatic;
+    std::optional<i64> UncompressedBlockCache;
+    std::optional<i64> VersionedChunkMeta;
+
+    REGISTER_YSON_STRUCT(TMemoryLimits);
+
+    static void Register(TRegistrar registrar);
+};
+
+DEFINE_REFCOUNTED_TYPE(TMemoryLimits)
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TInstanceResources
+    : public NYTree::TYsonStruct
+{
+    i64 Memory;
+    std::optional<i64> Net;
+
+    TString Type;
+    int Vcpu;
+
+    bool operator==(const TInstanceResources& resources) const;
+
+    void Clear();
+
+    REGISTER_YSON_STRUCT(TInstanceResources);
+
+    static void Register(TRegistrar registrar);
+};
+
+DEFINE_REFCOUNTED_TYPE(TInstanceResources)
+
+////////////////////////////////////////////////////////////////////////////////
+
+namespace NProto {
+
+////////////////////////////////////////////////////////////////////////////////
+
+void ToProto(NBundleController::NProto::TCpuLimits* protoCpuLimits, const NBundleControllerClient::TCpuLimitsPtr cpuLimits);
+void FromProto(NBundleControllerClient::TCpuLimitsPtr cpuLimits, const NBundleController::NProto::TCpuLimits* protoCpuLimits);
+
+void ToProto(NBundleController::NProto::TMemoryLimits* protoMemoryLimits, const NBundleControllerClient::TMemoryLimitsPtr memoryLimits);
+void FromProto(NBundleControllerClient::TMemoryLimitsPtr memoryLimits, const NBundleController::NProto::TMemoryLimits* protoMemoryLimits);
+
+void ToProto(NBundleController::NProto::TInstanceResources* protoInstanceResources, const NBundleControllerClient::TInstanceResourcesPtr instanceResources);
+void FromProto(NBundleControllerClient::TInstanceResourcesPtr instanceResources, const NBundleController::NProto::TInstanceResources* protoInstanceResources);
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NProto
+
+} // namespace NYT::NBundleControllerClient
diff --git a/yt/yt/client/driver/bundle_controller_commands.cpp b/yt/yt/client/driver/bundle_controller_commands.cpp
index d7591c8635..b0f84ab542 100644
--- a/yt/yt/client/driver/bundle_controller_commands.cpp
+++ b/yt/yt/client/driver/bundle_controller_commands.cpp
@@ -22,12 +22,7 @@ void TGetBundleConfigCommand::DoExecute(ICommandContextPtr context)
         Options))
         .ValueOrThrow();
 
-    context->ProduceOutputValue(BuildYsonStringFluently()
-        .BeginMap()
-            .Item("bundle_name").Value(result.BundleName)
-            .Item("rpc_proxy_count").Value(result.RpcProxyCount)
-            .Item("tablet_node_count").Value(result.TabletNodeCount)
-        .EndMap());
+    context->ProduceOutputValue(ConvertToYsonString(result));
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/client/federated/client.cpp b/yt/yt/client/federated/client.cpp
index 98efe5b9a7..f2606da4d6 100644
--- a/yt/yt/client/federated/client.cpp
+++ b/yt/yt/client/federated/client.cpp
@@ -407,7 +407,7 @@ public:
     UNIMPLEMENTED_METHOD(TFuture<TQuery>, GetQuery, (NQueryTrackerClient::TQueryId, const TGetQueryOptions&));
     UNIMPLEMENTED_METHOD(TFuture<TListQueriesResult>, ListQueries, (const TListQueriesOptions&));
     UNIMPLEMENTED_METHOD(TFuture<void>, AlterQuery, (NQueryTrackerClient::TQueryId, const TAlterQueryOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<TBundleConfigDescriptor>, GetBundleConfig, (const TString&, const TGetBundleConfigOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<TBundleConfigDescriptorPtr>, GetBundleConfig, (const TString&, const TGetBundleConfigOptions&));
 
     UNIMPLEMENTED_METHOD(TFuture<ITableReaderPtr>, CreateTableReader, (const NYPath::TRichYPath&, const TTableReaderOptions&));
     UNIMPLEMENTED_METHOD(TFuture<ITableWriterPtr>, CreateTableWriter, (const NYPath::TRichYPath&, const TTableWriterOptions&));
diff --git a/yt/yt/client/hedging/hedging.cpp b/yt/yt/client/hedging/hedging.cpp
index f27c17cc78..b76b740345 100644
--- a/yt/yt/client/hedging/hedging.cpp
+++ b/yt/yt/client/hedging/hedging.cpp
@@ -210,7 +210,7 @@ public:
     UNSUPPORTED_METHOD(TFuture<TQuery>, GetQuery, (NQueryTrackerClient::TQueryId, const TGetQueryOptions&));
     UNSUPPORTED_METHOD(TFuture<TListQueriesResult>, ListQueries, (const TListQueriesOptions&));
     UNSUPPORTED_METHOD(TFuture<void>, AlterQuery, (NQueryTrackerClient::TQueryId, const TAlterQueryOptions&));
-    UNSUPPORTED_METHOD(TFuture<TBundleConfigDescriptor>, GetBundleConfig, (const TString&, const TGetBundleConfigOptions&));
+    UNSUPPORTED_METHOD(TFuture<TBundleConfigDescriptorPtr>, GetBundleConfig, (const TString&, const TGetBundleConfigOptions&));
 
 private:
     THedgingExecutorPtr Executor_;
diff --git a/yt/yt/client/security_client/acl.cpp b/yt/yt/client/security_client/acl.cpp
index 69db3d80cc..ee06bafcd0 100644
--- a/yt/yt/client/security_client/acl.cpp
+++ b/yt/yt/client/security_client/acl.cpp
@@ -54,6 +54,7 @@ void Serialize(const TSerializableAccessControlEntry& ace, NYson::IYsonConsumer*
             // TODO(max42): YT-16347.
             // Do not serialize this field by default
             .Item("inheritance_mode").Value(ace.InheritanceMode)
+            .OptionalItem("subject_tag_filter", ace.SubjectTagFilter)
             .OptionalItem("columns", ace.Columns)
             .OptionalItem("vital", ace.Vital)
         .EndMap();
@@ -102,6 +103,11 @@ void Deserialize(TSerializableAccessControlEntry& ace, NYTree::INodePtr node)
     } else {
         ace.InheritanceMode = EAceInheritanceMode::ObjectAndDescendants;
     }
+    if (auto tagFilterNode = mapNode->FindChild("subject_tag_filter")) {
+        Deserialize(ace.SubjectTagFilter, tagFilterNode);
+    } else {
+        ace.SubjectTagFilter = {};
+    }
     if (auto columnsNode = mapNode->FindChild("columns")) {
         Deserialize(ace.Columns, columnsNode);
     } else {
@@ -121,6 +127,7 @@ void Deserialize(TSerializableAccessControlEntry& ace, NYson::TYsonPullParserCur
     auto HasSubjects = false;
     auto HasPermissions = false;
     ace.InheritanceMode = EAceInheritanceMode::ObjectAndDescendants;
+    ace.SubjectTagFilter = {};
     cursor->ParseMap([&] (TYsonPullParserCursor* cursor) {
         auto key = cursor->GetCurrent().UncheckedAsString();
         if (key == TStringBuf("action")) {
@@ -138,6 +145,9 @@ void Deserialize(TSerializableAccessControlEntry& ace, NYson::TYsonPullParserCur
         } else if (key == TStringBuf("inheritance_mode")) {
             cursor->Next();
             Deserialize(ace.InheritanceMode, cursor);
+        } else if (key == TStringBuf("subject_tag_filter")) {
+            cursor->Next();
+            Deserialize(ace.SubjectTagFilter, cursor);
         } else if (key == TStringBuf("columns")) {
             cursor->Next();
             Deserialize(ace.Columns, cursor);
@@ -163,6 +173,12 @@ void TSerializableAccessControlEntry::Persist(const TStreamPersistenceContext& c
     Persist(context, Subjects);
     Persist(context, Permissions);
     Persist(context, InheritanceMode);
+    // COMPAT(vovamelnikov)
+    if (context.IsLoad() && context.GetVersion() < 301305) {
+        SubjectTagFilter =  {};
+    } else {
+        Persist(context, SubjectTagFilter);
+    }
     // NB: Columns and Vital are not persisted since this method is intended only for use in controller.
 }
 
diff --git a/yt/yt/client/security_client/acl.h b/yt/yt/client/security_client/acl.h
index 04ea66221a..5615dff7fa 100644
--- a/yt/yt/client/security_client/acl.h
+++ b/yt/yt/client/security_client/acl.h
@@ -2,6 +2,8 @@
 
 #include "public.h"
 
+#include <yt/yt/core/misc/arithmetic_formula.h>
+
 #include <yt/yt/core/yson/consumer.h>
 
 #include <yt/yt/core/ytree/permission.h>
@@ -18,6 +20,7 @@ struct TSerializableAccessControlEntry
     std::vector<TString> Subjects;
     NYTree::EPermissionSet Permissions;
     EAceInheritanceMode InheritanceMode = EAceInheritanceMode::ObjectAndDescendants;
+    TString SubjectTagFilter;
     std::optional<std::vector<TString>> Columns;
     std::optional<bool> Vital;
 
diff --git a/yt/yt/client/unittests/mock/client.h b/yt/yt/client/unittests/mock/client.h
index 49dbe4c741..49c221d8ad 100644
--- a/yt/yt/client/unittests/mock/client.h
+++ b/yt/yt/client/unittests/mock/client.h
@@ -620,7 +620,7 @@ public:
     MOCK_METHOD(TFuture<void>, AlterQuery, (
         NQueryTrackerClient::TQueryId queryId, const TAlterQueryOptions& options), (override));
 
-    MOCK_METHOD(TFuture<TBundleConfigDescriptor>, GetBundleConfig, (
+    MOCK_METHOD(TFuture<TBundleConfigDescriptorPtr>, GetBundleConfig, (
         const TString& bundleName,
         const TGetBundleConfigOptions& options), (override));
 };
diff --git a/yt/yt/client/ya.make b/yt/yt/client/ya.make
index 4f6f2101a1..73a8deb291 100644
--- a/yt/yt/client/ya.make
+++ b/yt/yt/client/ya.make
@@ -13,7 +13,6 @@ SRCS(
     api/client_cache.cpp
     api/delegating_client.cpp
     api/etc_client.cpp
-    api/bundle_controller_client.cpp
     api/journal_client.cpp
     api/operation_client.cpp
     api/security_client.cpp
@@ -50,6 +49,9 @@ SRCS(
     api/rpc_proxy/row_stream.cpp
     api/rpc_proxy/wire_row_stream.cpp
 
+    bundle_controller_client/bundle_controller_client.cpp
+    bundle_controller_client/bundle_controller_settings.cpp
+
     election/public.cpp
 
     hive/timestamp_map.cpp
diff --git a/yt/yt/core/misc/error.cpp b/yt/yt/core/misc/error.cpp
index 17cf40583b..ce9a942841 100644
--- a/yt/yt/core/misc/error.cpp
+++ b/yt/yt/core/misc/error.cpp
@@ -5,6 +5,8 @@
 
 #include <yt/yt_proto/yt/core/misc/proto/error.pb.h>
 
+#include <yt/yt/core/actions/callback.h>
+
 #include <yt/yt/core/misc/protobuf_helpers.h>
 #include <yt/yt/core/misc/proc.h>
 
@@ -67,13 +69,16 @@ TString ToString(TErrorCode code)
 
 YT_THREAD_LOCAL(bool) ErrorSanitizerEnabled = false;
 YT_THREAD_LOCAL(TInstant) ErrorSanitizerDatetimeOverride = {};
+YT_THREAD_LOCAL(TErrorSanitizerGuard::THostNameSanitizer) LocalHostNameSanitizer = {};
 
-TErrorSanitizerGuard::TErrorSanitizerGuard(TInstant datetimeOverride)
+TErrorSanitizerGuard::TErrorSanitizerGuard(TInstant datetimeOverride, THostNameSanitizer localHostNameSanitizer)
     : SavedEnabled_(ErrorSanitizerEnabled)
     , SavedDatetimeOverride_(GetTlsRef(ErrorSanitizerDatetimeOverride))
+    , SavedLocalHostNameSanitizer_(GetTlsRef(LocalHostNameSanitizer))
 {
     ErrorSanitizerEnabled = true;
     GetTlsRef(ErrorSanitizerDatetimeOverride) = datetimeOverride;
+    GetTlsRef(LocalHostNameSanitizer) = localHostNameSanitizer;
 }
 
 TErrorSanitizerGuard::~TErrorSanitizerGuard()
@@ -150,7 +155,7 @@ public:
         return &Message_;
     }
 
-    bool HasOriginAttributes() const
+    bool HasHost() const
     {
         return Host_.operator bool();
     }
@@ -160,6 +165,11 @@ public:
         return Host_;
     }
 
+    bool HasOriginAttributes() const
+    {
+        return ThreadName_.Length > 0;
+    }
+
     bool HasDatetime() const
     {
         return Datetime_ != TInstant();
@@ -288,6 +298,8 @@ private:
     {
         if (ErrorSanitizerEnabled) {
             Datetime_ = GetTlsRef(ErrorSanitizerDatetimeOverride);
+            auto handleLocalHostName = GetTlsRef(LocalHostNameSanitizer);
+            Host_ = handleLocalHostName(NNet::ReadLocalHostName());
             return;
         }
 
@@ -472,12 +484,12 @@ TError& TError::SetMessage(TString message)
     return *this;
 }
 
-bool TError::HasOriginAttributes() const
+bool TError::HasHost() const
 {
     if (!Impl_) {
         return false;
     }
-    return Impl_->HasOriginAttributes();
+    return Impl_->HasHost();
 }
 
 TStringBuf TError::GetHost() const
@@ -488,6 +500,14 @@ TStringBuf TError::GetHost() const
     return Impl_->GetHost();
 }
 
+bool TError::HasOriginAttributes() const
+{
+    if (!Impl_) {
+        return false;
+    }
+    return Impl_->HasOriginAttributes();
+}
+
 bool TError::HasDatetime() const
 {
     if (!Impl_) {
@@ -932,6 +952,12 @@ void AppendError(TStringBuilderBase* builder, const TError& error, int indent)
                 (!error.GetThreadName().empty() ? error.GetThreadName() : ToString(error.GetTid())),
                 error.GetFid()),
             indent);
+    } else if (ErrorSanitizerEnabled && error.HasHost()) {
+        AppendAttribute(
+            builder,
+            "host",
+            Format("%v", error.GetHost()),
+            indent);
     }
 
     if (error.HasDatetime()) {
@@ -1048,6 +1074,9 @@ void ToProto(NYT::NProto::TError* protoError, const TError& error)
 
         static const TString FidKey("fid");
         addAttribute(FidKey, error.GetFid());
+    } else if (ErrorSanitizerEnabled && error.HasHost()) {
+        static const TString HostKey("host");
+        addAttribute(HostKey, error.GetHost());
     }
 
     if (error.HasDatetime()) {
@@ -1149,6 +1178,9 @@ void Serialize(
                         .Item("tid").Value(error.GetTid())
                         .Item("thread").Value(error.GetThreadName())
                         .Item("fid").Value(error.GetFid());
+                } else if (ErrorSanitizerEnabled && error.HasHost()) {
+                    fluent
+                        .Item("host").Value(error.GetHost());
                 }
                 if (error.HasDatetime()) {
                     fluent
diff --git a/yt/yt/core/misc/error.h b/yt/yt/core/misc/error.h
index 0716ceb461..be3a5d9ec7 100644
--- a/yt/yt/core/misc/error.h
+++ b/yt/yt/core/misc/error.h
@@ -69,12 +69,16 @@ class TErrorSanitizerGuard
     : public TNonCopyable
 {
 public:
-    explicit TErrorSanitizerGuard(TInstant datetimeOverride);
+    using TLocalHostNameSanitizerSignature = TString (TStringBuf);
+    using THostNameSanitizer = TCallback<TLocalHostNameSanitizerSignature>;
+
+    explicit TErrorSanitizerGuard(TInstant datetimeOverride, THostNameSanitizer localHostNameSanitizer);
     ~TErrorSanitizerGuard();
 
 private:
     const bool SavedEnabled_;
     const TInstant SavedDatetimeOverride_;
+    const THostNameSanitizer SavedLocalHostNameSanitizer_;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -146,8 +150,10 @@ public:
     const TString& GetMessage() const;
     TError& SetMessage(TString message);
 
-    bool HasOriginAttributes() const;
+    bool HasHost() const;
     TStringBuf GetHost() const;
+
+    bool HasOriginAttributes() const;
     TProcessId GetPid() const;
     TStringBuf GetThreadName() const;
     NThreading::TThreadId GetTid() const;
diff --git a/yt/yt/core/misc/protobuf_helpers-inl.h b/yt/yt/core/misc/protobuf_helpers-inl.h
index c292cd6dfc..0c8118f3ef 100644
--- a/yt/yt/core/misc/protobuf_helpers-inl.h
+++ b/yt/yt/core/misc/protobuf_helpers-inl.h
@@ -382,6 +382,19 @@ void FromProtoArrayImpl(
     }
 }
 
+template <class TOriginalKey, class TOriginalValue, class TSerializedArray>
+void FromProtoArrayImpl(
+    THashMap<TOriginalKey, TOriginalValue>* originalArray,
+    const TSerializedArray& serializedArray)
+{
+    originalArray->clear();
+    originalArray->reserve(serializedArray.size());
+    for (int i = 0; i < serializedArray.size(); ++i) {
+        originalArray->emplace(
+            FromProto<std::pair<TOriginalKey, TOriginalValue>>(serializedArray.Get(i)));
+    }
+}
+
 template <class TOriginal, class TSerializedArray>
 void CheckedFromProtoArrayImpl(
     THashSet<TOriginal>* originalArray,
@@ -420,90 +433,18 @@ void FromProtoArrayImpl(
 
 } // namespace NDetail
 
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const std::vector<TOriginal>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const std::vector<TOriginal>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal, size_t N>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const std::array<TOriginal, N>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal, size_t N>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const std::array<TOriginal, N>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal, size_t Size>
+template <class TSerialized, class TOriginalArray>
 void ToProto(
     ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const TCompactVector<TOriginal, Size>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal, size_t Size>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const TCompactVector<TOriginal, Size>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    TRange<TOriginal> originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    TRange<TOriginal> originalArray)
+    const TOriginalArray& originalArray)
 {
     NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
 }
 
-template <class TSerialized, class T, class E, E Min, E Max>
+template <class TSerialized, class TOriginalArray>
 void ToProto(
     ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const TEnumIndexedVector<E, T, Min, Max>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class T, class E, E Min, E Max>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const TEnumIndexedVector<E, T, Min, Max>& originalArray)
-{
-    NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
-}
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const THashSet<TOriginal>& originalArray)
+    const TOriginalArray& originalArray)
 {
     NYT::NDetail::ToProtoArrayImpl(serializedArray, originalArray);
 }
diff --git a/yt/yt/core/misc/protobuf_helpers.h b/yt/yt/core/misc/protobuf_helpers.h
index 4c8c8532cc..8c9d9046be 100644
--- a/yt/yt/core/misc/protobuf_helpers.h
+++ b/yt/yt/core/misc/protobuf_helpers.h
@@ -62,56 +62,15 @@ template <class T>
 void FromProto(T* original, ui64 serialized);
 
 ////////////////////////////////////////////////////////////////////////////////
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const std::vector<TOriginal>& originalArray);
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const std::vector<TOriginal>& originalArray);
-
-template <class TSerialized, class TOriginal, size_t N>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const std::array<TOriginal, N>& originalArray);
-
-template <class TSerialized, class TOriginal, size_t N>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const std::array<TOriginal, N>& originalArray);
-
-template <class TSerialized, class TOriginal, size_t Size>
+template <class TSerialized, class TOriginalArray>
 void ToProto(
     ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const TCompactVector<TOriginal, Size>& originalArray);
+    const TOriginalArray& originalArray);
 
-template <class TSerialized, class TOriginal, size_t Size>
+template <class TSerialized, class TOriginalArray>
 void ToProto(
     ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const TCompactVector<TOriginal, Size>& originalArray);
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const THashSet<TOriginal>& originalArray);
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const THashSet<TOriginal>& originalArray);
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    TRange<TOriginal> originalArray);
-
-template <class TSerialized, class TOriginal>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    TRange<TOriginal> originalArray);
+    const TOriginalArray& originalArray);
 
 template <class TOriginalArray, class TSerialized>
 void FromProto(
@@ -133,16 +92,6 @@ void CheckedHashSetFromProto(
     THashSet<TOriginal>* originalHashSet,
     const ::google::protobuf::RepeatedField<TSerialized>& serializedHashSet);
 
-template <class TSerialized, class T, class E, E Min, E Max>
-void ToProto(
-    ::google::protobuf::RepeatedPtrField<TSerialized>* serializedArray,
-    const TEnumIndexedVector<E, T, Min, Max>& originalArray);
-
-template <class TSerialized, class T, class E, E Min, E Max>
-void ToProto(
-    ::google::protobuf::RepeatedField<TSerialized>* serializedArray,
-    const TEnumIndexedVector<E, T, Min, Max>& originalArray);
-
 template <class TSerialized, class T, class TTag>
 void FromProto(TStrongTypedef<T, TTag>* original, const TSerialized& serialized);
 
diff --git a/yt/yt/core/misc/unittests/error_ut.cpp b/yt/yt/core/misc/unittests/error_ut.cpp
index 905a35d414..f8dad10b3c 100644
--- a/yt/yt/core/misc/unittests/error_ut.cpp
+++ b/yt/yt/core/misc/unittests/error_ut.cpp
@@ -682,6 +682,11 @@ TEST(TErrorTest, CompositeYTExceptionToError)
     }
 }
 
+TString HostSanitizer(TStringBuf)
+{
+    return "";
+}
+
 TEST(TErrorTest, ErrorSanitizer)
 {
     auto checkSantizied = [&] (const TError& error) {
@@ -711,7 +716,7 @@ TEST(TErrorTest, ErrorSanitizer)
 
     {
         auto instant1 = TInstant::Days(123);
-        TErrorSanitizerGuard guard1(instant1);
+        TErrorSanitizerGuard guard1(instant1, BIND(&HostSanitizer));
 
         auto error2 = TError("error2");
         checkSantizied(error2);
@@ -719,7 +724,7 @@ TEST(TErrorTest, ErrorSanitizer)
 
         {
             auto instant2 = TInstant::Days(234);
-            TErrorSanitizerGuard guard2(instant2);
+            TErrorSanitizerGuard guard2(instant2, BIND(&HostSanitizer));
 
             auto error3 = TError("error3");
             checkSantizied(error3);
diff --git a/yt/yt/core/rpc/balancing_channel.cpp b/yt/yt/core/rpc/balancing_channel.cpp
index 549c4bcc1b..7c1555f381 100644
--- a/yt/yt/core/rpc/balancing_channel.cpp
+++ b/yt/yt/core/rpc/balancing_channel.cpp
@@ -208,9 +208,6 @@ public:
             Config_->Addresses &&
             Config_->Addresses->size() == 1)
         {
-            // Disable discovery and balancing when just one address is given.
-            // This is vital for jobs since node's redirector is incapable of handling
-            // Discover requests properly.
             return MakeFuture(ChannelFactory_->CreateChannel((*Config_->Addresses)[0]));
         } else {
             return GetSubprovider(request->GetService())->GetChannel(request);
@@ -314,6 +311,7 @@ IRoamingChannelProviderPtr CreateBalancingChannelProvider(
 {
     YT_VERIFY(config);
     YT_VERIFY(channelFactory);
+    YT_VERIFY(peerDiscovery);
 
     return New<TBalancingChannelProvider>(
         std::move(config),
diff --git a/yt/yt/core/rpc/client.cpp b/yt/yt/core/rpc/client.cpp
index 107148dd50..74d42fc7d4 100644
--- a/yt/yt/core/rpc/client.cpp
+++ b/yt/yt/core/rpc/client.cpp
@@ -103,7 +103,7 @@ TSharedRefArray TClientRequest::Serialize()
 
     if (!retry) {
         auto output = CreateRequestMessage(Header_, headerlessMessage);
-        return TrackMemory(std::move(output));
+        return std::move(output);
     }
 
     if (StreamingEnabled_) {
@@ -114,7 +114,7 @@ TSharedRefArray TClientRequest::Serialize()
     patchedHeader.set_retry(true);
 
     auto output = CreateRequestMessage(patchedHeader, headerlessMessage);
-    return TrackMemory(std::move(output));
+    return std::move(output);
 }
 
 IClientRequestControlPtr TClientRequest::Send(IClientResponseHandlerPtr responseHandler)
@@ -491,11 +491,6 @@ bool IsRequestSticky(const IClientRequestPtr& request)
     return balancingExt.enable_stickiness();
 }
 
-TSharedRefArray TClientRequest::TrackMemory(TSharedRefArray array) const
-{
-    return NYT::TrackMemory(MemoryReferenceTracker_, std::move(array));
-}
-
 ////////////////////////////////////////////////////////////////////////////////
 
 TClientResponse::TClientResponse(TClientContextPtr clientContext)
diff --git a/yt/yt/core/rpc/client.h b/yt/yt/core/rpc/client.h
index 05d1c30495..5a04b9baca 100644
--- a/yt/yt/core/rpc/client.h
+++ b/yt/yt/core/rpc/client.h
@@ -137,7 +137,6 @@ public:
     DEFINE_BYVAL_RW_PROPERTY(NCompression::ECodec, ResponseCodec, NCompression::ECodec::None);
     DEFINE_BYVAL_RW_PROPERTY(bool, EnableLegacyRpcCodecs, true);
     DEFINE_BYVAL_RW_PROPERTY(bool, GenerateAttachmentChecksums, true);
-    DEFINE_BYVAL_RW_PROPERTY(IMemoryReferenceTrackerPtr, MemoryReferenceTracker);
     // Field is used on client side only. So it is never serialized.
     DEFINE_BYREF_RW_PROPERTY(NTracing::TTraceContext::TTagList, TracingTags);
     // For testing purposes only.
@@ -251,8 +250,6 @@ private:
 
     void PrepareHeader();
     TSharedRefArray GetHeaderlessMessage() const;
-
-    TSharedRefArray TrackMemory(TSharedRefArray array) const;
 };
 
 DEFINE_REFCOUNTED_TYPE(TClientRequest)
diff --git a/yt/yt/core/rpc/config.h b/yt/yt/core/rpc/config.h
index 7e7d7b2d70..be8c1fdec5 100644
--- a/yt/yt/core/rpc/config.h
+++ b/yt/yt/core/rpc/config.h
@@ -292,6 +292,10 @@ class TBalancingChannelConfig
 public:
     //! First option: static list of addresses.
     std::optional<std::vector<TString>> Addresses;
+
+    //! Disables discovery and balancing when just one address is given.
+    //! This is vital for jobs since node's redirector is incapable of handling
+    //! discover requests properly.
     bool DisableBalancingOnSingleAddress;
 
     //! Second option: SD endpoints.
diff --git a/yt/yt/library/column_converters/boolean_column_converter.cpp b/yt/yt/library/column_converters/boolean_column_converter.cpp
index 7462f940ec..abfe730d7a 100644
--- a/yt/yt/library/column_converters/boolean_column_converter.cpp
+++ b/yt/yt/library/column_converters/boolean_column_converter.cpp
@@ -31,9 +31,13 @@ class TBooleanColumnConverter
     : public IColumnConverter
 {
 public:
-    TBooleanColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+    TBooleanColumnConverter(
+        int columnIndex,
+        const NTableClient::TColumnSchema& columnSchema,
+        int columnOffset)
         : ColumnIndex_(columnIndex)
         , ColumnSchema_(columnSchema)
+        , ColumnOffset_(columnOffset)
     { }
 
     TConvertedColumn Convert(TRange<TUnversionedRowValues> rowsValues) override
@@ -64,6 +68,7 @@ public:
 private:
     const int ColumnIndex_;
     const NTableClient::TColumnSchema ColumnSchema_;
+    const int ColumnOffset_;
 
     TBitmapOutput Values_;
     TBitmapOutput NullBitmap_;
@@ -77,7 +82,7 @@ private:
     void AddValues(TRange<TUnversionedRowValues> rowsValues)
     {
         for (const auto& rowValues : rowsValues) {
-            auto value = rowValues[ColumnIndex_];
+            auto value = rowValues[ColumnOffset_];
             bool isNull = !value || value->Type == NTableClient::EValueType::Null;
             bool data = isNull ? false : value->Data.Boolean;
             NullBitmap_.Append(isNull);
@@ -90,9 +95,9 @@ private:
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateBooleanColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+IColumnConverterPtr CreateBooleanColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset)
 {
-    return std::make_unique<TBooleanColumnConverter>(columnIndex, columnSchema);
+    return std::make_unique<TBooleanColumnConverter>(columnId, columnSchema, columnOffset);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/library/column_converters/boolean_column_converter.h b/yt/yt/library/column_converters/boolean_column_converter.h
index 0495c4a188..8d917840c7 100644
--- a/yt/yt/library/column_converters/boolean_column_converter.h
+++ b/yt/yt/library/column_converters/boolean_column_converter.h
@@ -8,7 +8,7 @@ namespace NYT::NColumnConverters {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateBooleanColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema);
+IColumnConverterPtr CreateBooleanColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/column_converters/column_converter.cpp b/yt/yt/library/column_converters/column_converter.cpp
index 4a9665d48b..7442ea0fd4 100644
--- a/yt/yt/library/column_converters/column_converter.cpp
+++ b/yt/yt/library/column_converters/column_converter.cpp
@@ -18,37 +18,38 @@ using namespace NTableClient;
 
 IColumnConverterPtr CreateColumnConvert(
     const NTableClient::TColumnSchema& columnSchema,
-    int columnIndex)
+    int columnId,
+    int columnOffset)
 {
     switch (columnSchema.GetWireType()) {
         case EValueType::Int64:
-            return CreateInt64ColumnConverter(columnIndex, columnSchema);
+            return CreateInt64ColumnConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Uint64:
-            return CreateUint64ColumnConverter(columnIndex, columnSchema);
+            return CreateUint64ColumnConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Double:
             switch (columnSchema.CastToV1Type()) {
                 case NTableClient::ESimpleLogicalValueType::Float:
-                    return CreateFloatingPoint32ColumnConverter(columnIndex, columnSchema);
+                    return CreateFloatingPoint32ColumnConverter(columnId, columnSchema, columnOffset);
                 default:
-                    return CreateFloatingPoint64ColumnConverter(columnIndex, columnSchema);
+                    return CreateFloatingPoint64ColumnConverter(columnId, columnSchema, columnOffset);
             }
 
         case EValueType::String:
-            return CreateStringConverter(columnIndex, columnSchema);
+            return CreateStringConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Boolean:
-            return CreateBooleanColumnConverter(columnIndex, columnSchema);
+            return CreateBooleanColumnConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Any:
-            return CreateAnyConverter(columnIndex, columnSchema);
+            return CreateAnyConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Composite:
-            return CreateCompositeConverter(columnIndex, columnSchema);
+            return CreateCompositeConverter(columnId, columnSchema, columnOffset);
 
         case EValueType::Null:
-            return CreateNullConverter(columnIndex);
+            return CreateNullConverter(columnId);
 
         case EValueType::Min:
         case EValueType::TheBottom:
@@ -61,24 +62,43 @@ IColumnConverterPtr CreateColumnConvert(
 ////////////////////////////////////////////////////////////////////////////////
 
 
-TConvertedColumnRange ConvertRowsToColumns(
+TConvertedColumnRange TColumnConverters::ConvertRowsToColumns(
     TRange<TUnversionedRow> rows,
     const std::vector<TColumnSchema>& columnSchema)
 {
     TConvertedColumnRange convertedColumnsRange;
+    if (rows.size() == 0) {
+        return convertedColumnsRange;
+    }
+
+    if(IsFirstBatch_) {
+        // Initialize mapping column ids to indexes, since, for example, in the case of column specification (//path/to/table{column1,column3}), not all column ids will exist
+
+        auto firstRow = rows[0];
+
+        for (const auto* item = firstRow.Begin(); item != firstRow.End(); ++item) {
+            IdsToIndexes_[item->Id] = std::ssize(ColumnIds_);
+            ColumnIds_.push_back(item->Id);
+        }
+    }
+    IsFirstBatch_ = false;
+
     std::vector<TUnversionedRowValues> rowsValues;
     rowsValues.reserve(rows.size());
 
     for (auto row : rows) {
-        TUnversionedRowValues rowValues(columnSchema.size(), nullptr);
+        TUnversionedRowValues rowValues(ColumnIds_.size(), nullptr);
         for (const auto* item = row.Begin(); item != row.End(); ++item) {
-            rowValues[item->Id] = item;
+            auto iter = IdsToIndexes_.find(item->Id);
+            YT_VERIFY(iter != IdsToIndexes_.end());
+            rowValues[iter->second] = item;
         }
         rowsValues.push_back(std::move(rowValues));
     }
 
-    for (int columnId = 0; columnId < std::ssize(columnSchema); columnId++) {
-        auto converter = CreateColumnConvert(columnSchema[columnId], columnId);
+    for (int offset = 0; offset < std::ssize(ColumnIds_); offset++) {
+        YT_VERIFY(ColumnIds_[offset] >= 0 && ColumnIds_[offset] < std::ssize(columnSchema));
+        auto converter = CreateColumnConvert(columnSchema[ColumnIds_[offset]], ColumnIds_[offset], offset);
         auto columns = converter->Convert(rowsValues);
         convertedColumnsRange.push_back(columns);
     }
diff --git a/yt/yt/library/column_converters/column_converter.h b/yt/yt/library/column_converters/column_converter.h
index 13f4b5d133..46515d8eb4 100644
--- a/yt/yt/library/column_converters/column_converter.h
+++ b/yt/yt/library/column_converters/column_converter.h
@@ -45,9 +45,17 @@ using IColumnConverterPtr = std::unique_ptr<IColumnConverter>;
 
 ////////////////////////////////////////////////////////////////////////////////
 
-TConvertedColumnRange ConvertRowsToColumns(
-    TRange<NTableClient::TUnversionedRow> rows,
-    const std::vector<NTableClient::TColumnSchema>& columnSchema);
+class TColumnConverters
+{
+public:
+    TConvertedColumnRange ConvertRowsToColumns(
+        TRange<NTableClient::TUnversionedRow> rows,
+        const std::vector<NTableClient::TColumnSchema>& columnSchema);
+private:
+    THashMap<int, int> IdsToIndexes_;
+    std::vector<int> ColumnIds_;
+    bool IsFirstBatch_ = true;
+};
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/column_converters/floating_point_column_converter.cpp b/yt/yt/library/column_converters/floating_point_column_converter.cpp
index 0dcb86d21a..3613a6ba9a 100644
--- a/yt/yt/library/column_converters/floating_point_column_converter.cpp
+++ b/yt/yt/library/column_converters/floating_point_column_converter.cpp
@@ -52,9 +52,13 @@ class TFloatingPointColumnConverter
 public:
     static_assert(std::is_floating_point_v<TValue>);
 
-    TFloatingPointColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
-        : ColumnIndex_(columnIndex)
+    TFloatingPointColumnConverter(
+        int columnId,
+        const NTableClient::TColumnSchema& columnSchema,
+        int columnOffset)
+        : ColumnId_(columnId)
         , ColumnSchema_(columnSchema)
+        , ColumnOffset_(columnOffset)
 
     { }
 
@@ -80,7 +84,7 @@ public:
             nullBitmapRef);
 
         column->Type = ColumnSchema_.LogicalType();
-        column->Id = ColumnIndex_;
+        column->Id = ColumnId_;
 
         TOwningColumn owner = {
             .Column = std::move(column),
@@ -92,8 +96,9 @@ public:
     }
 
 private:
-    const int ColumnIndex_;
+    const int ColumnId_;
     const TColumnSchema ColumnSchema_;
+    const int ColumnOffset_;
 
     std::vector<TValue> Values_;
     TBitmapOutput NullBitmap_;
@@ -107,7 +112,7 @@ private:
     void AddValues(TRange<TUnversionedRowValues> rowsValues)
     {
         for (const auto& rowValues : rowsValues) {
-            auto value = rowValues[ColumnIndex_];
+            auto value = rowValues[ColumnOffset_];
             bool isNull = !value || value->Type == NTableClient::EValueType::Null;
             TValue valueData = isNull ? 0 : value->Data.Double;
             NullBitmap_.Append(isNull);
@@ -120,14 +125,14 @@ private:
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateFloatingPoint32ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+IColumnConverterPtr CreateFloatingPoint32ColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset)
 {
-    return std::make_unique<TFloatingPointColumnConverter<float, NTableClient::EValueType::Double>>(columnIndex, columnSchema);
+    return std::make_unique<TFloatingPointColumnConverter<float, NTableClient::EValueType::Double>>(columnId, columnSchema, columnOffset);
 }
 
-IColumnConverterPtr CreateFloatingPoint64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+IColumnConverterPtr CreateFloatingPoint64ColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset)
 {
-    return std::make_unique<TFloatingPointColumnConverter<double, NTableClient::EValueType::Double>>(columnIndex, columnSchema);
+    return std::make_unique<TFloatingPointColumnConverter<double, NTableClient::EValueType::Double>>(columnId, columnSchema, columnOffset);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/library/column_converters/floating_point_column_converter.h b/yt/yt/library/column_converters/floating_point_column_converter.h
index 3739d4e729..33387f1d9f 100644
--- a/yt/yt/library/column_converters/floating_point_column_converter.h
+++ b/yt/yt/library/column_converters/floating_point_column_converter.h
@@ -6,9 +6,9 @@ namespace NYT::NColumnConverters {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateFloatingPoint32ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema);
+IColumnConverterPtr CreateFloatingPoint32ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema, int columnOffset);
 
-IColumnConverterPtr CreateFloatingPoint64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema);
+IColumnConverterPtr CreateFloatingPoint64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema, int columnOffset);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/column_converters/integer_column_converter.cpp b/yt/yt/library/column_converters/integer_column_converter.cpp
index 6b3a2c6cb0..f29f8ec108 100644
--- a/yt/yt/library/column_converters/integer_column_converter.cpp
+++ b/yt/yt/library/column_converters/integer_column_converter.cpp
@@ -68,12 +68,14 @@ public:
     static_assert(std::is_integral_v<TValue>);
 
     TIntegerColumnConverter(
-        int columnIndex,
+        int columnId,
         NTableClient::EValueType ValueType,
-        NTableClient::TColumnSchema columnSchema)
-        : ColumnIndex_(columnIndex)
+        NTableClient::TColumnSchema columnSchema,
+        int columnOffset)
+        : ColumnId_(columnId)
         , ColumnSchema_(columnSchema)
         , ValueType_(ValueType)
+        , ColumnOffset_(columnOffset)
     { }
 
     TConvertedColumn Convert(TRange<TUnversionedRowValues> rowsValues) override
@@ -105,7 +107,7 @@ public:
             nullBitmapRef);
 
         column->Type = ColumnSchema_.LogicalType();
-        column->Id = ColumnIndex_;
+        column->Id = ColumnId_;
 
         TOwningColumn owner = {
             .Column = std::move(column),
@@ -118,9 +120,10 @@ public:
 
 
 private:
-    const int ColumnIndex_;
+    const int ColumnId_;
     const NTableClient::TColumnSchema ColumnSchema_;
     const NTableClient::EValueType ValueType_;
+    const int ColumnOffset_;
 
     i64 RowCount_ = 0;
     TBitmapOutput NullBitmap_;
@@ -141,7 +144,7 @@ private:
     void AddValues(TRange<TUnversionedRowValues> rowsValues)
     {
         for (const auto& rowValues : rowsValues) {
-            auto value = rowValues[ColumnIndex_];
+            auto value = rowValues[ColumnOffset_];
             bool isNull = !value || value->Type == NTableClient::EValueType::Null;
             ui64 data = 0;
             if (!isNull) {
@@ -159,15 +162,15 @@ private:
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateInt64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+IColumnConverterPtr CreateInt64ColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset)
 {
-    return std::make_unique<TIntegerColumnConverter<i64>>(columnIndex, NTableClient::EValueType::Int64, columnSchema);
+    return std::make_unique<TIntegerColumnConverter<i64>>(columnId, NTableClient::EValueType::Int64, columnSchema, columnOffset);
 }
 
 
-IColumnConverterPtr CreateUint64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema)
+IColumnConverterPtr CreateUint64ColumnConverter(int columnId, const NTableClient::TColumnSchema& columnSchema, int columnOffset)
 {
-    return std::make_unique<TIntegerColumnConverter<ui64>>(columnIndex, NTableClient::EValueType::Uint64, columnSchema);
+    return std::make_unique<TIntegerColumnConverter<ui64>>(columnId, NTableClient::EValueType::Uint64, columnSchema, columnOffset);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/library/column_converters/integer_column_converter.h b/yt/yt/library/column_converters/integer_column_converter.h
index 99b9d86342..3e41b560cb 100644
--- a/yt/yt/library/column_converters/integer_column_converter.h
+++ b/yt/yt/library/column_converters/integer_column_converter.h
@@ -8,9 +8,15 @@ namespace NYT::NColumnConverters {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateInt64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema);
-
-std::unique_ptr<IColumnConverter> CreateUint64ColumnConverter(int columnIndex, const NTableClient::TColumnSchema& columnSchema);
+IColumnConverterPtr CreateInt64ColumnConverter(
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset);
+
+std::unique_ptr<IColumnConverter> CreateUint64ColumnConverter(
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/column_converters/null_column_converter.cpp b/yt/yt/library/column_converters/null_column_converter.cpp
index f76886ec98..e581f76157 100644
--- a/yt/yt/library/column_converters/null_column_converter.cpp
+++ b/yt/yt/library/column_converters/null_column_converter.cpp
@@ -12,8 +12,8 @@ class TNullColumnWriterConverter
     : public IColumnConverter
 {
 public:
-    TNullColumnWriterConverter(int columnIndex)
-        : ColumnIndex_(columnIndex)
+    TNullColumnWriterConverter(int columnId)
+        : ColumnId_(columnId)
     { }
 
     TConvertedColumn Convert(TRange<TUnversionedRowValues> rowsValues) override
@@ -22,7 +22,7 @@ public:
 
         auto column = std::make_shared<TBatchColumn>();
 
-        column->Id = ColumnIndex_;
+        column->Id = ColumnId_;
         column->Type = SimpleLogicalType(ESimpleLogicalValueType::Null);
         column->ValueCount = rowCount;
 
@@ -34,14 +34,14 @@ public:
     }
 
 private:
-    const int ColumnIndex_;
+    const int ColumnId_;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateNullConverter(int columnIndex)
+IColumnConverterPtr CreateNullConverter(int columnId)
 {
-    return std::make_unique<TNullColumnWriterConverter>(columnIndex);
+    return std::make_unique<TNullColumnWriterConverter>(columnId);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/library/column_converters/null_column_converter.h b/yt/yt/library/column_converters/null_column_converter.h
index a8f97c84a1..9546d8e799 100644
--- a/yt/yt/library/column_converters/null_column_converter.h
+++ b/yt/yt/library/column_converters/null_column_converter.h
@@ -6,7 +6,7 @@ namespace NYT::NColumnConverters {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-IColumnConverterPtr CreateNullConverter(int columnIndex);
+IColumnConverterPtr CreateNullConverter(int columnId);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/column_converters/string_column_converter.cpp b/yt/yt/library/column_converters/string_column_converter.cpp
index cc2154174d..68e1732386 100644
--- a/yt/yt/library/column_converters/string_column_converter.cpp
+++ b/yt/yt/library/column_converters/string_column_converter.cpp
@@ -54,10 +54,12 @@ class TStringConverter
 {
 public:
     TStringConverter(
-        int columnIndex,
-        const TColumnSchema& columnSchema)
-        : ColumnIndex_(columnIndex)
+        int columnId,
+        const TColumnSchema& columnSchema,
+        int columnOffset)
+        : ColumnId_(columnId)
         , ColumnSchema_(columnSchema)
+        , ColumnOffset_(columnOffset)
     { }
 
     TConvertedColumn Convert(TRange<TUnversionedRowValues> rowsValues) override
@@ -68,8 +70,9 @@ public:
     }
 
 private:
-    const int ColumnIndex_;
+    const int ColumnId_;
     const TColumnSchema ColumnSchema_;
+    const int ColumnOffset_;
 
     ui32 RowCount_ = 0;
     ui64 AllStringsSize_ = 0;
@@ -159,7 +162,7 @@ private:
             TRef(nullBitmap));
 
         column->Type = ColumnSchema_.LogicalType();
-        column->Id = ColumnIndex_;
+        column->Id = ColumnId_;
 
         TOwningColumn owner = {
             .Column = std::move(column),
@@ -235,7 +238,7 @@ private:
 
         dictionaryColumn->Type = ColumnSchema_.LogicalType();
         primaryColumn->Type = ColumnSchema_.LogicalType();
-        primaryColumn->Id = ColumnIndex_;
+        primaryColumn->Id = ColumnId_;
 
         TOwningColumn dictOwner = {
             .Column = std::move(dictionaryColumn),
@@ -297,7 +300,7 @@ private:
     void AddValues(TRange<TUnversionedRowValues> rowsValues)
     {
         for (const auto& rowValues : rowsValues) {
-            auto unversionedValue = rowValues[ColumnIndex_];
+            auto unversionedValue = rowValues[ColumnOffset_];
             YT_VERIFY(unversionedValue);
             auto value = CaptureValue(*unversionedValue);
             Values_.push_back(value);
@@ -366,24 +369,27 @@ private:
 ////////////////////////////////////////////////////////////////////////////////
 
 IColumnConverterPtr CreateStringConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema)
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset)
 {
-    return std::make_unique<TStringConverter<EValueType::String>>(columnIndex, columnSchema);
+    return std::make_unique<TStringConverter<EValueType::String>>(columnId, columnSchema, columnOffset);
 }
 
 IColumnConverterPtr CreateAnyConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema)
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset)
 {
-    return std::make_unique<TStringConverter<EValueType::Any>>(columnIndex, columnSchema);
+    return std::make_unique<TStringConverter<EValueType::Any>>(columnId, columnSchema, columnOffset);
 }
 
 IColumnConverterPtr CreateCompositeConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema)
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset)
 {
-    return std::make_unique<TStringConverter<EValueType::Composite>>(columnIndex, columnSchema);
+    return std::make_unique<TStringConverter<EValueType::Composite>>(columnId, columnSchema, columnOffset);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/library/column_converters/string_column_converter.h b/yt/yt/library/column_converters/string_column_converter.h
index b9c3d2bdf7..0425ab802c 100644
--- a/yt/yt/library/column_converters/string_column_converter.h
+++ b/yt/yt/library/column_converters/string_column_converter.h
@@ -9,16 +9,19 @@ namespace NYT::NColumnConverters {
 ////////////////////////////////////////////////////////////////////////////////
 
 IColumnConverterPtr CreateStringConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema);
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset);
 
 IColumnConverterPtr CreateAnyConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema);
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset);
 
 IColumnConverterPtr CreateCompositeConverter(
-    int columnIndex,
-    const NTableClient::TColumnSchema& columnSchema);
+    int columnId,
+    const NTableClient::TColumnSchema& columnSchema,
+    int columnOffset);
 
 ////////////////////////////////////////////////////////////////////////////////
 
diff --git a/yt/yt/library/formats/arrow_writer.cpp b/yt/yt/library/formats/arrow_writer.cpp
index b4b8853571..6874a45ff5 100644
--- a/yt/yt/library/formats/arrow_writer.cpp
+++ b/yt/yt/library/formats/arrow_writer.cpp
@@ -906,6 +906,11 @@ public:
         YT_VERIFY(tableSchemas.size() > 0);
 
         auto tableSchema = tableSchemas[0];
+
+        for (const auto& columnSchema : tableSchema->Columns()) {
+            NameTable_->GetIdOrRegisterName(columnSchema.Name());
+        }
+
         auto columnCount = NameTable_->GetSize();
         SchemaExistenceFlags_.resize(columnCount, true);
 
@@ -933,7 +938,7 @@ private:
     {
         Reset();
 
-        auto convertedColumns = NColumnConverters::ConvertRowsToColumns(rows, ColumnSchemas_);
+        auto convertedColumns = ColumnConverters_.ConvertRowsToColumns(rows, ColumnSchemas_);
 
         std::vector<const TBatchColumn*> rootColumns;
         rootColumns.reserve( std::ssize(convertedColumns));
@@ -985,6 +990,7 @@ private:
     std::vector<TColumnSchema> ColumnSchemas_;
     std::vector<IUnversionedColumnarRowBatch::TDictionaryId> ArrowDictionaryIds_;
     std::vector<bool> SchemaExistenceFlags_;
+    NColumnConverters::TColumnConverters ColumnConverters_;
 
     struct TMessage
     {
diff --git a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
index e997eeed4f..8b0a758fd8 100644
--- a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
+++ b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
@@ -1313,6 +1313,55 @@ message TRspGetTablePivotKeys
 
 ////////////////////////////////////////////////////////////////////////////////
 
+message TTableBackupManifest
+{
+    required string source_path = 1; // YPath
+    required string destination_path = 2; // YPath
+    required int32 ordered_mode = 3;
+}
+
+message TBackupManifest
+{
+    message TClusterManifest
+    {
+        required string cluster = 1;
+        repeated TTableBackupManifest table_manifests = 2;
+    }
+
+    repeated TClusterManifest clusters = 1;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+message TReqCreateTableBackup
+{
+    required TBackupManifest manifest = 1;
+    required uint64 checkpoint_timestamp_delay = 2;
+    required uint64 checkpoint_check_period = 3;
+    required uint64 checkpoint_check_timeout = 4;
+    required bool force = 5;
+}
+
+message TRspCreateTableBackup
+{
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+message TReqRestoreTableBackup
+{
+    required TBackupManifest manifest = 1;
+    required bool force = 2;
+    required bool mount = 3;
+    required bool enable_replicas = 4;
+}
+
+message TRspRestoreTableBackup
+{
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 message TReqCreateNode
 {
     required string path = 1; // YPath
diff --git a/yt/yt_proto/yt/client/bundle_controller/proto/bundle_controller_service.proto b/yt/yt_proto/yt/client/bundle_controller/proto/bundle_controller_service.proto
new file mode 100644
index 0000000000..4629d1e559
--- /dev/null
+++ b/yt/yt_proto/yt/client/bundle_controller/proto/bundle_controller_service.proto
@@ -0,0 +1,52 @@
+package NYT.NBundleController.NProto;
+
+////////////////////////////////////////////////////////////////////////////////
+
+message TCpuLimits
+{
+    optional int32 lookup_thread_pool_size = 1;
+    optional int32 query_thread_pool_size = 2;
+    optional int32 write_thread_pool_size = 3;
+}
+
+message TMemoryLimits
+{
+    optional int64 compressed_block_cache = 1;
+    optional int64 key_filter_block_cache = 2;
+    optional int64 lookup_row_cache = 3;
+
+    optional int64 tablet_dynamic = 4;
+    optional int64 tablet_static = 5;
+
+    optional int64 uncompressed_block_cache = 6;
+    optional int64 versioned_chunk_meta = 7;
+}
+
+message TInstanceResources
+{
+    optional int64 memory = 1;
+    optional int64 net = 2;
+    optional string type = 3;
+    optional int32 vcpu = 4;
+}
+
+message TReqGetBundleConfig
+{
+    required string bundle_name = 1;
+}
+
+message TRspGetBundleConfig
+{
+    optional string bundle_name = 1;
+
+    optional TCpuLimits cpu_limits = 2;
+    optional TMemoryLimits memory_limits = 3;
+
+    optional int32 rpc_proxy_count = 4;
+    optional TInstanceResources rpc_proxy_resource_guarantee = 5;
+
+    optional int32 tablet_node_count = 6;
+    optional TInstanceResources tablet_node_resource_guarantee = 7;
+}
+
+////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt_proto/yt/client/ya.make b/yt/yt_proto/yt/client/ya.make
index 0ac05be114..8c742acce8 100644
--- a/yt/yt_proto/yt/client/ya.make
+++ b/yt/yt_proto/yt/client/ya.make
@@ -14,6 +14,8 @@ SRCS(
     api/rpc_proxy/proto/api_service.proto
     api/rpc_proxy/proto/discovery_service.proto
 
+    bundle_controller/proto/bundle_controller_service.proto
+
     cell_master/proto/cell_directory.proto
 
     chaos_client/proto/replication_card.proto