Memory control and search fix

ydb/core/protos/config.proto

@@ -1440,6 +1440,7 @@ message TColumnShardConfig {
     optional uint32 WritingBufferDurationMs = 8 [default = 0];
     optional bool UseChunkedMergeOnCompaction = 9 [default = true];
     optional uint64 CompactionMemoryLimit = 10 [default = 536870912];
+    optional uint64 TieringsMemoryLimit = 11 [default = 536870912];
 }
 
 message TSchemeShardConfig {

ydb/core/tablet/resource_broker.cpp

@@ -1269,6 +1269,7 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
     const ui64 KqpRmQueueCPU = 4;
     const ui64 KqpRmQueueMemory = 10ULL << 30;
 
+    const ui64 CSTTLCompactionMemoryLimit = 1ULL << 30;
     const ui64 CSInsertCompactionMemoryLimit = 1ULL << 30;
     const ui64 CSGeneralCompactionMemoryLimit = 3ULL << 30;
     const ui64 CSScanMemoryLimit = 3ULL << 30;
@@ -1314,6 +1315,12 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
     queue->MutableLimit()->SetCpu(3);
     queue->MutableLimit()->SetMemory(CSInsertCompactionMemoryLimit);
 
+    queue = config.AddQueues();
+    queue->SetName("queue_cs_ttl");
+    queue->SetWeight(100);
+    queue->MutableLimit()->SetCpu(3);
+    queue->MutableLimit()->SetMemory(CSTTLCompactionMemoryLimit);
+
     queue = config.AddQueues();
     queue->SetName("queue_cs_general");
     queue->SetWeight(100);
@@ -1413,6 +1420,11 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
     task->SetQueueName("queue_compaction_borrowed");
     task->SetDefaultDuration(TDuration::Minutes(10).GetValue());
 
+    task = config.AddTasks();
+    task->SetName("CS::TTL");
+    task->SetQueueName("queue_cs_ttl");
+    task->SetDefaultDuration(TDuration::Minutes(10).GetValue());
+
     task = config.AddTasks();
     task->SetName("CS::INDEXATION");
     task->SetQueueName("queue_cs_indexation");

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -89,6 +89,7 @@ TColumnShard::TColumnShard(TTabletStorageInfo* info, const TActorId& tablet)
     , SubscribeCounters(std::make_shared<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters>())
     , InsertTaskSubscription(NOlap::TInsertColumnEngineChanges::StaticTypeName(), SubscribeCounters)
     , CompactTaskSubscription(NOlap::TCompactColumnEngineChanges::StaticTypeName(), SubscribeCounters)
+    , TTLTaskSubscription(NOlap::TTTLColumnEngineChanges::StaticTypeName(), SubscribeCounters)
     , ReadCounters("Read")
     , ScanCounters("Scan")
     , WritesMonitor(*this)
@@ -744,21 +745,25 @@ bool TColumnShard::SetupTtl(const THashMap<ui64, NOlap::TTiering>& pathTtls, con
     }
 
     auto actualIndexInfo = TablesManager.GetPrimaryIndex()->GetVersionedIndex();
-    std::shared_ptr<NOlap::TTTLColumnEngineChanges> indexChanges = TablesManager.MutablePrimaryIndex().StartTtl(eviction, BackgroundController.GetConflictTTLPortions());
+    const ui64 memoryUsageLimit = HasAppData() ? AppDataVerified().ColumnShardConfig.GetTieringsMemoryLimit() : ((ui64)512 * 1024 * 1024);
+    std::shared_ptr<NOlap::TTTLColumnEngineChanges> indexChanges = TablesManager.MutablePrimaryIndex().StartTtl(
+        eviction, BackgroundController.GetConflictTTLPortions(), memoryUsageLimit);
 
     if (!indexChanges) {
         ACFL_DEBUG("background", "ttl")("skip_reason", "no_changes");
         return false;
     }
-
+    const TString externalTaskId = indexChanges->GetTaskIdentifier();
     const bool needWrites = indexChanges->NeedConstruction();
     ACFL_DEBUG("background", "ttl")("need_writes", needWrites);
 
     indexChanges->Start(*this);
     auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move(actualIndexInfo), indexChanges, false);
     NYDBTest::TControllers::GetColumnShardController()->OnWriteIndexStart(TabletID(), indexChanges->TypeString());
     if (needWrites) {
-        ActorContext().Register(new NOlap::NBlobOperations::NRead::TActor(std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters)));
+        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, TTLTaskSubscription));
     } else {
         ev->SetPutStatus(NKikimrProto::OK);
         ActorContext().Send(SelfId(), std::move(ev));

ydb/core/tx/columnshard/columnshard_impl.h

@@ -421,6 +421,7 @@ class TColumnShard
     std::shared_ptr<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters> SubscribeCounters;
     NOlap::NResourceBroker::NSubscribe::TTaskContext InsertTaskSubscription;
     NOlap::NResourceBroker::NSubscribe::TTaskContext CompactTaskSubscription;
+    NOlap::NResourceBroker::NSubscribe::TTaskContext TTLTaskSubscription;
     const TScanCounters ReadCounters;
     const TScanCounters ScanCounters;
     const TIndexationCounters CompactionCounters = TIndexationCounters("GeneralCompaction");

ydb/core/tx/columnshard/common/protos/snapshot.proto

@@ -0,0 +1,6 @@
+package NKikimrColumnShardProto;
+
+message TSnapshot {
+    optional uint64 PlanStep = 1;
+    optional uint64 TxId = 2;
+}

ydb/core/tx/columnshard/common/protos/ya.make

@@ -0,0 +1,10 @@
+PROTO_LIBRARY()
+
+SRCS(
+    snapshot.proto
+)
+
+PEERDIR(
+)
+
+END()

ydb/core/tx/columnshard/common/snapshot.cpp

@@ -1,4 +1,5 @@
 #include "snapshot.h"
+#include <ydb/core/tx/columnshard/common/protos/snapshot.pb.h>
 #include <util/string/builder.h>
 
 namespace NKikimr::NOlap {
@@ -7,4 +8,10 @@ TString TSnapshot::DebugString() const {
     return TStringBuilder() << "plan_step=" << PlanStep << ";tx_id=" << TxId << ";";
 }
 
+NKikimrColumnShardProto::TSnapshot TSnapshot::SerializeToProto() const {
+    NKikimrColumnShardProto::TSnapshot result;
+    SerializeToProto(result);
+    return result;
+}
+
 };

ydb/core/tx/columnshard/common/snapshot.h

@@ -1,6 +1,11 @@
 #pragma once
 #include <util/stream/output.h>
 #include <util/string/cast.h>
+#include <ydb/library/conclusion/status.h>
+
+namespace NKikimrColumnShardProto {
+class TSnapshot;
+}
 
 namespace NKikimr::NOlap {
 
@@ -47,6 +52,27 @@ class TSnapshot {
         return out << "{" << s.PlanStep << ':' << (s.TxId == std::numeric_limits<ui64>::max() ? "max" : ::ToString(s.TxId)) << "}";
     }
 
+    template <class TProto>
+    void SerializeToProto(TProto& result) const {
+        result.SetPlanStep(PlanStep);
+        result.SetTxId(TxId);
+    }
+
+    NKikimrColumnShardProto::TSnapshot SerializeToProto() const;
+
+    template <class TProto>
+    TConclusionStatus DeserializeFromProto(const TProto& proto) {
+        PlanStep = proto.GetPlanStep();
+        TxId = proto.GetTxId();
+        if (!PlanStep) {
+            return TConclusionStatus::Fail("incorrect planStep in proto");
+        }
+        if (!TxId) {
+            return TConclusionStatus::Fail("incorrect txId in proto");
+        }
+        return TConclusionStatus::Success();
+    }
+
     TString DebugString() const;
 };
 

ydb/core/tx/columnshard/common/ya.make

@@ -11,6 +11,7 @@ PEERDIR(
     ydb/core/protos
     contrib/libs/apache/arrow
     ydb/core/formats/arrow
+    ydb/core/tx/columnshard/common/protos
 )
 
 GENERATE_ENUM_SERIALIZATION(portion.h)

ydb/core/tx/columnshard/engines/changes/abstract/abstract.h

@@ -173,6 +173,12 @@ class TColumnEngineChanges {
     const TString TaskIdentifier = TGUID::Create().AsGuidString();
     virtual ui64 DoCalcMemoryForUsage() const = 0;
 public:
+    class IMemoryPredictor {
+    public:
+        virtual ui64 AddPortion(const TPortionInfo& portionInfo) = 0;
+        virtual ~IMemoryPredictor() = default;
+    };
+
     ui64 CalcMemoryForUsage() const {
         return DoCalcMemoryForUsage();
     }

ydb/core/tx/columnshard/engines/changes/general_compaction.h

@@ -29,12 +29,6 @@ class TGeneralCompactColumnEngineChanges: public TCompactColumnEngineChanges {
 public:
     using TBase::TBase;
 
-    class IMemoryPredictor {
-    public:
-        virtual ui64 AddPortion(const TPortionInfo& portionInfo) = 0;
-        virtual ~IMemoryPredictor() = default;
-    };
-
     class TMemoryPredictorSimplePolicy: public IMemoryPredictor {
     private:
         ui64 SumMemory = 0;

ydb/core/tx/columnshard/engines/changes/ttl.h

@@ -51,13 +51,32 @@ class TTTLColumnEngineChanges: public TChangesWithAppend {
     virtual TConclusionStatus DoConstructBlobs(TConstructionContext& context) noexcept override;
     virtual NColumnShard::ECumulativeCounters GetCounterIndex(const bool isSuccess) const override;
     virtual ui64 DoCalcMemoryForUsage() const override {
+        auto predictor = BuildMemoryPredictor();
         ui64 result = 0;
         for (auto& p : PortionsToEvict) {
-            result += 2 * p.GetPortionInfo().GetBlobBytes();
+            result = predictor->AddPortion(p.GetPortionInfo());
         }
         return result;
     }
 public:
+    class TMemoryPredictorSimplePolicy: public IMemoryPredictor {
+    private:
+        ui64 SumBlobsMemory = 0;
+        ui64 MaxRawMemory = 0;
+    public:
+        virtual ui64 AddPortion(const TPortionInfo& portionInfo) override {
+            if (MaxRawMemory < portionInfo.GetRawBytes()) {
+                MaxRawMemory = portionInfo.GetRawBytes();
+            }
+            SumBlobsMemory += portionInfo.GetBlobBytes();
+            return SumBlobsMemory + MaxRawMemory;
+        }
+    };
+
+    static std::shared_ptr<IMemoryPredictor> BuildMemoryPredictor() {
+        return std::make_shared<TMemoryPredictorSimplePolicy>();
+    }
+
     virtual bool NeedConstruction() const override {
         return PortionsToEvict.size();
     }

ydb/core/tx/columnshard/engines/column_engine.h

@@ -375,8 +375,8 @@ class IColumnEngine {
     virtual std::shared_ptr<TColumnEngineChanges> StartCompaction(const TCompactionLimits& limits, const THashSet<TPortionAddress>& busyPortions) noexcept = 0;
     virtual std::shared_ptr<TCleanupColumnEngineChanges> StartCleanup(const TSnapshot& snapshot, THashSet<ui64>& pathsToDrop,
                                                                ui32 maxRecords) noexcept = 0;
-    virtual std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions,
-                                                           ui64 maxBytesToEvict = TCompactionLimits::DEFAULT_EVICTION_BYTES) noexcept = 0;
+    virtual std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction,
+        const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept = 0;
     virtual bool ApplyChanges(IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> changes, const TSnapshot& snapshot) noexcept = 0;
     virtual void RegisterSchemaVersion(const TSnapshot& snapshot, TIndexInfo&& info) = 0;
     virtual const TMap<ui64, std::shared_ptr<TColumnEngineStats>>& GetStats() const = 0;

ydb/core/tx/columnshard/engines/column_engine_logs.cpp

@@ -312,7 +312,6 @@ std::shared_ptr<TCleanupColumnEngineChanges> TColumnEngineForLogs::StartCleanup(
 
 TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering& ttl, TTieringProcessContext& context) const {
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "ProcessTiering")("path_id", pathId)("ttl", ttl.GetDebugString());
-    ui64 evictionSize = 0;
     ui64 dropBlobs = 0;
     auto& indexInfo = VersionedIndex.GetLastSchema()->GetIndexInfo();
     Y_ABORT_UNLESS(context.Changes->Tiering.emplace(pathId, ttl).second);
@@ -340,9 +339,8 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
             continue;
         }
 
-        context.AllowEviction = (evictionSize <= context.MaxEvictBytes);
         context.AllowDrop = (dropBlobs <= TCompactionLimits::MAX_BLOBS_TO_DELETE);
-        const bool tryEvictPortion = context.AllowEviction && ttl.HasTiers();
+        const bool tryEvictPortion = ttl.HasTiers() && context.HasMemoryForEviction();
 
         if (auto max = info->MaxValue(ttlColumnId)) {
             bool keep = !expireTimestampOpt;
@@ -390,8 +388,8 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
                 }
                 if (currentTierName != tierName) {
                     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "tiering switch detected")("from", currentTierName)("to", tierName);
-                    evictionSize += info->BlobsSizes().first;
                     context.Changes->AddPortionToEvict(*info, TPortionEvictionFeatures(tierName, pathId, StoragesManager->GetOperator(tierName)));
+                    context.AppPortionForCheckMemoryUsage(*info);
                     SignalCounters.OnPortionToEvict(info->BlobsBytes());
                 }
             }
@@ -406,7 +404,7 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
             SignalCounters.OnPortionNoBorder(info->BlobsBytes());
         }
     }
-    if (dWaiting > TDuration::MilliSeconds(500) && (!context.AllowEviction || !context.AllowDrop)) {
+    if (dWaiting > TDuration::MilliSeconds(500) && (!context.HasMemoryForEviction() || !context.AllowDrop)) {
         dWaiting = TDuration::MilliSeconds(500);
     }
     Y_ABORT_UNLESS(!!dWaiting);
@@ -447,7 +445,7 @@ bool TColumnEngineForLogs::DrainEvictionQueue(std::map<TMonotonic, std::vector<T
     return hasChanges;
 }
 
-std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions, ui64 maxEvictBytes) noexcept {
+std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept {
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "StartTtl")("external", pathEviction.size())
         ("internal", EvictionsController.MutableNextCheckInstantForTierings().size())
         ;
@@ -456,7 +454,7 @@ std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const TH
 
     auto changes = std::make_shared<TTTLColumnEngineChanges>(TSplitSettings(), saverContext);
 
-    TTieringProcessContext context(maxEvictBytes, changes, busyPortions);
+    TTieringProcessContext context(memoryUsageLimit, changes, busyPortions, TTTLColumnEngineChanges::BuildMemoryPredictor());
     bool hasExternalChanges = false;
     for (auto&& i : pathEviction) {
         context.DurationsForced[i.first] = ProcessTiering(i.first, i.second, context);
@@ -575,9 +573,11 @@ void TColumnEngineForLogs::DoRegisterTable(const ui64 pathId) {
     AFL_VERIFY(Tables.emplace(pathId, std::make_shared<TGranuleMeta>(pathId, GranulesStorage, SignalCounters.RegisterGranuleDataCounters(), VersionedIndex)).second);
 }
 
-TColumnEngineForLogs::TTieringProcessContext::TTieringProcessContext(const ui64 maxEvictBytes, std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions)
-    : Now(TlsActivationContext ? AppData()->TimeProvider->Now() : TInstant::Now())
-    , MaxEvictBytes(maxEvictBytes)
+TColumnEngineForLogs::TTieringProcessContext::TTieringProcessContext(const ui64 memoryUsageLimit,
+    std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions, const std::shared_ptr<TColumnEngineChanges::IMemoryPredictor>& memoryPredictor)
+    : MemoryUsageLimit(memoryUsageLimit)
+    , MemoryPredictor(memoryPredictor)
+    , Now(TlsActivationContext ? AppData()->TimeProvider->Now() : TInstant::Now())
     , Changes(changes)
     , BusyPortions(busyPortions)
 {

ydb/core/tx/columnshard/engines/column_engine_logs.h

@@ -84,20 +84,37 @@ class TColumnEngineForLogs : public IColumnEngine {
     std::shared_ptr<IStoragesManager> StoragesManager;
     TEvictionsController EvictionsController;
     class TTieringProcessContext {
+    private:
+        const ui64 MemoryUsageLimit;
+        ui64 MemoryUsage = 0;
+        std::shared_ptr<TColumnEngineChanges::IMemoryPredictor> MemoryPredictor;
     public:
         bool AllowEviction = true;
         bool AllowDrop = true;
         const TInstant Now;
-        const ui64 MaxEvictBytes;
         std::shared_ptr<TTTLColumnEngineChanges> Changes;
         std::map<ui64, TDuration> DurationsForced;
         const THashSet<TPortionAddress>& BusyPortions;
-        TTieringProcessContext(const ui64 maxEvictBytes, std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions);
+
+        void AppPortionForCheckMemoryUsage(const TPortionInfo& info) {
+            MemoryUsage = MemoryPredictor->AddPortion(info);
+        }
+
+        bool HasMemoryForEviction() const {
+            return MemoryUsage < MemoryUsageLimit;
+        }
+
+        TTieringProcessContext(const ui64 memoryUsageLimit, std::shared_ptr<TTTLColumnEngineChanges> changes,
+            const THashSet<TPortionAddress>& busyPortions, const std::shared_ptr<TColumnEngineChanges::IMemoryPredictor>& memoryPredictor);
     };
 
     TDuration ProcessTiering(const ui64 pathId, const TTiering& tiering, TTieringProcessContext& context) const;
     bool DrainEvictionQueue(std::map<TMonotonic, std::vector<TEvictionsController::TTieringWithPathId>>& evictionsQueue, TTieringProcessContext& context) const;
 public:
+    ui64* GetLastPortionPointer() {
+        return &LastPortion;
+    }
+
     enum ETableIdx {
         GRANULES = 0,
     };
@@ -135,8 +152,8 @@ class TColumnEngineForLogs : public IColumnEngine {
     std::shared_ptr<TInsertColumnEngineChanges> StartInsert(std::vector<TInsertedData>&& dataToIndex) noexcept override;
     std::shared_ptr<TColumnEngineChanges> StartCompaction(const TCompactionLimits& limits, const THashSet<TPortionAddress>& busyPortions) noexcept override;
     std::shared_ptr<TCleanupColumnEngineChanges> StartCleanup(const TSnapshot& snapshot, THashSet<ui64>& pathsToDrop, ui32 maxRecords) noexcept override;
-    std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions,
-                                                   ui64 maxEvictBytes = TCompactionLimits::DEFAULT_EVICTION_BYTES) noexcept override;
+    std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction,
+        const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept override;
 
     bool ApplyChanges(IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> indexChanges,
                       const TSnapshot& snapshot) noexcept override;
@@ -164,9 +181,7 @@ class TColumnEngineForLogs : public IColumnEngine {
     }
 
     const TGranuleMeta& GetGranuleVerified(const ui64 pathId) const {
-        auto it = Tables.find(pathId);
-        AFL_VERIFY(it != Tables.end())("path_id", pathId)("count", Tables.size());
-        return *it->second;
+        return *GetGranulePtrVerified(pathId);
     }
 
     std::shared_ptr<TGranuleMeta> GetGranulePtrVerified(const ui64 pathId) const {

ydb/core/tx/columnshard/engines/portions/portion_info.h

@@ -364,6 +364,9 @@ class TPortionInfo {
         for (auto&& i : Records) {
             result.emplace(i.BlobRange.BlobId);
         }
+        for (auto&& i : Indexes) {
+            result.emplace(i.GetBlobRange().BlobId);
+        }
         return result;
     }
 

ydb/core/tx/columnshard/engines/reader/plain_reader/context.cpp

@@ -79,7 +79,10 @@ std::shared_ptr<NKikimr::NOlap::NPlainReader::IFetchingStep> TSpecialReadContext
             }
             current = current->AttachNext(std::make_shared<TFilterProgramStep>(i));
         }
-        const TColumnsSet columnsAdditionalFetch = *FFColumns - *EFColumns - *SpecColumns - *PredicateColumns;
+        TColumnsSet columnsAdditionalFetch = *FFColumns - *EFColumns - *SpecColumns;
+        if (partialUsageByPredicate) {
+            columnsAdditionalFetch = columnsAdditionalFetch - *PredicateColumns;
+        }
         if (columnsAdditionalFetch.GetColumnsCount()) {
             current = current->AttachNext(std::make_shared<TBlobsFetchingStep>(std::make_shared<TColumnsSet>(columnsAdditionalFetch)));
             current = current->AttachNext(std::make_shared<TAssemblerStep>(std::make_shared<TColumnsSet>(columnsAdditionalFetch)));