Skip to content

Commit 3335da7

Browse files
ivanmorozov333ivanmorozov333
committed
fixes
1 parent 716dba8 commit 3335da7

File tree

33 files changed

+244
-69
lines changed

33 files changed

+244
-69
lines changed

‎ydb/core/protos/config.proto‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1725,6 +1725,7 @@ message TColumnShardConfig {
17251725
optional uint64 MemoryLimitScanPortion = 27 [default = 100000000];
17261726
optional string ReaderClassName = 28;
17271727
optional bool AllowNullableColumnsInPK = 29 [default = false];
1728+
optional uint32 RestoreDataOnWriteTimeoutSeconds = 30;
17281729
}
17291730

17301731
message TSchemeShardConfig {

‎ydb/core/tx/columnshard/columnshard.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,11 @@ void TColumnShard::CleanupActors(const TActorContext& ctx) {
3030
if (BackgroundSessionsManager) {
3131
BackgroundSessionsManager->Stop();
3232
}
33+
InFlightReadsTracker.Stop(this);
3334
ctx.Send(ResourceSubscribeActor, new TEvents::TEvPoisonPill);
3435
ctx.Send(BufferizationWriteActorId, new TEvents::TEvPoisonPill);
3536
ctx.Send(DataAccessorsControlActorId, new TEvents::TEvPoisonPill);
37+
OperationsManager->StopWriting();
3638
if (PrioritizationClientId) {
3739
NPrioritiesQueue::TCompServiceOperator::UnregisterClient(PrioritizationClientId);
3840
}

‎ydb/core/tx/columnshard/columnshard.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@ namespace TEvColumnShard {
108108
public:
109109
std::optional<NOlap::TSnapshot> ReadFromSnapshot;
110110
std::optional<NOlap::TSnapshot> ReadToSnapshot;
111+
TString TaskIdentifier;
111112
std::shared_ptr<NOlap::TPKRangesFilter> RangesFilter;
112113
public:
113114
void AddColumn(const ui32 id, const TString& columnName) {

‎ydb/core/tx/columnshard/columnshard__write.cpp‎

Lines changed: 16 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -93,12 +93,15 @@ void TColumnShard::Handle(NPrivateEvents::NWrite::TEvWritePortionResult::TPtr& e
9393
NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_WRITE)("tablet_id", TabletID())("event", "TEvWritePortionResult");
9494
std::vector<TNoDataWrite> noDataWrites = ev->Get()->DetachNoDataWrites();
9595
for (auto&& i : noDataWrites) {
96+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "no_data_write_finished")("writing_size", i.GetDataSize())("writing_id", i.GetWriteMeta().GetId());
9697
Counters.GetWritesMonitor()->OnFinishWrite(i.GetDataSize(), 1);
9798
}
9899
if (ev->Get()->GetWriteStatus() == NKikimrProto::OK) {
99100
std::vector<TInsertedPortions> writtenPacks = ev->Get()->DetachInsertedPacks();
100101
const TMonotonic now = TMonotonic::Now();
101102
for (auto&& i : writtenPacks) {
103+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", i.GetDataSize())("event", "data_write_finished")(
104+
"writing_id", i.GetWriteMeta().GetId());
102105
Counters.OnWritePutBlobsSuccess(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
103106
Counters.GetWritesMonitor()->OnFinishWrite(i.GetDataSize(), 1);
104107
}
@@ -115,6 +118,8 @@ void TColumnShard::Handle(NPrivateEvents::NWrite::TEvWritePortionResult::TPtr& e
115118
for (auto&& i : writtenPacks) {
116119
Counters.OnWritePutBlobsFailed(now - i.GetWriteMeta().GetWriteStartInstant(), i.GetRecordsCount());
117120
Counters.GetCSCounters().OnWritePutBlobsFail(now - i.GetWriteMeta().GetWriteStartInstant());
121+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", i.GetDataSize())("event", "data_write_error")(
122+
"writing_id", i.GetWriteMeta().GetId());
118123
Counters.GetWritesMonitor()->OnFinishWrite(i.GetDataSize(), 1);
119124
}
120125
Execute(new TTxBlobsWritingFailed(this, ev->Get()->GetWriteStatus(), std::move(writtenPacks)), ctx);
@@ -131,10 +136,11 @@ void TColumnShard::Handle(TEvPrivate::TEvWriteBlobsResult::TPtr& ev, const TActo
131136
auto baseAggregations = wBuffer.GetAggregations();
132137
wBuffer.InitReplyReceived(TMonotonic::Now());
133138

134-
Counters.GetWritesMonitor()->OnFinishWrite(wBuffer.GetSumSize(), wBuffer.GetAggregations().size());
135-
136139
for (auto&& aggr : baseAggregations) {
137140
const auto& writeMeta = aggr->GetWriteMeta();
141+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("event", "blobs_write_finished")("writing_size", aggr->GetSize())(
142+
"writing_id", writeMeta.GetId())("status", putResult.GetPutStatus());
143+
Counters.GetWritesMonitor()->OnFinishWrite(aggr->GetSize(), 1);
138144

139145
if (!TablesManager.IsReadyForWrite(writeMeta.GetTableId())) {
140146
ACFL_ERROR("event", "absent_pathId")("path_id", writeMeta.GetTableId())("has_index", TablesManager.HasPrimaryIndex());
@@ -210,7 +216,7 @@ void TColumnShard::Handle(TEvColumnShard::TEvWrite::TPtr& ev, const TActorContex
210216
granuleShardingVersion = record.GetGranuleShardingVersion();
211217
}
212218

213-
NEvWrite::TWriteMeta writeMeta(writeId, pathId, source, granuleShardingVersion);
219+
NEvWrite::TWriteMeta writeMeta(writeId, pathId, source, granuleShardingVersion, TGUID::CreateTimebased().AsGuidString());
214220
if (record.HasModificationType()) {
215221
writeMeta.SetModificationType(TEnumOperator<NEvWrite::EModificationType>::DeserializeFromProto(record.GetModificationType()));
216222
}
@@ -569,14 +575,10 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
569575
if (overloadStatus != EOverloadStatus::None) {
570576
std::unique_ptr<NActors::IEventBase> result = NEvents::TDataEvents::TEvWriteResult::BuildError(
571577
TabletID(), 0, NKikimrDataEvents::TEvWriteResult::STATUS_OVERLOADED, "overload data error");
572-
OverloadWriteFail(overloadStatus, NEvWrite::TWriteMeta(0, pathId, source, {}), arrowData->GetSize(), cookie, std::move(result), ctx);
578+
OverloadWriteFail(overloadStatus, NEvWrite::TWriteMeta(0, pathId, source, {}, TGUID::CreateTimebased().AsGuidString()), arrowData->GetSize(), cookie, std::move(result), ctx);
573579
return;
574580
}
575581

576-
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", arrowData->GetSize())(
577-
"in_flight", Counters.GetWritesMonitor()->GetWritesInFlight())("sizE_in_flight", Counters.GetWritesMonitor()->GetWritesSizeInFlight());
578-
Counters.GetWritesMonitor()->OnStartWrite(arrowData->GetSize());
579-
580582
std::optional<ui32> granuleShardingVersionId;
581583
if (record.HasGranuleShardingVersionId()) {
582584
granuleShardingVersionId = record.GetGranuleShardingVersionId();
@@ -597,9 +599,14 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
597599
OperationsManager->RegisterLock(lockId, Generation());
598600
auto writeOperation = OperationsManager->RegisterOperation(
599601
pathId, lockId, cookie, granuleShardingVersionId, *mType, AppDataVerified().FeatureFlags.GetEnableWritePortionsOnInsert());
602+
603+
AFL_WARN(NKikimrServices::TX_COLUMNSHARD_WRITE)("writing_size", arrowData->GetSize())("operation_id", writeOperation->GetIdentifier())(
604+
"in_flight", Counters.GetWritesMonitor()->GetWritesInFlight())("size_in_flight", Counters.GetWritesMonitor()->GetWritesSizeInFlight());
605+
Counters.GetWritesMonitor()->OnStartWrite(arrowData->GetSize());
606+
600607
Y_ABORT_UNLESS(writeOperation);
601608
writeOperation->SetBehaviour(behaviour);
602-
NOlap::TWritingContext wContext(pathId, SelfId(), schema, StoragesManager, Counters.GetIndexationCounters().SplitterCounters,
609+
NOlap::TWritingContext wContext(TabletID(), SelfId(), schema, StoragesManager, Counters.GetIndexationCounters().SplitterCounters,
603610
Counters.GetCSCounters().WritingCounters, NOlap::TSnapshot::Max(), writeOperation->GetActivityChecker());
604611
arrowData->SetSeparationPoints(GetIndexAs<NOlap::TColumnEngineForLogs>().GetGranulePtrVerified(pathId)->GetBucketPositions());
605612
writeOperation->Start(*this, arrowData, source, wContext);

‎ydb/core/tx/columnshard/columnshard_impl.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -383,7 +383,7 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
383383
switch (ev->GetTypeRewrite()) {
384384
HFunc(TEvTablet::TEvTabletDead, HandleTabletDead);
385385
default:
386-
LOG_S_WARN("TColumnShard.StateBroken at " << TabletID() << " unhandled event type: " << ev->GetTypeRewrite()
386+
LOG_S_WARN("TColumnShard.StateBroken at " << TabletID() << " unhandled event type: " << ev->GetTypeName()
387387
<< " event: " << ev->ToString());
388388
Send(IEventHandle::ForwardOnNondelivery(std::move(ev), NActors::TEvents::TEvUndelivered::ReasonActorUnknown));
389389
break;
@@ -453,7 +453,7 @@ class TColumnShard: public TActor<TColumnShard>, public NTabletFlatExecutor::TTa
453453

454454
default:
455455
if (!HandleDefaultEvents(ev, SelfId())) {
456-
LOG_S_WARN("TColumnShard.StateWork at " << TabletID() << " unhandled event type: " << ev->GetTypeRewrite()
456+
LOG_S_WARN("TColumnShard.StateWork at " << TabletID() << " unhandled event type: " << ev->GetTypeName()
457457
<< " event: " << ev->ToString());
458458
}
459459
break;

‎ydb/core/tx/columnshard/counters/scan.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ class TScanCounters: public TCommonCountersOwner {
6464
UndeliveredEvent /* "UndeliveredEvent" */,
6565
CannotAddInFlight /* "CannotAddInFlight" */,
6666
ProblemOnStart /*ProblemOnStart*/,
67+
AbortFromTablet /*AbortFromTablet*/,
6768

6869
COUNT
6970
};

‎ydb/core/tx/columnshard/counters/writes_monitor.cpp‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
#include "writes_monitor.h"
2+
3+
#include <ydb/library/actors/core/log.h>
4+
5+
namespace NKikimr::NColumnShard {
6+
7+
TAtomicCounter TWritesMonitor::WritesInFlight = 0;
8+
TAtomicCounter TWritesMonitor::WritesSizeInFlight = 0;
9+
10+
void TWritesMonitor::OnStartWrite(const ui64 dataSize) {
11+
++WritesInFlightLocal;
12+
WritesSizeInFlightLocal += dataSize;
13+
WritesInFlight.Inc();
14+
WritesSizeInFlight.Add(dataSize);
15+
UpdateTabletCounters();
16+
}
17+
18+
void TWritesMonitor::OnFinishWrite(const ui64 dataSize, const ui32 writesCount /*= 1*/) {
19+
AFL_VERIFY(writesCount <= WritesInFlightLocal);
20+
AFL_VERIFY(dataSize <= WritesSizeInFlightLocal);
21+
WritesSizeInFlightLocal -= dataSize;
22+
WritesInFlightLocal -= writesCount;
23+
AFL_VERIFY(0 <= WritesInFlight.Sub(writesCount));
24+
AFL_VERIFY(0 <= WritesSizeInFlight.Sub(dataSize));
25+
UpdateTabletCounters();
26+
}
27+
28+
TString TWritesMonitor::DebugString() const {
29+
return TStringBuilder() << "{object=write_monitor;count_local=" << WritesInFlightLocal << ";size_local=" << WritesSizeInFlightLocal << ";"
30+
<< "count_node=" << WritesInFlight.Val() << ";size_node=" << WritesSizeInFlight.Val() << "}";
31+
}
32+
33+
} // namespace NKikimr::NColumnShard

‎ydb/core/tx/columnshard/counters/writes_monitor.h‎

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -6,40 +6,40 @@
66

77
namespace NKikimr::NColumnShard {
88

9-
class TWritesMonitor {
9+
class TWritesMonitor: TNonCopyable {
1010
private:
1111
TTabletCountersBase& Stats;
1212

13-
YDB_READONLY(ui64, WritesInFlight, 0);
14-
YDB_READONLY(ui64, WritesSizeInFlight, 0);
13+
static TAtomicCounter WritesInFlight;
14+
static TAtomicCounter WritesSizeInFlight;
15+
ui64 WritesInFlightLocal = 0;
16+
ui64 WritesSizeInFlightLocal = 0;
1517

1618
public:
1719
TWritesMonitor(TTabletCountersBase& stats)
1820
: Stats(stats) {
1921
}
2022

21-
void OnStartWrite(const ui64 dataSize) {
22-
++WritesInFlight;
23-
WritesSizeInFlight += dataSize;
24-
UpdateTabletCounters();
23+
~TWritesMonitor() {
24+
OnFinishWrite(WritesSizeInFlightLocal, WritesInFlightLocal);
2525
}
2626

27-
void OnFinishWrite(const ui64 dataSize, const ui32 writesCount = 1) {
28-
Y_ABORT_UNLESS(WritesInFlight > 0);
29-
Y_ABORT_UNLESS(WritesSizeInFlight >= dataSize);
30-
WritesInFlight -= writesCount;
31-
WritesSizeInFlight -= dataSize;
32-
UpdateTabletCounters();
33-
}
27+
void OnStartWrite(const ui64 dataSize);
28+
29+
void OnFinishWrite(const ui64 dataSize, const ui32 writesCount = 1);
3430

35-
TString DebugString() const {
36-
return TStringBuilder() << "{object=write_monitor;count=" << WritesInFlight << ";size=" << WritesSizeInFlight
37-
<< "}";
31+
TString DebugString() const;
32+
33+
ui64 GetWritesInFlight() const {
34+
return WritesInFlight.Val();
35+
}
36+
ui64 GetWritesSizeInFlight() const {
37+
return WritesSizeInFlight.Val();
3838
}
3939

4040
private:
4141
void UpdateTabletCounters() {
42-
Stats.Simple()[COUNTER_WRITES_IN_FLY].Set(WritesInFlight);
42+
Stats.Simple()[COUNTER_WRITES_IN_FLY].Set(WritesInFlightLocal);
4343
}
4444
};
4545

‎ydb/core/tx/columnshard/counters/ya.make‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ SRCS(
1414
scan.cpp
1515
splitter.cpp
1616
portions.cpp
17+
writes_monitor.cpp
1718
)
1819

1920
PEERDIR(

‎ydb/core/tx/columnshard/data_reader/actor.cpp‎

Lines changed: 60 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -3,50 +3,104 @@
33
namespace NKikimr::NOlap::NDataReader {
44

55
void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanData::TPtr& ev) {
6+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "scan_data");
7+
LastAck = std::nullopt;
8+
if (!CheckActivity()) {
9+
TBase::Send(*ScanActorId, new NKqp::TEvKqp::TEvAbortExecution(NYql::NDqProto::StatusIds::ABORTED, "external task aborted"));
10+
return;
11+
}
612
SwitchStage(EStage::WaitData, EStage::WaitData);
713
auto data = ev->Get()->ArrowBatch;
814
AFL_VERIFY(!!data || ev->Get()->Finished);
915
if (data) {
1016
AFL_VERIFY(ScanActorId);
1117
const auto status = RestoreTask->OnDataChunk(data);
1218
if (status.IsSuccess()) {
13-
TBase::Send(*ScanActorId, new NKqp::TEvKqpCompute::TEvScanDataAck(FreeSpace, 1, 1));
19+
TBase::Send(*ScanActorId, new NKqp::TEvKqpCompute::TEvScanDataAck(FreeSpace, 1, 1), NActors::IEventHandle::FlagTrackDelivery);
20+
LastAck = TMonotonic::Now();
1421
} else {
22+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "scan_data_restore_fail")("message", status.GetErrorMessage());
1523
SwitchStage(EStage::WaitData, EStage::Finished);
1624
TBase::Send(*ScanActorId, NKqp::TEvKqp::TEvAbortExecution::Aborted("task finished: " + status.GetErrorMessage()).Release());
25+
PassAway();
1726
}
1827
} else {
1928
SwitchStage(EStage::WaitData, EStage::Finished);
2029
auto status = RestoreTask->OnFinished();
2130
if (status.IsFail()) {
22-
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "restore_task_finished_error")("reason", status.GetErrorMessage());
31+
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "restore_task_finished_error")("reason", status.GetErrorMessage());
2332
} else {
24-
AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "restore_task_finished")("reason", status.GetErrorMessage());
33+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "restore_task_finished");
2534
}
2635
PassAway();
2736
}
2837
}
2938

3039
void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanInitActor::TPtr& ev) {
40+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "init_actor");
41+
LastAck = std::nullopt;
42+
if (!CheckActivity()) {
43+
TBase::Send(*ScanActorId, new NKqp::TEvKqp::TEvAbortExecution(NYql::NDqProto::StatusIds::ABORTED, "external task aborted"));
44+
return;
45+
}
3146
SwitchStage(EStage::Initialization, EStage::WaitData);
3247
AFL_VERIFY(!ScanActorId);
3348
auto& msg = ev->Get()->Record;
3449
ScanActorId = ActorIdFromProto(msg.GetScanActorId());
35-
TBase::Send(*ScanActorId, new NKqp::TEvKqpCompute::TEvScanDataAck(FreeSpace, 1, 1));
50+
TBase::Send(*ScanActorId, new NKqp::TEvKqpCompute::TEvScanDataAck(FreeSpace, 1, 1), NActors::IEventHandle::FlagTrackDelivery);
51+
LastAck = TMonotonic::Now();
3652
}
3753

3854
void TActor::HandleExecute(NKqp::TEvKqpCompute::TEvScanError::TPtr& ev) {
3955
SwitchStage(std::nullopt, EStage::Finished);
40-
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "problem_on_restore_data")(
56+
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "problem_on_restore_data")(
4157
"reason", NYql::IssuesFromMessageAsString(ev->Get()->Record.GetIssues()));
4258
RestoreTask->OnError(NYql::IssuesFromMessageAsString(ev->Get()->Record.GetIssues()));
4359
PassAway();
4460
}
4561

62+
void TActor::HandleExecute(NActors::TEvents::TEvWakeup::TPtr& /*ev*/) {
63+
if (!CheckActivity()) {
64+
TBase::Send(*ScanActorId, new NKqp::TEvKqp::TEvAbortExecution(NYql::NDqProto::StatusIds::ABORTED, "external task aborted"));
65+
return;
66+
}
67+
68+
if (LastAck && TMonotonic::Now() - *LastAck > RestoreTask->GetTimeout()) {
69+
SwitchStage(std::nullopt, EStage::Finished);
70+
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "problem_timeout");
71+
RestoreTask->OnError("timeout on restore data");
72+
TBase::Send(*ScanActorId, new NKqp::TEvKqp::TEvAbortExecution(NYql::NDqProto::StatusIds::ABORTED, "external task aborted"));
73+
PassAway();
74+
return;
75+
}
76+
Schedule(TDuration::Seconds(1), new NActors::TEvents::TEvWakeup());
77+
}
78+
4679
void TActor::Bootstrap(const TActorContext& /*ctx*/) {
80+
if (!CheckActivity()) {
81+
return;
82+
}
83+
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "start_restore")("tablet_actor_id", RestoreTask->GetTabletActorId())("this", (ui64)this);
4784
auto evStart = RestoreTask->BuildRequestInitiator();
48-
Send(RestoreTask->GetTabletActorId(), evStart.release());
85+
Send(RestoreTask->GetTabletActorId(), evStart.release(), NActors::IEventHandle::FlagTrackDelivery);
86+
LastAck = TMonotonic::Now();
4987
Become(&TActor::StateFunc);
88+
Schedule(TDuration::Seconds(1), new NActors::TEvents::TEvWakeup());
89+
}
90+
91+
bool TActor::CheckActivity() {
92+
if (AbortedFlag) {
93+
return false;
94+
}
95+
if (RestoreTask->IsActive()) {
96+
return true;
97+
}
98+
AbortedFlag = true;
99+
AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("event", "restoring_aborted");
100+
SwitchStage(std::nullopt, EStage::Finished);
101+
RestoreTask->OnError("restore task aborted");
102+
PassAway();
103+
return false;
50104
}
51105

52106
}

0 commit comments

Comments
 (0)