Skip to content

Commit eb0f09e

Browse files
gridnevvvitgridnevvvit
committed
remove result channel proxies (#11382)
1 parent 61dc134 commit eb0f09e

File tree

16 files changed

+46
-64
lines changed

16 files changed

+46
-64
lines changed

‎ydb/core/grpc_services/query/rpc_execute_query.cpp

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,8 @@ struct TProducerState {
3030
ui64 ChannelId = 0;
3131

3232
void SendAck(const NActors::TActorIdentity& actor) const {
33-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
34-
resp->Record.SetSeqNo(*LastSeqNo);
33+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(*LastSeqNo, ChannelId);
3534
resp->Record.SetFreeSpace(AckedFreeSpaceBytes);
36-
resp->Record.SetChannelId(ChannelId);
3735

3836
actor.Send(ActorId, resp.Release());
3937
}

‎ydb/core/grpc_services/rpc_stream_execute_scan_query.cpp

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -268,12 +268,10 @@ class TStreamExecuteScanQueryRPC : public TActorBootstrapped<TStreamExecuteScanQ
268268
<< ", freeSpace: " << freeSpaceBytes
269269
<< ", to: " << ExecuterActorId_);
270270

271-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
272-
resp->Record.SetSeqNo(*LastSeqNo_);
271+
// scan query has single result set, so it's ok to put zero as channelId here.
272+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(*LastSeqNo_, 0);
273273
resp->Record.SetFreeSpace(freeSpaceBytes);
274-
275274
ctx.Send(ExecuterActorId_, resp.Release());
276-
277275
AckedFreeSpaceBytes_ = freeSpaceBytes;
278276
}
279277
}
@@ -326,6 +324,7 @@ class TStreamExecuteScanQueryRPC : public TActorBootstrapped<TStreamExecuteScanQ
326324
}
327325

328326
void Handle(NKqp::TEvKqp::TEvAbortExecution::TPtr& ev, const TActorContext& ctx) {
327+
329328
auto& record = ev->Get()->Record;
330329
NYql::TIssues issues = ev->Get()->GetIssues();
331330

@@ -361,8 +360,7 @@ class TStreamExecuteScanQueryRPC : public TActorBootstrapped<TStreamExecuteScanQ
361360
<< ", to: " << ev->Sender
362361
<< ", queue: " << FlowControl_.QueueSize());
363362

364-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
365-
resp->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
363+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
366364
resp->Record.SetFreeSpace(freeSpaceBytes);
367365

368366
ctx.Send(ev->Sender, resp.Release());

‎ydb/core/grpc_services/rpc_stream_execute_yql_script.cpp

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -279,8 +279,7 @@ class TStreamExecuteYqlScriptRPC
279279
<< ", to: " << ev->Sender
280280
<< ", queue: " << FlowControl_.QueueSize());
281281

282-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
283-
resp->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
282+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
284283
resp->Record.SetFreeSpace(freeSpaceBytes);
285284

286285
ctx.Send(ev->Sender, resp.Release());
@@ -320,8 +319,7 @@ class TStreamExecuteYqlScriptRPC
320319
<< ", freeSpace: " << freeSpaceBytes
321320
<< ", to: " << GatewayRequestHandlerActorId_);
322321

323-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
324-
resp->Record.SetSeqNo(*LastSeqNo_);
322+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(*LastSeqNo_, 0);
325323
resp->Record.SetFreeSpace(freeSpaceBytes);
326324

327325
ctx.Send(GatewayRequestHandlerActorId_, resp.Release());

‎ydb/core/kqp/executer_actor/kqp_executer.h

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,19 @@ struct TEvKqpExecuter {
7070
TKqpExecuterEvents::EvStreamData> {};
7171

7272
struct TEvStreamDataAck : public TEventPB<TEvStreamDataAck, NKikimrKqp::TEvExecuterStreamDataAck,
73-
TKqpExecuterEvents::EvStreamDataAck> {};
73+
TKqpExecuterEvents::EvStreamDataAck>
74+
{
75+
friend class TEventPBBase;
76+
explicit TEvStreamDataAck(ui64 seqno, ui64 channelId)
77+
{
78+
Record.SetSeqNo(seqno);
79+
Record.SetChannelId(channelId);
80+
}
81+
82+
private:
83+
// using a little hack to hide default empty constructor
84+
TEvStreamDataAck() = default;
85+
};
7486

7587
struct TEvStreamProfile : public TEventPB<TEvStreamProfile, NKikimrKqp::TEvExecuterStreamProfile,
7688
TKqpExecuterEvents::EvStreamProfile> {};

‎ydb/core/kqp/gateway/kqp_ic_gateway.cpp

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -202,9 +202,8 @@ class TKqpScanQueryRequestHandler : public TRequestHandlerBase<
202202
ResultSet.set_truncated(true);
203203
}
204204

205-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
205+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
206206
resp->Record.SetEnough(truncated);
207-
resp->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
208207
resp->Record.SetFreeSpace(ResultSetBytesLimit);
209208
ctx.Send(ev->Sender, resp.Release());
210209
}
@@ -497,8 +496,7 @@ class TKqpGenericQueryRequestHandler: public TRequestHandlerBase<
497496
}
498497
}
499498

500-
auto response = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
501-
response->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
499+
auto response = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
502500
response->Record.SetFreeSpace(SizeLimit && SizeLimit < std::numeric_limits<i64>::max() ? SizeLimit : std::numeric_limits<i64>::max());
503501
Send(ev->Sender, response.Release());
504502
}

‎ydb/core/kqp/run_script_actor/kqp_run_script_actor.cpp

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,11 +43,8 @@ struct TProducerState {
4343
ui64 ChannelId = 0;
4444

4545
void SendAck(const NActors::TActorIdentity& actor) const {
46-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
47-
resp->Record.SetSeqNo(*LastSeqNo);
46+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(*LastSeqNo, ChannelId);
4847
resp->Record.SetFreeSpace(AckedFreeSpaceBytes);
49-
resp->Record.SetChannelId(ChannelId);
50-
5148
actor.Send(ActorId, resp.Release());
5249
}
5350

‎ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1404,9 +1404,8 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
14041404
Y_ASSERT(record.GetResultSet().rows().size() == 1);
14051405
result = 1;
14061406

1407-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
1407+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
14081408
resp->Record.SetEnough(false);
1409-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
14101409
resp->Record.SetFreeSpace(100);
14111410
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
14121411
return TTestActorRuntime::EEventAction::DROP;
@@ -1472,9 +1471,8 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
14721471
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
14731472
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
14741473

1475-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
1474+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
14761475
resp->Record.SetEnough(false);
1477-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
14781476
resp->Record.SetFreeSpace(100);
14791477
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
14801478
return TTestActorRuntime::EEventAction::DROP;
@@ -1535,9 +1533,8 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
15351533
Y_ASSERT(record.GetResultSet().rows().size() == 0);
15361534
hasResult = true;
15371535

1538-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
1536+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
15391537
resp->Record.SetEnough(false);
1540-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
15411538
resp->Record.SetFreeSpace(100);
15421539
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
15431540
return TTestActorRuntime::EEventAction::DROP;
@@ -1606,9 +1603,8 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
16061603
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
16071604
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
16081605

1609-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
1606+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
16101607
resp->Record.SetEnough(false);
1611-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
16121608
resp->Record.SetFreeSpace(100);
16131609
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
16141610
return TTestActorRuntime::EEventAction::DROP;
@@ -2210,7 +2206,7 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
22102206

22112207
auto result = CollectStreamResult(it);
22122208
auto ast = result.QueryStats->Getquery_ast();
2213-
2209+
22142210
pushdown = ast.find("KqpOlapFilter") != std::string::npos;
22152211
} else {
22162212
// Error means that predicate not pushed down

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2414,9 +2414,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
24142414
auto& record = ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record;
24152415
Y_ASSERT(record.GetResultSet().rows().size() == 0);
24162416

2417-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
2417+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
24182418
resp->Record.SetEnough(false);
2419-
resp->Record.SetSeqNo(record.GetSeqNo());
24202419
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
24212420
return true;
24222421
}

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

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -326,9 +326,8 @@ Y_UNIT_TEST_SUITE(KqpSplit) {
326326
collectedKeys->push_back(row.items(0).uint64_value());
327327
}
328328

329-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
329+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
330330
resp->Record.SetEnough(false);
331-
resp->Record.SetSeqNo(record.GetSeqNo());
332331
runtime->Send(new IEventHandle(ev->Sender, sender, resp.Release()));
333332
return true;
334333
}
@@ -401,7 +400,7 @@ Y_UNIT_TEST_SUITE(KqpSplit) {
401400
} else if (testActorType == ETestActorType::StreamLookup) {
402401
InterceptStreamLookupActorPipeCache(MakePipePerNodeCacheID(false));
403402
}
404-
403+
405404
if (providedServer) {
406405
Server = providedServer;
407406
} else {
@@ -890,7 +889,7 @@ Y_UNIT_TEST_SUITE(KqpSplit) {
890889
);
891890

892891
shim->ReadsReceived.WaitI();
893-
892+
894893
UNIT_ASSERT_EQUAL(shards.size(), 1);
895894
auto undelivery = MakeHolder<TEvPipeCache::TEvDeliveryProblem>(shards[0], true);
896895

@@ -937,7 +936,7 @@ Y_UNIT_TEST_SUITE(KqpSplit) {
937936
);
938937

939938
shim->ReadsReceived.WaitI();
940-
939+
941940
UNIT_ASSERT_EQUAL(shards.size(), 1);
942941
auto undelivery = MakeHolder<TEvPipeCache::TEvDeliveryProblem>(shards[0], true);
943942

‎ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.cpp

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,8 +86,7 @@ class TQueryRunnerActor : public TActorBootstrapped<TQueryRunnerActor> {
8686
SendNotification<TEvQueryRunner::TEvExecutionStarted>();
8787
}
8888

89-
auto response = std::make_unique<TEvKqpExecuter::TEvStreamDataAck>();
90-
response->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
89+
auto response = std::make_unique<TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
9190
response->Record.SetFreeSpace(std::numeric_limits<i64>::max());
9291

9392
auto resultSetIndex = ev->Get()->Record.GetQueryResultIndex();

‎ydb/core/local_pgwire/pgwire_kqp_proxy.cpp

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -273,8 +273,7 @@ class TPgwireKqpProxy : public TActorBootstrapped<Base> {
273273
TBase::Send(EventRequest_->Sender, response.release(), 0, EventRequest_->Cookie);
274274

275275
BLOG_D(this->SelfId() << " Send stream data ack to " << ev->Sender);
276-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
277-
resp->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
276+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
278277
resp->Record.SetFreeSpace(std::numeric_limits<i64>::max());
279278
TBase::Send(ev->Sender, resp.Release());
280279
}

‎ydb/core/tx/datashard/datashard_ut_kqp_scan.cpp

Lines changed: 7 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -116,9 +116,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
116116
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
117117
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
118118

119-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
119+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
120120
resp->Record.SetEnough(false);
121-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
122121
resp->Record.SetFreeSpace(100);
123122
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
124123
return TTestActorRuntime::EEventAction::DROP;
@@ -213,9 +212,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
213212
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
214213
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
215214

216-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
215+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
217216
resp->Record.SetEnough(false);
218-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
219217
resp->Record.SetFreeSpace(100);
220218
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
221219
return TTestActorRuntime::EEventAction::DROP;
@@ -316,9 +314,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
316314
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
317315
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
318316

319-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
317+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
320318
resp->Record.SetEnough(false);
321-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
322319
resp->Record.SetFreeSpace(100);
323320

324321
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
@@ -434,9 +431,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
434431
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
435432
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
436433

437-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
434+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
438435
resp->Record.SetEnough(false);
439-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
440436
resp->Record.SetFreeSpace(100);
441437

442438
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
@@ -564,9 +560,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
564560
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
565561
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
566562

567-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
563+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
568564
resp->Record.SetEnough(false);
569-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
570565
resp->Record.SetFreeSpace(100);
571566

572567
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
@@ -692,9 +687,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
692687
}
693688
}
694689

695-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
690+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
696691
resp->Record.SetEnough(false);
697-
resp->Record.SetSeqNo(ev->Get<NKqp::TEvKqpExecuter::TEvStreamData>()->Record.GetSeqNo());
698692
resp->Record.SetFreeSpace(100);
699693
runtime.Send(new IEventHandle(ev->Sender, sender, resp.Release()));
700694
return TTestActorRuntime::EEventAction::DROP;
@@ -802,9 +796,8 @@ Y_UNIT_TEST_SUITE(KqpScan) {
802796
Y_ASSERT(record.GetResultSet().rows().at(0).items().size() == 1);
803797
result = record.GetResultSet().rows().at(0).items().at(0).uint64_value();
804798

805-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
799+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(record.GetSeqNo(), record.GetChannelId());
806800
resp->Record.SetEnough(false);
807-
resp->Record.SetSeqNo(record.GetSeqNo());
808801
resp->Record.SetFreeSpace(100);
809802
ctx.Send(ev->Sender, resp.Release());
810803
break;

‎ydb/core/tx/datashard/datashard_ut_volatile.cpp

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1309,8 +1309,7 @@ Y_UNIT_TEST_SUITE(DataShardVolatile) {
13091309
if (msg->Record.GetResultSet().rows().size()) {
13101310
observedResults.emplace_back(FormatResult(msg->Record.GetResultSet()));
13111311
}
1312-
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
1313-
resp->Record.SetSeqNo(msg->Record.GetSeqNo());
1312+
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(msg->Record.GetSeqNo(), msg->Record.GetChannelId());
13141313
resp->Record.SetFreeSpace(1);
13151314
ctx.Send(ev->Sender, resp.Release());
13161315
break;

‎ydb/core/viewer/viewer_query.h

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -567,8 +567,7 @@ class TJsonQuery : public TViewerPipeClient {
567567
ResultSets[data.GetQueryResultIndex()].emplace_back() = std::move(*data.MutableResultSet());
568568
}
569569

570-
THolder<NKqp::TEvKqpExecuter::TEvStreamDataAck> ack = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
571-
ack->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
570+
THolder<NKqp::TEvKqpExecuter::TEvStreamDataAck> ack = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
572571
if (TotalRows >= LimitRows) {
573572
ack->Record.SetEnough(true);
574573
}

‎ydb/core/viewer/viewer_query_old.h

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -440,8 +440,7 @@ class TJsonQueryOld : public TViewerPipeClient {
440440
ResultSets.emplace_back();
441441
ResultSets.back() = std::move(data.GetResultSet());
442442

443-
THolder<NKqp::TEvKqpExecuter::TEvStreamDataAck> ack = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
444-
ack->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
443+
THolder<NKqp::TEvKqpExecuter::TEvStreamDataAck> ack = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
445444
Send(ev->Sender, ack.Release());
446445
}
447446

‎ydb/tests/tools/kqprun/src/actors.cpp

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,7 @@ class TRunScriptActorMock : public NActors::TActorBootstrapped<TRunScriptActorMo
4242
)
4343

4444
void Handle(NKikimr::NKqp::TEvKqpExecuter::TEvStreamData::TPtr& ev) {
45-
auto response = MakeHolder<NKikimr::NKqp::TEvKqpExecuter::TEvStreamDataAck>();
46-
response->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
45+
auto response = MakeHolder<NKikimr::NKqp::TEvKqpExecuter::TEvStreamDataAck>(ev->Get()->Record.GetSeqNo(), ev->Get()->Record.GetChannelId());
4746
response->Record.SetFreeSpace(ResultSizeLimit_);
4847

4948
auto resultSetIndex = ev->Get()->Record.GetQueryResultIndex();

0 commit comments

Comments
 (0)