Merge 852f18a into ea324a3

Author: aavdonkin

ydb/core/tx/columnshard/normalizer/abstract/abstract.h

@@ -59,6 +59,7 @@ enum class ENormalizerSequentialId: ui32 {
     EmptyPortionsCleaner,
     CleanInsertionDedup,
     GCCountersNormalizer,
+    SchemaVersionCleaner,
 
     MAX
 };

ydb/core/tx/columnshard/normalizer/schema_version/version.cpp

@@ -0,0 +1,136 @@
+#include "version.h"
+
+namespace NKikimr::NOlap {
+
+class TSchemaVersionNormalizer::TNormalizerResult : public INormalizerChanges {
+private:
+    class TKey {
+    public:
+        ui64 Step;
+        ui64 TxId;
+        ui64 Version;
+        ui32 Id;
+
+    public:
+        TKey() = default;
+
+        TKey(ui32 id, ui64 step, ui64 txId, ui64 version)
+            : Step(step)
+            , TxId(txId)
+            , Version(version)
+            , Id(id)
+        {
+        }
+    };
+
+    std::vector<TKey> VersionsToRemove;
+    std::optional<ui64> LastVersion;
+
+public:
+    TNormalizerResult(std::vector<TKey>&& versions, std::optional<ui64>& lastVersion)
+        : VersionsToRemove(versions)
+        , LastVersion(lastVersion)
+    {
+    }
+
+    bool ApplyOnExecute(NTabletFlatExecutor::TTransactionContext& txc, const TNormalizationController& /* normController */) const override {
+        using namespace NColumnShard;
+        NIceDb::TNiceDb db(txc.DB);
+        for (auto& key: VersionsToRemove) {
+            if ((!LastVersion.has_value()) || (key.Version != *LastVersion)) {
+                db.Table<Schema::SchemaPresetVersionInfo>().Key(key.Id, key.Step, key.TxId).Delete();
+            }
+        }
+        return true;
+    }
+
+    ui64 GetSize() const override {
+        return VersionsToRemove.size();
+    }
+
+    static std::optional<std::vector<INormalizerChanges::TPtr>> Init(NTabletFlatExecutor::TTransactionContext& txc) {
+        using namespace NColumnShard;
+        THashSet<ui64> usedSchemaVersions;
+        NIceDb::TNiceDb db(txc.DB);
+        {
+            auto rowset = db.Table<Schema::IndexPortions>().Select();
+            if (rowset.IsReady()) {
+                while (!rowset.EndOfSet()) {
+                    usedSchemaVersions.insert(rowset.GetValue<Schema::IndexPortions::SchemaVersion>());
+                    if (!rowset.Next()) {
+                        return std::nullopt;
+                    }
+                }
+            } else {
+                return std::nullopt;
+            }
+        }
+        {
+            auto rowset = db.Table<Schema::InsertTable>().Select();
+            if (rowset.IsReady()) {
+                while (!rowset.EndOfSet()) {
+                    if (rowset.HaveValue<Schema::InsertTable::SchemaVersion>()) {
+                        usedSchemaVersions.insert(rowset.GetValue<Schema::InsertTable::SchemaVersion>());
+                        if (!rowset.Next()) {
+                            return std::nullopt;
+                        }
+                    }
+                }
+            } else {
+                return std::nullopt;
+            }
+        }
+
+        std::vector<TKey> unusedSchemaIds;
+        std::optional<ui64> maxVersion;
+        std::vector<INormalizerChanges::TPtr> changes;
+
+        {
+            auto rowset = db.Table<Schema::SchemaPresetVersionInfo>().Select();
+            if (rowset.IsReady()) {
+                while (!rowset.EndOfSet()) {
+                    const ui32 id = rowset.GetValue<Schema::SchemaPresetVersionInfo::Id>();
+                    NKikimrTxColumnShard::TSchemaPresetVersionInfo info;
+                    Y_ABORT_UNLESS(info.ParseFromString(rowset.GetValue<Schema::SchemaPresetVersionInfo::InfoProto>()));
+                    if (info.HasSchema()) {
+                        ui64 version = info.GetSchema().GetVersion();
+                        if (!maxVersion.has_value() || (version > *maxVersion)) {
+                            maxVersion = version;
+                        }
+                        if (!usedSchemaVersions.contains(version)) {
+                            unusedSchemaIds.emplace_back(id, rowset.GetValue<Schema::SchemaPresetVersionInfo::SinceStep>(), rowset.GetValue<Schema::SchemaPresetVersionInfo::SinceTxId>(), version);
+                        }
+                    }
+
+                    if (!rowset.Next()) {
+                        return std::nullopt;
+                    }
+                }
+            } else {
+                return std::nullopt;
+            }
+        }
+
+        for (size_t start = 0; start < unusedSchemaIds.size(); start += 10000) {
+            std::vector<TKey> portion;
+            size_t end = std::min(unusedSchemaIds.size(), start + 10000);
+            portion.insert(portion.begin(), &unusedSchemaIds[start], &unusedSchemaIds[end]);
+            changes.emplace_back(std::make_shared<TNormalizerResult>(std::move(portion), maxVersion));
+        }
+        return changes;
+    }
+};
+
+TConclusion<std::vector<INormalizerTask::TPtr>> TSchemaVersionNormalizer::DoInit(const TNormalizationController&, NTabletFlatExecutor::TTransactionContext& txc) {
+    auto changes = TNormalizerResult::Init(txc);
+    if (!changes) {
+        return TConclusionStatus::Fail("Not ready");;
+    }
+    std::vector<INormalizerTask::TPtr> tasks;
+    for (auto&& c : *changes) {
+        tasks.emplace_back(std::make_shared<TTrivialNormalizerTask>(c));
+    }
+    return tasks;
+}
+
+}

ydb/core/tx/columnshard/normalizer/schema_version/version.h

@@ -0,0 +1,42 @@
+#pragma once
+
+#include <ydb/core/tx/columnshard/normalizer/abstract/abstract.h>
+#include <ydb/core/tx/columnshard/columnshard_schema.h>
+#include <ydb/core/tx/columnshard/defs.h>
+
+
+namespace NKikimr::NColumnShard {
+    class TTablesManager;
+}
+
+namespace NKikimr::NOlap {
+
+class TSchemaVersionNormalizer : public TNormalizationController::INormalizerComponent {
+public:
+    static TString GetClassNameStatic() {
+        return ::ToString(ENormalizerSequentialId::SchemaVersionCleaner);
+    }
+
+private:
+    static inline TFactory::TRegistrator<TSchemaVersionNormalizer> Registrator = TFactory::TRegistrator<TSchemaVersionNormalizer>(
+        GetClassNameStatic());
+public:
+    class TNormalizerResult;
+    class TTask;
+
+public:
+    virtual std::optional<ENormalizerSequentialId> DoGetEnumSequentialId() const override {
+        return ENormalizerSequentialId::SchemaVersionCleaner;
+    }
+
+    virtual TString GetClassName() const override {
+        return GetClassNameStatic();
+    }
+
+    TSchemaVersionNormalizer(const TNormalizationController::TInitContext&) {
+    }
+
+    virtual TConclusion<std::vector<INormalizerTask::TPtr>> DoInit(const TNormalizationController& controller, NTabletFlatExecutor::TTransactionContext& txc) override;
+};
+
+}

ydb/core/tx/columnshard/normalizer/schema_version/ya.make

@@ -0,0 +1,11 @@
+LIBRARY()
+
+SRCS(
+    GLOBAL version.cpp
+)
+
+PEERDIR(
+    ydb/core/tx/columnshard/normalizer/abstract
+)
+
+END()

ydb/core/tx/columnshard/normalizer/ya.make

@@ -6,6 +6,7 @@ PEERDIR(
     ydb/core/tx/columnshard/normalizer/tablet
     ydb/core/tx/columnshard/normalizer/tables
     ydb/core/tx/columnshard/normalizer/portion
+    ydb/core/tx/columnshard/normalizer/schema_version
     ydb/core/tx/columnshard/normalizer/insert_table
 )