WIP

ydb/core/protos/flat_scheme_op.proto

@@ -1070,7 +1070,7 @@ message TAlterCdcStream {
 
 message TDropCdcStream {
     optional string TableName = 1;
-    optional string StreamName = 2;
+    repeated string StreamName = 2;
 }
 
 message TRotateCdcStream {

ydb/core/protos/tx_datashard.proto

@@ -433,7 +433,7 @@ message TAlterCdcStreamNotice {
 message TDropCdcStreamNotice {
     optional NKikimrProto.TPathID PathId = 1;
     optional uint64 TableSchemaVersion = 2;
-    optional NKikimrProto.TPathID StreamPathId = 3;
+    repeated NKikimrProto.TPathID StreamPathId = 3;
     optional TSnapshot DropSnapshot = 4;
 }
 

ydb/core/tx/datashard/drop_cdc_stream_unit.cpp

@@ -7,7 +7,7 @@ namespace NKikimr {
 namespace NDataShard {
 
 class TDropCdcStreamUnit : public TExecutionUnit {
-    THolder<TEvChangeExchange::TEvRemoveSender> RemoveSender;
+    TVector<THolder<TEvChangeExchange::TEvRemoveSender>> RemoveSenders;
 
 public:
     TDropCdcStreamUnit(TDataShard& self, TPipeline& pipeline)
@@ -35,12 +35,33 @@ class TDropCdcStreamUnit : public TExecutionUnit {
         const auto pathId = TPathId::FromProto(params.GetPathId());
         Y_ENSURE(pathId.OwnerId == DataShard.GetPathOwnerId());
 
-        const auto streamPathId = TPathId::FromProto(params.GetStreamPathId());
+        // Collect stream IDs to drop - works for both single and multiple
+        TVector<TPathId> streamPathIds;
+        for (const auto& streamId : params.GetStreamPathId()) {
+            streamPathIds.push_back(TPathId::FromProto(streamId));
+        }
 
         const auto version = params.GetTableSchemaVersion();
         Y_ENSURE(version);
 
-        auto tableInfo = DataShard.AlterTableDropCdcStream(ctx, txc, pathId, version, streamPathId);
+        // Process all streams atomically
+        TUserTable::TPtr tableInfo;
+        for (const auto& streamPathId : streamPathIds) {
+            tableInfo = DataShard.AlterTableDropCdcStream(ctx, txc, pathId, version, streamPathId);
+
+            auto& scanManager = DataShard.GetCdcStreamScanManager();
+            scanManager.Forget(txc.DB, pathId, streamPathId);
+            if (const auto* info = scanManager.Get(streamPathId)) {
+                DataShard.CancelScan(tableInfo->LocalTid, info->ScanId);
+                scanManager.Complete(streamPathId);
+            }
+
+            DataShard.GetCdcStreamHeartbeatManager().DropCdcStream(txc.DB, pathId, streamPathId);
+
+            RemoveSenders.emplace_back(new TEvChangeExchange::TEvRemoveSender(streamPathId));
+        }
+
+        // Update table info once after processing all streams
         TDataShardLocksDb locksDb(DataShard, txc);
         DataShard.AddUserTable(pathId, tableInfo, &locksDb);
 
@@ -56,28 +77,18 @@ class TDropCdcStreamUnit : public TExecutionUnit {
             DataShard.GetSnapshotManager().RemoveSnapshot(txc.DB, key);
         }
 
-        auto& scanManager = DataShard.GetCdcStreamScanManager();
-        scanManager.Forget(txc.DB, pathId, streamPathId);
-        if (const auto* info = scanManager.Get(streamPathId)) {
-            DataShard.CancelScan(tableInfo->LocalTid, info->ScanId);
-            scanManager.Complete(streamPathId);
-        }
-
-        DataShard.GetCdcStreamHeartbeatManager().DropCdcStream(txc.DB, pathId, streamPathId);
-
-        RemoveSender.Reset(new TEvChangeExchange::TEvRemoveSender(streamPathId));
-
         BuildResult(op, NKikimrTxDataShard::TEvProposeTransactionResult::COMPLETE);
         op->Result()->SetStepOrderId(op->GetStepOrder().ToPair());
 
         return EExecutionStatus::DelayCompleteNoMoreRestarts;
     }
 
     void Complete(TOperation::TPtr, const TActorContext& ctx) override {
-        if (RemoveSender) {
-            ctx.Send(DataShard.GetChangeSender(), RemoveSender.Release());
+        for (auto& removeSender : RemoveSenders) {
+            ctx.Send(DataShard.GetChangeSender(), removeSender.Release());
         }
     }
+
 };
 
 THolder<TExecutionUnit> CreateDropCdcStreamUnit(TDataShard& self, TPipeline& pipeline) {

ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp

@@ -1868,7 +1868,7 @@ ui64 AsyncAlterDropStream(
     auto request = SchemeTxTemplate(NKikimrSchemeOp::ESchemeOpDropCdcStream, workingDir);
     auto& desc = *request->Record.MutableTransaction()->MutableModifyScheme()->MutableDropCdcStream();
     desc.SetTableName(tableName);
-    desc.SetStreamName(streamName);
+    desc.AddStreamName(streamName);
 
     return RunSchemeTx(*server->GetRuntime(), std::move(request));
 }

ydb/core/tx/schemeshard/schemeshard__backup_collection_common.cpp

@@ -38,6 +38,12 @@ std::optional<NBackup::TBackupCollectionPaths> ResolveBackupCollectionPaths(
 
     const TString& backupCollectionsDir = JoinPath({rootPath.GetDomainPathString(), ".backups/collections"});
 
+    // Validate the collection name
+    if (name.empty()) {
+        result->SetError(NKikimrScheme::EStatus::StatusInvalidParameter, "Backup collection name cannot be empty");
+        return std::nullopt;
+    }
+
     TPathSplitUnix absPathSplit(name);
 
     if (absPathSplit.size() > 1 && !absPathSplit.IsAbsolute) {

ydb/core/tx/schemeshard/schemeshard__operation.cpp

@@ -1572,7 +1572,7 @@ TVector<ISubOperation::TPtr> TDefaultOperationFactory::MakeOperationParts(
     case NKikimrSchemeOp::EOperationType::ESchemeOpAlterBackupCollection:
         Y_ABORT("TODO: implement");
     case NKikimrSchemeOp::EOperationType::ESchemeOpDropBackupCollection:
-        return {CreateDropBackupCollection(op.NextPartId(), tx)};
+        return CreateDropBackupCollectionCascade(op.NextPartId(), tx, context);
 
     case NKikimrSchemeOp::EOperationType::ESchemeOpBackupBackupCollection:
         return CreateBackupBackupCollection(op.NextPartId(), tx, context);