check fill generation in drop volume transaction in schemeshard

7 files changed, 118 insertions, 14 deletions

diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
index ca4cc30defa..9f87bab80e8 100644
--- a/ydb/core/protos/flat_scheme_op.proto
+++ b/ydb/core/protos/flat_scheme_op.proto
@@ -1162,6 +1162,10 @@ message TBlockStoreAssignOp {
     optional uint64 TokenVersion = 3;
 }
 
+message TDropBlockStoreVolume { 
+    optional uint64 FillGeneration = 1;
+}
+
 message TModifyACL {
     optional string Name = 1;
     optional bytes DiffACL = 2; // NACLibProto.TDiffACL
@@ -1478,6 +1482,8 @@ message TModifyScheme {
     optional TExternalTableDescription CreateExternalTable = 58;
 
     optional TExternalDataSourceDescription CreateExternalDataSource = 59;
+
+    optional TDropBlockStoreVolume DropBlockStoreVolume = 60;
 }
 
 // "Script", used by client to parse text files with multiple DDL commands
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_bsv.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_bsv.cpp
index 60482c33d09..59d60cb0fe2 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_bsv.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_bsv.cpp
@@ -138,7 +138,7 @@ public:
     THolder<TProposeResponse> Propose(const TString&, TOperationContext& context) override {
         const TTabletId ssId = context.SS->SelfTabletId();
 
-        const auto& drop = Transaction.GetDrop();
+        const NKikimrSchemeOp::TDrop& drop = Transaction.GetDrop();
 
         const TString& parentPathStr = Transaction.GetWorkingDir();
         const TString& name = drop.GetName();
@@ -179,6 +179,25 @@ public:
             }
         }
 
+        TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes.at(path.Base()->PathId);
+        Y_VERIFY(volume);
+
+        {
+            const NKikimrSchemeOp::TDropBlockStoreVolume& dropParams = Transaction.GetDropBlockStoreVolume();
+
+            ui64 proposedFillGeneration = dropParams.GetFillGeneration();
+            ui64 actualFillGeneration = volume->VolumeConfig.GetFillGeneration();
+
+            if (proposedFillGeneration > 0 &&
+                    proposedFillGeneration < actualFillGeneration) {
+                result->SetError(NKikimrScheme::StatusSuccess,
+                                 TStringBuilder() << "Proposed fill generation "
+                                    << "is less than fill generation of the volume: "
+                                    << proposedFillGeneration << " < " << actualFillGeneration);
+                return result;
+            }
+        }
+
         TString errStr;
         if (!context.SS->CheckApplyIf(Transaction, errStr)) {
             result->SetError(NKikimrScheme::StatusPreconditionFailed, errStr);
@@ -222,9 +241,6 @@ public:
 
         NIceDb::TNiceDb db(context.GetDB());
 
-        TBlockStoreVolumeInfo::TPtr volume = context.SS->BlockStoreVolumes.at(path.Base()->PathId);
-        Y_VERIFY(volume);
-
         TVector<TShardIdx> shards;
         shards.push_back(volume->VolumeShardIdx);
 
diff --git a/ydb/core/tx/schemeshard/ut_base/ut_base.cpp b/ydb/core/tx/schemeshard/ut_base/ut_base.cpp
index 99016d088d8..c75c7838175 100644
--- a/ydb/core/tx/schemeshard/ut_base/ut_base.cpp
+++ b/ydb/core/tx/schemeshard/ut_base/ut_base.cpp
@@ -2999,7 +2999,6 @@ Y_UNIT_TEST_SUITE(TSchemeShardTest) {
         TestModificationResult(runtime, txId, NKikimrScheme::StatusMultipleModifications);
         env.TestWaitNotification(runtime, {txId-1, txId});
 
-
         TestDescribeResult(DescribePath(runtime, "/MyRoot/Table"),
                            {NLs::PathVersionEqual(4)});
         TestDescribeResult(DescribePath(runtime, "/MyRoot/Copy4"),
@@ -8069,7 +8068,7 @@ Y_UNIT_TEST_SUITE(TSchemeShardTest) {
         TTestEnv env(runtime);
         ui64 txId = 100;
 
-        TestDropBlockStoreVolume(runtime, ++txId, "/MyRoot", "BSVolume", {NKikimrScheme::StatusPathDoesNotExist});
+        TestDropBlockStoreVolume(runtime, ++txId, "/MyRoot", "BSVolume", 0, {NKikimrScheme::StatusPathDoesNotExist});
 
         // Create volume with 1 partition
         NKikimrSchemeOp::TBlockStoreVolumeDescription vdescr;
@@ -8175,6 +8174,67 @@ Y_UNIT_TEST_SUITE(TSchemeShardTest) {
                            {NLs::PathNotExist});
     }
 
+    Y_UNIT_TEST(DropBlockStoreVolumeWithFillGeneration) { //+
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime);
+        ui64 txId = 100;
+
+        auto createVolume = [&](const TString& volumeName) {
+            // Create volume with fill generation 713
+            NKikimrSchemeOp::TBlockStoreVolumeDescription vdescr;
+            vdescr.SetName(volumeName);
+            auto& vc = *vdescr.MutableVolumeConfig();
+            vc.SetBlockSize(4096);
+            vc.AddPartitions()->SetBlockCount(16);
+            vc.AddExplicitChannelProfiles()->SetPoolKind("pool-kind-1");
+            vc.AddExplicitChannelProfiles()->SetPoolKind("pool-kind-1");
+            vc.AddExplicitChannelProfiles()->SetPoolKind("pool-kind-1");
+            vc.AddExplicitChannelProfiles()->SetPoolKind("pool-kind-1");
+            vc.SetFillGeneration(713);
+
+            TestCreateBlockStoreVolume(runtime, ++txId, "/MyRoot", vdescr.DebugString());
+            env.TestWaitNotification(runtime, txId);
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/" + volumeName),
+                               {NLs::Finished, NLs::PathsInsideDomain(1), NLs::ShardsInsideDomain(2)});
+        };
+
+        auto successfullyDropVolume = [&](const TString& volumeName, ui64 fillGeneration) {
+            TestDropBlockStoreVolume(runtime, ++txId, "/MyRoot", volumeName, fillGeneration, {NKikimrScheme::StatusAccepted});
+            env.TestWaitNotification(runtime, txId);
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/" + volumeName),
+                               {NLs::PathNotExist});
+            env.TestWaitTabletDeletion(runtime, {TTestTxConfig::FakeHiveTablets, TTestTxConfig::FakeHiveTablets+1});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot"),
+                               {NLs::Finished, NLs::PathsInsideDomain(0), NLs::ShardsInsideDomain(0)});
+        };
+
+        auto failToDropVolume = [&](const TString& volumeName, ui64 fillGeneration) {
+            TestDropBlockStoreVolume(runtime, ++txId, "/MyRoot", volumeName, fillGeneration, {NKikimrScheme::StatusSuccess});
+            env.TestWaitNotification(runtime, txId);
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/" + volumeName),
+                               {NLs::Finished, NLs::PathsInsideDomain(1), NLs::ShardsInsideDomain(2)});
+        };
+
+        createVolume("BSVolume");
+        // Try to drop the volume using smaller fill generation
+        failToDropVolume("BSVolume", 1);
+        failToDropVolume("BSVolume", 712);
+        // Drop the volume using equal fill generation
+        successfullyDropVolume("BSVolume", 713);
+
+        createVolume("BSVolume_2");
+        // Drop the volume using greater fill generation
+        successfullyDropVolume("BSVolume_2", 714);
+
+        createVolume("BSVolume_3");
+        // Drop the volume using greater fill generation
+        successfullyDropVolume("BSVolume_3", 777);
+
+        createVolume("BSVolume_4");
+        // Drop the volume using zero fill generation (should be successful)
+        successfullyDropVolume("BSVolume_4", 0);
+    }
+
     Y_UNIT_TEST(AssignBlockStoreVolume) { //+
         TTestBasicRuntime runtime;
         TTestEnv env(runtime);
diff --git a/ydb/core/tx/schemeshard/ut_bsvolume/ut_bsvolume.cpp b/ydb/core/tx/schemeshard/ut_bsvolume/ut_bsvolume.cpp
index 4bcc51a200d..ce46b103eef 100644
--- a/ydb/core/tx/schemeshard/ut_bsvolume/ut_bsvolume.cpp
+++ b/ydb/core/tx/schemeshard/ut_bsvolume/ut_bsvolume.cpp
@@ -176,7 +176,7 @@ Y_UNIT_TEST_SUITE(TBSV) {
         TestCreateBlockStoreVolume(runtime, ++txId, root, descr.DebugString());
         env.TestWaitNotification(runtime, txId);
         // drop should be throttled
-        TestDropBlockStoreVolume(runtime, ++txId, root, name, {throttled});
+        TestDropBlockStoreVolume(runtime, ++txId, root, name, 0, {throttled});
         env.TestWaitNotification(runtime, txId);
 
         mockTimeProvider->Time = TInstant::Seconds(1);
@@ -188,7 +188,7 @@ Y_UNIT_TEST_SUITE(TBSV) {
         TestCreateBlockStoreVolume(runtime, ++txId, root, descr.DebugString());
         env.TestWaitNotification(runtime, txId);
         // next drop should be throttled
-        TestDropBlockStoreVolume(runtime, ++txId, root, name, {throttled});
+        TestDropBlockStoreVolume(runtime, ++txId, root, name, 0, {throttled});
         env.TestWaitNotification(runtime, txId);
 
         // turn off rate limiter
diff --git a/ydb/core/tx/schemeshard/ut_bsvolume_reboots/ut_bsvolume_reboots.cpp b/ydb/core/tx/schemeshard/ut_bsvolume_reboots/ut_bsvolume_reboots.cpp
index 22ac1554691..61da9343471 100644
--- a/ydb/core/tx/schemeshard/ut_bsvolume_reboots/ut_bsvolume_reboots.cpp
+++ b/ydb/core/tx/schemeshard/ut_bsvolume_reboots/ut_bsvolume_reboots.cpp
@@ -57,14 +57,16 @@ Y_UNIT_TEST_SUITE(TBSVWithReboots) {
             AsyncAssignBlockStoreVolume(runtime, ++t.TxId, "/MyRoot", "BSVolume", "Owner123");
             t.TestEnv->TestWaitNotification(runtime, t.TxId);
 
-            TestDropBlockStoreVolume(runtime, ++t.TxId, "/MyRoot", "BSVolume", {NKikimrScheme::StatusMultipleModifications, NKikimrScheme::StatusAccepted});
+            TestDropBlockStoreVolume(runtime, ++t.TxId, "/MyRoot", "BSVolume", 0,
+                                     {NKikimrScheme::StatusMultipleModifications, NKikimrScheme::StatusAccepted});
             t.TestEnv->TestWaitNotification(runtime, t.TxId);
 
             t.TestEnv->TestWaitNotification(runtime, t.TxId - 2); // wait Alter
 
             {
                 TInactiveZone inactive(activeZone);
-                TestDropBlockStoreVolume(runtime, ++t.TxId, "/MyRoot", "BSVolume", {NKikimrScheme::StatusPathDoesNotExist, NKikimrScheme::StatusAccepted});
+                TestDropBlockStoreVolume(runtime, ++t.TxId, "/MyRoot", "BSVolume", 0,
+                                         {NKikimrScheme::StatusPathDoesNotExist, NKikimrScheme::StatusAccepted});
                 t.TestEnv->TestWaitNotification(runtime, t.TxId);
                 TestDescribeResult(DescribePath(runtime, "/MyRoot/BSVolume"),
                                    {NLs::PathNotExist});
diff --git a/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp b/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp
index 6327604dc52..c75b27ba0ab 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp
+++ b/ydb/core/tx/schemeshard/ut_helpers/helpers.cpp
@@ -886,8 +886,6 @@ namespace NSchemeShardUT_Private {
     // nbs
     GENERIC_HELPERS(CreateBlockStoreVolume, NKikimrSchemeOp::EOperationType::ESchemeOpCreateBlockStoreVolume, &NKikimrSchemeOp::TModifyScheme::MutableCreateBlockStoreVolume)
     GENERIC_HELPERS(AlterBlockStoreVolume, NKikimrSchemeOp::EOperationType::ESchemeOpAlterBlockStoreVolume, &NKikimrSchemeOp::TModifyScheme::MutableAlterBlockStoreVolume)
-    GENERIC_HELPERS(DropBlockStoreVolume, NKikimrSchemeOp::EOperationType::ESchemeOpDropBlockStoreVolume, &NKikimrSchemeOp::TModifyScheme::MutableDrop)
-    DROP_BY_PATH_ID_HELPERS(DropBlockStoreVolume, NKikimrSchemeOp::EOperationType::ESchemeOpDropBlockStoreVolume)
 
     // external table
     GENERIC_HELPERS(CreateExternalTable, NKikimrSchemeOp::EOperationType::ESchemeOpCreateExternalTable, &NKikimrSchemeOp::TModifyScheme::MutableCreateExternalTable)
@@ -921,6 +919,28 @@ namespace NSchemeShardUT_Private {
         return TestUserAttrs(runtime, txId, parentPath, name, {NKikimrScheme::StatusAccepted}, userAttrs);
     }
 
+    void AsyncDropBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name,
+            ui64 fillGeneration)
+    {
+        auto evTx = new TEvSchemeShard::TEvModifySchemeTransaction(txId, TTestTxConfig::SchemeShard);
+        auto transaction = evTx->Record.AddTransaction();
+        transaction->SetWorkingDir(parentPath);
+        transaction->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpDropBlockStoreVolume);
+
+        transaction->MutableDrop()->SetName(name);
+
+        transaction->MutableDropBlockStoreVolume()->SetFillGeneration(fillGeneration);
+
+        AsyncSend(runtime, TTestTxConfig::SchemeShard, evTx);
+    }
+
+    void TestDropBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name,
+            ui64 fillGeneration, const TVector<TExpectedResult>& expectedResults)
+    {
+        AsyncDropBlockStoreVolume(runtime, txId, parentPath, name, fillGeneration);
+        TestModificationResults(runtime, txId, expectedResults);
+    }
+
     void AsyncAssignBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name,
             const TString& mountToken, ui64 tokenVersion)
     {
diff --git a/ydb/core/tx/schemeshard/ut_helpers/helpers.h b/ydb/core/tx/schemeshard/ut_helpers/helpers.h
index 83b2b03290d..d9dd10febd5 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/helpers.h
+++ b/ydb/core/tx/schemeshard/ut_helpers/helpers.h
@@ -252,8 +252,8 @@ namespace NSchemeShardUT_Private {
     // nbs
     GENERIC_HELPERS(CreateBlockStoreVolume);
     GENERIC_HELPERS(AlterBlockStoreVolume);
-    GENERIC_HELPERS(DropBlockStoreVolume);
-    DROP_BY_PATH_ID_HELPERS(DropBlockStoreVolume);
+    void AsyncDropBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name, ui64 fillGeneration = 0);
+    void TestDropBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name, ui64 fillGeneration = 0, const TVector<TExpectedResult>& expectedResults = {NKikimrScheme::StatusAccepted});
     void AsyncAssignBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name, const TString& mountToken, ui64 tokenVersion = 0);
     void TestAssignBlockStoreVolume(TTestActorRuntime& runtime, ui64 txId, const TString& parentPath, const TString& name, const TString& mountToken, ui64 tokenVersion = 0, const TVector<TExpectedResult>& expectedResults = {NKikimrScheme::StatusSuccess});