Improve BlobDepot assimilation

8 files changed, 136 insertions, 75 deletions

diff --git a/ydb/core/blob_depot/agent/storage_get.cpp b/ydb/core/blob_depot/agent/storage_get.cpp
index 33229d481e..1a5ee96328 100644
--- a/ydb/core/blob_depot/agent/storage_get.cpp
+++ b/ydb/core/blob_depot/agent/storage_get.cpp
@@ -178,6 +178,8 @@ namespace NKikimr::NBlobDepot {
                     Agent.HandleGetResult(context, **p);
                 } else if (std::holds_alternative<TTabletDisconnected>(response)) {
                     if (auto *resolveContext = dynamic_cast<TResolveKeyContext*>(context.get())) {
+                        STLOG(PRI_DEBUG, BLOB_DEPOT_AGENT, BDA26, "TTabletDisconnected", (VirtualGroupId, Agent.VirtualGroupId),
+                            (QueryId, GetQueryId()), (QueryIdx, resolveContext->QueryIdx));
                         Response->Responses[resolveContext->QueryIdx].Status = NKikimrProto::ERROR;
                         --AnswersRemain;
                         CheckAndFinish();
diff --git a/ydb/core/blob_depot/assimilator.cpp b/ydb/core/blob_depot/assimilator.cpp
index 447e6d5b59..85afed1f7d 100644
--- a/ydb/core/blob_depot/assimilator.cpp
+++ b/ydb/core/blob_depot/assimilator.cpp
@@ -21,16 +21,19 @@ namespace NKikimr::NBlobDepot {
         const NKikimrProto::EReplyStatus Status;
         const TBlobSeqId BlobSeqId;
         const TData::TKey Key;
+        const ui64 GetId;
 
     public:
         TTxType GetTxType() const override { return NKikimrBlobDepot::TXTYPE_COMMIT_ASSIMILATED_BLOB; }
 
-        TTxCommitAssimilatedBlob(TAssimilator *self, NKikimrProto::EReplyStatus status, TBlobSeqId blobSeqId, TData::TKey key)
+        TTxCommitAssimilatedBlob(TAssimilator *self, NKikimrProto::EReplyStatus status, TBlobSeqId blobSeqId, TData::TKey key,
+                ui64 getId)
             : TTransactionBase(self->Self)
             , AssimilatorId(self->SelfId())
             , Status(status)
             , BlobSeqId(blobSeqId)
             , Key(std::move(key))
+            , GetId(getId)
         {}
 
         bool Execute(TTransactionContext& txc, const TActorContext&) override {
@@ -38,7 +41,7 @@ namespace NKikimr::NBlobDepot {
                 Y_VERIFY(!Self->Data->CanBeCollected(BlobSeqId));
                 Self->Data->BindToBlob(Key, BlobSeqId, txc, this);
             } else if (Status == NKikimrProto::NODATA) {
-                if (const TData::TValue *value = Self->Data->FindKey(Key); value && value->ValueChain.empty()) {
+                if (const TData::TValue *value = Self->Data->FindKey(Key); value && value->GoingToAssimilate) {
                     Self->Data->DeleteKey(Key, txc, this);
                 }
             }
@@ -57,7 +60,7 @@ namespace NKikimr::NBlobDepot {
                 }
             }
             Self->Data->CommitTrash(this);
-            TActivationContext::Send(new IEventHandle(TEvPrivate::EvTxComplete, 0, AssimilatorId, {}, nullptr, 0));
+            TActivationContext::Send(new IEventHandle(TEvPrivate::EvTxComplete, 0, AssimilatorId, {}, nullptr, GetId));
         }
     };
 
@@ -107,7 +110,7 @@ namespace NKikimr::NBlobDepot {
             hFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
             hFunc(TEvBlobStorage::TEvControllerGroupDecommittedResponse, Handle);
             cFunc(TEvPrivate::EvResume, Action);
-            cFunc(TEvPrivate::EvTxComplete, HandleTxComplete);
+            fFunc(TEvPrivate::EvTxComplete, HandleTxComplete);
             cFunc(TEvents::TSystem::Poison, PassAway);
 
             default:
@@ -273,73 +276,83 @@ namespace NKikimr::NBlobDepot {
 
     void TAssimilator::ScanDataForCopying() {
         STLOG(PRI_DEBUG, BLOB_DEPOT, BDT54, "TAssimilator::ScanDataForCopying", (Id, Self->GetLogId()),
-            (LastScannedKey, LastScannedKey));
+            (LastScannedKey, LastScannedKey), (NumGetsUnprocessed, GetIdToUnprocessedPuts.size()));
 
-        std::deque<TLogoBlobID> scanQ;
-        ui32 totalSize = 0;
         THPTimer timer;
-        ui32 numItems = 0;
 
-        auto callback = [&](const TData::TKey& key, const TData::TValue& value) {
-            if (++numItems == 1000) {
-                numItems = 0;
-                if (TDuration::Seconds(timer.Passed()) >= TDuration::MilliSeconds(1)) {
-                    return false;
+        while (GetIdToUnprocessedPuts.size() < MaxGetsUnprocessed) {
+            ui32 numItems = 0;
+            bool timeout = false;
+
+            auto callback = [&](const TData::TKey& key, const TData::TValue& value) {
+                if (++numItems == 1000) {
+                    numItems = 0;
+                    if (TDuration::Seconds(timer.Passed()) >= TDuration::MilliSeconds(1)) {
+                        timeout = true;
+                        return false;
+                    }
                 }
-            }
-            const TLogoBlobID& id = key.GetBlobId();
-            if (!value.ValueChain.empty()) {
-                LastScannedKey.emplace(id);
-                return true; // keep scanning
-            } else if (scanQ.empty() || scanQ.front().TabletID() == id.TabletID()) {
-                LastScannedKey.emplace(id);
-                scanQ.push_back(id);
-                totalSize += id.BlobSize();
-                EntriesToProcess = true;
-                return totalSize < MaxSizeToQuery;
-            } else {
-                return false; // a blob belonging to different tablet
-            }
-        };
+                const TLogoBlobID& id = key.GetBlobId();
+                if (!value.GoingToAssimilate) {
+                    LastScannedKey.emplace(id);
+                    return true; // keep scanning
+                } else if (ScanQ.empty() || ScanQ.front().TabletID() == id.TabletID()) {
+                    LastScannedKey.emplace(id);
+                    ScanQ.push_back(id);
+                    TotalSize += id.BlobSize();
+                    EntriesToProcess = true;
+                    return TotalSize < MaxSizeToQuery;
+                } else {
+                    return false; // a blob belonging to different tablet
+                }
+            };
 
-        // FIXME: reentrable as it shares mailbox with the BlobDepot tablet itself
-        const bool finished = Self->Data->ScanRange(
-            LastScannedKey ? std::make_optional<TData::TKey>(*LastScannedKey) : std::nullopt,
-            std::optional<TData::TKey>(), {}, callback);
-
-        STLOG(PRI_DEBUG, BLOB_DEPOT, BDT56, "ScanDataForCopying done", (Id, Self->GetLogId()),
-            (LastScannedKey, LastScannedKey), (ScanQ.size, scanQ.size()), (EntriesToProcess, EntriesToProcess),
-            (Finished, finished));
-
-        if (!scanQ.empty()) {
-            using TQuery = TEvBlobStorage::TEvGet::TQuery;
-            const ui32 sz = scanQ.size();
-            TArrayHolder<TQuery> queries(new TQuery[sz]);
-            TQuery *query = queries.Get();
-            for (const TLogoBlobID& id : scanQ) {
-                query->Set(id);
-                ++query;
+            Self->Data->ScanRange(LastScannedKey ? std::make_optional<TData::TKey>(*LastScannedKey) : std::nullopt,
+                std::optional<TData::TKey>(), {}, callback);
+
+            STLOG(PRI_DEBUG, BLOB_DEPOT, BDT56, "ScanDataForCopying step", (Id, Self->GetLogId()),
+                (LastScannedKey, LastScannedKey), (ScanQ.size, ScanQ.size()), (TotalSize, TotalSize),
+                (EntriesToProcess, EntriesToProcess), (Timeout, timeout), (NumGetsUnprocessed, GetIdToUnprocessedPuts.size()));
+
+            if (timeout) { // timeout hit, reschedule work
+                TActivationContext::Send(new IEventHandle(TEvPrivate::EvResume, 0, SelfId(), {}, nullptr, 0));
+            } else if (!ScanQ.empty()) {
+                using TQuery = TEvBlobStorage::TEvGet::TQuery;
+                const ui32 sz = ScanQ.size();
+                TArrayHolder<TQuery> queries(new TQuery[sz]);
+                TQuery *query = queries.Get();
+                for (const TLogoBlobID& id : ScanQ) {
+                    query->Set(id);
+                    ++query;
+                }
+                auto ev = std::make_unique<TEvBlobStorage::TEvGet>(queries, sz, TInstant::Max(), NKikimrBlobStorage::EGetHandleClass::FastRead);
+                ev->Decommission = true;
+                const ui64 getId = NextGetId++;
+                SendToBSProxy(SelfId(), Self->Config.GetVirtualGroupId(), ev.release(), getId);
+                GetIdToUnprocessedPuts.try_emplace(getId);
+                ScanQ.clear();
+                TotalSize = 0;
+            } else if (!GetIdToUnprocessedPuts.empty()) { // there are some unprocessed get queries, still have to wait
+                break;
+            } else if (!EntriesToProcess) { // we have finished scanning the whole table without any entries, copying is done
+                OnCopyDone();
+                break;
+            } else { // we have finished scanning, but we have replicated some data, restart scanning to ensure that nothing left
+                LastScannedKey.reset();
+                EntriesToProcess = false;
             }
-            auto ev = std::make_unique<TEvBlobStorage::TEvGet>(queries, sz, TInstant::Max(), NKikimrBlobStorage::EGetHandleClass::FastRead);
-            ev->Decommission = true;
-            SendToBSProxy(SelfId(), Self->Config.GetVirtualGroupId(), ev.release());
-        } else if (!finished) { // timeout hit, reschedule work
-            TActivationContext::Send(new IEventHandle(TEvPrivate::EvResume, 0, SelfId(), {}, nullptr, 0));
-        } else if (!EntriesToProcess) { // we have finished scanning the whole table without any entries, copying is done
-            OnCopyDone();
-        } else { // we have finished scanning, but we have replicated some data, restart scanning to ensure that nothing left
-            LastScannedKey.reset();
-            EntriesToProcess = false;
-            TActivationContext::Send(new IEventHandle(TEvPrivate::EvResume, 0, SelfId(), {}, nullptr, 0));
         }
     }
 
     void TAssimilator::Handle(TEvBlobStorage::TEvGetResult::TPtr ev) {
         auto& msg = *ev->Get();
+        const auto it = GetIdToUnprocessedPuts.find(ev->Cookie);
+        Y_VERIFY(it != GetIdToUnprocessedPuts.end());
+        ui32 getBytes = 0;
         for (ui32 i = 0; i < msg.ResponseSz; ++i) {
             auto& resp = msg.Responses[i];
             STLOG(PRI_DEBUG, BLOB_DEPOT, BDT34, "got TEvGetResult", (Id, Self->GetLogId()), (BlobId, resp.Id),
-                (Status, resp.Status), (NumPutsInFlight, NumPutsInFlight));
+                (Status, resp.Status), (NumGetsUnprocessed, GetIdToUnprocessedPuts.size()));
             if (resp.Status == NKikimrProto::OK) {
                 std::vector<ui8> channels(1);
                 Self->PickChannels(NKikimrBlobDepot::TChannelKind::Data, channels);
@@ -351,34 +364,46 @@ namespace NKikimr::NBlobDepot {
                 SendToBSProxy(SelfId(), channel.GroupId, new TEvBlobStorage::TEvPut(id, resp.Buffer, TInstant::Max()), putId);
                 const bool inserted = channel.AssimilatedBlobsInFlight.insert(value).second; // prevent from barrier advancing
                 Y_VERIFY(inserted);
-                const bool inserted1 = PutIdToKey.emplace(putId, TData::TKey(resp.Id)).second;
+                const bool inserted1 = PutIdToKey.try_emplace(putId, TData::TKey(resp.Id), it->first).second;
                 Y_VERIFY(inserted1);
-                ++NumPutsInFlight;
+                ++it->second;
+                getBytes += id.BlobSize();
             } else if (resp.Status == NKikimrProto::NODATA) {
                 Self->Execute(std::make_unique<TTxCommitAssimilatedBlob>(this, NKikimrProto::NODATA, TBlobSeqId(),
-                    TData::TKey(resp.Id)));
-                ++NumPutsInFlight;
+                    TData::TKey(resp.Id), it->first));
+                ++it->second;
             }
         }
-        if (!NumPutsInFlight) {
+        if (getBytes) {
+            Self->TabletCounters->Cumulative()[NKikimrBlobDepot::COUNTER_DECOMMIT_GET_BYTES] += getBytes;
+        }
+        if (!it->second) {
+            GetIdToUnprocessedPuts.erase(it);
             Action();
         }
     }
 
-    void TAssimilator::HandleTxComplete() {
-        if (!--NumPutsInFlight) {
+    void TAssimilator::HandleTxComplete(TAutoPtr<IEventHandle> ev) {
+        const auto it = GetIdToUnprocessedPuts.find(ev->Cookie);
+        Y_VERIFY(it != GetIdToUnprocessedPuts.end());
+        if (!--it->second) {
+            GetIdToUnprocessedPuts.erase(it);
             Action();
         }
     }
 
     void TAssimilator::Handle(TEvBlobStorage::TEvPutResult::TPtr ev) {
         auto& msg = *ev->Get();
+        if (msg.Status == NKikimrProto::OK) {
+            Self->TabletCounters->Cumulative()[NKikimrBlobDepot::COUNTER_DECOMMIT_PUT_OK_BYTES] += msg.Id.BlobSize();
+        }
         const auto it = PutIdToKey.find(ev->Cookie);
         Y_VERIFY(it != PutIdToKey.end());
+        const auto& [key, getId] = it->second;
         STLOG(PRI_DEBUG, BLOB_DEPOT, BDT37, "got TEvPutResult", (Id, Self->GetLogId()), (Msg, msg),
-            (NumPutsInFlight, NumPutsInFlight), (Key, it->second));
+            (NumGetsUnprocessed, GetIdToUnprocessedPuts.size()), (Key, key));
         Self->Execute(std::make_unique<TTxCommitAssimilatedBlob>(this, msg.Status, TBlobSeqId::FromLogoBlobId(msg.Id),
-            std::move(it->second)));
+            std::move(key), getId));
         PutIdToKey.erase(it);
     }
 
diff --git a/ydb/core/blob_depot/assimilator.h b/ydb/core/blob_depot/assimilator.h
index 0cde2fc524..427097f4d0 100644
--- a/ydb/core/blob_depot/assimilator.h
+++ b/ydb/core/blob_depot/assimilator.h
@@ -24,14 +24,19 @@ namespace NKikimr::NBlobDepot {
         std::optional<TLogoBlobID> LastScannedKey;
         bool EntriesToProcess = false;
 
-        static constexpr ui32 MaxSizeToQuery = 10'000'000;
+        static constexpr ui32 MaxSizeToQuery = 16'000'000;
 
-        ui32 NumPutsInFlight = 0;
+        static constexpr ui32 MaxGetsUnprocessed = 5;
+        ui64 NextGetId = 1;
+        std::unordered_map<ui64, ui32> GetIdToUnprocessedPuts;
+
+        std::deque<TLogoBlobID> ScanQ;
+        ui32 TotalSize = 0;
 
         TActorId PipeId;
 
         ui64 NextPutId = 1;
-        THashMap<ui64, TData::TKey> PutIdToKey;
+        THashMap<ui64, std::tuple<TData::TKey, ui64>> PutIdToKey;
 
         class TTxCommitAssimilatedBlob;
 
@@ -58,7 +63,7 @@ namespace NKikimr::NBlobDepot {
         void Handle(TEvBlobStorage::TEvAssimilateResult::TPtr ev);
         void ScanDataForCopying();
         void Handle(TEvBlobStorage::TEvGetResult::TPtr ev);
-        void HandleTxComplete();
+        void HandleTxComplete(TAutoPtr<IEventHandle> ev);
         void Handle(TEvBlobStorage::TEvPutResult::TPtr ev);
         void OnCopyDone();
         void CreatePipe();
diff --git a/ydb/core/blob_depot/data.cpp b/ydb/core/blob_depot/data.cpp
index c4814bc265..fd509ead95 100644
--- a/ydb/core/blob_depot/data.cpp
+++ b/ydb/core/blob_depot/data.cpp
@@ -90,6 +90,7 @@ namespace NKikimr::NBlobDepot {
             std::vector<TLogoBlobID> deleteQ;
             const bool uncertainWriteBefore = value.UncertainWrite;
             const bool wasUncertain = value.IsWrittenUncertainly();
+            const bool wasGoingToAssimilate = value.GoingToAssimilate;
 
             if (!inserted) {
                 EnumerateBlobsForValueChain(value.ValueChain, Self->TabletID(), [&](TLogoBlobID id, ui32, ui32) {
@@ -162,6 +163,9 @@ namespace NKikimr::NBlobDepot {
                     Data.erase(it);
                     row.Delete();
                     ValidateRecords();
+                    if (wasGoingToAssimilate) {
+                        Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_BYTES_TO_DECOMMIT] -= key.GetBlobId().BlobSize();
+                    }
                     return true;
 
                 case EUpdateOutcome::CHANGE:
@@ -176,6 +180,10 @@ namespace NKikimr::NBlobDepot {
                     if (wasUncertain && !value.IsWrittenUncertainly()) {
                         UncertaintyResolver->MakeKeyCertain(key);
                     }
+                    if (wasGoingToAssimilate != value.GoingToAssimilate) {
+                        const i64 sign = value.GoingToAssimilate - wasGoingToAssimilate;
+                        Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_BYTES_TO_DECOMMIT] += key.GetBlobId().BlobSize() * sign;
+                    }
                     ValidateRecords();
                     return true;
 
@@ -206,6 +214,9 @@ namespace NKikimr::NBlobDepot {
                 auto *chain = value.ValueChain.Add();
                 auto *locator = chain->MutableLocator();
                 locator->CopyFrom(item.GetBlobLocator());
+
+                // clear assimilation flag -- we have blob overwritten with fresh copy (of the same data)
+                value.GoingToAssimilate = false;
             }
 
             return EUpdateOutcome::CHANGE;
@@ -216,17 +227,17 @@ namespace NKikimr::NBlobDepot {
         STLOG(PRI_DEBUG, BLOB_DEPOT, BDT49, "BindToBlob", (Id, Self->GetLogId()), (Key, key), (BlobSeqId, blobSeqId));
         Y_VERIFY(Loaded || IsKeyLoaded(key));
         UpdateKey(key, txc, cookie, "BindToBlob", [&](TValue& value, bool inserted) {
-            if (!value.ValueChain.empty()) {
-                return EUpdateOutcome::NO_CHANGE;
-            } else {
+            if (inserted || value.GoingToAssimilate) {
                 auto *chain = value.ValueChain.Add();
                 auto *locator = chain->MutableLocator();
                 locator->SetGroupId(Self->Info()->GroupFor(blobSeqId.Channel, blobSeqId.Generation));
                 blobSeqId.ToProto(locator->MutableBlobSeqId());
                 locator->SetTotalDataLen(key.GetBlobId().BlobSize());
                 locator->SetFooterLen(0);
+                value.GoingToAssimilate = false;
                 return inserted ? EUpdateOutcome::DROP : EUpdateOutcome::CHANGE;
             }
+            return EUpdateOutcome::NO_CHANGE;
         });
     }
 
@@ -333,6 +344,9 @@ namespace NKikimr::NBlobDepot {
                 AddFirstMentionedBlob(id);
             }
         });
+        if (it->second.GoingToAssimilate) {
+            Self->TabletCounters->Simple()[NKikimrBlobDepot::COUNTER_BYTES_TO_DECOMMIT] += it->first.GetBlobId().BlobSize();
+        }
 
         ValidateRecords();
     }
@@ -353,6 +367,11 @@ namespace NKikimr::NBlobDepot {
                 change = true;
             }
 
+            if (value.ValueChain.empty() && !value.GoingToAssimilate) {
+                value.GoingToAssimilate = true;
+                change = true;
+            }
+
             return change ? EUpdateOutcome::CHANGE : EUpdateOutcome::NO_CHANGE;
         });
     }
diff --git a/ydb/core/blob_depot/data.h b/ydb/core/blob_depot/data.h
index 8d4366bfe0..536aafd09b 100644
--- a/ydb/core/blob_depot/data.h
+++ b/ydb/core/blob_depot/data.h
@@ -223,6 +223,7 @@ namespace NKikimr::NBlobDepot {
             TValueChain ValueChain;
             EKeepState KeepState = EKeepState::Default;
             bool Public = false;
+            bool GoingToAssimilate = false;
             bool UncertainWrite = false;
 
             TValue() = default;
@@ -237,6 +238,7 @@ namespace NKikimr::NBlobDepot {
                 , ValueChain(std::move(*proto.MutableValueChain()))
                 , KeepState(proto.GetKeepState())
                 , Public(proto.GetPublic())
+                , GoingToAssimilate(proto.GetGoingToAssimilate())
                 , UncertainWrite(uncertainWrite)
             {}
 
@@ -273,6 +275,9 @@ namespace NKikimr::NBlobDepot {
                 if (Public != proto->GetPublic()) {
                     proto->SetPublic(Public);
                 }
+                if (GoingToAssimilate != proto->GetGoingToAssimilate()) {
+                    proto->SetGoingToAssimilate(GoingToAssimilate);
+                }
             }
 
             static bool Validate(const NKikimrBlobDepot::TEvCommitBlobSeq::TItem& item);
@@ -298,6 +303,7 @@ namespace NKikimr::NBlobDepot {
                     << " ValueChain# " << FormatList(ValueChain)
                     << " KeepState# " << EKeepState_Name(KeepState)
                     << " Public# " << (Public ? "true" : "false")
+                    << " GoingToAssimilate# " << (GoingToAssimilate ? "true" : "false")
                     << " UncertainWrite# " << (UncertainWrite ? "true" : "false")
                     << "}";
             }
diff --git a/ydb/core/blob_depot/data_resolve.cpp b/ydb/core/blob_depot/data_resolve.cpp
index 87b8ea0131..c81d1756cb 100644
--- a/ydb/core/blob_depot/data_resolve.cpp
+++ b/ydb/core/blob_depot/data_resolve.cpp
@@ -343,7 +343,7 @@ namespace NKikimr::NBlobDepot {
                 item.SetCookie(*cookie);
             }
             item.SetKey(key.MakeBinaryKey());
-            if (value.ValueChain.empty() && Self->Config.GetIsDecommittingGroup()) {
+            if (value.GoingToAssimilate && Self->Config.GetIsDecommittingGroup()) {
                 Y_VERIFY(!value.UncertainWrite);
                 auto *out = item.AddValueChain();
                 out->SetGroupId(Self->Config.GetVirtualGroupId());
@@ -448,7 +448,7 @@ namespace NKikimr::NBlobDepot {
                     }
                     if (minId == maxId) {
                         const auto it = Data.find(TKey(minId));
-                        if (it != Data.end() && !it->second.ValueChain.empty()) {
+                        if (it != Data.end() && !it->second.GoingToAssimilate) {
                             continue; // fast path for extreme queries
                         }
                     }
diff --git a/ydb/core/protos/blob_depot.proto b/ydb/core/protos/blob_depot.proto
index d1233c4e96..8b0292a03f 100644
--- a/ydb/core/protos/blob_depot.proto
+++ b/ydb/core/protos/blob_depot.proto
@@ -36,6 +36,7 @@ message TValue {
     repeated TValueChain ValueChain = 2;
     optional EKeepState KeepState = 3;
     optional bool Public = 4;
+    optional bool GoingToAssimilate = 5;
 }
 
 message TGivenIdRange {
diff --git a/ydb/core/protos/counters_blob_depot.proto b/ydb/core/protos/counters_blob_depot.proto
index 16225814a8..bf79b1ff13 100644
--- a/ydb/core/protos/counters_blob_depot.proto
+++ b/ydb/core/protos/counters_blob_depot.proto
@@ -8,12 +8,15 @@ enum ESimpleCounters {
     COUNTER_TOTAL_STORED_DATA_SIZE = 0 [(NKikimr.CounterOpts) = {Name: "TotalStoredDataSize"}];
     COUNTER_TOTAL_STORED_TRASH_SIZE = 1 [(NKikimr.CounterOpts) = {Name: "TotalStoredTrashSize"}];
     COUNTER_IN_FLIGHT_TRASH_SIZE = 2 [(NKikimr.CounterOpts) = {Name: "InFlightTrashSize"}];
+    COUNTER_BYTES_TO_DECOMMIT = 3 [(NKikimr.CounterOpts) = {Name: "BytesToDecommit"}];
 }
 
 enum ECumulativeCounters {
     COUNTER_PUTS_INCOMING = 0 [(NKikimr.CounterOpts) = {Name: "Puts/Incoming"}];
     COUNTER_PUTS_OK = 1 [(NKikimr.CounterOpts) = {Name: "Puts/Ok"}];
     COUNTER_PUTS_ERROR = 2 [(NKikimr.CounterOpts) = {Name: "Puts/Error"}];
+    COUNTER_DECOMMIT_GET_BYTES = 3 [(NKikimr.CounterOpts) = {Name: "Decommit/GetBytes"}];
+    COUNTER_DECOMMIT_PUT_OK_BYTES = 4 [(NKikimr.CounterOpts) = {Name: "Decommit/PutOkBytes"}];
 }
 
 enum EPercentileCounters {