Skip to content

Commit 8250f33

Browse files
gridnevvvitgridnevvvit
committed
increase and make configurable some slj params
1 parent 6587ac8 commit 8250f33

11 files changed

+66
-15
lines changed

‎ydb/core/kqp/counters/kqp_counters.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -825,6 +825,8 @@ TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, co
825825
DataShardIteratorFails = KqpGroup->GetCounter("IteratorReads/DatashardFails", true);
826826
DataShardIteratorMessages = KqpGroup->GetCounter("IteratorReads/DatashardMessages", true);
827827
IteratorDeliveryProblems = KqpGroup->GetCounter("IteratorReads/DeliveryProblems", true);
828+
StreamLookupIteratorTotalQuotaBytesInFlight = KqpGroup->GetCounter("IteratorReads/StreamLookupIteratorTotalQuotaBytesInFlight", false);
829+
StreamLookupIteratorTotalQuotaBytesExceeded = KqpGroup->GetCounter("IteratorReads/StreamLookupIteratorTotalQuotaBytesExceeded", true);
828830

829831
/* sink writes */
830832
WriteActorsShardResolve = KqpGroup->GetCounter("SinkWrites/WriteActorShardResolve", true);

‎ydb/core/kqp/counters/kqp_counters.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -415,6 +415,8 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
415415
::NMonitoring::TDynamicCounters::TCounterPtr ReadActorRetries;
416416
::NMonitoring::TDynamicCounters::TCounterPtr DataShardIteratorFails;
417417
::NMonitoring::TDynamicCounters::TCounterPtr DataShardIteratorMessages;
418+
::NMonitoring::TDynamicCounters::TCounterPtr StreamLookupIteratorTotalQuotaBytesInFlight;
419+
::NMonitoring::TDynamicCounters::TCounterPtr StreamLookupIteratorTotalQuotaBytesExceeded;
418420
::NMonitoring::TDynamicCounters::TCounterPtr IteratorDeliveryProblems;
419421

420422
// Sink write counters

‎ydb/core/kqp/node_service/kqp_node_service.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -455,6 +455,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
455455
ptr->ReadResponseTimeout = TDuration::MilliSeconds(settings.GetIteratorResponseTimeoutMs());
456456
}
457457
ptr->MaxRetryDelay = TDuration::MilliSeconds(settings.GetMaxDelayMs());
458+
ptr->MaxRowsProcessingStreamLookup = settings.GetMaxRowsProcessingStreamLookup();
459+
ptr->MaxTotalBytesQuotaStreamLookup = settings.GetMaxTotalBytesQuotaStreamLookup();
458460
SetReadIteratorBackoffSettings(ptr);
459461
}
460462

‎ydb/core/kqp/runtime/kqp_buffer_lookup_actor.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -258,7 +258,7 @@ class TKqpBufferLookupActor : public NActors::TActorBootstrapped<TKqpBufferLooku
258258
auto reads = worker->BuildRequests(Partitioning, ReadId);
259259

260260
// lookup can't be overloaded
261-
AFL_ENSURE(!worker->IsOverloaded());
261+
AFL_ENSURE(!worker->IsOverloaded(std::numeric_limits<size_t>::max()));
262262

263263
for (auto& [shardId, read] : reads) {
264264
++state.ReadsInflight;

‎ydb/core/kqp/runtime/kqp_read_iterator_common.cpp‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ size_t MaxShardResolves() {
9191
return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardResolves;
9292
}
9393

94-
size_t MaxShardRetries() {
94+
size_t MaxShardRetries() {
9595
return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->MaxShardAttempts;
9696
}
9797

@@ -103,5 +103,13 @@ TMaybe<TDuration> ShardTimeout() {
103103
return Singleton<TBackoffStorage>()->SettingsPtr.AtomicLoad()->ReadResponseTimeout;
104104
}
105105

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+
106114
} // namespace NKqp
107115
} // namespace NKikimr

‎ydb/core/kqp/runtime/kqp_read_iterator_common.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ struct TIteratorReadBackoffSettings : TAtomicRefCount<TIteratorReadBackoffSettin
1717

1818
TMaybe<size_t> MaxTotalRetries;
1919
TMaybe<TDuration> ReadResponseTimeout;
20+
size_t MaxRowsProcessingStreamLookup = 65536;
21+
ui64 MaxTotalBytesQuotaStreamLookup = 5_MB * 512;
2022
};
2123

2224
struct TEvReadSettings : public TAtomicRefCount<TEvReadSettings> {
@@ -38,6 +40,8 @@ size_t MaxShardResolves();
3840
size_t MaxShardRetries();
3941
TMaybe<size_t> MaxTotalRetries();
4042
TMaybe<TDuration> ShardTimeout();
43+
size_t MaxRowsProcessingStreamLookup();
44+
ui64 MaxTotalBytesQuotaStreamLookup();
4145

4246
void SetDefaultIteratorQuotaSettings(ui32 rows, ui32 bytes);
4347
THolder<NKikimr::TEvDataShard::TEvRead> GetDefaultReadSettings();

‎ydb/core/kqp/runtime/kqp_stream_lookup_actor.cpp‎

Lines changed: 35 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
5050
, StreamLookupWorker(CreateStreamLookupWorker(std::move(settings), args.TaskId, args.TypeEnv, args.HolderFactory, args.InputDesc))
5151
, IsolationLevel(settings.GetIsolationLevel())
5252
, Database(settings.GetDatabase())
53+
, MaxTotalBytesQuota(MaxTotalBytesQuotaStreamLookup())
54+
, MaxRowsProcessing(MaxRowsProcessingStreamLookup())
5355
, Counters(counters)
5456
, LookupActorSpan(TWilsonKqp::LookupActor, std::move(args.TraceId), "LookupActor")
5557
{
@@ -308,7 +310,7 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
308310
ReadRowsCount += replyResultStats.ReadRowsCount;
309311
ReadBytesCount += replyResultStats.ReadBytesCount;
310312

311-
auto overloaded = StreamLookupWorker->IsOverloaded();
313+
auto overloaded = StreamLookupWorker->IsOverloaded(MaxRowsProcessing);
312314
if (!overloaded.has_value()) {
313315
FetchInputRows();
314316
} else {
@@ -451,6 +453,9 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
451453
}
452454
}
453455

456+
TotalBytesQuota -= MaxBytesDefaultQuota;
457+
Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Sub(MaxBytesDefaultQuota);
458+
454459
if (!Snapshot.IsValid()) {
455460
Snapshot = IKqpGateway::TKqpSnapshot(record.GetSnapshot().GetStep(), record.GetSnapshot().GetTxId());
456461
}
@@ -508,6 +513,7 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
508513
}
509514
}
510515

516+
511517
YQL_ENSURE(read.LastSeqNo < record.GetSeqNo());
512518
read.LastSeqNo = record.GetSeqNo();
513519

@@ -520,8 +526,14 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
520526
request->Record.SetSeqNo(record.GetSeqNo());
521527

522528
auto defaultSettings = GetDefaultReadAckSettings()->Record;
523-
request->Record.SetMaxRows(defaultSettings.GetMaxRows());
524-
request->Record.SetMaxBytes(defaultSettings.GetMaxBytes());
529+
request->Record.SetMaxRows(MaxRowsDefaultQuota);
530+
request->Record.SetMaxBytes(MaxBytesDefaultQuota);
531+
532+
TotalBytesQuota += MaxBytesDefaultQuota;
533+
Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Add(MaxBytesDefaultQuota);
534+
if (TotalBytesQuota > MaxTotalBytesQuota) {
535+
Counters->StreamLookupIteratorTotalQuotaBytesExceeded->Inc();
536+
}
525537

526538
const bool needToCreatePipe = Reads.NeedToCreatePipe(read.ShardId);
527539

@@ -663,10 +675,22 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
663675
}
664676

665677
auto defaultSettings = GetDefaultReadSettings()->Record;
666-
record.SetMaxRows(defaultSettings.GetMaxRows());
667-
record.SetMaxBytes(defaultSettings.GetMaxBytes());
678+
if (!MaxRowsDefaultQuota || !MaxBytesDefaultQuota) {
679+
MaxRowsDefaultQuota = defaultSettings.GetMaxRows();
680+
MaxBytesDefaultQuota = defaultSettings.GetMaxBytes();
681+
}
682+
683+
record.SetMaxRows(MaxRowsDefaultQuota);
684+
record.SetMaxBytes(MaxBytesDefaultQuota);
668685
record.SetResultFormat(NKikimrDataEvents::FORMAT_CELLVEC);
669686

687+
TotalBytesQuota += MaxBytesDefaultQuota;
688+
Counters->StreamLookupIteratorTotalQuotaBytesInFlight->Add(MaxBytesDefaultQuota);
689+
690+
if (TotalBytesQuota > MaxTotalBytesQuota) {
691+
Counters->StreamLookupIteratorTotalQuotaBytesExceeded->Inc();
692+
}
693+
670694
CA_LOG_D(TStringBuilder() << "Send EvRead (stream lookup) to shardId=" << shardId
671695
<< ", readId = " << record.GetReadId()
672696
<< ", tablePath: " << StreamLookupWorker->GetTablePath()
@@ -838,6 +862,12 @@ class TKqpStreamLookupActor : public NActors::TActorBootstrapped<TKqpStreamLooku
838862
ui64 ReadRowsCount = 0;
839863
ui64 ReadBytesCount = 0;
840864

865+
size_t TotalBytesQuota = 0;
866+
ui64 MaxTotalBytesQuota = 0;
867+
size_t MaxRowsProcessing = 0;
868+
size_t MaxBytesDefaultQuota = 0;
869+
size_t MaxRowsDefaultQuota = 0;
870+
841871
TIntrusivePtr<TKqpCounters> Counters;
842872
NWilson::TSpan LookupActorSpan;
843873
NWilson::TSpan LookupActorStateSpan;

‎ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@
1616
namespace NKikimr {
1717
namespace NKqp {
1818

19-
constexpr ui64 MAX_IN_FLIGHT_LIMIT = 500;
2019
constexpr ui64 SEQNO_SPACE = 40;
2120
constexpr ui64 MaxTaskId = (1ULL << (64 - SEQNO_SPACE));
2221

@@ -324,7 +323,7 @@ class TKqpLookupRows : public TKqpStreamLookupWorker {
324323
ReadResults.emplace_back(std::move(result));
325324
}
326325

327-
std::optional<TString> IsOverloaded() final {
326+
std::optional<TString> IsOverloaded(size_t) final {
328327
return std::nullopt;
329328
}
330329

@@ -608,10 +607,10 @@ class TKqpJoinRows : public TKqpStreamLookupWorker {
608607
YQL_ENSURE(false);
609608
}
610609

611-
std::optional<TString> IsOverloaded() final {
612-
if (UnprocessedRows.size() >= MAX_IN_FLIGHT_LIMIT ||
613-
PendingLeftRowsByKey.size() >= MAX_IN_FLIGHT_LIMIT ||
614-
ResultRowsBySeqNo.size() >= MAX_IN_FLIGHT_LIMIT)
610+
std::optional<TString> IsOverloaded(size_t maxRowsProcessing) final {
611+
if (UnprocessedRows.size() >= maxRowsProcessing ||
612+
PendingLeftRowsByKey.size() >= maxRowsProcessing ||
613+
ResultRowsBySeqNo.size() >= maxRowsProcessing)
615614
{
616615
TStringBuilder overloadDescriptor;
617616
overloadDescriptor << "unprocessed rows: " << UnprocessedRows.size()

‎ydb/core/kqp/runtime/kqp_stream_lookup_worker.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ class TKqpStreamLookupWorker {
8181
virtual bool AllRowsProcessed() = 0;
8282
virtual bool HasPendingResults() = 0;
8383
virtual void ResetRowsProcessing(ui64 readId) = 0;
84-
virtual std::optional<TString> IsOverloaded() = 0;
84+
virtual std::optional<TString> IsOverloaded(size_t maxRowsProcessing) = 0;
8585

8686
protected:
8787
const NMiniKQL::TTypeEnvironment& TypeEnv;

‎ydb/core/kqp/ut/scan/kqp_scan_ut.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2580,7 +2580,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
25802580
appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetUnsertaintyRatio(0.5);
25812581
appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMultiplier(2.0);
25822582
appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalRetries(100);
2583-
2583+
appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxRowsProcessingStreamLookup(500);
2584+
appConfig.MutableTableServiceConfig()->MutableIteratorReadsRetrySettings()->SetMaxTotalBytesQuotaStreamLookup(100);
25842585

25852586
TPortManager tp;
25862587
ui16 mbusport = tp.GetPort(2134);

0 commit comments

Comments
 (0)