@@ -61,8 +61,7 @@ void TScanHead::OnSourceReady(const std::shared_ptr<IDataSource>& source, std::s
61
61
AFL_VERIFY (FinishedSources.emplace (frontSource).second );
62
62
while (FinishedSources.size () && (*FinishedSources.begin ())->GetFinish () < SortedSources.front ()->GetStart ()) {
63
63
auto finishedSource = *FinishedSources.begin ();
64
- if (!finishedSource->GetResultRecordsCount () && Context->GetCommonContext ()->GetReadMetadata ()->HasLimit () &&
65
- InFlightLimit < MaxInFlight) {
64
+ if (!finishedSource->GetResultRecordsCount () && InFlightLimit < MaxInFlight) {
66
65
InFlightLimit = 2 * InFlightLimit;
67
66
}
68
67
FetchedCount += finishedSource->GetResultRecordsCount ();
@@ -75,6 +74,7 @@ void TScanHead::OnSourceReady(const std::shared_ptr<IDataSource>& source, std::s
75
74
AFL_NOTICE (NKikimrServices::TX_COLUMNSHARD)(" event" , " limit_exhausted" )(
76
75
" limit" , Context->GetCommonContext ()->GetReadMetadata ()->GetLimitRobust ())(" fetched" , FetchedCount);
77
76
SortedSources.clear ();
77
+ break ;
78
78
}
79
79
}
80
80
}
@@ -133,6 +133,8 @@ TConclusion<bool> TScanHead::BuildNextInterval() {
133
133
for (auto it = FetchingInFlightSources.begin (); it != FetchingInFlightSources.end (); ++it) {
134
134
if ((*it)->GetFinish () < SortedSources.front ()->GetStart ()) {
135
135
++inFlightCountLocal;
136
+ } else {
137
+ break ;
136
138
}
137
139
}
138
140
}
@@ -148,6 +150,8 @@ TConclusion<bool> TScanHead::BuildNextInterval() {
148
150
for (auto it = FetchingInFlightSources.begin (); it != FetchingInFlightSources.end (); ++it) {
149
151
if ((*it)->GetFinish () < SortedSources.front ()->GetStart ()) {
150
152
++inFlightCountLocalNew;
153
+ } else {
154
+ break ;
151
155
}
152
156
}
153
157
AFL_VERIFY (inFlightCountLocal <= inFlightCountLocalNew);
0 commit comments