use external data sources as tiers in cs (ydb-platform#11581)

ydb/core/kqp/ut/common/columnshard.cpp

@@ -11,37 +11,20 @@ extern "C" {
 
 namespace NKikimr {
 namespace NKqp {
-
-    TString GetConfigProtoWithName(const TString & tierName) {
-        return TStringBuilder() << "Name : \"" << tierName << "\"\n" <<
-        R"(
-            ObjectStorage : {
-                Endpoint: "fake"
-                Bucket: "fake"
-                SecretableAccessKey: {
-                    Value: {
-                        Data: "secretAccessKey"
-                    }
-                }
-                SecretableSecretKey: {
-                    Value: {
-                        Data: "fakeSecret"
-                    }
-                }
-            }
-        )";
-    }
-
     using namespace NYdb;
 
     TTestHelper::TTestHelper(const TKikimrSettings& settings) {
         TKikimrSettings kikimrSettings(settings);
         if (!kikimrSettings.FeatureFlags.HasEnableTieringInColumnShard()) {
             kikimrSettings.SetEnableTieringInColumnShard(true);
         }
+        if (!kikimrSettings.FeatureFlags.HasEnableExternalDataSources()) {
+            kikimrSettings.SetEnableExternalDataSources(true);
+        }
 
         Kikimr = std::make_unique<TKikimrRunner>(kikimrSettings);
-        TableClient = std::make_unique<NYdb::NTable::TTableClient>(Kikimr->GetTableClient());
+        TableClient =
+            std::make_unique<NYdb::NTable::TTableClient>(Kikimr->GetTableClient(NYdb::NTable::TClientSettings().AuthToken("root@builtin")));
         Session = std::make_unique<NYdb::NTable::TSession>(TableClient->CreateSession().GetValueSync().GetSession());
     }
 
@@ -64,7 +47,18 @@ namespace NKqp {
     }
 
     void TTestHelper::CreateTier(const TString& tierName) {
-        auto result = GetSession().ExecuteSchemeQuery("CREATE OBJECT " + tierName + " (TYPE TIER) WITH tierConfig = `" + GetConfigProtoWithName(tierName) + "`").GetValueSync();
+        auto result = GetSession().ExecuteSchemeQuery(R"(
+            UPSERT OBJECT `accessKey` (TYPE SECRET) WITH (value = `secretAccessKey`);
+            UPSERT OBJECT `secretKey` (TYPE SECRET) WITH (value = `fakeSecret`);
+            CREATE EXTERNAL DATA SOURCE `)" + tierName + R"(` WITH (
+                SOURCE_TYPE="ObjectStorage",
+                LOCATION="http://fake.fake/fake",
+                AUTH_METHOD="AWS",
+                AWS_ACCESS_KEY_ID_SECRET_NAME="accessKey",
+                AWS_SECRET_ACCESS_KEY_SECRET_NAME="secretKey",
+                AWS_REGION="ru-central1"
+        );
+        )").GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
     }
 

ydb/core/kqp/ut/common/kqp_ut_common.h

@@ -160,9 +160,9 @@ class TKikimrRunner {
 
     NYdb::TDriverConfig GetDriverConfig() const { return DriverConfig; }
 
-    NYdb::NTable::TTableClient GetTableClient() const {
-        return NYdb::NTable::TTableClient(*Driver, NYdb::NTable::TClientSettings()
-            .UseQueryCache(false));
+    NYdb::NTable::TTableClient GetTableClient(
+        NYdb::NTable::TClientSettings settings = NYdb::NTable::TClientSettings()) const {
+        return NYdb::NTable::TTableClient(*Driver, settings.UseQueryCache(false));
     }
 
     NYdb::NQuery::TQueryClient GetQueryClient(

ydb/core/kqp/ut/olap/tiering_ut.cpp

@@ -68,7 +68,7 @@ class TTestEvictionBase {
             UNIT_ASSERT_GT(columnRawBytes, 0);
         }
 
-        TestHelper->SetTiering("/Root/olapStore/olapTable", "tier1", "timestamp");
+        TestHelper->SetTiering("/Root/olapStore/olapTable", "/Root/tier1", "timestamp");
         csController->WaitActualization(TDuration::Seconds(5));
 
         {
@@ -82,7 +82,7 @@ class TTestEvictionBase {
 
             auto rows = ExecuteScanQuery(tableClient, selectQuery);
             UNIT_ASSERT_VALUES_EQUAL(rows.size(), 1);
-            UNIT_ASSERT_VALUES_EQUAL(GetUtf8(rows[0].at("TierName")), "tier1");
+            UNIT_ASSERT_VALUES_EQUAL(GetUtf8(rows[0].at("TierName")), "/Root/tier1");
             UNIT_ASSERT_VALUES_EQUAL_C(GetUint64(rows[0].at("RawBytes")), columnRawBytes,
                 TStringBuilder() << "RawBytes changed after eviction: before=" << columnRawBytes
                                 << " after=" << GetUint64(rows[0].at("RawBytes")));
@@ -121,7 +121,7 @@ class TTestEvictionResetTiering : public TTestEvictionBase {
 class TTestEvictionIncreaseDuration : public TTestEvictionBase {
     private:
     void UnevictAll() {
-        const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P30000D") TO EXTERNAL DATA SOURCE tier1 ON timestamp)";
+        const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P30000D") TO EXTERNAL DATA SOURCE `/Root/tier1` ON timestamp)";
         auto result = TestHelper->GetSession().ExecuteSchemeQuery(query).GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
     }
@@ -152,18 +152,18 @@ Y_UNIT_TEST_SUITE(KqpOlapTiering) {
         testHelper.CreateTier("tier1");
 
         {
-            const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P10D") TO EXTERNAL DATA SOURCE tier1 ON unknown_column;)";
+            const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P10D") TO EXTERNAL DATA SOURCE `/Root/tier1` ON unknown_column;)";
             auto result = testHelper.GetSession().ExecuteSchemeQuery(query).GetValueSync();
             UNIT_ASSERT_VALUES_UNEQUAL(result.GetStatus(), NYdb::EStatus::SUCCESS);
         }
 
         {
-            const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P10D") TO EXTERNAL DATA SOURCE tier1 ON uid;)";
+            const TString query = R"(ALTER TABLE `/Root/olapStore/olapTable` SET TTL Interval("P10D") TO EXTERNAL DATA SOURCE `/Root/tier1` ON uid;)";
             auto result = testHelper.GetSession().ExecuteSchemeQuery(query).GetValueSync();
             UNIT_ASSERT_VALUES_UNEQUAL(result.GetStatus(), NYdb::EStatus::SUCCESS);
         }
 
-        testHelper.SetTiering("/Root/olapStore/olapTable", "tier1", "timestamp");
+        testHelper.SetTiering("/Root/olapStore/olapTable", "/Root/tier1", "timestamp");
     }
 }
 

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -4777,7 +4777,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             WITH (
                 STORE = COLUMN,
                 AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = 10,
-                TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE tier1 ON Key
+                TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE `/Root/tier1` ON Key
             );)";
         auto result = session.ExecuteSchemeQuery(query).GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
@@ -4788,12 +4788,12 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             UNIT_ASSERT(desc.GetTableDescription().GetTtlSettings());
             auto ttl = desc.GetTableDescription().GetTtlSettings();
             UNIT_ASSERT_VALUES_EQUAL(ttl->GetTiers().size(), 1);
-            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "tier1");
+            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "/Root/tier1");
             UNIT_ASSERT_VALUES_EQUAL(std::get<TDateTypeColumnModeSettings>(ttl->GetTiers()[0].GetExpression()).GetExpireAfter(), TDuration::Seconds(10));
         }
         auto query2 = TStringBuilder() << R"(
             --!syntax_v1
-            ALTER TABLE `)" << tableName << R"(` SET (TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE tier2 ON Key);)";
+            ALTER TABLE `)" << tableName << R"(` SET (TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE `/Root/tier2` ON Key);)";
         result = session.ExecuteSchemeQuery(query2).GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
 
@@ -4804,7 +4804,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             UNIT_ASSERT(desc.GetTableDescription().GetTtlSettings());
             auto ttl = desc.GetTableDescription().GetTtlSettings();
             UNIT_ASSERT_VALUES_EQUAL(ttl->GetTiers().size(), 1);
-            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "tier2");
+            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "/Root/tier2");
             UNIT_ASSERT_VALUES_EQUAL(std::get<TDateTypeColumnModeSettings>(ttl->GetTiers()[0].GetExpression()).GetExpireAfter(), TDuration::Seconds(10));
         }
 
@@ -4824,7 +4824,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
 
         auto query4 = TStringBuilder() << R"(
             --!syntax_v1
-            ALTER TABLE `)" << tableName << R"(` SET (TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE tier1 ON Key);)";
+            ALTER TABLE `)" << tableName << R"(` SET (TTL = Interval("PT10S") TO EXTERNAL DATA SOURCE `/Root/tier1` ON Key);)";
         result = session.ExecuteSchemeQuery(query4).GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
 
@@ -4835,7 +4835,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             UNIT_ASSERT(desc.GetTableDescription().GetTtlSettings());
             auto ttl = desc.GetTableDescription().GetTtlSettings();
             UNIT_ASSERT_VALUES_EQUAL(ttl->GetTiers().size(), 1);
-            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "tier1");
+            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(ttl->GetTiers()[0].GetAction()).GetStorage(), "/Root/tier1");
             UNIT_ASSERT_VALUES_EQUAL(std::get<TDateTypeColumnModeSettings>(ttl->GetTiers()[0].GetExpression()).GetExpireAfter(), TDuration::Seconds(10));
         }
 
@@ -7357,7 +7357,7 @@ Y_UNIT_TEST_SUITE(KqpOlapScheme) {
             testHelper.BulkUpsert(testTable, tableInserter);
         }
 
-        testHelper.SetTiering(tableName, "tier1", "created_at");
+        testHelper.SetTiering(tableName, "/Root/tier1", "created_at");
 
         while (csController->GetTieringUpdates().Val() == 0) {
             Cout << "Wait tiering..." << Endl;
@@ -7425,7 +7425,7 @@ Y_UNIT_TEST_SUITE(KqpOlapScheme) {
             UNIT_ASSERT_VALUES_EQUAL(description.GetTtlSettings()->GetDateTypeColumn().GetExpireAfter(), TDuration::Hours(1));
         }
         {
-            auto alterQuery = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "`SET (TTL = Interval(\"PT10S\") TO EXTERNAL DATA SOURCE tier1, Interval(\"PT1H\") DELETE ON created_at);";
+            auto alterQuery = TStringBuilder() << "ALTER TABLE `" << testTable.GetName() << "`SET (TTL = Interval(\"PT10S\") TO EXTERNAL DATA SOURCE `/Root/tier1`, Interval(\"PT1H\") DELETE ON created_at);";
             auto alterResult = testHelper.GetSession().ExecuteSchemeQuery(alterQuery).GetValueSync();
             UNIT_ASSERT_VALUES_EQUAL_C(alterResult.GetStatus(), EStatus::SUCCESS, alterResult.GetIssues().ToString());
         }
@@ -7440,7 +7440,7 @@ Y_UNIT_TEST_SUITE(KqpOlapScheme) {
             UNIT_ASSERT_VALUES_EQUAL(ttl->GetTiers().size(), 2);
             auto evictTier = ttl->GetTiers()[0];
             UNIT_ASSERT(std::holds_alternative<TTtlEvictToExternalStorageAction>(evictTier.GetAction()));
-            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(evictTier.GetAction()).GetStorage(), "tier1");
+            UNIT_ASSERT_VALUES_EQUAL(std::get<TTtlEvictToExternalStorageAction>(evictTier.GetAction()).GetStorage(), "/Root/tier1");
             UNIT_ASSERT_VALUES_EQUAL(std::get<TDateTypeColumnModeSettings>(evictTier.GetExpression()).GetExpireAfter(), TDuration::Seconds(10));
             auto deleteTier = ttl->GetTiers()[1];
             UNIT_ASSERT(std::holds_alternative<TTtlDeleteAction>(deleteTier.GetAction()));

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

@@ -54,6 +54,7 @@ class IBlobsStorageOperator {
     virtual void DoStartGCAction(const std::shared_ptr<IBlobsGCAction>& counters) const = 0;
 
     void StartGCAction(const std::shared_ptr<IBlobsGCAction>& action) const {
+        AFL_VERIFY(IsReady());
         return DoStartGCAction(action);
     }
 
@@ -96,14 +97,17 @@ class IBlobsStorageOperator {
     }
 
     std::shared_ptr<IBlobsDeclareRemovingAction> StartDeclareRemovingAction(const NBlobOperations::EConsumer consumerId) {
+        AFL_VERIFY(IsReady());
         return DoStartDeclareRemovingAction(Counters->GetConsumerCounter(consumerId)->GetRemoveDeclareCounters());
     }
     std::shared_ptr<IBlobsWritingAction> StartWritingAction(const NBlobOperations::EConsumer consumerId) {
+        AFL_VERIFY(IsReady());
         auto result = DoStartWritingAction();
         result->SetCounters(Counters->GetConsumerCounter(consumerId)->GetWriteCounters());
         return result;
     }
     std::shared_ptr<IBlobsReadingAction> StartReadingAction(const NBlobOperations::EConsumer consumerId) {
+        AFL_VERIFY(IsReady());
         auto result = DoStartReadingAction();
         result->SetCounters(Counters->GetConsumerCounter(consumerId)->GetReadCounters());
         return result;
@@ -129,6 +133,8 @@ class IBlobsStorageOperator {
         CurrentGCAction = task;
         return CurrentGCAction;
     }
+
+    virtual bool IsReady() const = 0;
 };
 
 }

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

@@ -41,6 +41,10 @@ class TOperator: public IBlobsStorageOperator {
     virtual std::shared_ptr<IBlobInUseTracker> GetBlobsTracker() const override {
         return Manager;
     }
+
+    virtual bool IsReady() const override {
+        return true;
+    }
 };
 
 }

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

@@ -48,6 +48,9 @@ class TOperator: public IBlobsStorageOperator {
         return false;
     }
 
+    virtual bool IsReady() const override {
+        return true;
+    }
 };
 
 }

ydb/core/tx/columnshard/blobs_action/tier/storage.cpp

@@ -54,12 +54,14 @@ void TOperator::DoStartGCAction(const std::shared_ptr<IBlobsGCAction>& action) c
 }
 
 void TOperator::InitNewExternalOperator(const NColumnShard::NTiers::TManager* tierManager) {
-    NKikimrSchemeOp::TS3Settings settings;
-    if (tierManager) {
-        settings = tierManager->GetS3Settings();
-    } else {
-        settings.SetEndpoint("nowhere");
+    if (!tierManager || !tierManager->IsReady()) {
+        TGuard<TSpinLock> changeLock(ChangeOperatorLock);
+        CurrentS3Settings.reset();
+        ExternalStorageOperator = nullptr;
+        return;
     }
+
+    NKikimrSchemeOp::TS3Settings settings = tierManager->GetS3Settings();
     {
         TGuard<TSpinLock> changeLock(ChangeOperatorLock);
         if (CurrentS3Settings && CurrentS3Settings->SerializeAsString() == settings.SerializeAsString()) {
@@ -103,12 +105,7 @@ TOperator::TOperator(const TString& storageId, const TActorId& shardActorId, con
 
 void TOperator::DoOnTieringModified(const std::shared_ptr<NColumnShard::ITiersManager>& tiers) {
     auto* tierManager = tiers->GetManagerOptional(TBase::GetStorageId());
-    if (tierManager) {
-        InitNewExternalOperator(tierManager);
-    } else {
-        TGuard<TSpinLock> changeLock(ChangeOperatorLock);
-        ExternalStorageOperator = nullptr;
-    }
+    InitNewExternalOperator(tierManager);
 }
 
 bool TOperator::DoLoad(IBlobManagerDb& dbBlobs) {

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

@@ -56,6 +56,9 @@ class TOperator: public IBlobsStorageOperator {
         return GCInfo->HasToDelete(blobId, tabletId);
     }
 
+    virtual bool IsReady() const override {
+        return !!ExternalStorageOperator;
+    }
 };
 
 }

ydb/core/tx/columnshard/columnshard.cpp

@@ -58,25 +58,27 @@ void TColumnShard::BecomeBroken(const TActorContext& ctx) {
     CleanupActors(ctx);
 }
 
-void TColumnShard::SwitchToWork(const TActorContext& ctx) {
+void TColumnShard::TrySwitchToWork(const TActorContext& ctx) {
+    if (!Tiers->AreConfigsComplete()) {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "skip_switch_to_work")("reason", "tiering_metadata_not_ready");
+        return;
+    }
+    if (!IsTxInitFinished) {
+        AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "skip_switch_to_work")("reason", "db_reading_not_finished");
+        return;
+    }
+
+    ProgressTxController->OnTabletInit();
     {
         const TLogContextGuard gLogging =
             NActors::TLogContextBuilder::Build(NKikimrServices::TX_COLUMNSHARD)("tablet_id", TabletID())("self_id", SelfId())("process", "SwitchToWork");
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "initialize_shard")("step", "SwitchToWork");
-
-        for (const auto& [pathId, tiering] : TablesManager.GetTtl()) {
-            THashSet<TString> tiers;
-            for (const auto& [name, config] : tiering.GetTierByName()) {
-                tiers.emplace(name);
-            }
-            ActivateTiering(pathId, tiers);
-        }
-
         Become(&TThis::StateWork);
         SignalTabletActive(ctx);
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("event", "initialize_shard")("step", "SignalTabletActive");
         TryRegisterMediatorTimeCast();
         EnqueueProgressTx(ctx, std::nullopt);
+        OnTieringModified();
     }
     Counters.GetCSCounters().OnIndexMetadataLimit(NOlap::IColumnEngine::GetMetadataLimit());
     EnqueueBackgroundActivities();
@@ -87,6 +89,7 @@ void TColumnShard::SwitchToWork(const TActorContext& ctx) {
     NYDBTest::TControllers::GetColumnShardController()->OnSwitchToWork(TabletID());
     AFL_VERIFY(!!StartInstant);
     Counters.GetCSCounters().Initialization.OnSwitchToWork(TMonotonic::Now() - *StartInstant, TMonotonic::Now() - CreateInstant);
+    NYDBTest::TControllers::GetColumnShardController()->OnTabletInitCompleted(*this);
 }
 
 void TColumnShard::OnActivateExecutor(const TActorContext& ctx) {
@@ -104,8 +107,10 @@ void TColumnShard::OnActivateExecutor(const TActorContext& ctx) {
         ctx.Send(selfActorId, new TEvPrivate::TEvTieringModified);
     });
     Tiers->Start(Tiers);
-    if (!NMetadata::NProvider::TServiceOperator::IsEnabled()) {
-        Tiers->TakeConfigs(NYDBTest::TControllers::GetColumnShardController()->GetFallbackTiersSnapshot(), nullptr);
+    if (const auto& tiersSnapshot = NYDBTest::TControllers::GetColumnShardController()->GetOverrideTierConfigs(); !tiersSnapshot.empty()) {
+        for (const auto& [id, tier] : tiersSnapshot) {
+            Tiers->UpdateTierConfig(tier, CanonizePath(id), false);
+        }
     }
     BackgroundSessionsManager = std::make_shared<NOlap::NBackground::TSessionsManager>(
         std::make_shared<NBackground::TAdapter>(selfActorId, (NOlap::TTabletId)TabletID(), *this));
@@ -124,10 +129,20 @@ void TColumnShard::OnActivateExecutor(const TActorContext& ctx) {
 }
 
 void TColumnShard::Handle(TEvPrivate::TEvTieringModified::TPtr& /*ev*/, const TActorContext& /*ctx*/) {
+    if (const auto& tiersSnapshot = NYDBTest::TControllers::GetColumnShardController()->GetOverrideTierConfigs(); !tiersSnapshot.empty()) {
+        for (const auto& [id, tier] : tiersSnapshot) {
+            Tiers->UpdateTierConfig(tier, CanonizePath(id), false);
+        }
+    }
+
     OnTieringModified();
     NYDBTest::TControllers::GetColumnShardController()->OnTieringModified(Tiers);
 }
 
+void TColumnShard::HandleInit(TEvPrivate::TEvTieringModified::TPtr& /*ev*/, const TActorContext& ctx) {
+    TrySwitchToWork(ctx);
+}
+
 void TColumnShard::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev, const TActorContext&) {
     auto tabletId = ev->Get()->TabletId;
     auto clientId = ev->Get()->ClientId;