From a50edfed55b09d8bd668d2ed4688699529cfbc2d Mon Sep 17 00:00:00 2001 From: azevaykin <145343289+azevaykin@users.noreply.github.com> Date: Tue, 13 Aug 2024 18:26:14 +0300 Subject: Statistics: OperationId is generated by KQP (#7694) --- ydb/core/kqp/gateway/actors/analyze_actor.cpp | 29 ++++++++++- ydb/core/kqp/gateway/actors/analyze_actor.h | 7 +-- ydb/core/protos/out/out.cpp | 5 ++ ydb/core/protos/statistics.proto | 8 +-- ydb/core/statistics/aggregator/aggregator_impl.cpp | 19 +++---- ydb/core/statistics/aggregator/aggregator_impl.h | 8 +-- ydb/core/statistics/aggregator/schema.h | 4 +- .../statistics/aggregator/tx_analyze_table.cpp | 28 +++++----- .../statistics/aggregator/tx_finish_trasersal.cpp | 10 ++-- ydb/core/statistics/aggregator/tx_init.cpp | 14 +---- .../aggregator/ut/ut_analyze_columnshard.cpp | 59 ++++++++++++++++++++-- .../aggregator/ut/ut_analyze_datashard.cpp | 6 +-- ydb/core/statistics/service/http_request.cpp | 18 ++++++- ydb/core/statistics/service/http_request.h | 1 + ydb/core/statistics/ut_common/ut_common.cpp | 28 +++++++--- ydb/core/statistics/ut_common/ut_common.h | 10 +++- 16 files changed, 174 insertions(+), 80 deletions(-) diff --git a/ydb/core/kqp/gateway/actors/analyze_actor.cpp b/ydb/core/kqp/gateway/actors/analyze_actor.cpp index 15b11569773..2ababa6954b 100644 --- a/ydb/core/kqp/gateway/actors/analyze_actor.cpp +++ b/ydb/core/kqp/gateway/actors/analyze_actor.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -15,6 +16,18 @@ enum { using TNavigate = NSchemeCache::TSchemeCacheNavigate; +TString MakeOperationId() { + TULIDGenerator ulidGen; + return ulidGen.Next(TActivationContext::Now()).ToBinary(); +} + +TAnalyzeActor::TAnalyzeActor(TString tablePath, TVector columns, NThreading::TPromise promise) + : TablePath(tablePath) + , Columns(columns) + , Promise(promise) + , OperationId(MakeOperationId()) +{} + void TAnalyzeActor::Bootstrap() { using TNavigate = NSchemeCache::TSchemeCacheNavigate; auto navigate = std::make_unique(); @@ -34,7 +47,7 @@ void TAnalyzeActor::SendAnalyzeStatus() { auto getStatus = std::make_unique(); auto& record = getStatus->Record; - PathIdFromPathId(PathId, record.MutablePathId()); + record.SetOperationId(OperationId); Send( MakePipePerNodeCacheID(false), @@ -43,9 +56,19 @@ void TAnalyzeActor::SendAnalyzeStatus() { } void TAnalyzeActor::Handle(NStat::TEvStatistics::TEvAnalyzeResponse::TPtr& ev, const TActorContext& ctx) { - Y_UNUSED(ev); Y_UNUSED(ctx); + const auto& record = ev->Get()->Record; + const TString operationId = record.GetOperationId(); + + if (operationId != OperationId) { + ALOG_CRIT(NKikimrServices::KQP_GATEWAY, + "TAnalyzeActor, TEvAnalyzeResponse has operationId=" << operationId + << " , but expected " << OperationId); + } + + + // TODO Don't send EvAnalyzeStatus, EvAnalyzeResponse is already here SendAnalyzeStatus(); } @@ -172,6 +195,7 @@ void TAnalyzeActor::SendStatisticsAggregatorAnalyze(const NSchemeCache::TSchemeC auto analyzeRequest = std::make_unique(); auto& record = analyzeRequest->Record; + record.SetOperationId(OperationId); auto table = record.AddTables(); PathIdFromPathId(PathId, table->MutablePathId()); @@ -199,6 +223,7 @@ void TAnalyzeActor::SendStatisticsAggregatorAnalyze(const NSchemeCache::TSchemeC *table->MutableColumnTags()->Add() = tagByColumnName[columnName]; } + // TODO This request should be retried if StatisticsAggregator fails Send( MakePipePerNodeCacheID(false), new TEvPipeCache::TEvForward(analyzeRequest.release(), entry.DomainInfo->Params.GetStatisticsAggregator(), true), diff --git a/ydb/core/kqp/gateway/actors/analyze_actor.h b/ydb/core/kqp/gateway/actors/analyze_actor.h index af0b83cb1ce..8ea018c63d9 100644 --- a/ydb/core/kqp/gateway/actors/analyze_actor.h +++ b/ydb/core/kqp/gateway/actors/analyze_actor.h @@ -20,11 +20,7 @@ struct TEvAnalyzePrivate { class TAnalyzeActor : public NActors::TActorBootstrapped { public: - TAnalyzeActor(TString tablePath, TVector columns, NThreading::TPromise promise) - : TablePath(tablePath) - , Columns(columns) - , Promise(promise) - {} + TAnalyzeActor(TString tablePath, TVector columns, NThreading::TPromise promise); void Bootstrap(); @@ -61,6 +57,7 @@ private: // For Statistics Aggregator std::optional StatisticsAggregatorId; TPathId PathId; + TString OperationId; }; } // end of NKikimr::NKqp diff --git a/ydb/core/protos/out/out.cpp b/ydb/core/protos/out/out.cpp index eee0c8d61b2..735ac9b34a4 100644 --- a/ydb/core/protos/out/out.cpp +++ b/ydb/core/protos/out/out.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include @@ -238,3 +239,7 @@ Y_DECLARE_OUT_SPEC(, NKikimrDataEvents::TEvWrite::TOperation::EOperationType, st Y_DECLARE_OUT_SPEC(, NKikimrDataEvents::TEvWrite::ETxMode, stream, value) { stream << NKikimrDataEvents::TEvWrite::ETxMode_Name(value); } + +Y_DECLARE_OUT_SPEC(, NKikimrStat::TEvAnalyzeStatusResponse_EStatus, stream, value) { + stream << NKikimrStat::TEvAnalyzeStatusResponse_EStatus_Name(value); +} diff --git a/ydb/core/protos/statistics.proto b/ydb/core/protos/statistics.proto index 4e28439d41d..439feacc2c4 100644 --- a/ydb/core/protos/statistics.proto +++ b/ydb/core/protos/statistics.proto @@ -81,24 +81,24 @@ message TTable { // KQP -> SA message TEvAnalyze { - optional uint64 Cookie = 1; // request cookie to match response item + optional bytes OperationId = 1; // unique identifier to match response item repeated TTable Tables = 2; // list of analyzed tables and columns repeated EColumnStatisticType Types = 3; // list of statistics types requested. Empty means asking for all available. } // SA -> KQP message TEvAnalyzeResponse { - optional uint64 Cookie = 1; + optional bytes OperationId = 1; } // KQP -> SA message TEvAnalyzeStatus { - optional NKikimrProto.TPathID PathId = 1; + optional bytes OperationId = 1; // unique identifier to match response item } // SA -> KQP message TEvAnalyzeStatusResponse { - optional NKikimrProto.TPathID PathId = 1; + optional bytes OperationId = 1; enum EStatus { STATUS_UNSPECIFIED = 0; diff --git a/ydb/core/statistics/aggregator/aggregator_impl.cpp b/ydb/core/statistics/aggregator/aggregator_impl.cpp index 6a92bf8c253..12ba2dc9c1c 100644 --- a/ydb/core/statistics/aggregator/aggregator_impl.cpp +++ b/ydb/core/statistics/aggregator/aggregator_impl.cpp @@ -433,17 +433,18 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvStatTableCreationResponse:: } void TStatisticsAggregator::Handle(TEvStatistics::TEvAnalyzeStatus::TPtr& ev) { - auto& inRecord = ev->Get()->Record; - auto pathId = PathIdFromPathId(inRecord.GetPathId()); + const auto& inRecord = ev->Get()->Record; + const TString operationId = inRecord.GetOperationId(); auto response = std::make_unique(); auto& outRecord = response->Record; + outRecord.SetOperationId(operationId); - if (TraversalTableId.PathId == pathId) { + if (ForceTraversalOperationId == operationId) { outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_IN_PROGRESS); } else { if (std::any_of(ForceTraversals.begin(), ForceTraversals.end(), - [&pathId](const TForceTraversal& elem) { return elem.PathId == pathId;})) { + [&operationId](const TForceTraversal& elem) { return elem.OperationId == operationId;})) { outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_ENQUEUED); } else { outRecord.SetStatus(NKikimrStat::TEvAnalyzeStatusResponse::STATUS_NO_OPERATION); @@ -586,7 +587,6 @@ void TStatisticsAggregator::ScheduleNextTraversal(NIceDb::TNiceDb& db) { pathId = operation.PathId; ForceTraversalOperationId = operation.OperationId; - ForceTraversalCookie = operation.Cookie; ForceTraversalColumnTags = operation.ColumnTags; ForceTraversalTypes = operation.Types; ForceTraversalReplyToActorId = operation.ReplyToActorId; @@ -678,22 +678,17 @@ void TStatisticsAggregator::PersistStartKey(NIceDb::TNiceDb& db) { void TStatisticsAggregator::PersistForceTraversal(NIceDb::TNiceDb& db) { PersistSysParam(db, Schema::SysParam_ForceTraversalOperationId, ToString(ForceTraversalOperationId)); - PersistSysParam(db, Schema::SysParam_ForceTraversalCookie, ToString(ForceTraversalCookie)); + PersistSysParam(db, Schema::SysParam_ForceTraversalCookie, ForceTraversalOperationId); PersistSysParam(db, Schema::SysParam_ForceTraversalColumnTags, ToString(ForceTraversalColumnTags)); PersistSysParam(db, Schema::SysParam_ForceTraversalTypes, ToString(ForceTraversalTypes)); } -void TStatisticsAggregator::PersistNextForceTraversalOperationId(NIceDb::TNiceDb& db) { - PersistSysParam(db, Schema::SysParam_NextForceTraversalOperationId, ToString(NextForceTraversalOperationId)); -} - void TStatisticsAggregator::PersistGlobalTraversalRound(NIceDb::TNiceDb& db) { PersistSysParam(db, Schema::SysParam_GlobalTraversalRound, ToString(GlobalTraversalRound)); } void TStatisticsAggregator::ResetTraversalState(NIceDb::TNiceDb& db) { - ForceTraversalOperationId = 0; - ForceTraversalCookie = 0; + ForceTraversalOperationId.clear(); TraversalTableId.PathId = TPathId(); ForceTraversalColumnTags.clear(); ForceTraversalTypes.clear(); diff --git a/ydb/core/statistics/aggregator/aggregator_impl.h b/ydb/core/statistics/aggregator/aggregator_impl.h index ca8872dcb4e..60dc0998a85 100644 --- a/ydb/core/statistics/aggregator/aggregator_impl.h +++ b/ydb/core/statistics/aggregator/aggregator_impl.h @@ -147,7 +147,6 @@ private: void PersistTraversal(NIceDb::TNiceDb& db); void PersistForceTraversal(NIceDb::TNiceDb& db); void PersistStartKey(NIceDb::TNiceDb& db); - void PersistNextForceTraversalOperationId(NIceDb::TNiceDb& db); void PersistGlobalTraversalRound(NIceDb::TNiceDb& db); void ResetTraversalState(NIceDb::TNiceDb& db); @@ -306,15 +305,13 @@ private: private: // stored in local db - ui64 ForceTraversalOperationId = 0; - ui64 ForceTraversalCookie = 0; + TString ForceTraversalOperationId; TString ForceTraversalColumnTags; TString ForceTraversalTypes; TTableId TraversalTableId; bool TraversalIsColumnTable = false; TSerializedCellVec TraversalStartKey; TInstant TraversalStartTime; - ui64 NextForceTraversalOperationId = 0; size_t GlobalTraversalRound = 1; @@ -327,8 +324,7 @@ private: // stored in local db TTraversalsByTime ScheduleTraversalsByTime; struct TForceTraversal { - ui64 OperationId = 0; - ui64 Cookie = 0; + TString OperationId; TPathId PathId; TString ColumnTags; TString Types; diff --git a/ydb/core/statistics/aggregator/schema.h b/ydb/core/statistics/aggregator/schema.h index dc843a425cc..036902199f1 100644 --- a/ydb/core/statistics/aggregator/schema.h +++ b/ydb/core/statistics/aggregator/schema.h @@ -50,7 +50,7 @@ struct TAggregatorSchema : NIceDb::Schema { struct OperationId : Column<1, NScheme::NTypeIds::Uint64> {}; struct OwnerId : Column<2, NScheme::NTypeIds::Uint64> {}; struct LocalPathId : Column<3, NScheme::NTypeIds::Uint64> {}; - struct Cookie : Column<4, NScheme::NTypeIds::Uint64> {}; + struct Cookie : Column<4, NScheme::NTypeIds::String> {}; struct ColumnTags : Column<5, NScheme::NTypeIds::String> {}; struct Types : Column<6, NScheme::NTypeIds::String> {}; @@ -87,7 +87,7 @@ struct TAggregatorSchema : NIceDb::Schema { static constexpr ui64 SysParam_ForceTraversalColumnTags = 7; static constexpr ui64 SysParam_ForceTraversalTypes = 8; static constexpr ui64 SysParam_TraversalStartTime = 9; - static constexpr ui64 SysParam_NextForceTraversalOperationId = 10; + // deprecated 10 static constexpr ui64 SysParam_TraversalIsColumnTable = 11; static constexpr ui64 SysParam_GlobalTraversalRound = 12; }; diff --git a/ydb/core/statistics/aggregator/tx_analyze_table.cpp b/ydb/core/statistics/aggregator/tx_analyze_table.cpp index aecbdb8a53c..6dde96c8e54 100644 --- a/ydb/core/statistics/aggregator/tx_analyze_table.cpp +++ b/ydb/core/statistics/aggregator/tx_analyze_table.cpp @@ -19,7 +19,7 @@ struct TStatisticsAggregator::TTxAnalyzeTable : public TTxBase { TTxType GetTxType() const override { return TXTYPE_ANALYZE_TABLE; } bool Execute(TTransactionContext& txc, const TActorContext&) override { - SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTable::Execute"); + SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTable::Execute. ReplyToActorId " << ReplyToActorId << " , Record " << Record); if (!Self->EnableColumnStatistics) { return true; @@ -27,27 +27,31 @@ struct TStatisticsAggregator::TTxAnalyzeTable : public TTxBase { NIceDb::TNiceDb db(txc.DB); - const ui64 cookie = Record.GetCookie(); + const TString operationId = Record.GetOperationId(); const TString types = JoinVectorIntoString(TVector(Record.GetTypes().begin(), Record.GetTypes().end()), ","); for (const auto& table : Record.GetTables()) { const TPathId pathId = PathIdFromPathId(table.GetPathId()); const TString columnTags = JoinVectorIntoString(TVector{table.GetColumnTags().begin(),table.GetColumnTags().end()},","); - // drop request with the same cookie and path from this sender - if (std::any_of(Self->ForceTraversals.begin(), Self->ForceTraversals.end(), - [this, &pathId, &cookie](const TForceTraversal& elem) { + // check existing force traversal with the same cookie and path + auto forceTraversal = std::find_if(Self->ForceTraversals.begin(), Self->ForceTraversals.end(), + [&pathId, &operationId](const TForceTraversal& elem) { return elem.PathId == pathId - && elem.Cookie == cookie - && elem.ReplyToActorId == ReplyToActorId - ;})) { + && elem.OperationId == operationId;}); + + // update existing force traversal + if (forceTraversal != Self->ForceTraversals.end()) { + SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTable::Execute. Update existing force traversal. PathId " << pathId << " , ReplyToActorId " << ReplyToActorId); + forceTraversal->ReplyToActorId = ReplyToActorId; return true; } + SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTable::Execute. Create new force traversal operation for pathId " << pathId); + // create new force trasersal TForceTraversal operation { - .OperationId = Self->NextForceTraversalOperationId, - .Cookie = cookie, + .OperationId = operationId, .PathId = pathId, .ColumnTags = columnTags, .Types = types, @@ -66,8 +70,6 @@ struct TStatisticsAggregator::TTxAnalyzeTable : public TTxBase { */ } - Self->PersistNextForceTraversalOperationId(db); - return true; } @@ -77,8 +79,6 @@ struct TStatisticsAggregator::TTxAnalyzeTable : public TTxBase { }; void TStatisticsAggregator::Handle(TEvStatistics::TEvAnalyze::TPtr& ev) { - ++NextForceTraversalOperationId; - Execute(new TTxAnalyzeTable(this, ev->Get()->Record, ev->Sender), TActivationContext::AsActorContext()); } diff --git a/ydb/core/statistics/aggregator/tx_finish_trasersal.cpp b/ydb/core/statistics/aggregator/tx_finish_trasersal.cpp index 6d9d165def1..b9f7fc597fa 100644 --- a/ydb/core/statistics/aggregator/tx_finish_trasersal.cpp +++ b/ydb/core/statistics/aggregator/tx_finish_trasersal.cpp @@ -5,15 +5,13 @@ namespace NKikimr::NStat { struct TStatisticsAggregator::TTxFinishTraversal : public TTxBase { - ui64 OperationId; - ui64 Cookie; + TString OperationId; TPathId PathId; TActorId ReplyToActorId; TTxFinishTraversal(TSelf* self) : TTxBase(self) , OperationId(self->ForceTraversalOperationId) - , Cookie(self->ForceTraversalCookie) , PathId(self->TraversalTableId.PathId) , ReplyToActorId(self->ForceTraversalReplyToActorId) {} @@ -43,12 +41,12 @@ struct TStatisticsAggregator::TTxFinishTraversal : public TTxBase { if (operationsRemain) { SA_LOG_D("[" << Self->TabletID() << "] TTxFinishTraversal::Complete. Don't send TEvAnalyzeResponse. " << - "There are pending operations, Cookie " << Cookie << " , ActorId=" << ReplyToActorId); + "There are pending operations, OperationId " << OperationId << " , ActorId=" << ReplyToActorId); } else { SA_LOG_D("[" << Self->TabletID() << "] TTxFinishTraversal::Complete. " << - "Send TEvAnalyzeResponse, Cookie=" << Cookie << ", ActorId=" << ReplyToActorId); + "Send TEvAnalyzeResponse, OperationId=" << OperationId << ", ActorId=" << ReplyToActorId); auto response = std::make_unique(); - response->Record.SetCookie(Cookie); + response->Record.SetOperationId(OperationId); ctx.Send(ReplyToActorId, response.release()); } } diff --git a/ydb/core/statistics/aggregator/tx_init.cpp b/ydb/core/statistics/aggregator/tx_init.cpp index f50ea80d270..e4012d7bbe3 100644 --- a/ydb/core/statistics/aggregator/tx_init.cpp +++ b/ydb/core/statistics/aggregator/tx_init.cpp @@ -55,15 +55,10 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal start key"); break; case Schema::SysParam_ForceTraversalOperationId: { - Self->ForceTraversalOperationId = FromString(value); + Self->ForceTraversalOperationId = value; SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal operation id: " << value); break; } - case Schema::SysParam_ForceTraversalCookie: { - Self->ForceTraversalCookie = FromString(value); - SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal cookie: " << value); - break; - } case Schema::SysParam_TraversalTableOwnerId: Self->TraversalTableId.PathId.OwnerId = FromString(value); SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal table owner id: " @@ -90,11 +85,6 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal start time: " << us); break; } - case Schema::SysParam_NextForceTraversalOperationId: { - Self->NextForceTraversalOperationId = FromString(value); - SA_LOG_D("[" << Self->TabletID() << "] Loaded next traversal operation id: " << value); - break; - } case Schema::SysParam_TraversalIsColumnTable: { Self->TraversalIsColumnTable = FromString(value); SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal IsColumnTable: " << value); @@ -217,7 +207,7 @@ struct TStatisticsAggregator::TTxInit : public TTxBase { ui64 operationId = rowset.GetValue(); ui64 ownerId = rowset.GetValue(); ui64 localPathId = rowset.GetValue(); - ui64 cookie = rowset.GetValue(); + TString cookie = rowset.GetValue(); TString columnTags = rowset.GetValue(); TString types = rowset.GetValue(); diff --git a/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp b/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp index 85107784bc8..5b90aa0362f 100644 --- a/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp +++ b/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp @@ -46,9 +46,9 @@ Y_UNIT_TEST_SUITE(AnalyzeColumnshard) { auto& runtime = *env.GetServer().GetRuntime(); auto tableInfo = CreateDatabaseTables(env, 1, 1)[0]; - AnalyzeTable(runtime, tableInfo.PathId, tableInfo.ShardIds[0]); + AnalyzeTable(runtime, tableInfo.ShardIds[0], tableInfo.PathId); - Analyze(runtime, {tableInfo.PathId}, tableInfo.SaTabletId); + Analyze(runtime, tableInfo.SaTabletId, {tableInfo.PathId}); } Y_UNIT_TEST(AnalyzeAnalyzeOneColumnTableSpecificColumns) { @@ -56,7 +56,7 @@ Y_UNIT_TEST_SUITE(AnalyzeColumnshard) { auto& runtime = *env.GetServer().GetRuntime(); auto tableInfo = CreateDatabaseTables(env, 1, 1)[0]; - Analyze(runtime, {{tableInfo.PathId, {1, 2}}}, tableInfo.SaTabletId); + Analyze(runtime, tableInfo.SaTabletId, {{tableInfo.PathId, {1, 2}}}); } Y_UNIT_TEST(AnalyzeTwoColumnTables) { @@ -64,8 +64,59 @@ Y_UNIT_TEST_SUITE(AnalyzeColumnshard) { auto& runtime = *env.GetServer().GetRuntime(); auto tableInfos = CreateDatabaseTables(env, 2, 1); - Analyze(runtime, {tableInfos[0].PathId, tableInfos[1].PathId}, tableInfos[0].SaTabletId); + Analyze(runtime, tableInfos[0].SaTabletId, {tableInfos[0].PathId, tableInfos[1].PathId}); } + + Y_UNIT_TEST(AnalyzeStatus) { + TTestEnv env(1, 1); + auto& runtime = *env.GetServer().GetRuntime(); + auto sender = runtime.AllocateEdgeActor(); + + auto schemeShardStatsBlocker = runtime.AddObserver([&](auto& ev) { + ev.Reset(); + }); + + auto tableInfo = CreateDatabaseTables(env, 1, 1)[0]; + + const TString operationId = "operationId"; + + AnalyzeStatus(runtime, tableInfo.SaTabletId, operationId, NKikimrStat::TEvAnalyzeStatusResponse::STATUS_NO_OPERATION); + + auto analyzeRequest = MakeAnalyzeRequest({{tableInfo.PathId, {1, 2}}}, operationId); + runtime.SendToPipe(tableInfo.SaTabletId, sender, analyzeRequest.release()); + + AnalyzeStatus(runtime, tableInfo.SaTabletId, operationId, NKikimrStat::TEvAnalyzeStatusResponse::STATUS_ENQUEUED); + + schemeShardStatsBlocker.Remove(); + + auto analyzeResonse = runtime.GrabEdgeEventRethrow(sender); + UNIT_ASSERT_VALUES_EQUAL(analyzeResonse->Get()->Record.GetOperationId(), operationId); + + AnalyzeStatus(runtime, tableInfo.SaTabletId, operationId, NKikimrStat::TEvAnalyzeStatusResponse::STATUS_NO_OPERATION); + } + + Y_UNIT_TEST(AnalyzeSameOperationId) { + TTestEnv env(1, 1); + auto& runtime = *env.GetServer().GetRuntime(); + auto tableInfo = CreateDatabaseTables(env, 1, 1)[0]; + + auto sender = runtime.AllocateEdgeActor(); + const TString operationId = "operationId"; + + auto analyzeRequest1 = MakeAnalyzeRequest({tableInfo.PathId}, operationId); + runtime.SendToPipe(tableInfo.SaTabletId, sender, analyzeRequest1.release()); + + auto analyzeRequest2 = MakeAnalyzeRequest({tableInfo.PathId}, operationId); + runtime.SendToPipe(tableInfo.SaTabletId, sender, analyzeRequest2.release()); + + auto response1 = runtime.GrabEdgeEventRethrow(sender); + UNIT_ASSERT(response1); + UNIT_ASSERT_VALUES_EQUAL(response1->Get()->Record.GetOperationId(), operationId); + + auto response2 = runtime.GrabEdgeEventRethrow(sender, TDuration::Seconds(5)); + UNIT_ASSERT(!response2); + } + } } // NStat diff --git a/ydb/core/statistics/aggregator/ut/ut_analyze_datashard.cpp b/ydb/core/statistics/aggregator/ut/ut_analyze_datashard.cpp index 3f77cb85e61..d491aad9742 100644 --- a/ydb/core/statistics/aggregator/ut/ut_analyze_datashard.cpp +++ b/ydb/core/statistics/aggregator/ut/ut_analyze_datashard.cpp @@ -36,7 +36,7 @@ Y_UNIT_TEST_SUITE(AnalyzeDatashard) { runtime.SimulateSleep(TDuration::Seconds(30)); - Analyze(runtime, {{pathId}}, saTabletId); + Analyze(runtime, saTabletId, {{pathId}}); ValidateCountMin(runtime, pathId); } @@ -62,7 +62,7 @@ Y_UNIT_TEST_SUITE(AnalyzeDatashard) { auto pathId1 = ResolvePathId(runtime, "/Root/Database/Table1", nullptr, &saTabletId1); auto pathId2 = ResolvePathId(runtime, "/Root/Database/Table2"); - Analyze(runtime, {pathId1, pathId2}, saTabletId1); + Analyze(runtime, saTabletId1, {pathId1, pathId2}); ValidateCountMin(runtime, pathId1); ValidateCountMin(runtime, pathId2); @@ -92,7 +92,7 @@ Y_UNIT_TEST_SUITE(AnalyzeDatashard) { runtime.SimulateSleep(TDuration::Seconds(5)); init2Thread.join(); - Analyze(runtime, {pathId}, saTabletId); + Analyze(runtime, saTabletId, {pathId}); runtime.SimulateSleep(TDuration::Seconds(60)); diff --git a/ydb/core/statistics/service/http_request.cpp b/ydb/core/statistics/service/http_request.cpp index 6e7b28081f7..1c00bf0e39b 100644 --- a/ydb/core/statistics/service/http_request.cpp +++ b/ydb/core/statistics/service/http_request.cpp @@ -2,6 +2,7 @@ #include +#include #include #include #include @@ -9,10 +10,16 @@ namespace NKikimr { namespace NStat { +TString MakeOperationId() { + TULIDGenerator ulidGen; + return ulidGen.Next(TActivationContext::Now()).ToBinary(); +} + THttpRequest::THttpRequest(EType type, const TString& path, TActorId replyToActorId) : Type(type) , Path(path) , ReplyToActorId(replyToActorId) + , OperationId(MakeOperationId() ) {} void THttpRequest::Bootstrap() { @@ -102,6 +109,14 @@ void THttpRequest::Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& void THttpRequest::Handle(TEvStatistics::TEvAnalyzeStatusResponse::TPtr& ev) { auto& record = ev->Get()->Record; + + if (record.GetOperationId() != OperationId) { + ALOG_ERROR(NKikimrServices::STATISTICS, + "THttpRequest, TEvAnalyzeStatusResponse has operationId=" << record.GetOperationId() + << " , but expected " << OperationId); + HttpReply("Wrong OperationId"); + } + switch (record.GetStatus()) { case NKikimrStat::TEvAnalyzeStatusResponse::STATUS_UNSPECIFIED: HttpReply("Status is unspecified"); @@ -131,6 +146,7 @@ void THttpRequest::ResolveSuccess() { if (Type == ANALYZE) { auto analyze = std::make_unique(); auto& record = analyze->Record; + record.SetOperationId(OperationId); PathIdFromPathId(PathId, record.AddTables()->MutablePathId()); Send(MakePipePerNodeCacheID(false), @@ -140,7 +156,7 @@ void THttpRequest::ResolveSuccess() { } else { auto getStatus = std::make_unique(); auto& record = getStatus->Record; - PathIdFromPathId(PathId, record.MutablePathId()); + record.SetOperationId(OperationId); Send(MakePipePerNodeCacheID(false), new TEvPipeCache::TEvForward(getStatus.release(), StatisticsAggregatorId, true)); diff --git a/ydb/core/statistics/service/http_request.h b/ydb/core/statistics/service/http_request.h index 87472544089..24f6bddfb44 100644 --- a/ydb/core/statistics/service/http_request.h +++ b/ydb/core/statistics/service/http_request.h @@ -55,6 +55,7 @@ private: const TActorId ReplyToActorId; TPathId PathId; + TString OperationId; ui64 StatisticsAggregatorId = 0; static const ui64 FirstRoundCookie = 1; diff --git a/ydb/core/statistics/ut_common/ut_common.cpp b/ydb/core/statistics/ut_common/ut_common.cpp index 9a454c5d590..bdd07b99d9f 100644 --- a/ydb/core/statistics/ut_common/ut_common.cpp +++ b/ydb/core/statistics/ut_common/ut_common.cpp @@ -1,6 +1,5 @@ #include "ut_common.h" -#include #include #include @@ -340,24 +339,27 @@ void TAnalyzedTable::ToProto(NKikimrStat::TTable& tableProto) const { tableProto.MutableColumnTags()->Add(ColumnTags.begin(), ColumnTags.end()); } - -void Analyze(TTestActorRuntime& runtime, const std::vector& tables, ui64 saTabletId) { - const ui64 cookie = 555; +std::unique_ptr MakeAnalyzeRequest(const std::vector& tables, const TString operationId) { auto ev = std::make_unique(); NKikimrStat::TEvAnalyze& record = ev->Record; - record.SetCookie(cookie); + record.SetOperationId(operationId); record.AddTypes(NKikimrStat::EColumnStatisticType::TYPE_COUNT_MIN_SKETCH); for (const TAnalyzedTable& table : tables) table.ToProto(*record.AddTables()); + return ev; +} + +void Analyze(TTestActorRuntime& runtime, ui64 saTabletId, const std::vector& tables, const TString operationId) { + auto ev = MakeAnalyzeRequest(tables, operationId); auto sender = runtime.AllocateEdgeActor(); runtime.SendToPipe(saTabletId, sender, ev.release()); auto evResponse = runtime.GrabEdgeEventRethrow(sender); - UNIT_ASSERT_VALUES_EQUAL(evResponse->Get()->Record.GetCookie(), cookie); + UNIT_ASSERT_VALUES_EQUAL(evResponse->Get()->Record.GetOperationId(), operationId); } -void AnalyzeTable(TTestActorRuntime& runtime, const TAnalyzedTable& table, ui64 shardTabletId) { +void AnalyzeTable(TTestActorRuntime& runtime, ui64 shardTabletId, const TAnalyzedTable& table) { auto ev = std::make_unique(); auto& record = ev->Record; table.ToProto(*record.MutableTable()); @@ -368,6 +370,18 @@ void AnalyzeTable(TTestActorRuntime& runtime, const TAnalyzedTable& table, ui64 runtime.GrabEdgeEventRethrow(sender); } +void AnalyzeStatus(TTestActorRuntime& runtime, ui64 saTabletId, const TString operationId, const NKikimrStat::TEvAnalyzeStatusResponse::EStatus expectedStatus) { + auto analyzeStatusRequest = std::make_unique(); + analyzeStatusRequest->Record.SetOperationId(operationId); + auto sender = runtime.AllocateEdgeActor(); + runtime.SendToPipe(saTabletId, sender, analyzeStatusRequest.release()); + + auto analyzeStatusResponse = runtime.GrabEdgeEventRethrow(sender); + UNIT_ASSERT(analyzeStatusResponse); + UNIT_ASSERT_VALUES_EQUAL(analyzeStatusResponse->Get()->Record.GetOperationId(), operationId); + UNIT_ASSERT_VALUES_EQUAL(analyzeStatusResponse->Get()->Record.GetStatus(), expectedStatus); +} + } // NStat } // NKikimr diff --git a/ydb/core/statistics/ut_common/ut_common.h b/ydb/core/statistics/ut_common/ut_common.h index 509ddb84956..2f4c795359a 100644 --- a/ydb/core/statistics/ut_common/ut_common.h +++ b/ydb/core/statistics/ut_common/ut_common.h @@ -1,5 +1,7 @@ #pragma once +#include + #include #include @@ -87,8 +89,12 @@ struct TAnalyzedTable { void ToProto(NKikimrStat::TTable& tableProto) const; }; -void Analyze(TTestActorRuntime& runtime, const std::vector& table, ui64 saTabletId); -void AnalyzeTable(TTestActorRuntime& runtime, const TAnalyzedTable& table, ui64 shardTabletId); +std::unique_ptr MakeAnalyzeRequest(const std::vector& tables, const TString operationId = "operationId"); + +void Analyze(TTestActorRuntime& runtime, ui64 saTabletId, const std::vector& table, const TString operationId = "operationId"); +void AnalyzeTable(TTestActorRuntime& runtime, ui64 shardTabletId, const TAnalyzedTable& table); +void AnalyzeStatus(TTestActorRuntime& runtime, ui64 saTabletId, const TString operationId, const NKikimrStat::TEvAnalyzeStatusResponse::EStatus expectedStatus); + } // namespace NStat } // namespace NKikimr -- cgit v1.3