Merge 0b57472 into 186c199

alexd65536 · web-flow · commit 3c6f75903ac8 · 2024-05-15T12:35:21.000+03:00
diff --git a/ydb/core/protos/counters_statistics_aggregator.proto b/ydb/core/protos/counters_statistics_aggregator.proto
@@ -17,4 +17,5 @@ enum ETxTypes {
     TXTYPE_SCAN_RESPONSE = 7       [(TxTypeOpts) = {Name: "TxScanResponse"}];
     TXTYPE_SAVE_QUERY_RESPONSE = 8 [(TxTypeOpts) = {Name: "TxSaveQueryResponse"}];
     TXTYPE_SCHEDULE_SCAN = 9       [(TxTypeOpts) = {Name: "TxScheduleScan"}];
+    TXTYPE_DELETE_QUERY_RESPONSE = 10 [(TxTypeOpts) = {Name: "TxDeleteQueryResponse"}];
 }
diff --git a/ydb/core/statistics/aggregator/aggregator_impl.cpp b/ydb/core/statistics/aggregator/aggregator_impl.cpp
@@ -407,6 +407,10 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvStatTableCreationResponse::
         PendingSaveStatistics = false;
         SaveStatisticsToTable();
     }
+    if (PendingDeleteStatistics) {
+        PendingDeleteStatistics = false;
+        DeleteStatisticsFromTable();
+    }
 }
 
 void TStatisticsAggregator::Initialize() {
@@ -484,6 +488,17 @@ void TStatisticsAggregator::SaveStatisticsToTable() {
         std::move(columnNames), std::move(data)));
 }
 
+void TStatisticsAggregator::DeleteStatisticsFromTable() {
+    if (!IsStatisticsTableCreated) {
+        PendingDeleteStatistics = true;
+        return;
+    }
+
+    PendingDeleteStatistics = false;
+
+    Register(CreateDeleteStatisticsQuery(ScanTableId.PathId));
+}
+
 void TStatisticsAggregator::ScheduleNextScan() {
     while (!ScanTablesByTime.empty()) {
         auto& topTable = ScanTablesByTime.top();
@@ -539,6 +554,38 @@ void TStatisticsAggregator::ResetScanState(NIceDb::TNiceDb& db) {
     ColumnNames.clear();
 }
 
+void TStatisticsAggregator::RescheduleScanTable(NIceDb::TNiceDb& db) {
+    if (ScanTablesByTime.empty()) {
+        return;
+    }
+    auto& topTable = ScanTablesByTime.top();
+    auto pathId = topTable.PathId;
+    if (pathId == ScanTableId.PathId) {
+        TScanTable scanTable;
+        scanTable.PathId = pathId;
+        scanTable.SchemeShardId = topTable.SchemeShardId;
+        scanTable.LastUpdateTime = ScanStartTime;
+
+        ScanTablesByTime.pop();
+        ScanTablesByTime.push(scanTable);
+
+        db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Update(
+            NIceDb::TUpdate<Schema::ScanTables::LastUpdateTime>(ScanStartTime.MicroSeconds()));
+    }
+}
+
+void TStatisticsAggregator::DropScanTable(NIceDb::TNiceDb& db) {
+    if (ScanTablesByTime.empty()) {
+        return;
+    }
+    auto& topTable = ScanTablesByTime.top();
+    auto pathId = topTable.PathId;
+    if (pathId == ScanTableId.PathId) {
+        ScanTablesByTime.pop();
+        db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Delete();
+    }
+}
+
 template <typename T, typename S>
 void PrintContainerStart(const T& container, size_t count, TStringStream& str,
     std::function<S(const typename T::value_type&)> extractor)
diff --git a/ydb/core/statistics/aggregator/aggregator_impl.h b/ydb/core/statistics/aggregator/aggregator_impl.h
@@ -47,6 +47,7 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
     struct TTxStatisticsScanResponse;
     struct TTxSaveQueryResponse;
     struct TTxScheduleScan;
+    struct TTxDeleteQueryResponse;
 
     struct TEvPrivate {
         enum EEv {
@@ -107,20 +108,24 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
     void Handle(TEvPipeCache::TEvDeliveryProblem::TPtr& ev);
     void Handle(TEvStatistics::TEvStatTableCreationResponse::TPtr& ev);
     void Handle(TEvStatistics::TEvSaveStatisticsQueryResponse::TPtr& ev);
+    void Handle(TEvStatistics::TEvDeleteStatisticsQueryResponse::TPtr& ev);
     void Handle(TEvPrivate::TEvScheduleScan::TPtr& ev);
 
     void Initialize();
     void Navigate();
     void Resolve();
     void NextRange();
     void SaveStatisticsToTable();
+    void DeleteStatisticsFromTable();
     void ScheduleNextScan();
 
     void PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value);
     void PersistScanTableId(NIceDb::TNiceDb& db);
     void PersistScanStartTime(NIceDb::TNiceDb& db);
 
     void ResetScanState(NIceDb::TNiceDb& db);
+    void RescheduleScanTable(NIceDb::TNiceDb& db);
+    void DropScanTable(NIceDb::TNiceDb& db);
 
     STFUNC(StateInit) {
         StateInitImpl(ev, SelfId());
@@ -203,6 +208,7 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
 
     bool IsStatisticsTableCreated = false;
     bool PendingSaveStatistics = false;
+    bool PendingDeleteStatistics = false;
 
     std::vector<NScheme::TTypeInfo> KeyColumnTypes;
     TVector<TKeyDesc::TColumnOp> Columns;
diff --git a/ydb/core/statistics/aggregator/tx_delete_query_response.cpp b/ydb/core/statistics/aggregator/tx_delete_query_response.cpp
@@ -0,0 +1,35 @@
+#include "aggregator_impl.h"
+
+#include <ydb/core/tx/datashard/datashard.h>
+
+namespace NKikimr::NStat {
+
+struct TStatisticsAggregator::TTxDeleteQueryResponse : public TTxBase {
+
+    TTxDeleteQueryResponse(TSelf* self)
+        : TTxBase(self)
+    {}
+
+    TTxType GetTxType() const override { return TXTYPE_DELETE_QUERY_RESPONSE; }
+
+    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+        SA_LOG_D("[" << Self->TabletID() << "] TTxDeleteQueryResponse::Execute");
+
+        NIceDb::TNiceDb db(txc.DB);
+
+        Self->ResetScanState(db);
+
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {
+        SA_LOG_D("[" << Self->TabletID() << "] TTxDeleteQueryResponse::Complete");
+
+        Self->ScheduleNextScan();
+    }
+};
+void TStatisticsAggregator::Handle(TEvStatistics::TEvDeleteStatisticsQueryResponse::TPtr&) {
+    Execute(new TTxDeleteQueryResponse(this), TActivationContext::AsActorContext());
+}
+
+} // NKikimr::NStat
diff --git a/ydb/core/statistics/aggregator/tx_navigate.cpp b/ydb/core/statistics/aggregator/tx_navigate.cpp
@@ -26,13 +26,12 @@ struct TStatisticsAggregator::TTxNavigate : public TTxBase {
         if (entry.Status != NSchemeCache::TSchemeCacheNavigate::EStatus::Ok) {
             Cancelled = true;
 
-            if (!Self->ScanTablesByTime.empty()) {
-                auto& topTable = Self->ScanTablesByTime.top();
-                auto pathId = topTable.PathId;
-                if (pathId == Self->ScanTableId.PathId) {
-                    Self->ScanTablesByTime.pop();
-                    db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Delete();
-                }
+            if (entry.Status == NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown) {
+                Self->DropScanTable(db);
+                Self->DeleteStatisticsFromTable();
+            } else {
+                Self->RescheduleScanTable(db);
+                Self->ScheduleNextScan();
             }
 
             Self->ResetScanState(db);
@@ -70,7 +69,6 @@ struct TStatisticsAggregator::TTxNavigate : public TTxBase {
         SA_LOG_D("[" << Self->TabletID() << "] TTxNavigate::Complete");
 
         if (Cancelled) {
-            Self->ScheduleNextScan();
             return;
         }
 
diff --git a/ydb/core/statistics/aggregator/tx_resolve.cpp b/ydb/core/statistics/aggregator/tx_resolve.cpp
@@ -26,13 +26,12 @@ struct TStatisticsAggregator::TTxResolve : public TTxBase {
         if (entry.Status != NSchemeCache::TSchemeCacheRequest::EStatus::OkData) {
             Cancelled = true;
 
-            if (!Self->ScanTablesByTime.empty()) {
-                auto& topTable = Self->ScanTablesByTime.top();
-                auto pathId = topTable.PathId;
-                if (pathId == Self->ScanTableId.PathId) {
-                    Self->ScanTablesByTime.pop();
-                    db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Delete();
-                }
+            if (entry.Status == NSchemeCache::TSchemeCacheRequest::EStatus::PathErrorNotExist) {
+                Self->DropScanTable(db);
+                Self->DeleteStatisticsFromTable();
+            } else {
+                Self->RescheduleScanTable(db);
+                Self->ScheduleNextScan();
             }
 
             Self->ResetScanState(db);
@@ -58,7 +57,6 @@ struct TStatisticsAggregator::TTxResolve : public TTxBase {
         SA_LOG_D("[" << Self->TabletID() << "] TTxResolve::Complete");
 
         if (Cancelled) {
-            Self->ScheduleNextScan();
             return;
         }
 
diff --git a/ydb/core/statistics/aggregator/tx_save_query_response.cpp b/ydb/core/statistics/aggregator/tx_save_query_response.cpp
@@ -17,30 +17,8 @@ struct TStatisticsAggregator::TTxSaveQueryResponse : public TTxBase {
 
         NIceDb::TNiceDb db(txc.DB);
 
-        if (!Self->ScanTablesByTime.empty()) {
-            auto& topTable = Self->ScanTablesByTime.top();
-            auto pathId = topTable.PathId;
-            if (pathId == Self->ScanTableId.PathId) {
-                TScanTable scanTable;
-                scanTable.PathId = pathId;
-                scanTable.SchemeShardId = topTable.SchemeShardId;
-                scanTable.LastUpdateTime = Self->ScanStartTime;
-
-                Self->ScanTablesByTime.pop();
-                Self->ScanTablesByTime.push(scanTable);
-
-                db.Table<Schema::ScanTables>().Key(pathId.OwnerId, pathId.LocalPathId).Update(
-                    NIceDb::TUpdate<Schema::ScanTables::LastUpdateTime>(Self->ScanStartTime.MicroSeconds()));
-            }
-        }
-
-        Self->ScanTableId.PathId = TPathId();
-        Self->PersistScanTableId(db);
-
-        for (auto& [tag, _] : Self->CountMinSketches) {
-            db.Table<Schema::Statistics>().Key(tag).Delete();
-        }
-        Self->CountMinSketches.clear();
+        Self->RescheduleScanTable(db);
+        Self->ResetScanState(db);
 
         return true;
     }
diff --git a/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp b/ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp
@@ -62,7 +62,9 @@ struct TStatisticsAggregator::TTxSchemeShardStats : public TTxBase {
     void Complete(const TActorContext&) override {
         SA_LOG_D("[" << Self->TabletID() << "] TTxSchemeShardStats::Complete");
 
-        Self->ScheduleNextScan();
+        if (!Self->ScanTableId.PathId) {
+            Self->ScheduleNextScan();
+        }
     }
 };
 
diff --git a/ydb/core/statistics/aggregator/ya.make b/ydb/core/statistics/aggregator/ya.make
@@ -13,6 +13,7 @@ SRCS(
     schema.h
     schema.cpp
     tx_configure.cpp
+    tx_delete_query_response.cpp
     tx_init.cpp
     tx_init_schema.cpp
     tx_navigate.cpp
diff --git a/ydb/core/statistics/events.h b/ydb/core/statistics/events.h
@@ -69,6 +69,8 @@ struct TEvStatistics {
         EvScanTable,
         EvScanTableResponse,
 
+        EvDeleteStatisticsQueryResponse,
+
         EvEnd
     };
 
@@ -159,6 +161,13 @@ struct TEvStatistics {
         std::optional<TString> Data;
     };
 
+    struct TEvDeleteStatisticsQueryResponse : public TEventLocal<
+        TEvDeleteStatisticsQueryResponse,
+        EvDeleteStatisticsQueryResponse>
+    {
+        bool Success = true;
+    };
+
     struct TEvScanTable : public TEventPB<
         TEvScanTable,
         NKikimrStat::TEvScanTable,
diff --git a/ydb/core/statistics/save_load_stats.cpp b/ydb/core/statistics/save_load_stats.cpp
@@ -106,8 +106,6 @@ class TSaveStatisticsQuery : public NKikimr::TQueryBase {
             sql << (i == Data.size() - 1 ? ";" : ",");
         }
 
-        Y_UNUSED(StatType);
-        Y_UNUSED(PathId);
         NYdb::TParamsBuilder params;
         params
             .AddParam("$owner_id")
@@ -211,12 +209,12 @@ class TLoadStatisticsQuery : public NKikimr::TQueryBase {
 
     void OnQueryResult() override {
         if (ResultSets.size() != 1) {
-            Finish(Ydb::StatusIds::INTERNAL_ERROR, "Unexpected read response");
+            Finish(Ydb::StatusIds::INTERNAL_ERROR, "Unexpected read response", false);
             return;
         }
         NYdb::TResultSetParser result(ResultSets[0]);
         if (result.RowsCount() == 0) {
-            Finish(Ydb::StatusIds::BAD_REQUEST, "No data");
+            Finish(Ydb::StatusIds::BAD_REQUEST, "No data", false);
             return;
         }
         result.TryNextRow();
@@ -242,4 +240,56 @@ NActors::IActor* CreateLoadStatisticsQuery(const TPathId& pathId, ui64 statType,
     return new TLoadStatisticsQuery(pathId, statType, columnName, cookie);
 }
 
+
+class TDeleteStatisticsQuery : public NKikimr::TQueryBase {
+private:
+    const TPathId PathId;
+
+public:
+    TDeleteStatisticsQuery(const TPathId& pathId)
+        : NKikimr::TQueryBase(NKikimrServices::STATISTICS, {})
+        , PathId(pathId)
+    {
+    }
+
+    void OnRunQuery() override {
+        TString sql = R"(
+            DECLARE $owner_id AS Uint64;
+            DECLARE $local_path_id AS Uint64;
+
+            DELETE FROM `.metadata/statistics`
+            WHERE
+                owner_id = $owner_id AND
+                local_path_id = $local_path_id;
+        )";
+
+        NYdb::TParamsBuilder params;
+        params
+            .AddParam("$owner_id")
+                .Uint64(PathId.OwnerId)
+                .Build()
+            .AddParam("$local_path_id")
+                .Uint64(PathId.LocalPathId)
+                .Build();
+
+        RunDataQuery(sql, &params);
+    }
+
+    void OnQueryResult() override {
+        Finish();
+    }
+
+    void OnFinish(Ydb::StatusIds::StatusCode status, NYql::TIssues&& issues) override {
+        Y_UNUSED(issues);
+        auto response = std::make_unique<TEvStatistics::TEvDeleteStatisticsQueryResponse>();
+        response->Success = (status == Ydb::StatusIds::SUCCESS);
+        Send(Owner, response.release());
+    }
+};
+
+NActors::IActor* CreateDeleteStatisticsQuery(const TPathId& pathId)
+{
+    return new TDeleteStatisticsQuery(pathId);
+}
+
 } // NKikimr::NStat
diff --git a/ydb/core/statistics/save_load_stats.h b/ydb/core/statistics/save_load_stats.h
@@ -13,4 +13,6 @@ NActors::IActor* CreateSaveStatisticsQuery(const TPathId& pathId, ui64 statType,
 NActors::IActor* CreateLoadStatisticsQuery(const TPathId& pathId, ui64 statType,
     const TString& columnName, ui64 cookie);
 
+NActors::IActor* CreateDeleteStatisticsQuery(const TPathId& pathId);
+
 };
diff --git a/ydb/core/statistics/ut/ut_aggregator.cpp b/ydb/core/statistics/ut/ut_aggregator.cpp
diff --git a/ydb/core/statistics/ut/ut_save_load_stats.cpp b/ydb/core/statistics/ut/ut_save_load_stats.cpp

Original file line number	Diff line number	Diff line change
`@@ -17,4 +17,5 @@ enum ETxTypes {`
`17`	`17`	`TXTYPE_SCAN_RESPONSE = 7 [(TxTypeOpts) = {Name: "TxScanResponse"}];`
`18`	`18`	`TXTYPE_SAVE_QUERY_RESPONSE = 8 [(TxTypeOpts) = {Name: "TxSaveQueryResponse"}];`
`19`	`19`	`TXTYPE_SCHEDULE_SCAN = 9 [(TxTypeOpts) = {Name: "TxScheduleScan"}];`
	`20`	`+ TXTYPE_DELETE_QUERY_RESPONSE = 10 [(TxTypeOpts) = {Name: "TxDeleteQueryResponse"}];`
`20`	`21`	`}`
Original file line number	Diff line number	Diff line change
`@@ -62,7 +62,9 @@ struct TStatisticsAggregator::TTxSchemeShardStats : public TTxBase {`
`62`	`62`	`void Complete(const TActorContext&) override {`
`63`	`63`	`SA_LOG_D("[" << Self->TabletID() << "] TTxSchemeShardStats::Complete");`
`64`	`64`
`65`		`- Self->ScheduleNextScan();`
	`65`	`+ if (!Self->ScanTableId.PathId) {`
	`66`	`+ Self->ScheduleNextScan();`
	`67`	`+ }`
`66`	`68`	`}`
`67`	`69`	`};`
`68`	`70`