increase and make configurable some slj params (#29319)

gridnevvvit · gridnevvvit · commit db7f56afd2f9 · 2025-11-21T22:31:43.000+03:00
diff --git a/ydb/core/kqp/counters/kqp_counters.cpp b/ydb/core/kqp/counters/kqp_counters.cpp
@@ -815,6 +815,8 @@ TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, co
     DataShardIteratorFails = KqpGroup->GetCounter("IteratorReads/DatashardFails", true);
     DataShardIteratorMessages = KqpGroup->GetCounter("IteratorReads/DatashardMessages", true);
     IteratorDeliveryProblems = KqpGroup->GetCounter("IteratorReads/DeliveryProblems", true);
+    StreamLookupIteratorTotalQuotaBytesInFlight = KqpGroup->GetCounter("IteratorReads/StreamLookupIteratorTotalQuotaBytesInFlight", false);
+    StreamLookupIteratorTotalQuotaBytesExceeded = KqpGroup->GetCounter("IteratorReads/StreamLookupIteratorTotalQuotaBytesExceeded", true);
 
     /* sink writes */
     WriteActorsShardResolve = KqpGroup->GetCounter("SinkWrites/WriteActorShardResolve", true);
diff --git a/ydb/core/kqp/counters/kqp_counters.h b/ydb/core/kqp/counters/kqp_counters.h
@@ -409,6 +409,8 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
     ::NMonitoring::TDynamicCounters::TCounterPtr ReadActorRetries;
     ::NMonitoring::TDynamicCounters::TCounterPtr DataShardIteratorFails;
     ::NMonitoring::TDynamicCounters::TCounterPtr DataShardIteratorMessages;
+    ::NMonitoring::TDynamicCounters::TCounterPtr StreamLookupIteratorTotalQuotaBytesInFlight;
+    ::NMonitoring::TDynamicCounters::TCounterPtr StreamLookupIteratorTotalQuotaBytesExceeded;
     ::NMonitoring::TDynamicCounters::TCounterPtr IteratorDeliveryProblems;
 
     // Sink write counters
diff --git a/ydb/core/kqp/node_service/kqp_node_service.cpp b/ydb/core/kqp/node_service/kqp_node_service.cpp
@@ -434,6 +434,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
             ptr->ReadResponseTimeout = TDuration::MilliSeconds(settings.GetIteratorResponseTimeoutMs());
         }
         ptr->MaxRetryDelay = TDuration::MilliSeconds(settings.GetMaxDelayMs());
+        ptr->MaxRowsProcessingStreamLookup = settings.GetMaxRowsProcessingStreamLookup();
+        ptr->MaxTotalBytesQuotaStreamLookup = settings.GetMaxTotalBytesQuotaStreamLookup();
         SetReadIteratorBackoffSettings(ptr);
     }
 
diff --git a/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp b/ydb/core/kqp/runtime/kqp_read_iterator_common.cpp
@@ -91,7 +91,7 @@ size_t MaxShardResolves() {
     return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardResolves;
 }
 
-size_t MaxShardRetries() {  
+size_t MaxShardRetries() {
     return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardAttempts;
 }
 
@@ -103,5 +103,13 @@ TMaybe<TDuration> ShardTimeout() {
     return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->ReadResponseTimeout;
 }
 
+size_t MaxRowsProcessingStreamLookup() {
+    return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxRowsProcessingStreamLookup;
+}
+
+ui64 MaxTotalBytesQuotaStreamLookup() {
+    return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxTotalBytesQuotaStreamLookup;
+}
+
 } // namespace NKqp
 } // namespace NKikimr
diff --git a/ydb/core/kqp/runtime/kqp_read_iterator_common.h b/ydb/core/kqp/runtime/kqp_read_iterator_common.h
@@ -17,6 +17,8 @@ struct TIteratorReadBackoffSettings : TAtomicRefCount<TIteratorReadBackoffSettin
 
     TMaybe<size_t> MaxTotalRetries;
     TMaybe<TDuration> ReadResponseTimeout;
+    size_t MaxRowsProcessingStreamLookup = 65536;
+    ui64 MaxTotalBytesQuotaStreamLookup = 5_MB * 512;
 };
 
 struct TEvReadSettings : public TAtomicRefCount<TEvReadSettings> {
@@ -38,6 +40,8 @@ size_t MaxShardResolves();
 size_t MaxShardRetries();
 TMaybe<size_t> MaxTotalRetries();
 TMaybe<TDuration> ShardTimeout();
+size_t MaxRowsProcessingStreamLookup();
+ui64 MaxTotalBytesQuotaStreamLookup();
 
 void SetDefaultIteratorQuotaSettings(ui32 rows, ui32 bytes);
 THolder<NKikimr::TEvDataShard::TEvRead> GetDefaultReadSettings();
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp b/ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp
@@ -47,6 +47,8 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
         , SchemeCacheRequestTimeout(SCHEME_CACHE_REQUEST_TIMEOUT)
         , LookupStrategy(settings.GetLookupStrategy())
         , StreamLookupWorker(CreateStreamLookupWorker(std::move(settings), args.TaskId, args.TypeEnv, args.HolderFactory, args.InputDesc))
+        , MaxTotalBytesQuota(MaxTotalBytesQuotaStreamLookup())
+        , MaxRowsProcessing(MaxRowsProcessingStreamLookup())
         , Counters(counters)
         , LookupActorSpan(TWilsonKqp::LookupActor, std::move(args.TraceId), "LookupActor")
     {
@@ -305,7 +307,7 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
         ReadRowsCount += replyResultStats.ReadRowsCount;
         ReadBytesCount += replyResultStats.ReadBytesCount;
 
-        auto overloaded = StreamLookupWorker->IsOverloaded();
+        auto overloaded = StreamLookupWorker->IsOverloaded(MaxRowsProcessing);
         if (!overloaded.has_value()) {
             FetchInputRows();
         } else {
@@ -445,6 +447,9 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
             }
         }
 
+        TotalBytesQuota -= MaxBytesDefaultQuota;
+        Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Sub(MaxBytesDefaultQuota);
+
         if (!Snapshot.IsValid()) {
             Snapshot = IKqpGateway::TKqpSnapshot(record.GetSnapshot().GetStep(), record.GetSnapshot().GetTxId());
         }
@@ -502,6 +507,7 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
             }
         }
 
+
         YQL_ENSURE(read.LastSeqNo < record.GetSeqNo());
         read.LastSeqNo = record.GetSeqNo();
 
@@ -514,8 +520,14 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
             request->Record.SetSeqNo(record.GetSeqNo());
 
             auto defaultSettings = GetDefaultReadAckSettings()->Record;
-            request->Record.SetMaxRows(defaultSettings.GetMaxRows());
-            request->Record.SetMaxBytes(defaultSettings.GetMaxBytes());
+            request->Record.SetMaxRows(MaxRowsDefaultQuota);
+            request->Record.SetMaxBytes(MaxBytesDefaultQuota);
+
+            TotalBytesQuota += MaxBytesDefaultQuota;
+            Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Add(MaxBytesDefaultQuota);
+            if (TotalBytesQuota > MaxTotalBytesQuota) {
+                Counters->StreamLookupIteratorTotalQuotaBytesExceeded->Inc();
+            }
 
             const bool needToCreatePipe = Reads.NeedToCreatePipe(read.ShardId);
 
@@ -657,10 +669,22 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
         }
 
         auto defaultSettings = GetDefaultReadSettings()->Record;
-        record.SetMaxRows(defaultSettings.GetMaxRows());
-        record.SetMaxBytes(defaultSettings.GetMaxBytes());
+        if (!MaxRowsDefaultQuota || !MaxBytesDefaultQuota) {
+            MaxRowsDefaultQuota = defaultSettings.GetMaxRows();
+            MaxBytesDefaultQuota = defaultSettings.GetMaxBytes();
+        }
+
+        record.SetMaxRows(MaxRowsDefaultQuota);
+        record.SetMaxBytes(MaxBytesDefaultQuota);
         record.SetResultFormat(NKikimrDataEvents::FORMAT_CELLVEC);
 
+        TotalBytesQuota += MaxBytesDefaultQuota;
+        Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Add(MaxBytesDefaultQuota);
+
+        if (TotalBytesQuota > MaxTotalBytesQuota) {
+            Counters->StreamLookupIteratorTotalQuotaBytesExceeded->Inc();
+        }
+
         CA_LOG_D(TStringBuilder() << "Send EvRead (stream lookup) to shardId=" << shardId
             << ", readId = " << record.GetReadId()
             << ", tablePath: " << StreamLookupWorker->GetTablePath()
@@ -828,6 +852,12 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
     ui64 ReadRowsCount = 0;
     ui64 ReadBytesCount = 0;
 
+    size_t TotalBytesQuota = 0;
+    ui64 MaxTotalBytesQuota = 0;
+    size_t MaxRowsProcessing = 0;
+    size_t MaxBytesDefaultQuota = 0;
+    size_t MaxRowsDefaultQuota = 0;
+
     TIntrusivePtr<TKqpCounters> Counters;
     NWilson::TSpan LookupActorSpan;
     NWilson::TSpan LookupActorStateSpan;
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp
@@ -15,7 +15,6 @@
 namespace NKikimr {
 namespace NKqp {
 
-constexpr ui64 MAX_IN_FLIGHT_LIMIT = 500;
 constexpr ui64 SEQNO_SPACE = 40;
 constexpr ui64 MaxTaskId = (1ULL << (64 - SEQNO_SPACE));
 
@@ -323,7 +322,7 @@ class TKqpLookupRows : public TKqpStreamLookupWorker {
         ReadResults.emplace_back(std::move(result));
     }
 
-    std::optional<TString> IsOverloaded() final {
+    std::optional<TString> IsOverloaded(size_t) final {
         return std::nullopt;
     }
 
@@ -596,10 +595,10 @@ class TKqpJoinRows : public TKqpStreamLookupWorker {
         YQL_ENSURE(false);
     }
 
-    std::optional<TString> IsOverloaded() final {
-        if (UnprocessedRows.size() >= MAX_IN_FLIGHT_LIMIT ||
-            PendingLeftRowsByKey.size() >= MAX_IN_FLIGHT_LIMIT ||
-            ResultRowsBySeqNo.size() >= MAX_IN_FLIGHT_LIMIT)
+    std::optional<TString> IsOverloaded(size_t maxRowsProcessing) final {
+        if (UnprocessedRows.size() >= maxRowsProcessing ||
+            PendingLeftRowsByKey.size() >= maxRowsProcessing ||
+            ResultRowsBySeqNo.size() >= maxRowsProcessing)
         {
             TStringBuilder overloadDescriptor;
             overloadDescriptor << "unprocessed rows: " << UnprocessedRows.size()
diff --git a/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h b/ydb/core/kqp/runtime/kqp_stream_lookup_worker.h
@@ -80,7 +80,7 @@ class TKqpStreamLookupWorker {
     virtual bool AllRowsProcessed() = 0;
     virtual bool HasPendingResults() = 0;
     virtual void ResetRowsProcessing(ui64 readId) = 0;
-    virtual std::optional<TString> IsOverloaded() = 0;
+    virtual std::optional<TString> IsOverloaded(size_t maxRowsProcessing) = 0;
 
 protected:
     const NMiniKQL::TTypeEnvironment& TypeEnv;
diff --git a/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp b/ydb/core/kqp/ut/scan/kqp_scan_ut.cpp
@@ -2579,7 +2579,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
         appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0.5);
         appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(2.0);
         appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(100);
-
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxRowsProcessingStreamLookup(500);
+        appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalBytesQuotaStreamLookup(100);
 
         TPortManager tp;
         ui16 mbusport = tp.GetPort(2134);
diff --git a/ydb/core/protos/table_service_config.proto b/ydb/core/protos/table_service_config.proto
@@ -206,6 +206,9 @@ message TTableServiceConfig {
         optional double Multiplier = 5;
         optional uint32 IteratorResponseTimeoutMs = 6;
         optional uint32 MaxTotalRetries = 7;
+
+        optional uint64 MaxRowsProcessingStreamLookup = 9 [default = 65536];
+        optional uint64 MaxTotalBytesQuotaStreamLookup = 10 [default = 2684354560]; // 5_MB * 512
     }
 
     message TIteratorReadQuotaSettings {

Original file line number	Diff line number	Diff line change
`@@ -434,6 +434,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {`
`434`	`434`	`ptr->ReadResponseTimeout = TDuration::MilliSeconds(settings.GetIteratorResponseTimeoutMs());`
`435`	`435`	`}`
`436`	`436`	`ptr->MaxRetryDelay = TDuration::MilliSeconds(settings.GetMaxDelayMs());`
	`437`	`+ ptr->MaxRowsProcessingStreamLookup = settings.GetMaxRowsProcessingStreamLookup();`
	`438`	`+ ptr->MaxTotalBytesQuotaStreamLookup = settings.GetMaxTotalBytesQuotaStreamLookup();`
`437`	`439`	`SetReadIteratorBackoffSettings(ptr);`
`438`	`440`	`}`
`439`	`441`
Original file line number	Diff line number	Diff line change
`@@ -91,7 +91,7 @@ size_t MaxShardResolves() {`
`91`	`91`	`return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardResolves;`
`92`	`92`	`}`
`93`	`93`
`94`		`-size_t MaxShardRetries() {`
	`94`	`+size_t MaxShardRetries() {`
`95`	`95`	`return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardAttempts;`
`96`	`96`	`}`
`97`	`97`
`@@ -103,5 +103,13 @@ TMaybe<TDuration> ShardTimeout() {`
`103`	`103`	`return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->ReadResponseTimeout;`
`104`	`104`	`}`
`105`	`105`
	`106`	`+size_t MaxRowsProcessingStreamLookup() {`
	`107`	`+ return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxRowsProcessingStreamLookup;`
	`108`	`+}`
	`109`	`+`
	`110`	`+ui64 MaxTotalBytesQuotaStreamLookup() {`
	`111`	`+ return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxTotalBytesQuotaStreamLookup;`
	`112`	`+}`
	`113`	`+`
`106`	`114`	`} // namespace NKqp`
`107`	`115`	`} // namespace NKikimr`
Original file line number	Diff line number	Diff line change
`@@ -206,6 +206,9 @@ message TTableServiceConfig {`
`206`	`206`	`optional double Multiplier = 5;`
`207`	`207`	`optional uint32 IteratorResponseTimeoutMs = 6;`
`208`	`208`	`optional uint32 MaxTotalRetries = 7;`
	`209`	`+`
	`210`	`+ optional uint64 MaxRowsProcessingStreamLookup = 9 [default = 65536];`
	`211`	`+ optional uint64 MaxTotalBytesQuotaStreamLookup = 10 [default = 2684354560]; // 5_MB * 512`
`209`	`212`	`}`
`210`	`213`
`211`	`214`	`message TIteratorReadQuotaSettings {`