fix analyze for serverless case (#8843)

alexd65536 · web-flow · commit ac22cb5b0ee7 · 2024-09-06T17:55:15.000+03:00
diff --git a/ydb/core/kqp/gateway/actors/analyze_actor.cpp b/ydb/core/kqp/gateway/actors/analyze_actor.cpp
@@ -105,9 +105,8 @@ void TAnalyzeActor::Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr&
                     TStringBuilder() << "Can't get statistics aggregator ID.", {}
                 )
             );
+            this->Die(ctx);
         }
-
-        this->Die(ctx);
         return;
     }
     
diff --git a/ydb/core/statistics/aggregator/tx_analyze_table_response.cpp b/ydb/core/statistics/aggregator/tx_analyze_table_response.cpp
@@ -24,30 +24,32 @@ struct TStatisticsAggregator::TTxAnalyzeTableResponse : public TTxBase {
         const TPathId pathId = PathIdFromPathId(Record.GetPathId());
         auto operationTable = Self->ForceTraversalTable(operationId, pathId);
         if (!operationTable) {
-            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Complete. Unknown OperationTable. Record: " << Record.ShortDebugString());
+            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Execute. Unknown OperationTable. Record: " << Record.ShortDebugString());
             return true;
         }
 
         auto analyzedShard = std::find_if(operationTable->AnalyzedShards.begin(), operationTable->AnalyzedShards.end(), 
             [tabletId = Record.GetShardTabletId()] (TAnalyzedShard& analyzedShard) { return analyzedShard.ShardTabletId == tabletId;});
         if (analyzedShard == operationTable->AnalyzedShards.end()) {
-            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Complete. Unknown AnalyzedShards. Record: " << Record.ShortDebugString() << ", ShardTabletId " << Record.GetShardTabletId());
+            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Execute. Unknown AnalyzedShards. Record: " << Record.ShortDebugString() << ", ShardTabletId " << Record.GetShardTabletId());
             return true;
         }
         if (analyzedShard->Status != TAnalyzedShard::EStatus::AnalyzeStarted) {
-            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Complete. Unknown AnalyzedShards Status. Record: " << Record.ShortDebugString() << ", ShardTabletId " << Record.GetShardTabletId());
+            SA_LOG_E("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Execute. Unknown AnalyzedShards Status. Record: " << Record.ShortDebugString() << ", ShardTabletId " << Record.GetShardTabletId());
         }
 
         analyzedShard->Status = TAnalyzedShard::EStatus::AnalyzeFinished;
 
         bool completeResponse = std::any_of(operationTable->AnalyzedShards.begin(), operationTable->AnalyzedShards.end(), 
             [] (const TAnalyzedShard& analyzedShard) { return analyzedShard.Status == TAnalyzedShard::EStatus::AnalyzeFinished;});
 
-        if (!completeResponse)
+        if (!completeResponse) {
+            SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Execute. There are shards which are not analyzed");
             return true;   
-        
+        }
         NIceDb::TNiceDb db(txc.DB);
         Self->UpdateForceTraversalTableStatus(TForceTraversalTable::EStatus::AnalyzeFinished, operationId, *operationTable,  db);
+        SA_LOG_D("[" << Self->TabletID() << "] TTxAnalyzeTableResponse::Execute. All shards are analyzed");
         return true;
     }
 
diff --git a/ydb/core/statistics/aggregator/tx_navigate.cpp b/ydb/core/statistics/aggregator/tx_navigate.cpp
@@ -59,10 +59,8 @@ struct TStatisticsAggregator::TTxNavigate : public TTxBase {
         }
 
         if (Self->TraversalIsColumnTable) {
-            // TODO: serverless case
-            if (entry.DomainInfo->Params.HasHive()) {
-                Self->HiveId = entry.DomainInfo->Params.GetHive();
-            } else {
+            Self->HiveId = entry.DomainInfo->ExtractHive();
+            if (Self->HiveId == 0) {
                 Self->HiveId = AppData()->DomainsInfo->GetHive();
             }
         }
diff --git a/ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp b/ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp
@@ -55,13 +55,14 @@ struct TStatisticsAggregator::TTxResponseTabletDistribution : public TTxBase {
     }
 
     bool Execute(TTransactionContext& txc, const TActorContext&) override {
-        SA_LOG_D("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Execute");
+        SA_LOG_D("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Execute. Node count = " << HiveRecord.NodesSize());
 
         auto distribution = Self->TabletsForReqDistribution;
         for (auto& inNode : HiveRecord.GetNodes()) {
             if (inNode.GetNodeId() == 0) {
                 // these tablets are probably in Hive boot queue
                 if (Self->HiveRequestRound < Self->MaxHiveRequestRoundCount) {
+                    SA_LOG_W("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Execute. Some tablets are probably in Hive boot queue");
                     Action = EAction::ScheduleReqDistribution;
                 }
                 continue;
@@ -76,6 +77,7 @@ struct TStatisticsAggregator::TTxResponseTabletDistribution : public TTxBase {
         }
 
         if (!distribution.empty() && Self->ResolveRound < Self->MaxResolveRoundCount) {
+            SA_LOG_W("[" << Self->TabletID() << "] TTxResponseTabletDistribution::Execute. Some tablets do not exist in Hive anymore; tablet count = " << distribution.size());
             // these tablets do not exist in Hive anymore
             Self->NavigatePathId = Self->TraversalPathId;
             Action = EAction::ScheduleResolve;
diff --git a/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp b/ydb/core/statistics/aggregator/ut/ut_analyze_columnshard.cpp
@@ -19,7 +19,7 @@ Y_UNIT_TEST_SUITE(AnalyzeColumnshard) {
 
         AnalyzeTable(runtime, tableInfo.ShardIds[0], tableInfo.PathId);
     }
-    
+
     Y_UNIT_TEST(Analyze) {
         TTestEnv env(1, 1);
         auto& runtime = *env.GetServer().GetRuntime();
@@ -28,6 +28,14 @@ Y_UNIT_TEST_SUITE(AnalyzeColumnshard) {
         Analyze(runtime, tableInfo.SaTabletId, {tableInfo.PathId});
     }
 
+    Y_UNIT_TEST(AnalyzeServerless) {
+        TTestEnv env(1, 1);
+        auto& runtime = *env.GetServer().GetRuntime();
+        auto tableInfo = CreateServerlessDatabaseColumnTables(env, 1, 1)[0];
+
+        Analyze(runtime, tableInfo.SaTabletId, {tableInfo.PathId});
+    }
+
     Y_UNIT_TEST(AnalyzeAnalyzeOneColumnTableSpecificColumns) {
         TTestEnv env(1, 1);
         auto& runtime = *env.GetServer().GetRuntime();
diff --git a/ydb/core/statistics/ut_common/ut_common.cpp b/ydb/core/statistics/ut_common/ut_common.cpp
@@ -155,14 +155,36 @@ TPathId ResolvePathId(TTestActorRuntime& runtime, const TString& path, TPathId*
         *domainKey = resultEntry.DomainInfo->DomainKey;
     }
 
-    if (saTabletId && resultEntry.DomainInfo->Params.HasStatisticsAggregator()) {
-        *saTabletId = resultEntry.DomainInfo->Params.GetStatisticsAggregator();
+    if (saTabletId) {
+        if (resultEntry.DomainInfo->Params.HasStatisticsAggregator()) {
+            *saTabletId = resultEntry.DomainInfo->Params.GetStatisticsAggregator();
+        } else {
+            auto resourcesDomainKey = resultEntry.DomainInfo->ResourcesDomainKey;
+            auto request = std::make_unique<TNavigate>();
+            auto& entry = request->ResultSet.emplace_back();
+            entry.TableId = TTableId(resourcesDomainKey.OwnerId, resourcesDomainKey.LocalPathId);
+            entry.RequestType = TNavigate::TEntry::ERequestType::ByTableId;
+            entry.Operation = TNavigate::EOp::OpPath;
+            entry.RedirectRequired = false;
+            runtime.Send(MakeSchemeCacheID(), sender, new TEvRequest(request.release()));
+
+            auto ev = runtime.GrabEdgeEventRethrow<TEvResponse>(sender);
+            UNIT_ASSERT(ev);
+            UNIT_ASSERT(ev->Get());
+            std::unique_ptr<TNavigate> response(ev->Get()->Request.Release());
+            UNIT_ASSERT(response->ResultSet.size() == 1);
+            auto& secondResultEntry = response->ResultSet[0];
+
+            if (secondResultEntry.DomainInfo->Params.HasStatisticsAggregator()) {
+                *saTabletId = secondResultEntry.DomainInfo->Params.GetStatisticsAggregator();
+            }
+        }
     }
 
     return resultEntry.TableId.PathId;
 }
 
-NKikimrScheme::TEvDescribeSchemeResult DescribeTable(TTestActorRuntime& runtime, TActorId sender, const TString &path)
+NKikimrScheme::TEvDescribeSchemeResult DescribeTable(TTestActorRuntime& runtime, TActorId sender, const TString& path)
 {
     TAutoPtr<IEventHandle> handle;
 
@@ -175,7 +197,7 @@ NKikimrScheme::TEvDescribeSchemeResult DescribeTable(TTestActorRuntime& runtime,
     return *reply->MutableRecord();
 }
 
-TVector<ui64> GetTableShards(TTestActorRuntime& runtime, TActorId sender, const TString &path)
+TVector<ui64> GetTableShards(TTestActorRuntime& runtime, TActorId sender, const TString& path)
 {
     TVector<ui64> shards;
     auto lsResult = DescribeTable(runtime, sender, path);
@@ -185,7 +207,7 @@ TVector<ui64> GetTableShards(TTestActorRuntime& runtime, TActorId sender, const
     return shards;
 }
 
-TVector<ui64> GetColumnTableShards(TTestActorRuntime& runtime, TActorId sender,const TString &path)
+TVector<ui64> GetColumnTableShards(TTestActorRuntime& runtime, TActorId sender, const TString& path)
 {
     TVector<ui64> shards;
     auto lsResult = DescribeTable(runtime, sender, path);
@@ -275,6 +297,21 @@ void CreateColumnStoreTable(TTestEnv& env, const TString& databaseName, const TS
     env.GetController()->WaitActualization(TDuration::Seconds(1));
 }
 
+std::vector<TTableInfo> GatherColumnTablesInfo(TTestEnv& env, ui8 tableCount) {
+    auto& runtime = *env.GetServer().GetRuntime();
+    auto sender = runtime.AllocateEdgeActor();
+
+    std::vector<TTableInfo> ret;
+    for (ui8 tableId = 1; tableId <= tableCount; tableId++) {
+        TTableInfo tableInfo;
+        const TString path = Sprintf("/Root/Database/Table%u", tableId);
+        tableInfo.ShardIds = GetColumnTableShards(runtime, sender, path);
+        tableInfo.PathId = ResolvePathId(runtime, path, &tableInfo.DomainKey, &tableInfo.SaTabletId);
+        ret.emplace_back(tableInfo);
+    }
+    return ret;
+}
+
 std::vector<TTableInfo> CreateDatabaseColumnTables(TTestEnv& env, ui8 tableCount, ui8 shardCount) {
     auto init = [&] () {
         CreateDatabase(env, "Database");
@@ -285,20 +322,38 @@ std::vector<TTableInfo> CreateDatabaseColumnTables(TTestEnv& env, ui8 tableCount
     std::thread initThread(init);
 
     auto& runtime = *env.GetServer().GetRuntime();
-    auto sender = runtime.AllocateEdgeActor();
 
     runtime.SimulateSleep(TDuration::Seconds(10));
     initThread.join();
 
-    std::vector<TTableInfo> ret;
-    for (ui8 tableId = 1; tableId <= tableCount; tableId++) {
-        TTableInfo tableInfo;
-        const TString path = Sprintf("/Root/Database/Table%u", tableId);
-        tableInfo.ShardIds = GetColumnTableShards(runtime, sender, path);
-        tableInfo.PathId = ResolvePathId(runtime, path, &tableInfo.DomainKey, &tableInfo.SaTabletId);
-        ret.emplace_back(tableInfo);
-    }
-    return ret;
+    return GatherColumnTablesInfo(env, tableCount);
+}
+
+std::vector<TTableInfo> CreateServerlessDatabaseColumnTables(TTestEnv& env, ui8 tableCount, ui8 shardCount) {
+    auto init = [&] () {
+        CreateDatabase(env, "Shared");
+    };
+    std::thread initThread(init);
+
+    auto& runtime = *env.GetServer().GetRuntime();
+    runtime.SimulateSleep(TDuration::Seconds(5));
+    initThread.join();
+
+    TPathId domainKey;
+    ResolvePathId(runtime, "/Root/Shared", &domainKey);
+
+    auto init2 = [&] () {
+        CreateServerlessDatabase(env, "Database", domainKey);
+        for (ui8 tableId = 1; tableId <= tableCount; tableId++) {
+            CreateColumnStoreTable(env, "Database", Sprintf("Table%u", tableId), shardCount);
+        }
+    };
+    std::thread init2Thread(init2);
+
+    runtime.SimulateSleep(TDuration::Seconds(5));
+    init2Thread.join();
+
+    return GatherColumnTablesInfo(env, tableCount);
 }
 
 void DropTable(TTestEnv& env, const TString& databaseName, const TString& tableName) {
diff --git a/ydb/core/statistics/ut_common/ut_common.h b/ydb/core/statistics/ut_common/ut_common.h
@@ -82,6 +82,7 @@ struct TTableInfo {
     TPathId PathId;
 };
 std::vector<TTableInfo> CreateDatabaseColumnTables(TTestEnv& env, ui8 tableCount, ui8 shardCount);
+std::vector<TTableInfo> CreateServerlessDatabaseColumnTables(TTestEnv& env, ui8 tableCount, ui8 shardCount);
 
 TPathId ResolvePathId(TTestActorRuntime& runtime, const TString& path, TPathId* domainKey = nullptr, ui64* saTabletId = nullptr);
 
diff --git a/ydb/core/tx/scheme_cache/scheme_cache.h b/ydb/core/tx/scheme_cache/scheme_cache.h
@@ -66,6 +66,10 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {
         if (descr.HasServerlessComputeResourcesMode()) {
             ServerlessComputeResourcesMode = descr.GetServerlessComputeResourcesMode();
         }
+
+        if (descr.HasSharedHive()) {
+            SharedHiveId = descr.GetSharedHive();
+        }
     }
 
     inline ui64 GetVersion() const {
@@ -80,6 +84,14 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {
         }
     }
 
+    inline ui64 ExtractHive() const {
+        if (IsServerless()) {
+            return SharedHiveId;
+        } else {
+            return Params.GetHive();
+        }
+    }
+
     inline bool IsServerless() const {
         return DomainKey != ResourcesDomainKey;
     }
@@ -89,6 +101,7 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {
     NKikimrSubDomains::TProcessingParams Params;
     TCoordinators Coordinators;
     TMaybeServerlessComputeResourcesMode ServerlessComputeResourcesMode;
+    ui64 SharedHiveId = 0;
 
     TString ToString() const;
 

Original file line number	Diff line number	Diff line change
`@@ -105,9 +105,8 @@ void TAnalyzeActor::Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr&`
`105`	`105`	`TStringBuilder() << "Can't get statistics aggregator ID.", {}`
`106`	`106`	`)`
`107`	`107`	`);`
	`108`	`+ this->Die(ctx);`
`108`	`109`	`}`
`109`		`-`
`110`		`- this->Die(ctx);`
`111`	`110`	`return;`
`112`	`111`	`}`
`113`	`112`
Original file line number	Diff line number	Diff line change
`@@ -59,10 +59,8 @@ struct TStatisticsAggregator::TTxNavigate : public TTxBase {`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`if (Self->TraversalIsColumnTable) {`
`62`		`- // TODO: serverless case`
`63`		`- if (entry.DomainInfo->Params.HasHive()) {`
`64`		`- Self->HiveId = entry.DomainInfo->Params.GetHive();`
`65`		`- } else {`
	`62`	`+ Self->HiveId = entry.DomainInfo->ExtractHive();`
	`63`	`+ if (Self->HiveId == 0) {`
`66`	`64`	`Self->HiveId = AppData()->DomainsInfo->GetHive();`
`67`	`65`	`}`
`68`	`66`	`}`
Original file line number	Diff line number	Diff line change
`@@ -66,6 +66,10 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {`
`66`	`66`	`if (descr.HasServerlessComputeResourcesMode()) {`
`67`	`67`	`ServerlessComputeResourcesMode = descr.GetServerlessComputeResourcesMode();`
`68`	`68`	`}`
	`69`	`+`
	`70`	`+ if (descr.HasSharedHive()) {`
	`71`	`+ SharedHiveId = descr.GetSharedHive();`
	`72`	`+ }`
`69`	`73`	`}`
`70`	`74`
`71`	`75`	`inline ui64 GetVersion() const {`
`@@ -80,6 +84,14 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {`
`80`	`84`	`}`
`81`	`85`	`}`
`82`	`86`
	`87`	`+ inline ui64 ExtractHive() const {`
	`88`	`+ if (IsServerless()) {`
	`89`	`+ return SharedHiveId;`
	`90`	`+ } else {`
	`91`	`+ return Params.GetHive();`
	`92`	`+ }`
	`93`	`+ }`
	`94`	`+`
`83`	`95`	`inline bool IsServerless() const {`
`84`	`96`	`return DomainKey != ResourcesDomainKey;`
`85`	`97`	`}`
`@@ -89,6 +101,7 @@ struct TDomainInfo : public TAtomicRefCount<TDomainInfo> {`
`89`	`101`	`NKikimrSubDomains::TProcessingParams Params;`
`90`	`102`	`TCoordinators Coordinators;`
`91`	`103`	`TMaybeServerlessComputeResourcesMode ServerlessComputeResourcesMode;`
	`104`	`+ ui64 SharedHiveId = 0;`
`92`	`105`
`93`	`106`	`TString ToString() const;`
`94`	`107`