deduplication memory fixes (ydb-platform#22532)

Author: swalrus1

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -2260,6 +2260,29 @@ void TCompGroupedMemoryLimiterInitializer::InitializeServices(NActors::TActorSys
     }
 }
 
+TDeduplicationGroupedMemoryLimiterInitializer::TDeduplicationGroupedMemoryLimiterInitializer(const TKikimrRunConfig& runConfig)
+    : IKikimrServicesInitializer(runConfig)
+{
+}
+
+void TDeduplicationGroupedMemoryLimiterInitializer::InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) {
+    NOlap::NGroupedMemoryManager::TConfig serviceConfig;
+    if (Config.GetDeduplicationGroupedMemoryLimiterConfig().GetCountBuckets() == 0) {
+        Config.MutableDeduplicationGroupedMemoryLimiterConfig()->SetCountBuckets(1);
+    }
+    Y_ABORT_UNLESS(serviceConfig.DeserializeFromProto(Config.GetDeduplicationGroupedMemoryLimiterConfig()));
+
+    if (serviceConfig.IsEnabled()) {
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> tabletGroup = GetServiceCounters(appData->Counters, "tablets");
+        TIntrusivePtr<::NMonitoring::TDynamicCounters> countersGroup = tabletGroup->GetSubgroup("type", "TX_DEDU_GROUPED_MEMORY_LIMITER");
+
+        auto service = NOlap::NGroupedMemoryManager::TDeduplicationMemoryLimiterOperator::CreateService(serviceConfig, countersGroup);
+
+        setup->LocalServices.push_back(std::make_pair(NOlap::NGroupedMemoryManager::TDeduplicationMemoryLimiterOperator::MakeServiceId(NodeId),
+            TActorSetupCmd(service, TMailboxType::HTSwap, appData->UserPoolId)));
+    }
+}
+
 TCompDiskLimiterInitializer::TCompDiskLimiterInitializer(const TKikimrRunConfig& runConfig)
     : IKikimrServicesInitializer(runConfig) {
 }

ydb/core/driver_lib/run/kikimr_services_initializers.h

@@ -415,6 +415,12 @@ class TCompGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
     void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
 };
 
+class TDeduplicationGroupedMemoryLimiterInitializer: public IKikimrServicesInitializer {
+public:
+    TDeduplicationGroupedMemoryLimiterInitializer(const TKikimrRunConfig& runConfig);
+    void InitializeServices(NActors::TActorSystemSetup* setup, const NKikimr::TAppData* appData) override;
+};
+
 class TCompPrioritiesInitializer: public IKikimrServicesInitializer {
 public:
     TCompPrioritiesInitializer(const TKikimrRunConfig& runConfig);

ydb/core/driver_lib/run/run.cpp

@@ -1741,6 +1741,7 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
     if (serviceMask.EnableGroupedMemoryLimiter) {
         sil->AddServiceInitializer(new TScanGroupedMemoryLimiterInitializer(runConfig));
         sil->AddServiceInitializer(new TCompGroupedMemoryLimiterInitializer(runConfig));
+        sil->AddServiceInitializer(new TDeduplicationGroupedMemoryLimiterInitializer(runConfig));
     }
 
     if (serviceMask.EnableCompPriorities) {

ydb/core/protos/config.proto

@@ -2523,6 +2523,7 @@ message TAppConfig {
     optional TCompositeConveyorConfig CompositeConveyorConfig = 109;
     optional TGeneralCacheConfig PortionsMetadataCache = 110;
     optional TGeneralCacheConfig ColumnDataCache = 111;
+    optional TGroupedMemoryLimiterConfig DeduplicationGroupedMemoryLimiterConfig = 112;
 }
 
 message TYdbVersion {

ydb/core/protos/console_config.proto

@@ -154,6 +154,7 @@ message TConfigItem {
 
         CompositeConveyorConfigItem = 109;
         PortionsMetadataCacheItem = 110;
+        DeduGroupedMemoryLimiterConfig = 112;
 
         // synthetic kinds for audit purposes only
         DatabaseYamlConfigChangeItem = 32767;

ydb/core/tx/columnshard/engines/reader/simple_reader/duplicates/context.cpp

@@ -4,17 +4,10 @@
 
 namespace NKikimr::NOlap::NReader::NSimple::NDuplicateFiltering {
 
-    namespace {
-
-        static std::shared_ptr<NOlap::NGroupedMemoryManager::TStageFeatures> DeduplicationStageFeatures =
-            NOlap::NGroupedMemoryManager::TDeduplicationMemoryLimiterOperator::BuildStageFeatures("DEFAULT", 1000000000);
-
-    }
-
 TInternalFilterConstructor::TInternalFilterConstructor(const TEvRequestFilter::TPtr& request, TColumnDataSplitter&& splitter)
     : OriginalRequest(request)
     , Intervals(std::move(splitter))
-    , ProcessGuard(NGroupedMemoryManager::TDeduplicationMemoryLimiterOperator::BuildProcessGuard({ DeduplicationStageFeatures }))
+    , ProcessGuard(NGroupedMemoryManager::TDeduplicationMemoryLimiterOperator::BuildProcessGuard({}))
     , ScopeGuard(ProcessGuard->BuildScopeGuard(1))
     , GroupGuard(ScopeGuard->BuildGroupGuard())
 {

ydb/core/tx/columnshard/engines/reader/simple_reader/duplicates/context.h

@@ -101,10 +101,10 @@ class TInternalFilterConstructor: TMoveOnly {
         return ProcessGuard->GetProcessId();
     }
     ui64 GetMemoryScopeId() const {
-        return ScopeGuard->GetProcessId();
+        return ScopeGuard->GetScopeId();
     }
     ui64 GetMemoryGroupId() const {
-        return GroupGuard->GetProcessId();
+        return GroupGuard->GetGroupId();
     }
 };
 

ydb/core/tx/columnshard/engines/reader/simple_reader/duplicates/manager.cpp

@@ -69,7 +69,7 @@ class TColumnDataAllocation: public NGroupedMemoryManager::IAllocation {
 private:
     virtual void DoOnAllocationImpossible(const TString& errorMessage) override {
         AFL_VERIFY(Callback);
-        Callback->OnError(errorMessage);
+        Callback->OnError(TStringBuilder() << "cannot allocate memory: " << errorMessage);
     }
     virtual bool DoOnAllocated(std::shared_ptr<NGroupedMemoryManager::TAllocationGuard>&& guard,
         const std::shared_ptr<NGroupedMemoryManager::IAllocation>& /*allocation*/) override {

ydb/tests/library/harness/kikimr_config.py

@@ -165,6 +165,7 @@ def __init__(
             enable_resource_pools=None,
             scan_grouped_memory_limiter_config=None,
             comp_grouped_memory_limiter_config=None,
+            deduplication_grouped_memory_limiter_config=None,
             query_service_config=None,
             domain_login_only=None,
             use_self_management=False,
@@ -408,6 +409,8 @@ def __init__(
             self.yaml_config["scan_grouped_memory_limiter_config"] = scan_grouped_memory_limiter_config
         if comp_grouped_memory_limiter_config:
             self.yaml_config["comp_grouped_memory_limiter_config"] = comp_grouped_memory_limiter_config
+        if deduplication_grouped_memory_limiter_config:
+            self.yaml_config["deduplication_grouped_memory_limiter_config"] = deduplication_grouped_memory_limiter_config
 
         self.__build()