fix actualization(ttl, tiering, sync) tasks count limit usage (#3234)

Author: ivanmorozov333

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -595,7 +595,7 @@ class TChangesReadTask: public NOlap::NBlobOperations::NRead::ITask {
         }
     }
     virtual bool DoOnError(const TString& storageId, const NOlap::TBlobRange& range, const NOlap::IBlobsReadingAction::TErrorStatus& status) override {
-        AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "DoOnError")("storage_id", storageId)("blob_id", range)("status", status.GetErrorMessage())("status_code", status.GetStatus());
+        AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "DoOnError")("storage_id", storageId)("blob_id", range)("status", status.GetErrorMessage())("status_code", status.GetStatus());
         AFL_VERIFY(false)("blob_id", range)("status", status.GetStatus())("error", status.GetErrorMessage())("type", TxEvent->IndexChanges->TypeString())("task_id", TxEvent->IndexChanges->GetTaskIdentifier())
             ("debug", TxEvent->IndexChanges->DebugString());
         TxEvent->SetPutStatus(NKikimrProto::ERROR);
@@ -614,6 +614,27 @@ class TChangesReadTask: public NOlap::NBlobOperations::NRead::ITask {
     }
 };
 
+class TInsertChangesReadTask: public TChangesReadTask, public TMonitoringObjectsCounter<TInsertChangesReadTask> {
+private:
+    using TBase = TChangesReadTask;
+public:
+    using TBase::TBase;
+};
+
+class TCompactChangesReadTask: public TChangesReadTask, public TMonitoringObjectsCounter<TCompactChangesReadTask> {
+private:
+    using TBase = TChangesReadTask;
+public:
+    using TBase::TBase;
+};
+
+class TTTLChangesReadTask: public TChangesReadTask, public TMonitoringObjectsCounter<TTTLChangesReadTask> {
+private:
+    using TBase = TChangesReadTask;
+public:
+    using TBase::TBase;
+};
+
 void TColumnShard::StartIndexTask(std::vector<const NOlap::TInsertedData*>&& dataToIndex, const i64 bytesToIndex) {
     CSCounters.IndexationInput(bytesToIndex);
 
@@ -637,7 +658,7 @@ void TColumnShard::StartIndexTask(std::vector<const NOlap::TInsertedData*>&& dat
 
     NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription(
         ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
-                                std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), IndexationCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, InsertTaskSubscription));
+                                std::make_shared<TInsertChangesReadTask>(std::move(ev), SelfId(), TabletID(), IndexationCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, InsertTaskSubscription));
 }
 
 void TColumnShard::SetupIndexation() {
@@ -714,7 +735,7 @@ void TColumnShard::SetupCompaction() {
 
         NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription(
             ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
-                std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, CompactTaskSubscription));
+                std::make_shared<TCompactChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, CompactTaskSubscription));
     }
 
     LOG_S_DEBUG("ActiveCompactions: " << BackgroundController.GetCompactionsCount() << " at tablet " << TabletID());
@@ -748,7 +769,7 @@ bool TColumnShard::SetupTtl(const THashMap<ui64, NOlap::TTiering>& pathTtls) {
         if (needWrites) {
             NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription(
                 ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
-                    std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, i->CalcMemoryForUsage(), externalTaskId, TTLTaskSubscription));
+                    std::make_shared<TTTLChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, i->CalcMemoryForUsage(), externalTaskId, TTLTaskSubscription));
         } else {
             ev->SetPutStatus(NKikimrProto::OK);
             ActorContext().Send(SelfId(), std::move(ev));

ydb/core/tx/columnshard/engines/changes/actualization/controller/controller.h

@@ -11,7 +11,7 @@ class TController {
 
 public:
     void StartActualization(const NActualizer::TRWAddress& address) {
-        ++ActualizationsInProgress[address];
+        AFL_VERIFY(++ActualizationsInProgress[address] <= (i32)GetLimitForAddress(address));
     }
 
     void FinishActualization(const NActualizer::TRWAddress& address) {
@@ -23,7 +23,7 @@ class TController {
         if (it == ActualizationsInProgress.end()) {
             return readyTemporaryTasks < GetLimitForAddress(address);
         } else {
-            return readyTemporaryTasks < GetLimitForAddress(address) + it->second;
+            return it->second + readyTemporaryTasks < GetLimitForAddress(address);
         }
     }
 };

ydb/core/tx/columnshard/engines/insert_table/data.h

@@ -153,7 +153,7 @@ class TCommittedBlob {
         , Last(last)
     {}
 
-    /// It uses trick then we place key wtih planStep:txId in container and find them later by BlobId only.
+    /// It uses trick then we place key with planStep:txId in container and find them later by BlobId only.
     /// So hash() and equality should depend on BlobId only.
     bool operator == (const TCommittedBlob& key) const { return BlobRange == key.BlobRange; }
     ui64 Hash() const noexcept { return BlobRange.Hash(); }

ydb/core/tx/columnshard/engines/storage/actualizer/tiering/counters.h

@@ -16,6 +16,7 @@ class TTieringGlobalCounters: public NColumnShard::TCommonCountersOwner {
     std::shared_ptr<NColumnShard::TValueAggregationAgent> DifferenceWaitToEvict;
     std::shared_ptr<NColumnShard::TValueAggregationAgent> DifferenceWaitToDelete;
     NMonitoring::TDynamicCounters::TCounterPtr SkipEvictionForCompaction;
+    NMonitoring::TDynamicCounters::TCounterPtr SkipEvictionForLimit;
 public:
     TTieringGlobalCounters()
         : TBase("TieringActualizer")
@@ -25,6 +26,11 @@ class TTieringGlobalCounters: public NColumnShard::TCommonCountersOwner {
         DifferenceWaitToEvict = TBase::GetValueAutoAggregations("Granule/Eviction/WaitingInSeconds");
         DifferenceWaitToDelete = TBase::GetValueAutoAggregations("Granule/Deletion/WaitingInSeconds");
         SkipEvictionForCompaction = TBase::GetDeriviative("Eviction/SkipForCompaction");
+        SkipEvictionForLimit = TBase::GetDeriviative("Eviction/SkipForLimit");
+    }
+
+    static NMonitoring::TDynamicCounters::TCounterPtr GetSkipEvictionForLimit() {
+        return Singleton<TTieringGlobalCounters>()->SkipEvictionForLimit;
     }
 
     static NMonitoring::TDynamicCounters::TCounterPtr GetSkipEvictionForCompaction() {
@@ -56,14 +62,15 @@ class TTieringCounters {
     const std::shared_ptr<NColumnShard::TValueAggregationClient> DifferenceWaitToEvict;
     const std::shared_ptr<NColumnShard::TValueAggregationClient> DifferenceWaitToDelete;
     const NMonitoring::TDynamicCounters::TCounterPtr SkipEvictionForCompaction;
+    const NMonitoring::TDynamicCounters::TCounterPtr SkipEvictionForLimit;
 
     TTieringCounters()
         : QueueSizeToEvict(TTieringGlobalCounters::BuildQueueSizeToEvict())
         , QueueSizeToDelete(TTieringGlobalCounters::BuildQueueSizeToDelete())
         , DifferenceWaitToEvict(TTieringGlobalCounters::BuildDifferenceWaitToEvict())
         , DifferenceWaitToDelete(TTieringGlobalCounters::BuildDifferenceWaitToDelete())
         , SkipEvictionForCompaction(TTieringGlobalCounters::GetSkipEvictionForCompaction())
-    {
+        , SkipEvictionForLimit(TTieringGlobalCounters::GetSkipEvictionForLimit()) {
     }
 
 };

ydb/core/tx/columnshard/engines/storage/actualizer/tiering/tiering.cpp

@@ -106,6 +106,7 @@ void TTieringActualizer::DoExtractTasks(TTieringProcessContext& tasksContext, co
     THashSet<ui64> portionIds;
     for (auto&& [address, addressPortions] : PortionIdByWaitDuration) {
         if (!tasksContext.IsRWAddressAvailable(address)) {
+            Counters.SkipEvictionForLimit->Add(1);
             continue;
         }
         for (auto&& [duration, portions] : addressPortions.GetPortions()) {