Statistics: OperationId is generated by KQP (#7694)

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 <ydb/core/base/path.h>
 #include <ydb/core/base/tablet_pipecache.h>
+#include <ydb/core/util/ulid.h>
 #include <ydb/library/actors/core/log.h>
 #include <ydb/library/services/services.pb.h>
 
@@ -15,6 +16,18 @@ enum {
 
 using TNavigate = NSchemeCache::TSchemeCacheNavigate;
 
+TString MakeOperationId() {
+    TULIDGenerator ulidGen;
+    return ulidGen.Next(TActivationContext::Now()).ToBinary();
+}
+
+TAnalyzeActor::TAnalyzeActor(TString tablePath, TVector<TString> columns, NThreading::TPromise<NYql::IKikimrGateway::TGenericResult> promise)
+    : TablePath(tablePath)
+    , Columns(columns) 
+    , Promise(promise)
+    , OperationId(MakeOperationId())
+{}
+
 void TAnalyzeActor::Bootstrap() {
     using TNavigate = NSchemeCache::TSchemeCacheNavigate;
     auto navigate = std::make_unique<TNavigate>();
@@ -34,7 +47,7 @@ void TAnalyzeActor::SendAnalyzeStatus() {
 
     auto getStatus = std::make_unique<NStat::TEvStatistics::TEvAnalyzeStatus>();
     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<NStat::TEvStatistics::TEvAnalyze>();
     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<TAnalyzeActor> { 
 public:
-    TAnalyzeActor(TString tablePath, TVector<TString> columns, NThreading::TPromise<NYql::IKikimrGateway::TGenericResult> promise)
-        : TablePath(tablePath)
-        , Columns(columns) 
-        , Promise(promise)
-    {}
+    TAnalyzeActor(TString tablePath, TVector<TString> columns, NThreading::TPromise<NYql::IKikimrGateway::TGenericResult> promise);
 
     void Bootstrap();
 
@@ -61,6 +57,7 @@ private:
     // For Statistics Aggregator
     std::optional<ui64> 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 <ydb/core/protos/flat_scheme_op.pb.h>
 #include <ydb/core/protos/subdomains.pb.h>
 #include <ydb/core/protos/data_events.pb.h>
+#include <ydb/core/protos/statistics.pb.h>
 
 #include <util/stream/output.h>
 
@@ -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<TEvStatistics::TEvAnalyzeStatusResponse>();
     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<ui32>(Record.GetTypes().begin(), Record.GetTypes().end()), ",");
         
         for (const auto& table : Record.GetTables()) {
             const TPathId pathId = PathIdFromPathId(table.GetPathId());
             const TString columnTags = JoinVectorIntoString(TVector<ui32>{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<TEvStatistics::TEvAnalyzeResponse>();
-            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<ui64>(value);
+                        Self->ForceTraversalOperationId = value;
                         SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal operation id: " << value);
                         break;
                     }  
-                    case Schema::SysParam_ForceTraversalCookie: {
-                        Self->ForceTraversalCookie = FromString<ui64>(value);
-                        SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal cookie: " << value);
-                        break;
-                    }
                     case Schema::SysParam_TraversalTableOwnerId:
                         Self->TraversalTableId.PathId.OwnerId = FromString<ui64>(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<ui64>(value);
-                        SA_LOG_D("[" << Self->TabletID() << "] Loaded next traversal operation id: " << value);
-                        break;
-                    }
                     case Schema::SysParam_TraversalIsColumnTable: {
                         Self->TraversalIsColumnTable = FromString<bool>(value);
                         SA_LOG_D("[" << Self->TabletID() << "] Loaded traversal IsColumnTable: " << value);
@@ -217,7 +207,7 @@ struct TStatisticsAggregator::TTxInit : public TTxBase {
                 ui64 operationId = rowset.GetValue<Schema::ForceTraversals::OperationId>();
                 ui64 ownerId = rowset.GetValue<Schema::ForceTraversals::OwnerId>();
                 ui64 localPathId = rowset.GetValue<Schema::ForceTraversals::LocalPathId>();
-                ui64 cookie = rowset.GetValue<Schema::ForceTraversals::Cookie>();
+                TString cookie = rowset.GetValue<Schema::ForceTraversals::Cookie>();
                 TString columnTags = rowset.GetValue<Schema::ForceTraversals::ColumnTags>();
                 TString types = rowset.GetValue<Schema::ForceTraversals::Types>();
 
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<TEvStatistics::TEvSchemeShardStats>([&](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<TEvStatistics::TEvAnalyzeResponse>(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<TEvStatistics::TEvAnalyzeResponse>(sender);
+        UNIT_ASSERT(response1);
+        UNIT_ASSERT_VALUES_EQUAL(response1->Get()->Record.GetOperationId(), operationId);
+
+        auto response2 = runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeResponse>(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 <ydb/core/base/path.h>
+#include <ydb/core/util/ulid.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/actors/core/log.h>
@@ -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<TEvStatistics::TEvAnalyze>();
         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<TEvStatistics::TEvAnalyzeStatus>();
         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 <ydb/core/statistics/events.h>
 #include <ydb/core/statistics/service/service.h>
 
 #include <ydb/core/tx/scheme_cache/scheme_cache.h>
@@ -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<TAnalyzedTable>& tables, ui64 saTabletId) {
-    const ui64 cookie = 555;
+std::unique_ptr<TEvStatistics::TEvAnalyze> MakeAnalyzeRequest(const std::vector<TAnalyzedTable>& tables, const TString operationId) {
     auto ev = std::make_unique<TEvStatistics::TEvAnalyze>();
     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<TAnalyzedTable>& tables, const TString operationId) {
+    auto ev = MakeAnalyzeRequest(tables, operationId);
 
     auto sender = runtime.AllocateEdgeActor();
     runtime.SendToPipe(saTabletId, sender, ev.release());
     auto evResponse = runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeResponse>(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<TEvStatistics::TEvAnalyzeTable>();
     auto& record = ev->Record;
     table.ToProto(*record.MutableTable());
@@ -368,6 +370,18 @@ void AnalyzeTable(TTestActorRuntime& runtime, const TAnalyzedTable& table, ui64
     runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeTableResponse>(sender);
 }
 
+void AnalyzeStatus(TTestActorRuntime& runtime, ui64 saTabletId, const TString operationId, const NKikimrStat::TEvAnalyzeStatusResponse::EStatus expectedStatus) {
+    auto analyzeStatusRequest = std::make_unique<TEvStatistics::TEvAnalyzeStatus>();
+    analyzeStatusRequest->Record.SetOperationId(operationId);
+    auto sender = runtime.AllocateEdgeActor();
+    runtime.SendToPipe(saTabletId, sender, analyzeStatusRequest.release());
+
+    auto analyzeStatusResponse = runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeStatusResponse>(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 <ydb/core/statistics/events.h>
+
 #include <ydb/core/testlib/test_client.h>
 #include <library/cpp/testing/unittest/registar.h>
 
@@ -87,8 +89,12 @@ struct TAnalyzedTable {
     void ToProto(NKikimrStat::TTable& tableProto) const;
 };
 
-void Analyze(TTestActorRuntime& runtime, const std::vector<TAnalyzedTable>& table, ui64 saTabletId);
-void AnalyzeTable(TTestActorRuntime& runtime, const TAnalyzedTable& table, ui64 shardTabletId);
+std::unique_ptr<TEvStatistics::TEvAnalyze> MakeAnalyzeRequest(const std::vector<TAnalyzedTable>& tables, const TString operationId = "operationId");
+
+void Analyze(TTestActorRuntime& runtime, ui64 saTabletId, const std::vector<TAnalyzedTable>& 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