Two-stage ANALYZE with adaptive count-min sketch params (#30206)

Author: ztlpn

ydb/core/protos/statistics.proto

@@ -198,3 +198,8 @@ message TEvAggregateStatisticsResponse {
     }
     repeated TFailedTablet FailedTablets = 3;
 }
+
+message TSimpleColumnStatistics {
+    optional uint64 Count = 1;
+    optional uint64 CountDistinct = 2;
+};

ydb/core/statistics/aggregator/aggregator_impl.cpp

@@ -608,25 +608,23 @@ void TStatisticsAggregator::SaveStatisticsToTable() {
 
     PendingSaveStatistics = false;
 
-    std::vector<ui32> columnTags;
-    std::vector<TString> data;
-    auto count = CountMinSketches.size();
-    if (count == 0) {
-        Send(SelfId(), new TEvStatistics::TEvSaveStatisticsQueryResponse(
-            Ydb::StatusIds::SUCCESS, {}, TraversalPathId));
-        return;
-    }
-    columnTags.reserve(count);
-    data.reserve(count);
+    std::vector<TStatisticsItem> items = std::exchange(StatisticsToSave, {});
 
     for (auto& [tag, sketch] : CountMinSketches) {
-        columnTags.push_back(tag);
+        if (!ColumnNames.contains(tag)) {
+            continue;
+        }
         TString strSketch(sketch->AsStringBuf());
-        data.push_back(strSketch);
+        items.emplace_back(tag, EStatType::COUNT_MIN_SKETCH, std::move(strSketch));
+    }
+
+    if (items.empty()) {
+        Send(SelfId(), new TEvStatistics::TEvSaveStatisticsQueryResponse(
+            Ydb::StatusIds::SUCCESS, {}, TraversalPathId));
+        return;
     }
 
-    Register(CreateSaveStatisticsQuery(SelfId(), Database,
-        TraversalPathId, EStatType::COUNT_MIN_SKETCH, std::move(columnTags), std::move(data)));
+    Register(CreateSaveStatisticsQuery(SelfId(), Database, TraversalPathId, std::move(items)));
 }
 
 void TStatisticsAggregator::DeleteStatisticsFromTable() {
@@ -677,6 +675,8 @@ void TStatisticsAggregator::ScheduleNextAnalyze(NIceDb::TNiceDb& db, const TActo
                 UpdateForceTraversalTableStatus(
                     TForceTraversalTable::EStatus::AnalyzeStarted, operation.OperationId, operationTable, db);
 
+                // operation.Types field is not used, TAnalyzeActor will determine suitable
+                // statistic types itself.
                 ctx.RegisterWithSameMailbox(new TAnalyzeActor(
                     SelfId(), operation.OperationId, operation.DatabaseName, operationTable.PathId,
                     operationTable.ColumnTags));

ydb/core/statistics/aggregator/aggregator_impl.h

@@ -284,6 +284,7 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
 
     bool IsStatisticsTableCreated = false;
     bool PendingSaveStatistics = false;
+    std::vector<TStatisticsItem> StatisticsToSave;
     bool PendingDeleteStatistics = false;
 
     std::vector<NScheme::TTypeInfo> KeyColumnTypes;