Skip to content

Commit d24978e

Browse files
azevaykinazevaykin
authored
Statistics: Two force traversal collections in local DB (#7765)
1 parent 929e8f5 commit d24978e

8 files changed

+246
-125
lines changed

‎ydb/core/statistics/aggregator/aggregator_impl.cpp

Lines changed: 83 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -443,8 +443,8 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvAnalyzeStatus::TPtr& ev) {
443443
if (ForceTraversalOperationId == operationId) {
444444
outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_IN_PROGRESS);
445445
} else {
446-
if (std::any_of(ForceTraversals.begin(), ForceTraversals.end(),
447-
[&operationId](const TForceTraversal& elem) { return elem.OperationId == operationId;})) {
446+
auto forceTraversalOperation = ForceTraversalOperation(operationId);
447+
if (forceTraversalOperation) {
448448
outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_ENQUEUED);
449449
} else {
450450
outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_NO_OPERATION);
@@ -580,22 +580,35 @@ void TStatisticsAggregator::ScheduleNextTraversal(NIceDb::TNiceDb& db) {
580580

581581
TPathId pathId;
582582

583-
if (!ForceTraversals.empty() && !LastTraversalWasForce) {
583+
if (!LastTraversalWasForce) {
584584
LastTraversalWasForce = true;
585585

586-
TForceTraversal& operation = ForceTraversals.front();
587-
pathId = operation.PathId;
586+
for (TForceTraversalOperation& operation : ForceTraversals) {
587+
for (TForceTraversalTable& operationTable : operation.Tables) {
588+
if (operationTable.Status == TForceTraversalTable::EStatus::None) {
589+
operationTable.Status = TForceTraversalTable::EStatus::RequestSent;
590+
db.Table<Schema::ForceTraversalTables>().Key(operation.OperationId, operationTable.PathId.OwnerId, operationTable.PathId.LocalPathId)
591+
.Update(NIceDb::TUpdate<Schema::ForceTraversalTables::Status>((ui64)operationTable.Status));
588592

589-
ForceTraversalOperationId = operation.OperationId;
590-
ForceTraversalColumnTags = operation.ColumnTags;
591-
ForceTraversalTypes = operation.Types;
592-
ForceTraversalReplyToActorId = operation.ReplyToActorId;
593+
pathId = operationTable.PathId;
594+
break;
595+
}
596+
}
597+
598+
if (!pathId) {
599+
SA_LOG_D("[" << TabletID() << "] All the force traversal tables sent the requests. OperationId=" << operation.OperationId);
600+
continue;
601+
}
593602

594-
PersistForceTraversal(db);
603+
ForceTraversalOperationId = operation.OperationId;
604+
}
595605

596-
// db.Table<Schema::ForceTraversals>().Key(operation.OperationId, operation.PathId.OwnerId, operation.PathId.LocalPathId).Delete();
597-
ForceTraversals.pop_front();
598-
} else if (!ScheduleTraversalsByTime.Empty()){
606+
if (!pathId) {
607+
SA_LOG_D("[" << TabletID() << "] All the force traversal operations sent the requests.");
608+
}
609+
}
610+
611+
if (!pathId && !ScheduleTraversalsByTime.Empty()){
599612
LastTraversalWasForce = false;
600613

601614
auto* oldestTable = ScheduleTraversalsByTime.Top();
@@ -606,8 +619,10 @@ void TStatisticsAggregator::ScheduleNextTraversal(NIceDb::TNiceDb& db) {
606619
}
607620

608621
pathId = oldestTable->PathId;
609-
} else {
610-
SA_LOG_E("[" << TabletID() << "] No schedule traversal from schemeshard.");
622+
}
623+
624+
if (!pathId) {
625+
SA_LOG_E("[" << TabletID() << "] No traversal from schemeshard.");
611626
return;
612627
}
613628

@@ -653,13 +668,66 @@ void TStatisticsAggregator::FinishTraversal(NIceDb::TNiceDb& db) {
653668
}
654669
}
655670

671+
auto forceTraversalOperation = CurrentForceTraversalOperation();
672+
if (forceTraversalOperation) {
673+
bool tablesRemained = std::any_of(forceTraversalOperation->Tables.begin(), forceTraversalOperation->Tables.end(),
674+
[](const TForceTraversalTable& elem) { return elem.Status == TForceTraversalTable::EStatus::None;});
675+
if (!tablesRemained) {
676+
DeleteForceTraversalOperation(ForceTraversalOperationId, db);
677+
}
678+
}
679+
656680
ResetTraversalState(db);
657681
}
658682

659683
TString TStatisticsAggregator::LastTraversalWasForceString() const {
660684
return LastTraversalWasForce ? "force" : "schedule";
661685
}
662686

687+
TStatisticsAggregator::TForceTraversalOperation* TStatisticsAggregator::CurrentForceTraversalOperation() {
688+
return ForceTraversalOperation(ForceTraversalOperationId);
689+
}
690+
691+
TStatisticsAggregator::TForceTraversalOperation* TStatisticsAggregator::ForceTraversalOperation(const TString& operationId) {
692+
auto forceTraversalOperation = std::find_if(ForceTraversals.begin(), ForceTraversals.end(),
693+
[operationId](const TForceTraversalOperation& elem) { return elem.OperationId == operationId;});
694+
695+
if (forceTraversalOperation == ForceTraversals.end()) {
696+
return nullptr;
697+
} else {
698+
return &*forceTraversalOperation;
699+
}
700+
}
701+
702+
void TStatisticsAggregator::DeleteForceTraversalOperation(const TString& operationId, NIceDb::TNiceDb& db) {
703+
db.Table<Schema::ForceTraversalOperations>().Key(ForceTraversalOperationId).Delete();
704+
705+
auto operation = ForceTraversalOperation(operationId);
706+
for(const TForceTraversalTable& table : operation->Tables) {
707+
db.Table<Schema::ForceTraversalTables>().Key(operationId, table.PathId.OwnerId, table.PathId.LocalPathId).Delete();
708+
}
709+
710+
ForceTraversals.remove_if([operationId](const TForceTraversalOperation& elem) { return elem.OperationId == operationId;});
711+
}
712+
713+
TStatisticsAggregator::TForceTraversalTable* TStatisticsAggregator::ForceTraversalTable(const TString& operationId, const TPathId& pathId) {
714+
for (TForceTraversalOperation& operation : ForceTraversals) {
715+
if (operation.OperationId == operationId) {
716+
for (TForceTraversalTable& operationTable : operation.Tables) {
717+
if (operationTable.PathId == pathId) {
718+
return &operationTable;
719+
}
720+
}
721+
}
722+
}
723+
724+
return nullptr;
725+
}
726+
727+
TStatisticsAggregator::TForceTraversalTable* TStatisticsAggregator::CurrentForceTraversalTable() {
728+
return ForceTraversalTable(ForceTraversalOperationId, TraversalTableId.PathId);
729+
}
730+
663731
void TStatisticsAggregator::PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value) {
664732
db.Table<Schema::SysParams>().Key(id).Update(
665733
NIceDb::TUpdate<Schema::SysParams::Value>(value));
@@ -676,30 +744,19 @@ void TStatisticsAggregator::PersistStartKey(NIceDb::TNiceDb& db) {
676744
PersistSysParam(db, Schema::SysParam_TraversalStartKey, TraversalStartKey.GetBuffer());
677745
}
678746

679-
void TStatisticsAggregator::PersistForceTraversal(NIceDb::TNiceDb& db) {
680-
PersistSysParam(db, Schema::SysParam_ForceTraversalOperationId, ToString(ForceTraversalOperationId));
681-
PersistSysParam(db, Schema::SysParam_ForceTraversalCookie, ForceTraversalOperationId);
682-
PersistSysParam(db, Schema::SysParam_ForceTraversalColumnTags, ToString(ForceTraversalColumnTags));
683-
PersistSysParam(db, Schema::SysParam_ForceTraversalTypes, ToString(ForceTraversalTypes));
684-
}
685-
686747
void TStatisticsAggregator::PersistGlobalTraversalRound(NIceDb::TNiceDb& db) {
687748
PersistSysParam(db, Schema::SysParam_GlobalTraversalRound, ToString(GlobalTraversalRound));
688749
}
689750

690751
void TStatisticsAggregator::ResetTraversalState(NIceDb::TNiceDb& db) {
691752
ForceTraversalOperationId.clear();
692753
TraversalTableId.PathId = TPathId();
693-
ForceTraversalColumnTags.clear();
694-
ForceTraversalTypes.clear();
695754
TraversalStartTime = TInstant::MicroSeconds(0);
696755
PersistTraversal(db);
697756

698757
TraversalStartKey = TSerializedCellVec();
699758
PersistStartKey(db);
700759

701-
ForceTraversalReplyToActorId = {};
702-
703760
for (auto& [tag, _] : CountMinSketches) {
704761
db.Table<Schema::ColumnStatistics>().Key(tag).Delete();
705762
}

‎ydb/core/statistics/aggregator/aggregator_impl.h

Lines changed: 22 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,6 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
145145

146146
void PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value);
147147
void PersistTraversal(NIceDb::TNiceDb& db);
148-
void PersistForceTraversal(NIceDb::TNiceDb& db);
149148
void PersistStartKey(NIceDb::TNiceDb& db);
150149
void PersistGlobalTraversalRound(NIceDb::TNiceDb& db);
151150

@@ -240,8 +239,6 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
240239
std::queue<TEvStatistics::TEvRequestStats::TPtr> PendingRequests;
241240
bool ProcessUrgentInFlight = false;
242241

243-
TActorId ForceTraversalReplyToActorId = {};
244-
245242
bool IsSchemeshardSeen = false;
246243
bool IsStatisticsTableCreated = false;
247244
bool PendingSaveStatistics = false;
@@ -306,8 +303,7 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
306303
private: // stored in local db
307304

308305
TString ForceTraversalOperationId;
309-
TString ForceTraversalColumnTags;
310-
TString ForceTraversalTypes;
306+
311307
TTableId TraversalTableId;
312308
bool TraversalIsColumnTable = false;
313309
TSerializedCellVec TraversalStartKey;
@@ -323,14 +319,32 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
323319
TTraversalsByTime;
324320
TTraversalsByTime ScheduleTraversalsByTime;
325321

326-
struct TForceTraversal {
327-
TString OperationId;
322+
struct TForceTraversalTable {
328323
TPathId PathId;
329324
TString ColumnTags;
325+
326+
enum class EStatus : ui8 {
327+
None,
328+
RequestSent,
329+
ResponseReceived,
330+
};
331+
EStatus Status = EStatus::None;
332+
};
333+
struct TForceTraversalOperation {
334+
TString OperationId;
335+
std::list<TForceTraversalTable> Tables;
330336
TString Types;
331337
TActorId ReplyToActorId;
332338
};
333-
std::list<TForceTraversal> ForceTraversals;
339+
std::list<TForceTraversalOperation> ForceTraversals;
340+
341+
private:
342+
TForceTraversalOperation* CurrentForceTraversalOperation();
343+
TForceTraversalOperation* ForceTraversalOperation(const TString& operationId);
344+
void DeleteForceTraversalOperation(const TString& operationId, NIceDb::TNiceDb& db);
345+
346+
TForceTraversalTable* ForceTraversalTable(const TString& operationId, const TPathId& pathId);
347+
TForceTraversalTable* CurrentForceTraversalTable();
334348
};
335349

336350
} // NKikimr::NStat

‎ydb/core/statistics/aggregator/schema.h

Lines changed: 28 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -45,32 +45,45 @@ struct TAggregatorSchema : NIceDb::Schema {
4545
IsColumnTable
4646
>;
4747
};
48-
/*
49-
struct ForceTraversals : Table<5> {
50-
struct OperationId : Column<1, NScheme::NTypeIds::Uint64> {};
48+
49+
// struct ForceTraversals : Table<5>
50+
51+
struct ForceTraversalOperations : Table<6> {
52+
struct OperationId : Column<1, NScheme::NTypeIds::String> {};
53+
struct Types : Column<2, NScheme::NTypeIds::String> {};
54+
55+
using TKey = TableKey<OperationId>;
56+
using TColumns = TableColumns<
57+
OperationId,
58+
Types
59+
>;
60+
};
61+
62+
struct ForceTraversalTables : Table<7> {
63+
struct OperationId : Column<1, NScheme::NTypeIds::String> {};
5164
struct OwnerId : Column<2, NScheme::NTypeIds::Uint64> {};
5265
struct LocalPathId : Column<3, NScheme::NTypeIds::Uint64> {};
53-
struct Cookie : Column<4, NScheme::NTypeIds::String> {};
54-
struct ColumnTags : Column<5, NScheme::NTypeIds::String> {};
55-
struct Types : Column<6, NScheme::NTypeIds::String> {};
66+
struct ColumnTags : Column<4, NScheme::NTypeIds::String> {};
67+
struct Status : Column<5, NScheme::NTypeIds::Uint64> {};
5668

5769
using TKey = TableKey<OperationId, OwnerId, LocalPathId>;
5870
using TColumns = TableColumns<
5971
OperationId,
6072
OwnerId,
6173
LocalPathId,
62-
Cookie,
6374
ColumnTags,
64-
Types
75+
Status
6576
>;
6677
};
67-
*/
78+
6879
using TTables = SchemaTables<
6980
SysParams,
7081
BaseStatistics,
7182
ColumnStatistics,
72-
ScheduleTraversals
73-
// ForceTraversals
83+
ScheduleTraversals,
84+
// ForceTraversals,
85+
ForceTraversalOperations,
86+
ForceTraversalTables
7487
>;
7588

7689
using TSettings = SchemaSettings<
@@ -80,12 +93,12 @@ struct TAggregatorSchema : NIceDb::Schema {
8093

8194
static constexpr ui64 SysParam_Database = 1;
8295
static constexpr ui64 SysParam_TraversalStartKey = 2;
83-
static constexpr ui64 SysParam_ForceTraversalOperationId = 3;
96+
// deprecated 3
8497
static constexpr ui64 SysParam_TraversalTableOwnerId = 4;
8598
static constexpr ui64 SysParam_TraversalTableLocalPathId = 5;
86-
static constexpr ui64 SysParam_ForceTraversalCookie = 6;
87-
static constexpr ui64 SysParam_ForceTraversalColumnTags = 7;
88-
static constexpr ui64 SysParam_ForceTraversalTypes = 8;
99+
// deprecated 6
100+
// deprecated 7
101+
// deprecated 8
89102
static constexpr ui64 SysParam_TraversalStartTime = 9;
90103
// deprecated 10
91104
static constexpr ui64 SysParam_TraversalIsColumnTable = 11;

0 commit comments

Comments
 (0)