streamlookup join: add lru size monitoring

Author: yumkam

ydb/library/yql/dq/actors/input_transforms/dq_input_transform_lookup.cpp

@@ -69,6 +69,7 @@ class TInputTransformStreamLookupBase
         , MaxDelayedRows(maxDelayedRows)
         , CacheTtl(cacheTtl)
         , ReadyQueue(OutputRowType)
+        , LastLruSize(0)
     {
         Y_ABORT_UNLESS(Alloc);
         for (size_t i = 0; i != LookupInputIndexes.size(); ++i) {
@@ -172,11 +173,14 @@ class TInputTransformStreamLookupBase
             LruCache->Update(NUdf::TUnboxedValue(const_cast<NUdf::TUnboxedValue&&>(k)), std::move(v), now + CacheTtl);
         }
         KeysForLookup->clear();
+        auto deltaLruSize = (i64)LruCache->Size() - LastLruSize;
         auto deltaTime = GetCpuTimeDelta(startCycleCount);
         CpuTime += deltaTime;
         if (CpuTimeUs) {
+            LruSize->Add(deltaLruSize); // Note: there can be several streamlookup tied to same counter, so Add instead of Set
             CpuTimeUs->Add(deltaTime.MicroSeconds());
         }
+        LastLruSize += deltaLruSize;
         Send(ComputeActorId, new TEvNewAsyncInputDataArrived{InputIndex});
     }
 
@@ -200,6 +204,10 @@ class TInputTransformStreamLookupBase
     }
 
     void Free() {
+        if (LruSize && LastLruSize) {
+            LruSize->Add(-LastLruSize);
+            LastLruSize = 0;
+        }
         auto guard = BindAllocator();
         //All resources, held by this class, that have been created with mkql allocator, must be deallocated here
         KeysForLookup.reset();
@@ -290,6 +298,7 @@ class TInputTransformStreamLookupBase
         auto component = taskCounters->GetSubgroup("component", "Lookup");
         LruHits = component->GetCounter("Hits");
         LruMiss = component->GetCounter("Miss");
+        LruSize = component->GetCounter("LruSize");
         CpuTimeUs = component->GetCounter("CpuUs");
         Batches = component->GetCounter("Batches");
     }
@@ -358,9 +367,11 @@ class TInputTransformStreamLookupBase
     NKikimr::NMiniKQL::TUnboxedValueBatch ReadyQueue;
     NYql::NDq::TDqAsyncStats IngressStats;
     std::shared_ptr<IDqAsyncLookupSource::TUnboxedValueMap> KeysForLookup;
+    i64 LastLruSize;
 
     ::NMonitoring::TDynamicCounters::TCounterPtr LruHits;
     ::NMonitoring::TDynamicCounters::TCounterPtr LruMiss;
+    ::NMonitoring::TDynamicCounters::TCounterPtr LruSize;
     ::NMonitoring::TDynamicCounters::TCounterPtr CpuTimeUs;
     ::NMonitoring::TDynamicCounters::TCounterPtr Batches;
     TDuration CpuTime;