Skip to content

Commit 8097409

Browse files
niksavelievniksaveliev
authored
Merge 40ac8dd into 2b91145 
2 parents 2b91145 + 40ac8dd commit 8097409

26 files changed

+647
-139
lines changed

‎ydb/core/kqp/executer_actor/kqp_data_executer.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2096,6 +2096,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
20962096
TTopicTabletTxs topicTxs;
20972097
TDatashardTxs datashardTxs;
20982098
TEvWriteTxs evWriteTxs;
2099+
20992100
if (!TxManager) {
21002101
BuildDatashardTxs(datashardTasks, datashardTxs, evWriteTxs, topicTxs);
21012102
}
@@ -2380,7 +2381,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
23802381
YQL_ENSURE(!ReadOnlyTx);
23812382
}
23822383
}
2383-
23842384
Request.TopicOperations.BuildTopicTxs(topicTxs);
23852385

23862386
const bool needRollback = Request.LocksOp == ELocksOp::Rollback;

‎ydb/core/kqp/session_actor/kqp_query_state.cpp‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -355,7 +355,7 @@ bool TKqpQueryState::PrepareNextStatementPart() {
355355
void TKqpQueryState::AddOffsetsToTransaction() {
356356
YQL_ENSURE(HasTopicOperations());
357357

358-
const auto& operations = GetTopicOperations();
358+
const auto& operations = GetTopicOperationsFromRequest();
359359

360360
TMaybe<TString> consumer;
361361
if (operations.HasConsumer()) {
@@ -368,7 +368,6 @@ void TKqpQueryState::AddOffsetsToTransaction() {
368368
}
369369

370370
TopicOperations = NTopic::TTopicOperations();
371-
372371
for (auto& topic : operations.GetTopics()) {
373372
auto path = CanonizePath(NPersQueue::GetFullTopicPath(TlsActivationContext->AsActorContext(),
374373
GetDatabase(), topic.path()));
@@ -379,8 +378,7 @@ void TKqpQueryState::AddOffsetsToTransaction() {
379378
} else {
380379
for (auto& range : partition.partition_offsets()) {
381380
YQL_ENSURE(consumer.Defined());
382-
383-
TopicOperations.AddOperation(path, partition.partition_id(), *consumer, range);
381+
TopicOperations.AddOperation(path, partition.partition_id(), *consumer, range, partition.force_commit(), partition.kill_read_session(), partition.only_check_commited_to_finish());
384382
}
385383
}
386384
}
@@ -401,7 +399,7 @@ std::unique_ptr<NSchemeCache::TSchemeCacheNavigate> TKqpQueryState::BuildSchemeC
401399
auto navigate = std::make_unique<NSchemeCache::TSchemeCacheNavigate>();
402400
navigate->DatabaseName = CanonizePath(GetDatabase());
403401

404-
const auto& operations = GetTopicOperations();
402+
const auto& operations = GetTopicOperationsFromRequest();
405403
TMaybe<TString> consumer;
406404
if (operations.HasConsumer())
407405
consumer = operations.GetConsumer();

‎ydb/core/kqp/session_actor/kqp_query_state.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -315,7 +315,7 @@ class TKqpQueryState : public TNonCopyable {
315315
return RequestEv->GetQuery();
316316
}
317317

318-
const ::NKikimrKqp::TTopicOperationsRequest& GetTopicOperations() const {
318+
const ::NKikimrKqp::TTopicOperationsRequest& GetTopicOperationsFromRequest() const {
319319
return RequestEv->GetTopicOperations();
320320
}
321321

‎ydb/core/kqp/session_actor/kqp_session_actor.cpp‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -773,7 +773,6 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
773773
QueryState->QueryData = std::make_shared<TQueryData>(QueryState->TxCtx->TxAlloc);
774774
QueryState->TxCtx->SetIsolationLevel(settings);
775775
QueryState->TxCtx->OnBeginQuery();
776-
777776
if (!Transactions.CreateNew(QueryState->TxId.GetValue(), QueryState->TxCtx)) {
778777
std::vector<TIssue> issues{
779778
YqlIssue({}, TIssuesIds::KIKIMR_TOO_MANY_TRANSACTIONS)};
@@ -1249,7 +1248,6 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
12491248
}
12501249
}
12511250
}
1252-
12531251
request.TopicOperations = std::move(txCtx.TopicOperations);
12541252
} else if (QueryState->ShouldAcquireLocks(tx) && (!txCtx.HasOlapTable || Settings.TableService.GetEnableOlapSink())) {
12551253
request.AcquireLocksTxId = txCtx.Locks.GetLockTxId();

‎ydb/core/kqp/topics/kqp_topics.cpp‎

Lines changed: 68 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
#include "kqp_topics.h"
22

33
#include <ydb/core/base/path.h>
4+
#include <ydb/core/protos/kqp.pb.h>
45
#include <ydb/core/persqueue/utils.h>
56
#include <ydb/library/actors/core/log.h>
67

@@ -26,21 +27,44 @@ static void UpdateSupportivePartition(TMaybe<ui32>& lhs, const TMaybe<ui32>& rhs
2627
//
2728
bool TConsumerOperations::IsValid() const
2829
{
29-
return Offsets_.GetNumIntervals() == 1;
30+
return Offsets_.GetNumIntervals() <= 1;
3031
}
3132

32-
std::pair<ui64, ui64> TConsumerOperations::GetRange() const
33+
std::pair<ui64, ui64> TConsumerOperations::GetOffsetsCommitRange() const
3334
{
3435
Y_ABORT_UNLESS(IsValid());
3536

36-
return {Offsets_.Min(), Offsets_.Max()};
37+
if (Offsets_.Empty()) {
38+
return {0,0};
39+
} else {
40+
return {Offsets_.Min(), Offsets_.Max()};
41+
}
42+
}
43+
44+
bool TConsumerOperations::GetForceCommit() const
45+
{
46+
return ForceCommit_;
47+
}
48+
49+
bool TConsumerOperations::GetKillReadSession() const
50+
{
51+
return KillReadSession_;
3752
}
3853

39-
void TConsumerOperations::AddOperation(const TString& consumer, const Ydb::Topic::OffsetsRange& range)
54+
bool TConsumerOperations::GetOnlyCheckCommitedToFinish() const
55+
{
56+
return OnlyCheckCommitedToFinish_;
57+
}
58+
59+
void TConsumerOperations::AddOperation(const TString& consumer,
60+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
61+
bool forceCommit,
62+
bool killReadSession,
63+
bool onlyCheckCommitedToFinish)
4064
{
4165
Y_ABORT_UNLESS(Consumer_.Empty() || Consumer_ == consumer);
4266

43-
AddOperationImpl(consumer, range.start(), range.end());
67+
AddOperationImpl(consumer, range.start(), range.end(), forceCommit, killReadSession, onlyCheckCommitedToFinish);
4468
}
4569

4670
void TConsumerOperations::Merge(const TConsumerOperations& rhs)
@@ -49,12 +73,16 @@ void TConsumerOperations::Merge(const TConsumerOperations& rhs)
4973
Y_ABORT_UNLESS(Consumer_.Empty() || Consumer_ == rhs.Consumer_);
5074

5175
for (auto& range : rhs.Offsets_) {
52-
AddOperationImpl(*rhs.Consumer_, range.first, range.second);
76+
AddOperationImpl(*rhs.Consumer_, range.first, range.second, rhs.GetForceCommit(), rhs.GetKillReadSession(), rhs.GetOnlyCheckCommitedToFinish());
5377
}
5478
}
5579

5680
void TConsumerOperations::AddOperationImpl(const TString& consumer,
57-
ui64 begin, ui64 end)
81+
ui64 begin,
82+
ui64 end,
83+
bool forceCommit,
84+
bool killReadSession,
85+
bool onlyCheckCommitedToFinish)
5886
{
5987
if (Offsets_.Intersects(begin, end)) {
6088
ythrow TOffsetsRangeIntersectExpection() << "offset ranges intersect";
@@ -64,7 +92,13 @@ void TConsumerOperations::AddOperationImpl(const TString& consumer,
6492
Consumer_ = consumer;
6593
}
6694

67-
Offsets_.InsertInterval(begin, end);
95+
if (end != 0) {
96+
Offsets_.InsertInterval(begin, end);
97+
}
98+
99+
ForceCommit_ = forceCommit;
100+
KillReadSession_ = killReadSession;
101+
OnlyCheckCommitedToFinish_ = onlyCheckCommitedToFinish;
68102
}
69103

70104
//
@@ -76,9 +110,13 @@ bool TTopicPartitionOperations::IsValid() const
76110
[](auto& x) { return x.second.IsValid(); });
77111
}
78112

79-
void TTopicPartitionOperations::AddOperation(const TString& topic, ui32 partition,
113+
void TTopicPartitionOperations::AddOperation(const TString& topic,
114+
ui32 partition,
80115
const TString& consumer,
81-
const Ydb::Topic::OffsetsRange& range)
116+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
117+
bool forceCommit,
118+
bool killReadSession,
119+
bool onlyCheckCommitedToFinish)
82120
{
83121
Y_ABORT_UNLESS(Topic_.Empty() || Topic_ == topic);
84122
Y_ABORT_UNLESS(Partition_.Empty() || Partition_ == partition);
@@ -88,7 +126,7 @@ void TTopicPartitionOperations::AddOperation(const TString& topic, ui32 partitio
88126
Partition_ = partition;
89127
}
90128

91-
Operations_[consumer].AddOperation(consumer, range);
129+
Operations_[consumer].AddOperation(consumer, range, forceCommit, killReadSession, onlyCheckCommitedToFinish);
92130
}
93131

94132
void TTopicPartitionOperations::AddOperation(const TString& topic, ui32 partition,
@@ -117,11 +155,14 @@ void TTopicPartitionOperations::BuildTopicTxs(TTopicOperationTransactions& txs)
117155
for (auto& [consumer, operations] : Operations_) {
118156
NKikimrPQ::TPartitionOperation* o = t.tx.MutableOperations()->Add();
119157
o->SetPartitionId(*Partition_);
120-
auto [begin, end] = operations.GetRange();
121-
o->SetBegin(begin);
122-
o->SetEnd(end);
158+
auto [begin, end] = operations.GetOffsetsCommitRange();
159+
o->SetCommitOffsetsBegin(begin);
160+
o->SetCommitOffsetsEnd(end);
123161
o->SetConsumer(consumer);
124162
o->SetPath(*Topic_);
163+
o->SetKillReadSession(operations.GetKillReadSession());
164+
o->SetForceCommit(operations.GetForceCommit());
165+
o->SetOnlyCheckCommitedToFinish(operations.GetOnlyCheckCommitedToFinish());
125166
}
126167

127168
if (HasWriteOperations_) {
@@ -251,14 +292,23 @@ bool TTopicOperations::TabletHasReadOperations(ui64 tabletId) const
251292
return false;
252293
}
253294

254-
void TTopicOperations::AddOperation(const TString& topic, ui32 partition,
295+
void TTopicOperations::AddOperation(const TString& topic,
296+
ui32 partition,
255297
const TString& consumer,
256-
const Ydb::Topic::OffsetsRange& range)
298+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
299+
bool forceCommit,
300+
bool killReadSession,
301+
bool onlyCheckCommitedToFinish
302+
)
257303
{
258304
TTopicPartition key{topic, partition};
259-
Operations_[key].AddOperation(topic, partition,
305+
Operations_[key].AddOperation(topic,
306+
partition,
260307
consumer,
261-
range);
308+
range,
309+
forceCommit,
310+
killReadSession,
311+
onlyCheckCommitedToFinish);
262312
HasReadOperations_ = true;
263313
}
264314

‎ydb/core/kqp/topics/kqp_topics.h‎

Lines changed: 32 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
#include <ydb/public/api/protos/ydb_topic.pb.h>
55
#include <ydb/core/protos/pqconfig.pb.h>
66

7+
#include <ydb/core/protos/kqp.pb.h>
78
#include <ydb/core/tx/long_tx_service/public/lock_handle.h>
89
#include <ydb/core/tx/scheme_cache/scheme_cache.h>
910

@@ -26,20 +27,36 @@ class TConsumerOperations {
2627
public:
2728
bool IsValid() const;
2829

29-
std::pair<ui64, ui64> GetRange() const;
30+
std::pair<ui64, ui64> GetOffsetsCommitRange() const;
3031

31-
ui64 GetBegin() const;
32-
ui64 GetEnd() const;
32+
ui64 GetOffsetCommitBegin() const;
33+
ui64 GetOffsetCommitEnd() const;
34+
35+
bool GetForceCommit() const;
36+
bool GetKillReadSession() const;
37+
bool GetOnlyCheckCommitedToFinish() const;
38+
39+
void AddOperation(const TString& consumer,
40+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
41+
bool forceCommit = false,
42+
bool killReadSession = false,
43+
bool onlyCheckCommitedToFinish = false);
3344

34-
void AddOperation(const TString& consumer, const Ydb::Topic::OffsetsRange& range);
3545
void Merge(const TConsumerOperations& rhs);
3646

3747
private:
3848
void AddOperationImpl(const TString& consumer,
39-
ui64 begin, ui64 end);
49+
ui64 begin,
50+
ui64 end,
51+
bool forceCommit = false,
52+
bool killReadSession = false,
53+
bool onlyCheckCommitedToFinish = false);
4054

4155
TMaybe<TString> Consumer_;
4256
TDisjointIntervalTree<ui64> Offsets_;
57+
bool ForceCommit_ = false;
58+
bool KillReadSession_ = false;
59+
bool OnlyCheckCommitedToFinish_ = false;
4360
};
4461

4562
struct TTopicOperationTransaction {
@@ -53,9 +70,13 @@ class TTopicPartitionOperations {
5370
public:
5471
bool IsValid() const;
5572

56-
void AddOperation(const TString& topic, ui32 partition,
73+
void AddOperation(const TString& topic,
74+
ui32 partition,
5775
const TString& consumer,
58-
const Ydb::Topic::OffsetsRange& range);
76+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
77+
bool forceCommit = false,
78+
bool killReadSession = false,
79+
bool onlyCheckCommitedToFinish = false);
5980
void AddOperation(const TString& topic, ui32 partition,
6081
TMaybe<ui32> supportivePartition);
6182

@@ -106,7 +127,10 @@ class TTopicOperations {
106127

107128
void AddOperation(const TString& topic, ui32 partition,
108129
const TString& consumer,
109-
const Ydb::Topic::OffsetsRange& range);
130+
const NKikimrKqp::TTopicOperationsRequest_TopicOffsets_PartitionOffsets_OffsetsRange& range,
131+
bool forceCommit,
132+
bool killReadSession,
133+
bool onlyCheckCommitedToFinish);
110134
void AddOperation(const TString& topic, ui32 partition,
111135
TMaybe<ui32> supportivePartition);
112136

‎ydb/core/persqueue/events/internal.h‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -810,11 +810,14 @@ struct TEvPQ {
810810
{
811811
}
812812

813-
void AddOperation(TString consumer, ui64 begin, ui64 end) {
813+
void AddOperation(TString consumer, ui64 begin, ui64 end, bool forceCommit = false, bool killReadSession = false, bool onlyCheckCommitedToFinish = false) {
814814
NKikimrPQ::TPartitionOperation operation;
815-
operation.SetBegin(begin);
816-
operation.SetEnd(end);
815+
operation.SetCommitOffsetsBegin(begin);
816+
operation.SetCommitOffsetsEnd(end);
817817
operation.SetConsumer(std::move(consumer));
818+
operation.SetForceCommit(forceCommit);
819+
operation.SetKillReadSession(killReadSession);
820+
operation.SetOnlyCheckCommitedToFinish(onlyCheckCommitedToFinish);
818821

819822
Operations.push_back(std::move(operation));
820823
}

0 commit comments

Comments
 (0)