tiering usage validations and control incorrect state usage

ydb/core/tx/columnshard/blobs_action/abstract/storage.h

@@ -77,6 +77,7 @@ class IBlobsStorageOperator {
         return DoLoad(dbBlobs);
     }
     void OnTieringModified(const std::shared_ptr<NColumnShard::ITiersManager>& tiers) {
+        AFL_VERIFY(tiers);
         return DoOnTieringModified(tiers);
     }
 

ydb/core/tx/columnshard/blobs_action/abstract/storages_manager.cpp

@@ -33,6 +33,7 @@ std::shared_ptr<NKikimr::NOlap::IBlobsStorageOperator> IStoragesManager::GetOper
 }
 
 void IStoragesManager::OnTieringModified(const std::shared_ptr<NColumnShard::ITiersManager>& tiers) {
+    AFL_VERIFY(tiers);
     for (auto&& i : tiers->GetManagers()) {
         GetOperatorGuarantee(i.first)->OnTieringModified(tiers);
     }

ydb/core/tx/columnshard/columnshard.cpp

@@ -39,9 +39,8 @@ void TColumnShard::SwitchToWork(const TActorContext& ctx) {
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "initialize_shard")("step", "SwitchToWork");
 
         for (auto&& i : TablesManager.GetTables()) {
-            ActivateTiering(i.first, i.second.GetTieringUsage(), true);
+            ActivateTiering(i.first, i.second.GetTieringUsage());
         }
-        OnTieringModified();
 
         Become(&TThis::StateWork);
         SignalTabletActive(ctx);

ydb/core/tx/columnshard/columnshard__init.cpp

@@ -381,7 +381,7 @@ bool TTxInitSchema::Execute(TTransactionContext& txc, const TActorContext&) {
 }
 
 void TTxInitSchema::Complete(const TActorContext& ctx) {
-    LOG_S_DEBUG("TxInitSchema.Complete at tablet " << Self->TabletID();)
+    LOG_S_DEBUG("TxInitSchema.Complete at tablet " << Self->TabletID(););
     Self->Execute(new TTxUpdateSchema(Self), ctx);
 }
 

ydb/core/tx/columnshard/columnshard__propose_transaction.cpp

@@ -109,7 +109,7 @@ TTxController::TProposeResult TTxProposeTransaction::ProposeTtlDeprecated(const
     if (!Self->SetupTtl(pathTtls)) {
         return TTxController::TProposeResult(NKikimrTxColumnShard::EResultStatus::SCHEMA_ERROR, "TTL not started");
     }
-    Self->TablesManager.MutablePrimaryIndex().OnTieringModified(Self->Tiers, Self->TablesManager.GetTtl());
+    Self->TablesManager.MutablePrimaryIndex().OnTieringModified(Self->Tiers, Self->TablesManager.GetTtl(), {});
 
     return TTxController::TProposeResult();
 }

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -401,21 +401,28 @@ void TColumnShard::RunEnsureTable(const NKikimrTxColumnShard::TCreateTable& tabl
         *tableVerProto.MutableSchema() = tableProto.GetSchema();
     }
 
-    TTableInfo table(pathId);
-    if (tableProto.HasTtlSettings()) {
-        const auto& ttlSettings = tableProto.GetTtlSettings();
-        *tableVerProto.MutableTtlSettings() = ttlSettings;
-        if (ttlSettings.HasUseTiering()) {
-            table.SetTieringUsage(ttlSettings.GetUseTiering());
-            ActivateTiering(pathId, table.GetTieringUsage());
+    {
+        bool needTieringActivation = false;
+        TTableInfo table(pathId);
+        if (tableProto.HasTtlSettings()) {
+            const auto& ttlSettings = tableProto.GetTtlSettings();
+            *tableVerProto.MutableTtlSettings() = ttlSettings;
+            if (ttlSettings.HasUseTiering()) {
+                table.SetTieringUsage(ttlSettings.GetUseTiering());
+                needTieringActivation = true;
+            }
+        }
+        const TString tieringName = table.GetTieringUsage();
+        TablesManager.RegisterTable(std::move(table), db);
+        if (needTieringActivation) {
+            ActivateTiering(pathId, tieringName);
         }
     }
 
     tableVerProto.SetSchemaPresetVersionAdj(tableProto.GetSchemaPresetVersionAdj());
     tableVerProto.SetTtlSettingsPresetVersionAdj(tableProto.GetTtlSettingsPresetVersionAdj());
 
-    TablesManager.RegisterTable(std::move(table), db);
-    TablesManager.AddTableVersion(pathId, version, tableVerProto, db);
+    TablesManager.AddTableVersion(pathId, version, tableVerProto, db, Tiers);
 
     SetCounter(COUNTER_TABLES, TablesManager.GetTables().size());
     SetCounter(COUNTER_TABLE_PRESETS, TablesManager.GetSchemaPresets().size());
@@ -452,7 +459,7 @@ void TColumnShard::RunAlterTable(const NKikimrTxColumnShard::TAlterTable& alterP
     Schema::SaveTableInfo(db, pathId, tieringUsage);
 
     tableVerProto.SetSchemaPresetVersionAdj(alterProto.GetSchemaPresetVersionAdj());
-    TablesManager.AddTableVersion(pathId, version, tableVerProto, db);
+    TablesManager.AddTableVersion(pathId, version, tableVerProto, db, Tiers);
 }
 
 void TColumnShard::RunDropTable(const NKikimrTxColumnShard::TDropTable& dropProto, const NOlap::TSnapshot& version,
@@ -1026,18 +1033,17 @@ void TColumnShard::Handle(NMetadata::NProvider::TEvRefreshSubscriberData::TPtr&
     Tiers->TakeConfigs(ev->Get()->GetSnapshot(), nullptr);
 }
 
-void TColumnShard::ActivateTiering(const ui64 pathId, const TString& useTiering, const bool onTabletInit) {
-    Y_ABORT_UNLESS(!!Tiers);
-    if (!!Tiers) {
-        if (useTiering) {
-            Tiers->EnablePathId(pathId, useTiering);
-        } else {
-            Tiers->DisablePathId(pathId);
-        }
+void TColumnShard::ActivateTiering(const ui64 pathId, const TString& useTiering) {
+    AFL_VERIFY(Tiers);
+    if (useTiering) {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "activate_tiering")("path_id", pathId)("tiering", useTiering);
     }
-    if (!onTabletInit) {
-        OnTieringModified();
+    if (useTiering) {
+        Tiers->EnablePathId(pathId, useTiering);
+    } else {
+        Tiers->DisablePathId(pathId);
     }
+    OnTieringModified(pathId);
 }
 
 void TColumnShard::Enqueue(STFUNC_SIG) {
@@ -1050,11 +1056,13 @@ void TColumnShard::Enqueue(STFUNC_SIG) {
     }
 }
 
-void TColumnShard::OnTieringModified() {
+void TColumnShard::OnTieringModified(const std::optional<ui64> pathId) {
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "OnTieringModified");
-    StoragesManager->OnTieringModified(Tiers);
-    if (TablesManager.HasPrimaryIndex()) {
-        TablesManager.MutablePrimaryIndex().OnTieringModified(Tiers, TablesManager.GetTtl());
+    if (Tiers->IsReady()) {
+        StoragesManager->OnTieringModified(Tiers);
+        if (TablesManager.HasPrimaryIndex()) {
+            TablesManager.MutablePrimaryIndex().OnTieringModified(Tiers, TablesManager.GetTtl(), pathId);
+        }
     }
 }
 

ydb/core/tx/columnshard/columnshard_impl.h

@@ -255,8 +255,8 @@ class TColumnShard
         TabletCounters->Cumulative()[counter].Increment(num);
     }
 
-    void ActivateTiering(const ui64 pathId, const TString& useTiering, const bool onTabletInit = false);
-    void OnTieringModified();
+    void ActivateTiering(const ui64 pathId, const TString& useTiering);
+    void OnTieringModified(const std::optional<ui64> pathId = {});
 public:
     enum class EOverloadStatus {
         ShardTxInFly /* "shard_tx" */,

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

@@ -4,6 +4,9 @@
 
 namespace NKikimr::NOlap {
 
+const std::shared_ptr<arrow::Schema>& IColumnEngine::GetReplaceKey() const {
+    return GetVersionedIndex().GetLastSchema()->GetIndexInfo().GetReplaceKey();
+}
 }
 
 template <>

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

@@ -361,7 +361,8 @@ class IColumnEngine {
     virtual ~IColumnEngine() = default;
 
     virtual const TVersionedIndex& GetVersionedIndex() const = 0;
-    virtual const std::shared_ptr<arrow::Schema>& GetReplaceKey() const { return GetVersionedIndex().GetLastSchema()->GetIndexInfo().GetReplaceKey(); }
+    virtual std::shared_ptr<TVersionedIndex> CopyVersionedIndexPtr() const = 0;
+    virtual const std::shared_ptr<arrow::Schema>& GetReplaceKey() const;
 
     virtual bool HasDataInPathId(const ui64 pathId) const = 0;
     virtual bool Load(IDbWrapper& db) = 0;
@@ -383,7 +384,7 @@ class IColumnEngine {
     virtual const TColumnEngineStats& GetTotalStats() = 0;
     virtual ui64 MemoryUsage() const { return 0; }
     virtual TSnapshot LastUpdate() const { return TSnapshot::Zero(); }
-    virtual void OnTieringModified(std::shared_ptr<NColumnShard::TTiersManager> manager, const NColumnShard::TTtl& ttl) = 0;
+    virtual void OnTieringModified(const std::shared_ptr<NColumnShard::TTiersManager>& manager, const NColumnShard::TTtl& ttl, const std::optional<ui64> pathId) = 0;
 };
 
 }

ydb/core/tx/columnshard/engines/column_engine_logs.cpp

@@ -573,17 +573,44 @@ std::shared_ptr<TSelectInfo> TColumnEngineForLogs::Select(ui64 pathId, TSnapshot
     return out;
 }
 
-void TColumnEngineForLogs::OnTieringModified(std::shared_ptr<NColumnShard::TTiersManager> manager, const NColumnShard::TTtl& ttl) {
-    AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "OnTieringModified");
-    std::optional<THashMap<ui64, TTiering>> tierings;
+void TColumnEngineForLogs::OnTieringModified(const std::shared_ptr<NColumnShard::TTiersManager>& manager, const NColumnShard::TTtl& ttl, const std::optional<ui64> pathId) {
+    if (!TiersInitialized) {
+        for (auto&& i : Tables) {
+            i.second->StartActualizationIndex();
+        }
+    }
+
+    TiersInitialized = true;
+    AFL_VERIFY(manager);
+    THashMap<ui64, TTiering> tierings;
     if (manager) {
         tierings = manager->GetTiering();
     }
+    ttl.AddTtls(tierings);
+
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "OnTieringModified")
-        ("new_count_tierings", tierings ? ::ToString(tierings->size()) : TString("undefined"))
+        ("new_count_tierings", tierings.size())
         ("new_count_ttls", ttl.PathsCount());
     EvictionsController.RefreshTierings(std::move(tierings), ttl);
-
+    if (pathId) {
+        auto itGranule = Tables.find(*pathId);
+        AFL_VERIFY(itGranule != Tables.end());
+        auto it = tierings.find(*pathId);
+        if (it == tierings.end()) {
+            itGranule->second->RefreshTiering({});
+        } else {
+            itGranule->second->RefreshTiering(it->second);
+        }
+    } else {
+        for (auto&& g : Tables) {
+            auto it = tierings.find(g.first);
+            if (it == tierings.end()) {
+                g.second->RefreshTiering({});
+            } else {
+                g.second->RefreshTiering(it->second);
+            }
+        }
+    }
 }
 
 void TColumnEngineForLogs::DoRegisterTable(const ui64 pathId) {
@@ -606,9 +633,7 @@ void TEvictionsController::RefreshTierings(std::optional<THashMap<ui64, TTiering
         OriginalTierings = std::move(*tierings);
     }
     auto copy = OriginalTierings;
-    if (!ttl.AddTtls(copy)) {
-        AFL_WARN(NKikimrServices::TX_COLUMNSHARD)("event", "Broken ttl");
-    }
+    ttl.AddTtls(copy);
     NextCheckInstantForTierings = BuildNextInstantCheckers(std::move(copy));
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "RefreshTierings")("count", NextCheckInstantForTierings.size());
 }

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

@@ -85,6 +85,7 @@ class TColumnEngineForLogs : public IColumnEngine {
     friend class TCleanupColumnEngineChanges;
     friend class NDataSharing::TDestinationSession;
 private:
+    bool TiersInitialized = false;
     const NColumnShard::TEngineLogsCounters SignalCounters;
     std::shared_ptr<TGranulesStorage> GranulesStorage;
     std::shared_ptr<IStoragesManager> StoragesManager;
@@ -148,7 +149,11 @@ class TColumnEngineForLogs : public IColumnEngine {
 
     TColumnEngineForLogs(ui64 tabletId, const TCompactionLimits& limits, const std::shared_ptr<IStoragesManager>& storagesManager);
 
-    virtual void OnTieringModified(std::shared_ptr<NColumnShard::TTiersManager> manager, const NColumnShard::TTtl& ttl) override;
+    virtual void OnTieringModified(const std::shared_ptr<NColumnShard::TTiersManager>& manager, const NColumnShard::TTtl& ttl, const std::optional<ui64> pathId) override;
+
+    virtual std::shared_ptr<TVersionedIndex> CopyVersionedIndexPtr() const override {
+        return std::make_shared<TVersionedIndex>(VersionedIndex);
+    }
 
     const TVersionedIndex& GetVersionedIndex() const override {
         return VersionedIndex;

ydb/core/tx/columnshard/engines/storage/granule.h

@@ -177,6 +177,12 @@ class TGranuleMeta: TNonCopyable {
     void OnAdditiveSummaryChange() const;
     YDB_READONLY(TMonotonic, LastCompactionInstant, TMonotonic::Zero());
 public:
+    void RefreshTiering(const std::optional<TTiering>& /*tiering*/) {
+    }
+
+    void StartActualizationIndex() {
+    }
+
     NJson::TJsonValue OptimizerSerializeToJson() const {
         return OptimizerPlanner->SerializeToJsonVisual();
     }

ydb/core/tx/columnshard/tables_manager.cpp

@@ -214,9 +214,6 @@ void TTablesManager::DropTable(const ui64 pathId, const NOlap::TSnapshot& versio
     table.SetDropVersion(version);
     PathsToDrop.insert(pathId);
     Ttl.DropPathTtl(pathId);
-    if (PrimaryIndex) {
-        PrimaryIndex->OnTieringModified(nullptr, Ttl);
-    }
     Schema::SaveTableDropVersion(db, pathId, version.GetPlanStep(), version.GetTxId());
 }
 
@@ -269,7 +266,7 @@ void TTablesManager::AddSchemaVersion(const ui32 presetId, const NOlap::TSnapsho
     }
 }
 
-void TTablesManager::AddTableVersion(const ui64 pathId, const NOlap::TSnapshot& version, const TTableInfo::TTableVersionInfo& versionInfo, NIceDb::TNiceDb& db) {
+void TTablesManager::AddTableVersion(const ui64 pathId, const NOlap::TSnapshot& version, const TTableInfo::TTableVersionInfo& versionInfo, NIceDb::TNiceDb& db, std::shared_ptr<TTiersManager>& manager) {
     auto it = Tables.find(pathId);
     AFL_VERIFY(it != Tables.end());
     auto& table = it->second;
@@ -296,7 +293,7 @@ void TTablesManager::AddTableVersion(const ui64 pathId, const NOlap::TSnapshot&
             Ttl.DropPathTtl(pathId);
         }
         if (PrimaryIndex) {
-            PrimaryIndex->OnTieringModified(nullptr, Ttl);
+            PrimaryIndex->OnTieringModified(manager, Ttl, pathId);
         }
     }
     Schema::SaveTableVersionInfo(db, pathId, version, versionInfo);
@@ -318,7 +315,6 @@ void TTablesManager::IndexSchemaVersion(const NOlap::TSnapshot& snapshot, const
             PrimaryIndex->RegisterTable(i.first);
         }
     }
-    PrimaryIndex->OnTieringModified(nullptr, Ttl);
 }
 
 TTablesManager::TTablesManager(const std::shared_ptr<NOlap::IStoragesManager>& storagesManager, const ui64 tabletId)

ydb/core/tx/columnshard/tables_manager.h

@@ -232,7 +232,7 @@ class TTablesManager {
     bool RegisterSchemaPreset(const TSchemaPreset& schemaPreset, NIceDb::TNiceDb& db);
 
     void AddSchemaVersion(const ui32 presetId, const NOlap::TSnapshot& version, const NKikimrSchemeOp::TColumnTableSchema& schema, NIceDb::TNiceDb& db);
-    void AddTableVersion(const ui64 pathId, const NOlap::TSnapshot& version, const TTableInfo::TTableVersionInfo& versionInfo, NIceDb::TNiceDb& db);
+    void AddTableVersion(const ui64 pathId, const NOlap::TSnapshot& version, const TTableInfo::TTableVersionInfo& versionInfo, NIceDb::TNiceDb& db, std::shared_ptr<TTiersManager>& manager);
     bool FillMonitoringReport(NTabletFlatExecutor::TTransactionContext& txc, NJson::TJsonValue& json);
 private:
     void IndexSchemaVersion(const NOlap::TSnapshot& version, const NKikimrSchemeOp::TColumnTableSchema& schema);

ydb/core/tx/tiering/manager.cpp

@@ -194,15 +194,14 @@ NMetadata::NFetcher::ISnapshotsFetcher::TPtr TTiersManager::GetExternalDataManip
 
 THashMap<ui64, NKikimr::NOlap::TTiering> TTiersManager::GetTiering() const {
     THashMap<ui64, NKikimr::NOlap::TTiering> result;
-    if (!IsReady()) {
-        return result;
-    }
+    AFL_VERIFY(IsReady());
     auto snapshotPtr = std::dynamic_pointer_cast<NTiers::TConfigsSnapshot>(Snapshot);
     Y_ABORT_UNLESS(snapshotPtr);
     auto& tierConfigs = snapshotPtr->GetTierConfigs();
     for (auto&& i : PathIdTiering) {
         auto* tiering = snapshotPtr->GetTieringById(i.second);
         if (tiering) {
+            AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("path_id", i.first)("tiering_name", i.second)("event", "activation");
             result.emplace(i.first, tiering->BuildOlapTiers());
             for (auto& [pathId, pathTiering] : result) {
                 for (auto& [name, tier] : pathTiering.GetTierByName()) {
@@ -213,6 +212,8 @@ THashMap<ui64, NKikimr::NOlap::TTiering> TTiersManager::GetTiering() const {
                     }
                 }
             }
+        } else {
+            AFL_ERROR(NKikimrServices::TX_COLUMNSHARD)("path_id", i.first)("tiering_name", i.second)("event", "not_found");
         }
     }
     return result;

ydb/core/tx/tiering/manager.h

@@ -85,18 +85,12 @@ class TTiersManager: public ITiersManager {
     TTiersManager& Start(std::shared_ptr<TTiersManager> ownerPtr);
     TTiersManager& Stop(const bool needStopActor);
     virtual const std::map<TString, NTiers::TManager>& GetManagers() const override {
+        AFL_VERIFY(IsReady());
         return Managers;
     }
     virtual const NTiers::TManager* GetManagerOptional(const TString& tierId) const override;
     NMetadata::NFetcher::ISnapshotsFetcher::TPtr GetExternalDataManipulation() const;
 
-    TManagers::const_iterator begin() const {
-        return Managers.begin();
-    }
-
-    TManagers::const_iterator end() const {
-        return Managers.end();
-    }
 };
 
 }