clean inserted portions normalizer has been added

ydb/core/tx/columnshard/normalizer/portion/clean_inserted_portions.cpp

@@ -0,0 +1,81 @@
+#include "clean_inserted_portions.h"
+
+#include <ydb/core/formats/arrow/arrow_helpers.h>
+#include <ydb/core/tx/columnshard/columnshard_schema.h>
+#include <ydb/core/tx/columnshard/engines/portions/data_accessor.h>
+#include <ydb/core/tx/columnshard/engines/portions/read_with_blobs.h>
+#include <ydb/core/tx/columnshard/engines/scheme/filtered_scheme.h>
+#include <ydb/core/tx/columnshard/tables_manager.h>
+
+#include <util/string/vector.h>
+
+namespace NKikimr::NOlap::NNormalizer::NCleanInsertedPortions {
+
+class TCleanInsertedPortionsNormalizer::TNormalizerResult: public INormalizerChanges {
+    std::vector<TPortionDataAccessor> InsertedPortions;
+
+public:
+    TNormalizerResult(std::vector<TPortionDataAccessor>&& portions)
+        : InsertedPortions(std::move(portions)) {
+    }
+
+    bool ApplyOnExecute(NTabletFlatExecutor::TTransactionContext& txc, const TNormalizationController& normController) const override {
+        NOlap::TBlobManagerDb blobManagerDb(txc.DB);
+        TDbWrapper db(txc.DB, nullptr);
+        for (auto&& portionInfo : InsertedPortions) {
+            auto copy = portionInfo.GetPortionInfo().MakeCopy();
+            copy->SetRemoveSnapshot(TSnapshot(1, 1));
+            db.WritePortion(portionInfo.GetBlobIds(), *copy);
+        }
+        if (InsertedPortions.size()) {
+            NIceDb::TNiceDb db(txc.DB);
+            normController.AddNormalizerEvent(db, "REMOVE_PORTIONS", DebugString());
+        }
+        return true;
+    }
+
+    void ApplyOnComplete(const TNormalizationController& /* normController */) const override {
+    }
+
+    ui64 GetSize() const override {
+        return InsertedPortions.size();
+    }
+
+    TString DebugString() const override {
+        TStringBuilder sb;
+        ui64 recordsCount = 0;
+        sb << "path_ids=[";
+        for (auto&& p : InsertedPortions) {
+            sb << p.GetPortionInfo().GetPathId() << ",";
+            recordsCount += p.GetPortionInfo().GetRecordsCount();
+        }
+        sb << "]";
+        sb << ";records_count=" << recordsCount;
+        sb << ";inserted_portions_count=" << InsertedPortions.size();
+        return sb;
+    }
+};
+
+bool TCleanInsertedPortionsNormalizer::CheckPortion(const NColumnShard::TTablesManager& /*tablesManager*/, const TPortionDataAccessor& /*portionInfo*/) const {
+    return false;
+}
+
+INormalizerTask::TPtr TCleanInsertedPortionsNormalizer::BuildTask(
+    std::vector<TPortionDataAccessor>&& portions, std::shared_ptr<THashMap<ui64, ISnapshotSchema::TPtr>>) const {
+    std::vector<TPortionDataAccessor> insertedPortions;
+    for (auto&& portion : portions) {
+        if (portion.GetPortionInfo().GetProduced() == NPortion::EProduced::INSERTED) {
+            insertedPortions.push_back(std::move(portion));
+        }
+    }
+    auto taskResult = std::make_shared<TNormalizerResult>(std::move(insertedPortions));
+    ACFL_WARN("normalizer", "TCleanInsertedPortionsNormalizer")("message", taskResult->DebugString());
+    ACFL_WARN("normalizer", "TCleanInsertedPortionsNormalizer")("all portions", portions.size());
+    return std::make_shared<TTrivialNormalizerTask>(taskResult);
+}
+
+TConclusion<bool> TCleanInsertedPortionsNormalizer::DoInitImpl(const TNormalizationController&, NTabletFlatExecutor::TTransactionContext&) {
+    return true;
+}
+
+}   // namespace NKikimr::NOlap::NNormalizer::NCleanInsertedPortions

ydb/core/tx/columnshard/normalizer/portion/clean_inserted_portions.h

@@ -0,0 +1,50 @@
+#pragma once
+
+#include "normalizer.h"
+
+#include <ydb/core/tx/columnshard/defs.h>
+#include <ydb/core/tx/columnshard/normalizer/abstract/abstract.h>
+
+namespace NKikimr::NColumnShard {
+class TTablesManager;
+}
+
+namespace NKikimr::NOlap::NNormalizer::NCleanInsertedPortions {
+
+class TCleanInsertedPortionsNormalizer: public TPortionsNormalizerBase {
+public:
+    static TString GetClassNameStatic() {
+        return "CleanInsertedPortions";
+    }
+
+private:
+    static inline TFactory::TRegistrator<TCleanInsertedPortionsNormalizer> Registrator = TFactory::TRegistrator<TCleanInsertedPortionsNormalizer>(GetClassNameStatic());
+
+public:
+    class TNormalizerResult;
+
+public:
+    virtual std::optional<ENormalizerSequentialId> DoGetEnumSequentialId() const override {
+        return {};
+    }
+
+    virtual TString GetClassName() const override {
+        return GetClassNameStatic();
+    }
+
+    TCleanInsertedPortionsNormalizer(const TNormalizationController::TInitContext& info)
+        : TPortionsNormalizerBase(info) {
+    }
+
+    virtual std::set<ui32> GetColumnsFilter(const ISnapshotSchema::TPtr& /*schema*/) const override {
+        return {};
+    }
+
+    virtual INormalizerTask::TPtr BuildTask(
+        std::vector<TPortionDataAccessor>&& portions, std::shared_ptr<THashMap<ui64, ISnapshotSchema::TPtr>> schemas) const override;
+    virtual TConclusion<bool> DoInitImpl(const TNormalizationController& controller, NTabletFlatExecutor::TTransactionContext& txc) override;
+
+    virtual bool CheckPortion(const NColumnShard::TTablesManager& tablesManager, const TPortionDataAccessor& portionInfo) const override;
+};
+
+}   // namespace NKikimr::NOlap::NNormalizer::NCleanInsertedPortions

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

@@ -2,24 +2,25 @@ LIBRARY()
 
 SRCS(
     normalizer.cpp
-    GLOBAL portion.cpp
-    GLOBAL chunks.cpp
-    GLOBAL clean.cpp
-    GLOBAL clean_empty.cpp
     GLOBAL broken_blobs.cpp
-    GLOBAL special_cleaner.cpp
+    GLOBAL chunks.cpp
     GLOBAL chunks_actualization.cpp
-    GLOBAL restore_v1_chunks.cpp
-    GLOBAL restore_v2_chunks.cpp
-    GLOBAL leaked_blobs.cpp
-    GLOBAL clean_deprecated_snapshot.cpp
     GLOBAL chunks_v0_meta.cpp
+    GLOBAL clean.cpp
+    GLOBAL clean_deprecated_snapshot.cpp
+    GLOBAL clean_empty.cpp
     GLOBAL clean_index_columns.cpp
-    GLOBAL clean_unused_tables_template.cpp
+    GLOBAL clean_inserted_portions.cpp
     GLOBAL clean_ttl_preset_setting_info.cpp
     GLOBAL clean_ttl_preset_setting_version_info.cpp
+    GLOBAL clean_unused_tables_template.cpp
     GLOBAL copy_blob_ids_to_v2.cpp
+    GLOBAL leaked_blobs.cpp
+    GLOBAL portion.cpp
     GLOBAL restore_appearance_snapshot.cpp
+    GLOBAL restore_v1_chunks.cpp
+    GLOBAL restore_v2_chunks.cpp
+    GLOBAL special_cleaner.cpp
 )
 
 PEERDIR(

ydb/core/tx/columnshard/ut_rw/ut_normalizer.cpp

@@ -100,6 +100,12 @@ class TPortionsCleaner: public NYDBTest::ILocalDBModifier {
     }
 };
 
+class TInsertedPortionsCleaner: public NYDBTest::ILocalDBModifier {
+public:
+    virtual void Apply(NTabletFlatExecutor::TTransactionContext&) const override {
+    }
+};
+
 class TEmptyPortionsCleaner: public NYDBTest::ILocalDBModifier {
 public:
     virtual void Apply(NTabletFlatExecutor::TTransactionContext& txc) const override {
@@ -329,6 +335,25 @@ Y_UNIT_TEST_SUITE(Normalizers) {
         };
         TestNormalizerImpl<TPortionsCleaner>(TLocalNormalizerChecker());
     }
+
+    Y_UNIT_TEST(InsertedPortionsCleanerNormalizer) {
+        class TLocalNormalizerChecker: public TNormalizerChecker {
+        public:
+            virtual ui64 RecordsCountAfterReboot(const ui64 /*initialRecordsCount*/) const override {
+                return 0;
+            }
+            virtual void CorrectFeatureFlagsOnStart(TFeatureFlags& /* featuresFlags */) const override {
+            }
+            virtual void CorrectConfigurationOnStart(NKikimrConfig::TColumnShardConfig& columnShardConfig) const override {
+                {
+                    auto* repair = columnShardConfig.MutableRepairs()->Add();
+                    repair->SetClassName("CleanInsertedPortions");
+                    repair->SetDescription("Removing inserted portions");
+                }
+            }
+        };
+        TestNormalizerImpl<TInsertedPortionsCleaner>(TLocalNormalizerChecker());
+    }
 
     Y_UNIT_TEST(SchemaVersionsNormalizer) {
         class TLocalNormalizerChecker: public TNormalizerChecker {