Maintan schema versions reference count and remove unused from local database and memory

ydb/core/tablet_flat/tablet_flat_executor.h

@@ -23,6 +23,10 @@ namespace NTable {
 
 namespace NTabletFlatExecutor {
 
+namespace NFlatExecutorSetup {
+    struct ITablet;
+}
+
 class TTransactionContext;
 class TExecutor;
 struct TPageCollectionTxEnv;
@@ -247,6 +251,7 @@ class TTransactionContext : public TTxMemoryProviderBase {
     NTable::TDatabase &DB;
     NWilson::TSpan &TransactionSpan;
     NWilson::TSpan TransactionExecutionSpan;
+    NFlatExecutorSetup::ITablet* Owner = nullptr;
 
 private:
     bool Rescheduled_ = false;
@@ -280,7 +285,6 @@ enum class ETerminationReason {
     MemoryLimitExceeded = 1,
 };
 
-
 class ITransaction : TNonCopyable {
 public:
     using TTransactionContext = NTabletFlatExecutor::TTransactionContext;

ydb/core/tx/columnshard/blobs_action/transaction/tx_write.cpp

@@ -32,6 +32,7 @@ bool TTxWrite::CommitOneBlob(TTransactionContext& txc, const NOlap::TWideSeriali
 }
 
 bool TTxWrite::Execute(TTransactionContext& txc, const TActorContext&) {
+    txc.Owner = Self;
     TMemoryProfileGuard mpg("TTxWrite::Execute");
     NActors::TLogContextGuard logGuard =
         NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_BLOBS)("tablet_id", Self->TabletID())("tx_state", "execute");

ydb/core/tx/columnshard/blobs_action/transaction/tx_write_index.cpp

@@ -7,6 +7,7 @@
 namespace NKikimr::NColumnShard {
 
 bool TTxWriteIndex::Execute(TTransactionContext& txc, const TActorContext& ctx) {
+    txc.Owner = Self;
     auto changes = Ev->Get()->IndexChanges;
     TMemoryProfileGuard mpg("TTxWriteIndex::Execute::" + changes->TypeString());
     TLogContextGuard gLogging = NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD_BLOBS)("tablet_id", Self->TabletID())("external_task_id", changes->GetTaskIdentifier());

ydb/core/tx/columnshard/columnshard__init.cpp

@@ -104,7 +104,7 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
     {
         ACFL_DEBUG("step", "TInsertTable::Load_Start");
         TMemoryProfileGuard g("TTxInit/InsertTable");
-        auto localInsertTable = std::make_unique<NOlap::TInsertTable>();
+        auto localInsertTable = std::make_unique<NOlap::TInsertTable>(Self);
         if (!localInsertTable->Load(db, dbTable, TAppData::TimeProvider->Now())) {
             ACFL_ERROR("step", "TInsertTable::Load_Fails");
             return false;
@@ -115,7 +115,7 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
 
     {
         ACFL_DEBUG("step", "TTablesManager::Load_Start");
-        TTablesManager tManagerLocal(Self->StoragesManager, Self->TabletID());
+        TTablesManager tManagerLocal(Self->StoragesManager, Self->TabletID(), Self);
         {
             TMemoryProfileGuard g("TTxInit/TTablesManager");
             if (!tManagerLocal.InitFromDB(db)) {
@@ -239,6 +239,7 @@ bool TTxInit::ReadEverything(TTransactionContext& txc, const TActorContext& ctx)
 }
 
 bool TTxInit::Execute(TTransactionContext& txc, const TActorContext& ctx) {
+    txc.Owner = Self;
     NActors::TLogContextGuard gLogging = NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD)("tablet_id", Self->TabletID())("event", "initialize_shard");
     LOG_S_DEBUG("TTxInit.Execute at tablet " << Self->TabletID());
 
@@ -340,6 +341,7 @@ class TTxApplyNormalizer : public TTransactionBase<TColumnShard> {
 };
 
 bool TTxApplyNormalizer::Execute(TTransactionContext& txc, const TActorContext&) {
+    txc.Owner = Self;
     NActors::TLogContextGuard gLogging = NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD)("tablet_id", Self->TabletID())("event", "initialize_shard");
     AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("step", "TTxApplyNormalizer.Execute")("details", Self->NormalizerController.DebugString());
     if (!Changes->ApplyOnExecute(txc, Self->NormalizerController)) {

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -76,10 +76,10 @@ TColumnShard::TColumnShard(TTabletStorageInfo* info, const TActorId& tablet)
     , PeriodicWakeupActivationPeriod(NYDBTest::TControllers::GetColumnShardController()->GetPeriodicWakeupActivationPeriod())
     , StatsReportInterval(NYDBTest::TControllers::GetColumnShardController()->GetStatsReportInterval())
     , InFlightReadsTracker(StoragesManager, Counters.GetRequestsTracingCounters())
-    , TablesManager(StoragesManager, info->TabletID)
+    , TablesManager(StoragesManager, info->TabletID, this)
     , Subscribers(std::make_shared<NSubscriber::TManager>(*this))
     , PipeClientCache(NTabletPipe::CreateBoundedClientCache(new NTabletPipe::TBoundedClientCacheConfig(), GetPipeClientConfig()))
-    , InsertTable(std::make_unique<NOlap::TInsertTable>())
+    , InsertTable(std::make_unique<NOlap::TInsertTable>(this))
     , InsertTaskSubscription(NOlap::TInsertColumnEngineChanges::StaticTypeName(), Counters.GetSubscribeCounters())
     , CompactTaskSubscription(NOlap::TCompactColumnEngineChanges::StaticTypeName(), Counters.GetSubscribeCounters())
     , TTLTaskSubscription(NOlap::TTTLColumnEngineChanges::StaticTypeName(), Counters.GetSubscribeCounters())

ydb/core/tx/columnshard/columnshard_impl.h

@@ -7,6 +7,8 @@
 #include "columnshard_private_events.h"
 #include "tables_manager.h"
 
+#include "common/log.h"
+
 #include "blobs_action/events/delete_blobs.h"
 #include "bg_tasks/events/local.h"
 #include "transactions/tx_controller.h"
@@ -45,6 +47,8 @@
 #include <ydb/services/metadata/service.h>
 #include <ydb/services/metadata/abstract/common.h>
 
+#include <util/generic/string.h>
+
 namespace NKikimr::NOlap {
 class TCleanupPortionsColumnEngineChanges;
 class TCleanupTablesColumnEngineChanges;
@@ -53,6 +57,8 @@ class TChangesWithAppend;
 class TCompactColumnEngineChanges;
 class TInsertColumnEngineChanges;
 class TStoragesManager;
+class TDbWrapper;
+class TColumnEngineForLogs;
 
 namespace NReader {
 class TTxScan;
@@ -136,6 +142,94 @@ using ITransaction = NTabletFlatExecutor::ITransaction;
 template <typename T>
 using TTransactionBase = NTabletFlatExecutor::TTransactionBase<T>;
 
+class TInsertKey {
+public:
+    ui64 PlanStep;
+    ui64 TxId;
+    ui64 PathId;
+    TString DedupId;
+    ui8 RecType;
+
+public:
+    TInsertKey() = default;
+
+    TInsertKey(ui64 planStep, ui64 txId, ui64 pathId, const TString& dedupId, ui8 recType)
+        : PlanStep(planStep)
+        , TxId(txId)
+        , PathId(pathId)
+        , DedupId(dedupId)
+        , RecType(recType)
+    {
+    }
+
+    bool operator==(const TInsertKey& other) const {
+        return (PlanStep == other.PlanStep) && (TxId == other.TxId) && (PathId == other.PathId) && (DedupId == other.DedupId) && (RecType == other.RecType);
+    }
+
+    ui64 Hash() const {
+        return CombineHashes(PlanStep, CombineHashes(TxId, CombineHashes(PathId, CombineHashes(hash<TString>()(DedupId), (ui64)RecType))));
+    }
+};
+
+class TVersionCounts {
+private:
+    struct Hasher {
+        inline size_t operator()(const TInsertKey& key) const noexcept {
+            return key.Hash();
+        }
+    };
+
+private:
+    THashMap<TInsertKey, ui64, Hasher> InsertVersions;
+    THashMap<std::pair<ui64, ui64>, ui64> PortionVersions;
+    THashMap<ui64, ui32> VersionCounts;
+
+public:
+    template<class Key, class Versions>
+    void VersionAddRef(Versions& versions, const Key& portion, ui64 version) {
+        ui64& curVer = versions[portion];
+        if (curVer != 0) {// Portion is already in the local database, no need to increase ref count
+            TEMPLOG("Schema version is already written");
+            AFL_VERIFY(version == curVer);
+            return;
+        }
+        curVer = version;
+        ui32& refCount = VersionCounts[version];
+        refCount++;
+        TEMPLOG("Ref count of schema version " << version << " changed from " << refCount - 1 << " to " << refCount << " this " << (ui64)this);
+    }
+
+    template<class Key, class Versions>
+    ui32 VersionRemoveRef(Versions& versions, const Key& portion, ui64 version) {
+        auto iter = versions.find(portion);
+        if (iter == versions.end()) { //Portion is already removed from local databae, no need to decrease ref count
+            return (ui32)-1;
+        }
+        versions.erase(iter);
+        ui32& refCount = VersionCounts[version];
+        AFL_VERIFY(refCount > 0);
+        TEMPLOG("Ref count of schema version " << version << " changed from " << refCount << " to " << refCount - 1 << " this " << (ui64)this);
+        return --refCount;
+    }
+
+    void VersionAddRef(ui64 portion, ui64 pathId, ui64 version) {
+        VersionAddRef(PortionVersions, std::pair<ui64, ui64>(portion, pathId), version);
+    }
+
+    ui32 VersionRemoveRef(ui64 portion, ui64 pathId, ui64 version) {
+        return VersionRemoveRef(PortionVersions, std::pair<ui64, ui64>(portion, pathId), version);
+    }
+
+    void VersionAddRef(const TInsertKey& key, ui64 version) {
+        VersionAddRef(InsertVersions, key, version);
+    }
+
+    ui32 VersionRemoveRef(const TInsertKey& key, ui64 version) {
+        return VersionRemoveRef(InsertVersions, key, version);
+    }
+
+};
+
 class TColumnShard
     : public TActor<TColumnShard>
     , public NTabletFlatExecutor::TTabletExecutedFlat
@@ -186,6 +280,9 @@ class TColumnShard
     friend class NOlap::NReader::TTxInternalScan;
     friend class NOlap::NReader::NPlain::TIndexScannerConstructor;
 
+    friend class NOlap::TDbWrapper;
+    friend class NOlap::TColumnEngineForLogs;
+
     class TStoragesManager;
     friend class TTxController;
 
@@ -620,6 +717,29 @@ class TColumnShard
     }
 
     TColumnShard(TTabletStorageInfo* info, const TActorId& tablet);
+
+    void VersionAddRef(ui64 portion, ui64 pathId, ui64 version) {
+        VersionCounts.VersionAddRef(portion, pathId, version);
+    }
+
+    ui32 VersionRemoveRef(ui64 portion, ui64 pathId, ui64 version) {
+        return VersionCounts.VersionRemoveRef(portion, pathId, version);
+    }
+
+    void VersionAddRef(ui64 planStep, ui64 txId, ui64 pathId, const TString& dedupId, ui8 recType, ui64 version) {
+        VersionCounts.VersionAddRef(TInsertKey(planStep, txId, pathId, dedupId, recType), version);
+    }
+
+    ui32 VersionRemoveRef(ui64 planStep, ui64 txId, ui64 pathId, const TString& dedupId, ui8 recType, ui64 version) {
+        return VersionCounts.VersionRemoveRef(TInsertKey(planStep, txId, pathId, dedupId, recType), version);
+    }
+
+    void RemoveUnusedSchemaVersion(NTable::TDatabase* database, ui64 version) {
+        TablesManager.RemoveUnusedSchemaVersion(database, version);
+    }
+
+private:
+     TVersionCounts VersionCounts;
 };
 
 }

ydb/core/tx/columnshard/columnshard_schema.cpp

@@ -1,8 +1,15 @@
 #include "columnshard_schema.h"
 #include "transactions/tx_controller.h"
+#include "columnshard_impl.h"
+
+#include "common/log.h"
 
 namespace NKikimr::NColumnShard {
 
+ui64 GetPlanStep(const NOlap::TInsertTableRecordLoadContext& constructor) {
+    return constructor.GetRecType() == Schema::EInsertTableIds::Committed ? constructor.GetPlanStep() : 0;
+}
+
 bool Schema::InsertTable_Load(NIceDb::TNiceDb& db, const IBlobGroupSelector* dsGroupSelector, NOlap::TInsertTableAccessor& insertTable, const TInstant& /*loadTime*/) {
     auto rowset = db.Table<InsertTable>().Select();
     if (!rowset.IsReady()) {
@@ -12,7 +19,10 @@ bool Schema::InsertTable_Load(NIceDb::TNiceDb& db, const IBlobGroupSelector* dsG
     while (!rowset.EndOfSet()) {
         NOlap::TInsertTableRecordLoadContext constructor;
         constructor.ParseFromDatabase(rowset);
+        ui64 planStep = GetPlanStep(constructor);
 
+        TEMPLOG("Loaded schema version " << constructor.GetSchemaVersion() << " planstep " << planStep << " txid " << constructor.GetWriteTxId() << " dedup id " << constructor.GetDedupId() << " rec type " << (ui32)constructor.GetRecType());
+        insertTable.CS->VersionAddRef(planStep, constructor.GetWriteTxId(), constructor.GetPathId(), constructor.GetDedupId(), (ui8)constructor.GetRecType(), constructor.GetSchemaVersion());
         switch (constructor.GetRecType()) {
             case Schema::EInsertTableIds::Inserted:
                 insertTable.AddInserted(constructor.BuildInsertedOrAborted(dsGroupSelector), true);

ydb/core/tx/columnshard/columnshard_schema.h

@@ -984,11 +984,11 @@ class TInsertTableRecordLoadContext {
 private:
     NColumnShard::Schema::EInsertTableIds RecType;
     ui64 PlanStep;
-    ui64 WriteTxId;
     TInsertWriteId InsertWriteId;
-    ui64 PathId;
+    YDB_ACCESSOR_DEF(ui64, WriteTxId);
+    YDB_ACCESSOR_DEF(ui64, PathId);
     YDB_ACCESSOR_DEF(TString, DedupId);
-    ui64 SchemaVersion;
+    YDB_ACCESSOR_DEF(ui64, SchemaVersion);
     TString BlobIdString;
     std::optional<NOlap::TUnifiedBlobId> BlobId;
     TString MetadataString;

ydb/core/tx/columnshard/common/log.h

@@ -0,0 +1,7 @@
+#pragma once
+
+#if 0
+#define TEMPLOG(x) LOG_S_CRIT(x)
+#else
+#define TEMPLOG(x)
+#endif

ydb/core/tx/columnshard/data_sharing/destination/transactions/tx_data_from_source.cpp

@@ -1,10 +1,12 @@
 #include "tx_data_from_source.h"
 #include <ydb/core/tx/columnshard/engines/column_engine_logs.h>
 #include <ydb/core/tx/columnshard/columnshard_schema.h>
+#include <ydb/core/tx/columnshard/common/log.h>
 
 namespace NKikimr::NOlap::NDataSharing {
 
 bool TTxDataFromSource::DoExecute(NTabletFlatExecutor::TTransactionContext& txc, const TActorContext& /*ctx*/) {
+    TEMPLOG("SaveToDatabase from TTxDataFromSource");
     using namespace NKikimr::NColumnShard;
     TDbWrapper dbWrapper(txc.DB, nullptr);
     auto& index = Self->TablesManager.MutablePrimaryIndexAsVerified<NOlap::TColumnEngineForLogs>();
@@ -22,6 +24,9 @@ bool TTxDataFromSource::DoExecute(NTabletFlatExecutor::TTransactionContext& txc,
     for (auto&& i : PortionsByPathId) {
         for (auto&& p : i.second.GetPortions()) {
             p.SaveToDatabase(dbWrapper, schemaPtr->GetIndexInfo().GetPKFirstColumnId(), false);
+            txc.DB.OnCommit([self = Self, portion = p.GetPortion(), pathId = p.GetPathId(), schema = p.GetSchemaVersionVerified()]() {
+                self->VersionAddRef(portion, pathId, schema);
+            });
         }
     }
     NIceDb::TNiceDb db(txc.DB);

ydb/core/tx/columnshard/engines/changes/cleanup_portions.cpp

@@ -1,5 +1,6 @@
 #include "cleanup_portions.h"
 #include <ydb/core/tx/columnshard/columnshard_impl.h>
+#include <ydb/core/tx/columnshard/common/log.h>
 #include <ydb/core/tx/columnshard/engines/column_engine_logs.h>
 #include <ydb/core/tx/columnshard/blobs_action/blob_manager_db.h>
 #include <ydb/core/tx/columnshard/columnshard_schema.h>
@@ -25,6 +26,14 @@ void TCleanupPortionsColumnEngineChanges::DoWriteIndexOnExecute(NColumnShard::TC
         p.RemoveFromDatabase(context.DBWrapper);
         p.FillBlobIdsByStorage(blobIdsByStorage, context.EngineLogs.GetVersionedIndex());
         pathIds.emplace(p.GetPathId());
+        context.DB->OnCommit([self, portion = p.GetPortion(), pathId = p.GetPathId(), schema = p.GetSchemaVersionVerified(), db = context.DB]() {
+            TEMPLOG("Removing portion from cleanup");
+            ui32 refCount = self->VersionRemoveRef(portion, pathId, schema);
+            if (refCount == 0) {
+                TEMPLOG("Ref count is set to 0 for version " << schema << " need to delete");
+                self->TablesManager.RemoveUnusedSchemaVersion(db, schema);
+            }
+        });
     }
     for (auto&& i : blobIdsByStorage) {
         auto action = BlobsAction.GetRemoving(i.first);

ydb/core/tx/columnshard/engines/changes/with_appended.cpp

@@ -16,6 +16,11 @@ void TChangesWithAppend::DoWriteIndexOnExecute(NColumnShard::TColumnShard* self,
         Y_ABORT_UNLESS(portionInfo.HasRemoveSnapshot());
         AFL_VERIFY(usedPortionIds.emplace(portionInfo.GetPortionId()).second)("portion_info", portionInfo.DebugString(true));
         portionInfo.SaveToDatabase(context.DBWrapper, schemaPtr->GetIndexInfo().GetPKFirstColumnId(), false);
+        if (self != nullptr) { // nullptr can happen in tests
+            context.DB->OnCommit([self, portion = portionInfo.GetPortion(), pathId = portionInfo.GetPathId(), schema = portionInfo.GetSchemaVersionVerified()]() {
+                self->VersionAddRef(portion, pathId, schema);
+            });
+        }
     }
     const auto predRemoveDroppedTable = [self](const TWritePortionInfoWithBlobsResult& item) {
         auto& portionInfo = item.GetPortionResult();
@@ -31,6 +36,11 @@ void TChangesWithAppend::DoWriteIndexOnExecute(NColumnShard::TColumnShard* self,
         auto& portionInfo = portionInfoWithBlobs.GetPortionResult();
         AFL_VERIFY(usedPortionIds.emplace(portionInfo.GetPortionId()).second)("portion_info", portionInfo.DebugString(true));
         portionInfo.SaveToDatabase(context.DBWrapper, schemaPtr->GetIndexInfo().GetPKFirstColumnId(), false);
+        if (self != nullptr) { // nullptr can happen in tests
+            context.DB->OnCommit([self, portion = portionInfo.GetPortion(), pathId = portionInfo.GetPathId(), schema = portionInfo.GetSchemaVersionVerified()]() {
+                self->VersionAddRef(portion, pathId, schema);
+            });
+        }
     }
 }
 

ydb/core/tx/columnshard/engines/column_engine.h

@@ -298,6 +298,10 @@ class IColumnEngine {
         return TSnapshot::Zero();
     }
     virtual void OnTieringModified(const std::shared_ptr<NColumnShard::TTiersManager>& manager, const NColumnShard::TTtl& ttl, const std::optional<ui64> pathId) = 0;
+
+    virtual bool RemoveSchemaVersion(ui64) {
+        return true;
+    }
 };
 
 }   // namespace NKikimr::NOlap