diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
index 0a5fa2284df1..d993f8a6bffd 100644
--- a/ydb/core/protos/flat_scheme_op.proto
+++ b/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 {
diff --git a/ydb/core/protos/tx_datashard.proto b/ydb/core/protos/tx_datashard.proto
index 6ed96ed00efe..c995fbb3613a 100644
--- a/ydb/core/protos/tx_datashard.proto
+++ b/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;
 }
 
diff --git a/ydb/core/tx/datashard/drop_cdc_stream_unit.cpp b/ydb/core/tx/datashard/drop_cdc_stream_unit.cpp
index c07ec5bb1774..2feb8c4edcd5 100644
--- a/ydb/core/tx/datashard/drop_cdc_stream_unit.cpp
+++ b/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,17 +77,6 @@ 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());
 
@@ -74,10 +84,11 @@ class TDropCdcStreamUnit : public TExecutionUnit {
     }
 
     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) {
diff --git a/ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp b/ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp
index 907e4683e7fc..40b688a22778 100644
--- a/ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp
+++ b/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));
 }
diff --git a/ydb/core/tx/schemeshard/schemeshard__backup_collection_common.cpp b/ydb/core/tx/schemeshard/schemeshard__backup_collection_common.cpp
index 49262f5c3cb7..28c22af19ece 100644
--- a/ydb/core/tx/schemeshard/schemeshard__backup_collection_common.cpp
+++ b/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) {
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation.cpp b/ydb/core/tx/schemeshard/schemeshard__operation.cpp
index a2c0b6ed2fcb..3d415af7d3eb 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation.cpp
+++ b/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);
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_backup_collection.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_backup_collection.cpp
index 7d57ae56287a..caa7838d2fbd 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_backup_collection.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_backup_collection.cpp
@@ -1,6 +1,11 @@
 #include "schemeshard__backup_collection_common.h"
 #include "schemeshard__operation_common.h"
+#include "schemeshard__operation_part.h"
+#include "schemeshard__operation.h"
 #include "schemeshard_impl.h"
+#include "schemeshard_utils.h"
+
+#include <algorithm>
 
 #define LOG_I(stream) LOG_INFO_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "[" << context.SS->TabletID() << "] " << stream)
 #define LOG_N(stream) LOG_NOTICE_S(context.Ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "[" << context.SS->TabletID() << "] " << stream)
@@ -9,6 +14,180 @@ namespace NKikimr::NSchemeShard {
 
 namespace {
 
+struct TDropPlan {
+    struct TCdcStreamInfo {
+        TPathId TablePathId;
+        TString StreamName;
+        TString TablePath;
+    };
+    
+    struct TTableCdcStreams {
+        TPathId TablePathId;
+        TString TablePath;
+        TVector<TString> StreamNames;
+    };
+    
+    THashMap<TPathId, TTableCdcStreams> CdcStreamsByTable;  // Grouped by table
+    TVector<TPath> BackupTables;
+    TVector<TPath> BackupTopics;
+    TPathId BackupCollectionId;
+    
+    bool HasExternalObjects() const {
+        return !CdcStreamsByTable.empty() || !BackupTables.empty() || !BackupTopics.empty();
+    }
+};
+
+THolder<TDropPlan> CollectExternalObjects(TOperationContext& context, const TPath& bcPath) {
+    auto plan = MakeHolder<TDropPlan>();
+    plan->BackupCollectionId = bcPath.Base()->PathId;
+    
+    LOG_I("DropPlan: Starting collection for backup collection: " << bcPath.PathString());
+    
+    // 1. Find CDC streams on source tables (these are OUTSIDE the backup collection)
+    // Group them by table for efficient multi-stream drops
+    for (const auto& [pathId, cdcStreamInfo] : context.SS->CdcStreams) {
+        if (!context.SS->PathsById.contains(pathId)) {
+            continue;
+        }
+        
+        auto streamPath = context.SS->PathsById.at(pathId);
+        if (!streamPath || streamPath->Dropped()) {
+            continue;
+        }
+        
+        if (streamPath->Name.EndsWith("_continuousBackupImpl")) {
+            if (!context.SS->PathsById.contains(streamPath->ParentPathId)) {
+                continue;
+            }
+            
+            auto tablePath = context.SS->PathsById.at(streamPath->ParentPathId);
+            if (!tablePath || !tablePath->IsTable() || tablePath->Dropped()) {
+                continue;
+            }
+            
+            TString tablePathStr = TPath::Init(streamPath->ParentPathId, context.SS).PathString();
+            
+            auto& tableEntry = plan->CdcStreamsByTable[streamPath->ParentPathId];
+            if (tableEntry.StreamNames.empty()) {
+                tableEntry.TablePathId = streamPath->ParentPathId;
+                tableEntry.TablePath = tablePathStr;
+            }
+            tableEntry.StreamNames.push_back(streamPath->Name);
+        }
+    }
+    
+    // 2. Find backup tables and topics UNDER the collection path recursively
+    TVector<TPath> toVisit = {bcPath};
+    while (!toVisit.empty()) {
+        TPath current = toVisit.back();
+        toVisit.pop_back();
+        
+        for (const auto& [childName, childPathId] : current.Base()->GetChildren()) {
+            TPath childPath = current.Child(childName);
+            
+            if (childPath.Base()->IsTable()) {
+                plan->BackupTables.push_back(childPath);
+            } else if (childPath.Base()->IsPQGroup()) {
+                plan->BackupTopics.push_back(childPath);
+            } else if (childPath.Base()->IsDirectory()) {
+                toVisit.push_back(childPath);
+            }
+        }
+    }
+    
+    return plan;
+}
+
+TTxTransaction CreateCdcDropTransaction(const TDropPlan::TTableCdcStreams& tableStreams, TOperationContext& context) {
+    TTxTransaction cdcDropTx;
+    TPath tablePath = TPath::Init(tableStreams.TablePathId, context.SS);
+    cdcDropTx.SetWorkingDir(tablePath.Parent().PathString());
+    cdcDropTx.SetOperationType(NKikimrSchemeOp::ESchemeOpDropCdcStream);
+    
+    auto* cdcDrop = cdcDropTx.MutableDropCdcStream();
+    cdcDrop->SetTableName(tablePath.LeafName());
+    
+    // Add all streams for this table using the new repeated field functionality
+    for (const auto& streamName : tableStreams.StreamNames) {
+        cdcDrop->AddStreamName(streamName);
+    }
+    
+    return cdcDropTx;
+}
+
+TTxTransaction CreateTableDropTransaction(const TPath& tablePath) {
+    TTxTransaction tableDropTx;
+    tableDropTx.SetWorkingDir(tablePath.Parent().PathString());
+    tableDropTx.SetOperationType(NKikimrSchemeOp::ESchemeOpDropTable);
+    
+    auto* drop = tableDropTx.MutableDrop();
+    drop->SetName(tablePath.LeafName());
+    
+    return tableDropTx;
+}
+
+// TODO: replace UGLY scan
+// Clean up incremental restore state for a backup collection
+void CleanupIncrementalRestoreState(const TPathId& backupCollectionPathId, TOperationContext& context, NIceDb::TNiceDb& db) {
+    LOG_I("CleanupIncrementalRestoreState for backup collection pathId: " << backupCollectionPathId);
+    
+    // Find all incremental restore states for this backup collection
+    TVector<ui64> statesToCleanup;
+    
+    for (auto it = context.SS->IncrementalRestoreStates.begin(); it != context.SS->IncrementalRestoreStates.end();) {
+        if (it->second.BackupCollectionPathId == backupCollectionPathId) {
+            const auto& stateId = it->first;  // it->first is ui64 (state ID)
+            statesToCleanup.push_back(stateId);
+            
+            // Remove from memory
+            auto toErase = it;
+            ++it;
+            context.SS->IncrementalRestoreStates.erase(toErase);
+        } else {
+            ++it;
+        }
+    }
+    
+    // Clean up database entries
+    for (const auto& stateId : statesToCleanup) {
+        // Delete from IncrementalRestoreState table
+        db.Table<Schema::IncrementalRestoreState>().Key(stateId).Delete();
+        
+        // Delete all shard progress records for this state
+        auto shardProgressRowset = db.Table<Schema::IncrementalRestoreShardProgress>().Range().Select();
+        if (!shardProgressRowset.IsReady()) {
+            return;
+        }
+        
+        while (!shardProgressRowset.EndOfSet()) {
+            ui64 operationId = shardProgressRowset.GetValue<Schema::IncrementalRestoreShardProgress::OperationId>();
+            ui64 shardIdx = shardProgressRowset.GetValue<Schema::IncrementalRestoreShardProgress::ShardIdx>();
+            
+            if (operationId == stateId) {
+                db.Table<Schema::IncrementalRestoreShardProgress>().Key(operationId, shardIdx).Delete();
+            }
+            
+            if (!shardProgressRowset.Next()) {
+                break;
+            }
+        }
+    }
+    
+    // Clean up operation-to-state mappings
+    for (auto opIt = context.SS->IncrementalRestoreOperationToState.begin(); 
+         opIt != context.SS->IncrementalRestoreOperationToState.end();) {
+        if (std::find(statesToCleanup.begin(), statesToCleanup.end(), opIt->second) != statesToCleanup.end()) {
+            auto toErase = opIt;
+            ++opIt;
+            context.SS->IncrementalRestoreOperationToState.erase(toErase);
+        } else {
+            ++opIt;
+        }
+    }
+    
+    LOG_I("CleanupIncrementalRestoreState: Cleaned up " << statesToCleanup.size() << " incremental restore states");
+}
+
 class TPropose : public TSubOperationState {
 public:
     explicit TPropose(TOperationId id)
@@ -45,9 +224,12 @@ class TPropose : public TSubOperationState {
         context.SS->PersistDropStep(db, pathId, step, OperationId);
         context.SS->PersistRemoveBackupCollection(db, pathId);
 
+        // Clean up incremental restore state for this backup collection
+        CleanupIncrementalRestoreState(pathId, context, db);
+
         auto domainInfo = context.SS->ResolveDomainInfo(pathId);
         domainInfo->DecPathsInside(context.SS);
-        DecAliveChildrenDirect(OperationId, parentDirPtr, context); // for correct discard of ChildrenExist prop
+        DecAliveChildrenDirect(OperationId, parentDirPtr, context);
         context.SS->TabletCounters->Simple()[COUNTER_BACKUP_COLLECTION_COUNT].Sub(1);
 
         ++parentDirPtr->DirAlterVersion;
@@ -119,6 +301,43 @@ class TDropBackupCollection : public TSubOperation {
         context.DbChanges.PersistPath(dstPath->ParentPathId);
     }
 
+    // TODO: replace UGLY scan
+    bool HasActiveBackupOperations(const TPath& bcPath, TOperationContext& context) const {
+        // Check if there are any active backup or restore operations for this collection
+        const TPathId& bcPathId = bcPath.Base()->PathId;
+        
+        // Check all active transactions to see if any involve this backup collection
+        for (const auto& [txId, txState] : context.SS->TxInFlight) {
+            if (txState.TxType == TTxState::TxBackup ||
+                txState.TxType == TTxState::TxRestore ||
+                txState.TxType == TTxState::TxCopyTable) {
+                
+                // Check if the transaction target is this backup collection or a child path
+                const TPathId& targetPathId = txState.TargetPathId;
+                if (targetPathId == bcPathId) {
+                    return true;
+                }
+                
+                // Check if target is a child of this backup collection
+                if (context.SS->PathsById.contains(targetPathId)) {
+                    auto targetPath = context.SS->PathsById.at(targetPathId);
+                    TPathId currentId = targetPathId;
+                    
+                    // Walk up the path hierarchy to check if bcPathId is an ancestor
+                    while (currentId && context.SS->PathsById.contains(currentId)) {
+                        if (currentId == bcPathId) {
+                            return true;
+                        }
+                        auto currentPath = context.SS->PathsById.at(currentId);
+                        currentId = currentPath->ParentPathId;
+                    }
+                }
+            }
+        }
+        
+        return false;
+    }
+
 public:
     using TSubOperation::TSubOperation;
 
@@ -137,7 +356,7 @@ class TDropBackupCollection : public TSubOperation {
             return result;
         }
 
-        auto& [_, dstPath] = *bcPaths;
+        auto& dstPath = bcPaths->DstPath;
 
         {
             auto checks = dstPath.Check();
@@ -156,9 +375,35 @@ class TDropBackupCollection : public TSubOperation {
                 const TBackupCollectionInfo::TPtr backupCollection = context.SS->BackupCollections.Value(dstPath->PathId, nullptr);
                 if (!backupCollection) {
                     result->SetError(NKikimrScheme::StatusSchemeError, "Backup collection doesn't exist");
+                    return result;
+                }
 
+                // Check for active backup/restore operations
+                if (HasActiveBackupOperations(dstPath, context)) {
+                    result->SetError(NKikimrScheme::StatusPreconditionFailed, 
+                                   "Cannot drop backup collection while backup or restore operations are active. Please wait for them to complete.");
                     return result;
                 }
+                
+                // Check for concurrent operations on the same path
+                // This catches the race condition where another drop operation is in progress
+                // but hasn't yet marked the path as "under deleting"
+                
+                for (const auto& [txId, txState] : context.SS->TxInFlight) {
+                    LOG_I("DropPlan: Found TxInFlight - txId: " << txId.GetTxId() 
+                         << ", targetPathId: " << txState.TargetPathId 
+                         << ", txType: " << (int)txState.TxType);
+                         
+                    if (txState.TargetPathId == dstPath.Base()->PathId && 
+                        txId.GetTxId() != OperationId.GetTxId()) {
+                        result->SetError(NKikimrScheme::StatusMultipleModifications, 
+                                       TStringBuilder() << "Check failed: path: '" << dstPath.PathString() 
+                                       << "', error: another operation is already in progress for this backup collection");
+                        result->SetPathDropTxId(ui64(txId.GetTxId()));
+                        result->SetPathId(dstPath.Base()->PathId.LocalPathId);
+                        return result;
+                    }
+                }
             }
 
             if (!checks) {
@@ -207,10 +452,185 @@ class TDropBackupCollection : public TSubOperation {
     }
 };
 
+// Cleanup operation for incremental restore state
+class TIncrementalRestoreCleanup : public TSubOperationState {
+public:
+    explicit TIncrementalRestoreCleanup(TOperationId id, TPathId backupCollectionPathId)
+        : OperationId(std::move(id))
+        , BackupCollectionPathId(backupCollectionPathId)
+    {}
+
+    bool ProgressState(TOperationContext& context) override {
+        LOG_I(DebugHint() << "ProgressState");
+
+        NIceDb::TNiceDb db(context.GetDB());
+        
+        // Clean up incremental restore state for this backup collection
+        TVector<ui64> operationsToCleanup;
+        
+        for (const auto& [opId, restoreState] : context.SS->IncrementalRestoreStates) {
+            if (restoreState.BackupCollectionPathId == BackupCollectionPathId) {
+                operationsToCleanup.push_back(opId);
+            }
+        }
+        
+        for (ui64 opId : operationsToCleanup) {
+            LOG_I(DebugHint() << "Cleaning up incremental restore state for operation: " << opId);
+            
+            // Remove from database
+            db.Table<Schema::IncrementalRestoreOperations>()
+                .Key(opId)
+                .Delete();
+            
+            // Remove from in-memory state
+            context.SS->IncrementalRestoreStates.erase(opId);
+            
+            // Clean up related mappings
+            auto txIt = context.SS->TxIdToIncrementalRestore.begin();
+            while (txIt != context.SS->TxIdToIncrementalRestore.end()) {
+                if (txIt->second == opId) {
+                    auto toErase = txIt;
+                    ++txIt;
+                    context.SS->TxIdToIncrementalRestore.erase(toErase);
+                } else {
+                    ++txIt;
+                }
+            }
+            
+            auto opIt = context.SS->IncrementalRestoreOperationToState.begin();
+            while (opIt != context.SS->IncrementalRestoreOperationToState.end()) {
+                if (opIt->second == opId) {
+                    auto toErase = opIt;
+                    ++opIt;
+                    context.SS->IncrementalRestoreOperationToState.erase(toErase);
+                } else {
+                    ++opIt;
+                }
+            }
+        }
+        
+        LOG_I(DebugHint() << "Cleaned up " << operationsToCleanup.size() << " incremental restore operations");
+        
+        return true;
+    }
+
+    bool HandleReply(TEvPrivate::TEvOperationPlan::TPtr&, TOperationContext&) override {
+        return true;
+    }
+
+private:
+    TOperationId OperationId;
+    TPathId BackupCollectionPathId;
+    
+    TString DebugHint() const override {
+        return TStringBuilder()
+            << "TIncrementalRestoreCleanup"
+            << " operationId: " << OperationId;
+    }
+};
+
 }  // anonymous namespace
 
-ISubOperation::TPtr CreateDropBackupCollection(TOperationId id, const TTxTransaction& tx) {
-    return MakeSubOperation<TDropBackupCollection>(id, tx);
+// Create multiple suboperations for dropping backup collection
+TVector<ISubOperation::TPtr> CreateDropBackupCollectionCascade(TOperationId nextId, const TTxTransaction& tx, TOperationContext& context) {
+    Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::EOperationType::ESchemeOpDropBackupCollection);
+
+    auto dropOperation = tx.GetDropBackupCollection();
+    const TString parentPathStr = tx.GetWorkingDir();
+    
+    // Check for empty backup collection name
+    if (dropOperation.GetName().empty()) {
+        return {CreateReject(nextId, NKikimrScheme::StatusInvalidParameter, "Backup collection name cannot be empty")};
+    }
+    
+    // Use the same validation logic as ResolveBackupCollectionPaths to be consistent
+    auto proposeResult = MakeHolder<TProposeResponse>(NKikimrScheme::StatusAccepted, static_cast<ui64>(nextId.GetTxId()), static_cast<ui64>(context.SS->SelfTabletId()));
+    auto bcPaths = NBackup::ResolveBackupCollectionPaths(parentPathStr, dropOperation.GetName(), false, context, proposeResult);
+    if (!bcPaths) {
+        return {CreateReject(nextId, proposeResult->Record.GetStatus(), proposeResult->Record.GetReason())};
+    }
+
+    auto& dstPath = bcPaths->DstPath;
+
+    {
+        TPath::TChecker checks = dstPath.Check();
+        checks
+            .NotEmpty()
+            .IsResolved()
+            .NotDeleted()
+            .IsBackupCollection()
+            .NotUnderDeleting()
+            .NotUnderOperation()
+            .IsCommonSensePath();
+
+        if (!checks) {
+            // TODO: is there more clean way to write it?
+            // Handle the special case where the path is being deleted
+            if (dstPath.IsResolved() && dstPath.Base()->IsBackupCollection() && 
+                (dstPath.Base()->PlannedToDrop() || dstPath.Base()->Dropped())) {
+                
+                auto errorResult = MakeHolder<TProposeResponse>(checks.GetStatus(), static_cast<ui64>(nextId.GetTxId()), static_cast<ui64>(context.SS->SelfTabletId()));
+                errorResult->SetError(checks.GetStatus(), checks.GetError());
+                errorResult->SetPathDropTxId(ui64(dstPath.Base()->DropTxId));
+                errorResult->SetPathId(dstPath.Base()->PathId.LocalPathId);
+                return {CreateReject(nextId, std::move(errorResult))};
+            } else {
+                return {CreateReject(nextId, checks.GetStatus(), checks.GetError())};
+            }
+        }
+    }
+
+    const TPathId& pathId = dstPath.Base()->PathId;
+    
+    // Check if any backup or restore operations are active for this collection
+    for (const auto& [txId, txState] : context.SS->TxInFlight) {
+        if (txState.TargetPathId == pathId && 
+            (txState.TxType == TTxState::TxBackup || 
+             txState.TxType == TTxState::TxRestore)) {
+            return {CreateReject(nextId, NKikimrScheme::StatusPreconditionFailed,
+                "Cannot drop backup collection while backup or restore operations are active. Please wait for them to complete.")};
+        }
+    }
+    
+    // Check for active incremental restore operations in IncrementalRestoreStates
+    for (const auto& [opId, restoreState] : context.SS->IncrementalRestoreStates) {
+        if (restoreState.BackupCollectionPathId == pathId) {
+            return {CreateReject(nextId, NKikimrScheme::StatusPreconditionFailed,
+                "Cannot drop backup collection while incremental restore operations are active. Please wait for them to complete.")};
+        }
+    }
+
+    TVector<ISubOperation::TPtr> result;
+
+    auto cleanupOp = MakeSubOperation<TDropBackupCollection>(nextId, tx);
+    result.push_back(cleanupOp);
+
+    auto dropPlan = CollectExternalObjects(context, dstPath);
+    if (dropPlan->HasExternalObjects()) {
+        // Create suboperations for CDC streams - grouped by table for atomic multi-stream drops
+        for (const auto& [tableId, tableStreams] : dropPlan->CdcStreamsByTable) {
+            TStringBuilder streamList;
+            for (size_t i = 0; i < tableStreams.StreamNames.size(); ++i) {
+                if (i > 0) streamList << ", ";
+                streamList << tableStreams.StreamNames[i];
+            }
+            
+            TTxTransaction cdcDropTx = CreateCdcDropTransaction(tableStreams, context);
+            if (!CreateDropCdcStream(nextId, cdcDropTx, context, result)) {
+                return result;
+            }
+        }
+
+        // Create suboperations for backup tables
+        for (const auto& tablePath : dropPlan->BackupTables) {
+            TTxTransaction tableDropTx = CreateTableDropTransaction(tablePath);
+            if (!CreateDropTable(nextId, tableDropTx, context, result)) {
+                return result;
+            }
+        }
+    }
+    
+    return result;
 }
 
 ISubOperation::TPtr CreateDropBackupCollection(TOperationId id, TTxState::ETxState state) {
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.cpp
index 8ee7fb4134fe..593de4172b0a 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.cpp
@@ -129,9 +129,32 @@ class TDropCdcStream: public TSubOperation {
                 .IsAtLocalSchemeShard()
                 .IsResolved()
                 .NotDeleted()
-                .IsCdcStream()
-                .NotUnderDeleting()
-                .NotUnderOperation();
+                .IsCdcStream();
+
+            // Allow processing streams that are being deleted/operated on by the same transaction
+            // (coordinated multi-stream drop within same transaction)
+            bool isSameTransaction = false;
+            
+            if (streamPath.Base()->PathState == TPathElement::EPathState::EPathStateDrop) {
+                // Check if the stream is being dropped by the same transaction
+                isSameTransaction = (streamPath.Base()->DropTxId == OperationId.GetTxId());
+                if (!isSameTransaction) {
+                    checks.NotUnderDeleting();
+                }
+            }
+            
+            // Check if stream is under operation by same transaction
+            // Allow if it's any suboperation of the same transaction
+            if (streamPath.Base()->LastTxId != InvalidTxId) {
+                if (streamPath.Base()->LastTxId != OperationId.GetTxId()) {
+                    checks.NotUnderOperation();
+                }
+            } else {
+                // No LastTxId set, check normal operation state
+                if (!isSameTransaction) {
+                    checks.NotUnderOperation();
+                }
+            }
 
             if (!checks) {
                 result->SetError(checks.GetStatus(), checks.GetError());
@@ -210,7 +233,8 @@ class TConfigurePartsAtTable: public NCdcStreamState::TConfigurePartsAtTable {
         pathId.ToProto(notice.MutablePathId());
         notice.SetTableSchemaVersion(table->AlterVersion + 1);
 
-        bool found = false;
+        // Collect all streams planned for drop on this table
+        TVector<TPathId> streamPathIds;
         for (const auto& [_, childPathId] : path->GetChildren()) {
             Y_ABORT_UNLESS(context.SS->PathsById.contains(childPathId));
             auto childPath = context.SS->PathsById.at(childPathId);
@@ -218,13 +242,13 @@ class TConfigurePartsAtTable: public NCdcStreamState::TConfigurePartsAtTable {
             if (!childPath->IsCdcStream() || !childPath->PlannedToDrop()) {
                 continue;
             }
-
-            Y_VERIFY_S(!found, "Too many cdc streams are planned to drop"
-                << ": found# " << TPathId::FromProto(notice.GetStreamPathId())
-                << ", another# " << childPathId);
-            found = true;
-
-            childPathId.ToProto(notice.MutableStreamPathId());
+            streamPathIds.push_back(childPathId);
+        }
+        
+        Y_VERIFY_S(!streamPathIds.empty(), "No CDC streams planned for drop");
+        
+        for (const auto& streamId : streamPathIds) {
+            streamId.ToProto(notice.AddStreamPathId());
         }
     }
 
@@ -304,6 +328,11 @@ class TDropCdcStreamAtTable: public TSubOperation {
         : TSubOperation(id, tx)
         , DropSnapshot(dropSnapshot)
     {
+        // Extract all stream names from transaction
+        const auto& op = tx.GetDropCdcStream();
+        for (const auto& name : op.GetStreamName()) {
+            StreamNames.push_back(name);
+        }
     }
 
     explicit TDropCdcStreamAtTable(TOperationId id, TTxState::ETxState state, bool dropSnapshot)
@@ -316,11 +345,11 @@ class TDropCdcStreamAtTable: public TSubOperation {
         const auto& workingDir = Transaction.GetWorkingDir();
         const auto& op = Transaction.GetDropCdcStream();
         const auto& tableName = op.GetTableName();
-        const auto& streamName = op.GetStreamName();
 
         LOG_N("TDropCdcStreamAtTable Propose"
             << ": opId# " << OperationId
-            << ", stream# " << workingDir << "/" << tableName << "/" << streamName);
+            << ", table# " << workingDir << "/" << tableName 
+            << ", streams# " << StreamNames.size());
 
         auto result = MakeHolder<TProposeResponse>(NKikimrScheme::StatusAccepted, ui64(OperationId.GetTxId()), context.SS->TabletID());
 
@@ -348,8 +377,10 @@ class TDropCdcStreamAtTable: public TSubOperation {
             }
         }
 
-        const auto streamPath = tablePath.Child(streamName);
-        {
+        // Validate all streams exist and are on same table
+        TVector<TPath> streamPaths;
+        for (const auto& streamName : StreamNames) {
+            const auto streamPath = tablePath.Child(streamName);
             const auto checks = streamPath.Check();
             checks
                 .NotEmpty()
@@ -365,6 +396,7 @@ class TDropCdcStreamAtTable: public TSubOperation {
                 result->SetError(checks.GetStatus(), checks.GetError());
                 return result;
             }
+            streamPaths.push_back(streamPath);
         }
 
         TString errStr;
@@ -397,11 +429,22 @@ class TDropCdcStreamAtTable: public TSubOperation {
         Y_ABORT_UNLESS(table->AlterVersion != 0);
         Y_ABORT_UNLESS(!table->AlterData);
 
-        Y_ABORT_UNLESS(context.SS->CdcStreams.contains(streamPath.Base()->PathId));
-        auto stream = context.SS->CdcStreams.at(streamPath.Base()->PathId);
-
-        Y_ABORT_UNLESS(stream->AlterVersion != 0);
-        Y_ABORT_UNLESS(!stream->AlterData);
+        // Validate and mark all streams for drop in single transaction
+        for (const auto& streamPath : streamPaths) {
+            Y_ABORT_UNLESS(context.SS->CdcStreams.contains(streamPath.Base()->PathId));
+            auto stream = context.SS->CdcStreams.at(streamPath.Base()->PathId);
+
+            Y_ABORT_UNLESS(stream->AlterVersion != 0);
+            Y_ABORT_UNLESS(!stream->AlterData);
+
+            streamPath.Base()->PathState = TPathElement::EPathState::EPathStateDrop;
+            streamPath.Base()->DropTxId = OperationId.GetTxId();
+            streamPath.Base()->LastTxId = OperationId.GetTxId();
+            
+            context.SS->TabletCounters->Simple()[COUNTER_CDC_STREAMS_COUNT].Sub(1);
+            context.SS->ClearDescribePathCaches(streamPath.Base());
+            context.OnComplete.PublishToSchemeBoard(OperationId, streamPath.Base()->PathId);
+        }
 
         const auto txType = DropSnapshot
             ? TTxState::TxDropCdcStreamAtTableDropSnapshot
@@ -437,6 +480,7 @@ class TDropCdcStreamAtTable: public TSubOperation {
     }
 
 private:
+    TVector<TString> StreamNames;  // All streams in this operation
     const bool DropSnapshot;
 
 }; // TDropCdcStreamAtTable
@@ -514,13 +558,13 @@ void DoDropStream(
         const TOperationId& opId,
         const TPath& workingDirPath,
         const TPath& tablePath,
-        const TPath& streamPath,
+        const TVector<TPath>& streamPaths,  // Now handles multiple streams
         const TTxId lockTxId,
         TOperationContext& context)
 {
     {
         auto outTx = TransactionTemplate(workingDirPath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStreamAtTable);
-        outTx.MutableDropCdcStream()->CopyFrom(op);
+        outTx.MutableDropCdcStream()->CopyFrom(op);  // Preserves all stream names
 
         if (lockTxId != InvalidTxId) {
             outTx.MutableLockGuard()->SetOwnerTxId(ui64(lockTxId));
@@ -547,7 +591,7 @@ void DoDropStream(
         result.push_back(CreateAlterTableIndex(NextPartId(opId, result), outTx));
     }
 
-    {
+    for (const auto& streamPath : streamPaths) {
         auto outTx = TransactionTemplate(tablePath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStreamImpl);
         outTx.MutableDrop()->SetName(streamPath.Base()->Name);
 
@@ -558,21 +602,23 @@ void DoDropStream(
         result.push_back(CreateDropCdcStreamImpl(NextPartId(opId, result), outTx));
     }
 
-    for (const auto& [name, pathId] : streamPath.Base()->GetChildren()) {
-        Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId));
-        auto implPath = context.SS->PathsById.at(pathId);
+    for (const auto& streamPath : streamPaths) {
+        for (const auto& [name, pathId] : streamPath.Base()->GetChildren()) {
+            Y_ABORT_UNLESS(context.SS->PathsById.contains(pathId));
+            auto implPath = context.SS->PathsById.at(pathId);
 
-        if (implPath->Dropped()) {
-            continue;
-        }
+            if (implPath->Dropped()) {
+                continue;
+            }
 
-        auto streamImpl = context.SS->PathsById.at(pathId);
-        Y_ABORT_UNLESS(streamImpl->IsPQGroup());
+            auto streamImpl = context.SS->PathsById.at(pathId);
+            Y_ABORT_UNLESS(streamImpl->IsPQGroup());
 
-        auto outTx = TransactionTemplate(streamPath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpDropPersQueueGroup);
-        outTx.MutableDrop()->SetName(name);
+            auto outTx = TransactionTemplate(streamPath.PathString(), NKikimrSchemeOp::EOperationType::ESchemeOpDropPersQueueGroup);
+            outTx.MutableDrop()->SetName(name);
 
-        result.push_back(CreateDropPQ(NextPartId(opId, result), outTx));
+            result.push_back(CreateDropPQ(NextPartId(opId, result), outTx));
+        }
     }
 }
 
@@ -596,44 +642,92 @@ ISubOperation::TPtr CreateDropCdcStreamAtTable(TOperationId id, TTxState::ETxSta
     return MakeSubOperation<TDropCdcStreamAtTable>(id, state, dropSnapshot);
 }
 
-TVector<ISubOperation::TPtr> CreateDropCdcStream(TOperationId opId, const TTxTransaction& tx, TOperationContext& context) {
+bool CreateDropCdcStream(TOperationId opId, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result) {
     Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStream);
 
+    const auto& op = tx.GetDropCdcStream();
+    const auto& tableName = op.GetTableName();
+    
+    TVector<TString> streamNames;
+    for (const auto& name : op.GetStreamName()) {
+        streamNames.push_back(name);
+    }
+    
     LOG_D("CreateDropCdcStream"
         << ": opId# " << opId
+        << ", table# " << tableName
+        << ", streams# " << streamNames.size()
         << ", tx# " << tx.ShortDebugString());
 
-    const auto& op = tx.GetDropCdcStream();
-    const auto& tableName = op.GetTableName();
-    const auto& streamName = op.GetStreamName();
-
     const auto workingDirPath = TPath::Resolve(tx.GetWorkingDir(), context.SS);
 
-    const auto checksResult = DoDropStreamPathChecks(opId, workingDirPath, tableName, streamName);
-    if (std::holds_alternative<ISubOperation::TPtr>(checksResult)) {
-        return {std::get<ISubOperation::TPtr>(checksResult)};
+    // Validate all streams exist on the same table
+    TVector<TPath> streamPaths;
+    
+    if (streamNames.empty()) {
+        result = {CreateReject(opId, NKikimrScheme::StatusInvalidParameter, 
+                             "At least one StreamName must be specified")};
+        return false;
+    }
+    
+    const auto firstStreamChecksResult = DoDropStreamPathChecks(opId, workingDirPath, tableName, streamNames[0]);
+    if (std::holds_alternative<ISubOperation::TPtr>(firstStreamChecksResult)) {
+        result = {std::get<ISubOperation::TPtr>(firstStreamChecksResult)};
+        return false;
     }
+    
+    const auto [tablePath, firstStreamPath] = std::get<TStreamPaths>(firstStreamChecksResult);
+    streamPaths.push_back(firstStreamPath);
+    
+    for (size_t i = 1; i < streamNames.size(); ++i) {
+        const auto checksResult = DoDropStreamPathChecks(opId, workingDirPath, tableName, streamNames[i]);
+        if (std::holds_alternative<ISubOperation::TPtr>(checksResult)) {
+            result = {std::get<ISubOperation::TPtr>(checksResult)};
+            return false;
+        }
 
-    const auto [tablePath, streamPath] = std::get<TStreamPaths>(checksResult);
+        const auto [currentTablePath, streamPath] = std::get<TStreamPaths>(checksResult);
+        streamPaths.push_back(streamPath);
+    }
 
     TString errStr;
     if (!context.SS->CheckApplyIf(tx, errStr)) {
-        return {CreateReject(opId, NKikimrScheme::StatusPreconditionFailed, errStr)};
+        result = {CreateReject(opId, NKikimrScheme::StatusPreconditionFailed, errStr)};
+        return false;
     }
 
-    Y_ABORT_UNLESS(context.SS->CdcStreams.contains(streamPath.Base()->PathId));
-    auto stream = context.SS->CdcStreams.at(streamPath.Base()->PathId);
+    // Check lock consistency across all streams
+    TTxId lockTxId = InvalidTxId;
+    for (const auto& streamPath : streamPaths) {
+        Y_ABORT_UNLESS(context.SS->CdcStreams.contains(streamPath.Base()->PathId));
+        auto stream = context.SS->CdcStreams.at(streamPath.Base()->PathId);
+
+        const auto streamLockTxId = stream->State == TCdcStreamInfo::EState::ECdcStreamStateScan
+            ? streamPath.Base()->CreateTxId : InvalidTxId;
+            
+        if (lockTxId == InvalidTxId) {
+            lockTxId = streamLockTxId;
+        } else if (lockTxId != streamLockTxId) {
+            result = {CreateReject(opId, NKikimrScheme::StatusPreconditionFailed, 
+                                 "Cannot drop CDC streams with different lock states in single operation")};
+            return false;
+        }
+    }
 
-    const auto lockTxId = stream->State == TCdcStreamInfo::EState::ECdcStreamStateScan
-        ? streamPath.Base()->CreateTxId
-        : InvalidTxId;
     if (const auto reject = DoDropStreamChecks(opId, tablePath, lockTxId, context); reject) {
-        return {reject};
+        result = {reject};
+        return false;
     }
 
+    DoDropStream(result, op, opId, workingDirPath, tablePath, streamPaths, lockTxId, context);
+
+    return true;
+}
+
+TVector<ISubOperation::TPtr> CreateDropCdcStream(TOperationId opId, const TTxTransaction& tx, TOperationContext& context) {
     TVector<ISubOperation::TPtr> result;
 
-    DoDropStream(result, op, opId, workingDirPath, tablePath, streamPath, lockTxId, context);
+    CreateDropCdcStream(opId, tx, context, result);
 
     return result;
 }
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.h b/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.h
index 12be7102684c..0f2ff591cdcb 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.h
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_cdc_stream.h
@@ -28,7 +28,7 @@ void DoDropStream(
     const TOperationId& opId,
     const TPath& workingDirPath,
     const TPath& tablePath,
-    const TPath& streamPath,
+    const TVector<TPath>& streamPaths,
     const TTxId lockTxId,
     TOperationContext& context);
 
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_continuous_backup.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_continuous_backup.cpp
index 214feccd4aa9..4452d1e7e404 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_continuous_backup.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_continuous_backup.cpp
@@ -59,9 +59,10 @@ TVector<ISubOperation::TPtr> CreateDropContinuousBackup(TOperationId opId, const
 
             NKikimrSchemeOp::TDropCdcStream dropCdcStreamOp;
             dropCdcStreamOp.SetTableName(tableName);
-            dropCdcStreamOp.SetStreamName(child);
+            dropCdcStreamOp.AddStreamName(child);
 
-            NCdc::DoDropStream(result, dropCdcStreamOp, opId, workingDirPath, tablePath, streamPath, InvalidTxId, context);
+            TVector<TPath> streamPaths = {streamPath};
+            NCdc::DoDropStream(result, dropCdcStreamOp, opId, workingDirPath, tablePath, streamPaths, InvalidTxId, context);
         }
     }
 
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_pq.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_pq.cpp
index b1ac1914dcdc..9782254da75e 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_pq.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_pq.cpp
@@ -1,5 +1,6 @@
 #include "schemeshard__operation_common.h"
 #include "schemeshard__operation_part.h"
+#include "schemeshard__operation.h"
 #include "schemeshard_impl.h"
 #include "schemeshard_utils.h"  // for PQGroupReserve
 
@@ -470,4 +471,12 @@ ISubOperation::TPtr CreateDropPQ(TOperationId id, TTxState::ETxState state) {
     return MakeSubOperation<TDropPQ>(id, state);
 }
 
+bool CreateDropPQ(TOperationId id, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result) {
+    Y_UNUSED(context);
+    Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::EOperationType::ESchemeOpDropPersQueueGroup);
+
+    result.push_back(CreateDropPQ(NextPartId(id, result), tx));
+    return true;
+}
+
 }
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_drop_table.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_drop_table.cpp
index e73282493496..c9ca6d0f9312 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_drop_table.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_drop_table.cpp
@@ -1,5 +1,6 @@
 #include "schemeshard__operation_common.h"
 #include "schemeshard__operation_part.h"
+#include "schemeshard__operation.h"
 #include "schemeshard_impl.h"
 
 #include <ydb/core/base/subdomain.h>
@@ -634,4 +635,12 @@ ISubOperation::TPtr CreateDropTable(TOperationId id, TTxState::ETxState state) {
     return MakeSubOperation<TDropTable>(id, state);
 }
 
+bool CreateDropTable(TOperationId id, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result) {
+    Y_UNUSED(context);
+    Y_ABORT_UNLESS(tx.GetOperationType() == NKikimrSchemeOp::EOperationType::ESchemeOpDropTable);
+
+    result.push_back(CreateDropTable(NextPartId(id, result), tx));
+    return true;
+}
+
 }
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_part.h b/ydb/core/tx/schemeshard/schemeshard__operation_part.h
index 99271caa43a4..c5801aaf0b6f 100644
--- a/ydb/core/tx/schemeshard/schemeshard__operation_part.h
+++ b/ydb/core/tx/schemeshard/schemeshard__operation_part.h
@@ -385,6 +385,7 @@ ISubOperation::TPtr CreateSplitMerge(TOperationId id, TTxState::ETxState state);
 
 ISubOperation::TPtr CreateDropTable(TOperationId id, const TTxTransaction& tx);
 ISubOperation::TPtr CreateDropTable(TOperationId id, TTxState::ETxState state);
+bool CreateDropTable(TOperationId id, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result);
 
 TVector<ISubOperation::TPtr> CreateBuildColumn(TOperationId id, const TTxTransaction& tx, TOperationContext& context);
 
@@ -444,6 +445,7 @@ ISubOperation::TPtr CreateAlterCdcStreamAtTable(TOperationId id, const TTxTransa
 ISubOperation::TPtr CreateAlterCdcStreamAtTable(TOperationId id, TTxState::ETxState state, bool dropSnapshot);
 // Drop
 TVector<ISubOperation::TPtr> CreateDropCdcStream(TOperationId id, const TTxTransaction& tx, TOperationContext& context);
+bool CreateDropCdcStream(TOperationId id, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result);
 ISubOperation::TPtr CreateDropCdcStreamImpl(TOperationId id, const TTxTransaction& tx);
 ISubOperation::TPtr CreateDropCdcStreamImpl(TOperationId id, TTxState::ETxState state);
 ISubOperation::TPtr CreateDropCdcStreamAtTable(TOperationId id, const TTxTransaction& tx, bool dropSnapshot);
@@ -521,6 +523,7 @@ ISubOperation::TPtr CreateAlterPQ(TOperationId id, TTxState::ETxState state);
 
 ISubOperation::TPtr CreateDropPQ(TOperationId id, const TTxTransaction& tx);
 ISubOperation::TPtr CreateDropPQ(TOperationId id, TTxState::ETxState state);
+bool CreateDropPQ(TOperationId id, const TTxTransaction& tx, TOperationContext& context, TVector<ISubOperation::TPtr>& result);
 
 ISubOperation::TPtr CreateAllocatePQ(TOperationId id, const TTxTransaction& tx);
 ISubOperation::TPtr CreateAllocatePQ(TOperationId id, TTxState::ETxState state);
@@ -689,6 +692,7 @@ ISubOperation::TPtr CreateNewBackupCollection(TOperationId id, TTxState::ETxStat
 // Drop
 ISubOperation::TPtr CreateDropBackupCollection(TOperationId id, const TTxTransaction& tx);
 ISubOperation::TPtr CreateDropBackupCollection(TOperationId id, TTxState::ETxState state);
+TVector<ISubOperation::TPtr> CreateDropBackupCollectionCascade(TOperationId nextId, const TTxTransaction& tx, TOperationContext& context);
 // Restore
 TVector<ISubOperation::TPtr> CreateRestoreBackupCollection(TOperationId opId, const TTxTransaction& tx, TOperationContext& context);
 ISubOperation::TPtr CreateLongIncrementalRestoreOpControlPlane(TOperationId opId, const TTxTransaction& tx);
diff --git a/ydb/core/tx/schemeshard/schemeshard_audit_log_fragment.cpp b/ydb/core/tx/schemeshard/schemeshard_audit_log_fragment.cpp
index 85c8326d8d37..62c9be2752dc 100644
--- a/ydb/core/tx/schemeshard/schemeshard_audit_log_fragment.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard_audit_log_fragment.cpp
@@ -497,7 +497,13 @@ TVector<TString> ExtractChangingPaths(const NKikimrSchemeOp::TModifyScheme& tx)
         result.emplace_back(NKikimr::JoinPath({tx.GetWorkingDir(), tx.GetAlterCdcStream().GetTableName()}));
         break;
     case NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStream:
-        result.emplace_back(NKikimr::JoinPath({tx.GetWorkingDir(), tx.GetDropCdcStream().GetTableName(), tx.GetDropCdcStream().GetStreamName()}));
+        {
+            const auto& tablePath = NKikimr::JoinPath({tx.GetWorkingDir(), tx.GetDropCdcStream().GetTableName()});
+            // Add entry for each stream being dropped
+            for (const auto& streamName : tx.GetDropCdcStream().GetStreamName()) {
+                result.emplace_back(NKikimr::JoinPath({tx.GetWorkingDir(), tx.GetDropCdcStream().GetTableName(), streamName}));
+            }
+        }
         break;
     case NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStreamImpl:
         result.emplace_back(NKikimr::JoinPath({tx.GetWorkingDir(), tx.GetDrop().GetName()}));
diff --git a/ydb/core/tx/schemeshard/ut_backup_collection/ut_backup_collection.cpp b/ydb/core/tx/schemeshard/ut_backup_collection/ut_backup_collection.cpp
index 2b8d1c79ec31..57c1426378b1 100644
--- a/ydb/core/tx/schemeshard/ut_backup_collection/ut_backup_collection.cpp
+++ b/ydb/core/tx/schemeshard/ut_backup_collection/ut_backup_collection.cpp
@@ -1,4 +1,7 @@
 #include <ydb/core/tx/schemeshard/ut_helpers/helpers.h>
+#include <ydb/core/protos/schemeshard/operations.pb.h>
+#include <library/cpp/string_utils/base64/base64.h>
+#include <util/string/printf.h>
 
 #define DEFAULT_NAME_1 "MyCollection1"
 #define DEFAULT_NAME_2 "MyCollection2"
@@ -38,6 +41,20 @@ Y_UNIT_TEST_SUITE(TBackupCollectionTests) {
         )", name.c_str());
     }
 
+    TString DefaultCollectionSettingsWithName(const TString& name) {
+        return Sprintf(R"(
+            Name: "%s"
+
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/Table1"
+                }
+            }
+            Cluster: {}
+        )", name.c_str());
+    }
+
     void PrepareDirs(TTestBasicRuntime& runtime, TTestEnv& env, ui64& txId) {
         TestMkDir(runtime, ++txId, "/MyRoot", ".backups");
         env.TestWaitNotification(runtime, txId);
@@ -45,6 +62,41 @@ Y_UNIT_TEST_SUITE(TBackupCollectionTests) {
         env.TestWaitNotification(runtime, txId);
     }
 
+    void AsyncBackupBackupCollection(TTestBasicRuntime& runtime, ui64 txId, const TString& workingDir, const TString& request) {
+        auto modifyTx = std::make_unique<TEvSchemeShard::TEvModifySchemeTransaction>(txId, TTestTxConfig::SchemeShard);
+        auto transaction = modifyTx->Record.AddTransaction();
+        transaction->SetWorkingDir(workingDir);
+        transaction->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpBackupBackupCollection);
+        
+        bool parseOk = ::google::protobuf::TextFormat::ParseFromString(request, transaction->MutableBackupBackupCollection());
+        UNIT_ASSERT(parseOk);
+        
+        AsyncSend(runtime, TTestTxConfig::SchemeShard, modifyTx.release(), 0);
+    }
+
+    void TestBackupBackupCollection(TTestBasicRuntime& runtime, ui64 txId, const TString& workingDir, const TString& request, const TExpectedResult& expectedResult = {NKikimrScheme::StatusAccepted}) {
+        AsyncBackupBackupCollection(runtime, txId, workingDir, request);
+        TestModificationResults(runtime, txId, {expectedResult});
+    }
+
+    void AsyncBackupIncrementalBackupCollection(TTestBasicRuntime& runtime, ui64 txId, const TString& workingDir, const TString& request) {
+        TActorId sender = runtime.AllocateEdgeActor();
+        
+        auto request2 = MakeHolder<TEvSchemeShard::TEvModifySchemeTransaction>(txId, TTestTxConfig::SchemeShard);
+        auto transaction = request2->Record.AddTransaction();
+        transaction->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpBackupIncrementalBackupCollection);
+        transaction->SetWorkingDir(workingDir);
+        bool parseOk = ::google::protobuf::TextFormat::ParseFromString(request, transaction->MutableBackupIncrementalBackupCollection());
+        UNIT_ASSERT(parseOk);
+        
+        AsyncSend(runtime, TTestTxConfig::SchemeShard, request2.Release(), 0, sender);
+    }
+
+    ui64 TestBackupIncrementalBackupCollection(TTestBasicRuntime& runtime, ui64 txId, const TString& workingDir, const TString& request, const TExpectedResult& expectedResult = {NKikimrScheme::StatusAccepted}) {
+        AsyncBackupIncrementalBackupCollection(runtime, txId, workingDir, request);
+        return TestModificationResults(runtime, txId, {expectedResult});
+    }
+
     Y_UNIT_TEST(HiddenByFeatureFlag) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions());
@@ -60,310 +112,1286 @@ Y_UNIT_TEST_SUITE(TBackupCollectionTests) {
 
         TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
             NLs::PathNotExist,
-        });
+            });
 
-        // must not be there in any case, smoke test
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/" DEFAULT_NAME_1), {
-            NLs::PathNotExist,
-        });
-    }
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/" DEFAULT_NAME_1), {
+                NLs::PathNotExist,
+            });
+        }
 
-    Y_UNIT_TEST(DisallowedPath) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+        Y_UNIT_TEST(DisallowedPath) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
 
-        SetupLogging(runtime);
+            SetupLogging(runtime);
 
-        PrepareDirs(runtime, env, txId);
+            PrepareDirs(runtime, env, txId);
+
+            {
+                TestCreateBackupCollection(runtime, ++txId, "/MyRoot", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+
+                env.TestWaitNotification(runtime, txId);
+
+                TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                    NLs::PathNotExist,
+                });
+
+                TestDescribeResult(DescribePath(runtime, "/MyRoot/" DEFAULT_NAME_1), {
+                    NLs::PathNotExist,
+                });
+            }
+
+            {
+                TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+
+                env.TestWaitNotification(runtime, txId);
+
+                TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                    NLs::PathNotExist,
+                });
+
+                TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/" DEFAULT_NAME_1), {
+                    NLs::PathNotExist,
+                });
+            }
+
+            {
+                TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", CollectionSettings("SomePrefix/MyCollection1"), {NKikimrScheme::EStatus::StatusSchemeError});
+
+                env.TestWaitNotification(runtime, txId);
+
+                TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/SomePrefix/MyCollection1"), {
+                    NLs::PathNotExist,
+                });
+            }
+        }
 
-        {
-            TestCreateBackupCollection(runtime, ++txId, "/MyRoot", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+        Y_UNIT_TEST(CreateAbsolutePath) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot", CollectionSettings("/MyRoot/.backups/collections/" DEFAULT_NAME_1));
 
             env.TestWaitNotification(runtime, txId);
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-                NLs::PathNotExist,
+                NLs::PathExist,
+                NLs::IsBackupCollection,
             });
+        }
 
-            TestDescribeResult(DescribePath(runtime, "/MyRoot/" DEFAULT_NAME_1), {
-                NLs::PathNotExist,
+        Y_UNIT_TEST(Create) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
             });
         }
 
-        {
-            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+        Y_UNIT_TEST(CreateTwice) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
 
             env.TestWaitNotification(runtime, txId);
 
             TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-                NLs::PathNotExist,
+                NLs::PathExist,
+                NLs::IsBackupCollection,
             });
 
-            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/" DEFAULT_NAME_1), {
-                NLs::PathNotExist,
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+
+            env.TestWaitNotification(runtime, txId);
+        }
+
+        Y_UNIT_TEST(ParallelCreate) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            PrepareDirs(runtime, env, txId);
+
+            AsyncCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", CollectionSettings(DEFAULT_NAME_1));
+            AsyncCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", CollectionSettings(DEFAULT_NAME_2));
+            TestModificationResult(runtime, txId - 1, NKikimrScheme::StatusAccepted);
+            TestModificationResult(runtime, txId, NKikimrScheme::StatusAccepted);
+
+            env.TestWaitNotification(runtime, {txId, txId - 1});
+
+            TestDescribe(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1);
+            TestDescribe(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_2);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections"),
+                               {NLs::PathVersionEqual(7)});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                               {NLs::PathVersionEqual(1)});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_2),
+                               {NLs::PathVersionEqual(1)});
+        }
+
+        Y_UNIT_TEST(Drop) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathExist);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathNotExist);
+        }
+
+        Y_UNIT_TEST(DropTwice) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            AsyncDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            AsyncDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            TestModificationResult(runtime, txId - 1);
+
+            auto ev = runtime.GrabEdgeEvent<TEvSchemeShard::TEvModifySchemeTransactionResult>();
+            UNIT_ASSERT(ev);
+
+            const auto& record = ev->Record;
+            UNIT_ASSERT_VALUES_EQUAL(record.GetTxId(), txId);
+            UNIT_ASSERT_VALUES_EQUAL(record.GetStatus(), NKikimrScheme::StatusMultipleModifications);
+            UNIT_ASSERT_VALUES_EQUAL(record.GetPathDropTxId(), txId - 1);
+
+            env.TestWaitNotification(runtime, txId - 1);
+            TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathNotExist);
+        }
+
+        Y_UNIT_TEST(TableWithSystemColumns) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
             });
+
+            TestCreateTable(runtime, ++txId, "/MyRoot/.backups/collections", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Utf8" }
+                Columns { Name: "__ydb_system_column" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: ".backups/collections/Table2"
+                Columns { Name: "key" Type: "Utf8" }
+                Columns { Name: "__ydb_system_column" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot/.backups/collections", R"(
+                Name: "somepath/Table3"
+                Columns { Name: "key" Type: "Utf8" }
+                Columns { Name: "__ydb_system_column" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
         }
 
-        {
-            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", CollectionSettings("SomePrefix/MyCollection1"), {NKikimrScheme::EStatus::StatusSchemeError});
+        Y_UNIT_TEST(BackupAbsentCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
 
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
+                {NKikimrScheme::EStatus::StatusPathDoesNotExist});
             env.TestWaitNotification(runtime, txId);
+        }
 
-            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/SomePrefix/MyCollection1"), {
-                NLs::PathNotExist,
+        Y_UNIT_TEST(BackupDroppedCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
+                {NKikimrScheme::EStatus::StatusPathDoesNotExist});
+            env.TestWaitNotification(runtime, txId);
+        }
+
+        Y_UNIT_TEST(BackupAbsentDirs) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
+                {NKikimrScheme::EStatus::StatusPathDoesNotExist});
+            env.TestWaitNotification(runtime, txId);
+        }
+
+        Y_UNIT_TEST(BackupNonIncrementalCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupIncrementalBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
+                {NKikimrScheme::EStatus::StatusInvalidParameter});
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+                NLs::ChildrenCount(1),
+                NLs::Finished,
             });
         }
-    }
 
-    Y_UNIT_TEST(CreateAbsolutePath) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+        Y_UNIT_TEST(DropEmptyBackupCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
 
-        SetupLogging(runtime);
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
 
-        PrepareDirs(runtime, env, txId);
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot", CollectionSettings("/MyRoot/.backups/collections/" DEFAULT_NAME_1));
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
 
-        env.TestWaitNotification(runtime, txId);
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-        });
-    }
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
 
-    Y_UNIT_TEST(Create) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+        Y_UNIT_TEST(DropCollectionWithFullBackup) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
 
-        SetupLogging(runtime);
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
 
-        PrepareDirs(runtime, env, txId);
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
 
-        env.TestWaitNotification(runtime, txId);
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-        });
-    }
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+                NLs::ChildrenCount(1),
+            });
 
-    Y_UNIT_TEST(CreateTwice) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
 
-        SetupLogging(runtime);
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
 
-        PrepareDirs(runtime, env, txId);
+        Y_UNIT_TEST(DropCollectionWithIncrementalBackup) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
 
-        env.TestWaitNotification(runtime, txId);
+            TString collectionSettingsWithIncremental = R"(
+                Name: ")" DEFAULT_NAME_1 R"("
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-        });
+                ExplicitEntryList {
+                    Entries {
+                        Type: ETypeTable
+                        Path: "/MyRoot/Table1"
+                    }
+                }
+                Cluster: {}
+                IncrementalBackupConfig: {}
+            )";
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings(), {NKikimrScheme::EStatus::StatusSchemeError});
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", collectionSettingsWithIncremental);
+            env.TestWaitNotification(runtime, txId);
 
-        env.TestWaitNotification(runtime, txId);
-    }
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
 
-    Y_UNIT_TEST(ParallelCreate) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
 
-        PrepareDirs(runtime, env, txId);
+            runtime.AdvanceCurrentTime(TDuration::Seconds(1));
 
-        AsyncCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", CollectionSettings(DEFAULT_NAME_1));
-        AsyncCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", CollectionSettings(DEFAULT_NAME_2));
-        TestModificationResult(runtime, txId - 1, NKikimrScheme::StatusAccepted);
-        TestModificationResult(runtime, txId, NKikimrScheme::StatusAccepted);
+            TestBackupIncrementalBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
 
-        env.TestWaitNotification(runtime, {txId, txId - 1});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
 
-        TestDescribe(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1);
-        TestDescribe(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_2);
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections"),
-                           {NLs::PathVersionEqual(7)});
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
-                           {NLs::PathVersionEqual(1)});
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_2),
-                           {NLs::PathVersionEqual(1)});
-    }
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
 
-    Y_UNIT_TEST(Drop) {
-        TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
-        ui64 txId = 100;
+        Y_UNIT_TEST(DropCollectionDuringActiveBackup) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
 
-        PrepareDirs(runtime, env, txId);
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
-        env.TestWaitNotification(runtime, txId);
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
 
-        TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathExist);
+            AsyncBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
 
-        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
-        env.TestWaitNotification(runtime, txId);
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+                "Name: \"" DEFAULT_NAME_1 "\"", 
+                {NKikimrScheme::StatusPreconditionFailed});
+            env.TestWaitNotification(runtime, txId);
 
-        TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathNotExist);
-    }
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            env.TestWaitNotification(runtime, txId - 1);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropNonExistentCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+                "Name: \"NonExistentCollection\"", 
+                {NKikimrScheme::StatusPathDoesNotExist});
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/NonExistentCollection"),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropCollectionWithMultipleBackups) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+
+            runtime.AdvanceCurrentTime(TDuration::Seconds(1));
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropCollectionWithNestedTables) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestMkDir(runtime, ++txId, "/MyRoot", "SubDir");
+            env.TestWaitNotification(runtime, txId);
+
+            TString collectionSettingsNested = R"(
+                Name: ")" DEFAULT_NAME_1 R"("
+
+                ExplicitEntryList {
+                    Entries {
+                        Type: ETypeTable
+                        Path: "/MyRoot/Table1"
+                    }
+                    Entries {
+                        Type: ETypeTable
+                        Path: "/MyRoot/SubDir/Table2"
+                    }
+                }
+                Cluster: {}
+            )";
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", collectionSettingsNested);
+            env.TestWaitNotification(runtime, txId);
 
-    Y_UNIT_TEST(DropTwice) {
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot/SubDir", R"(
+                Name: "Table2"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropCollectionVerifyCDCCleanup) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateCdcStream(runtime, ++txId, "/MyRoot", R"(
+                TableName: "Table1"
+                StreamDescription {
+                  Name: "Stream1"
+                  Mode: ECdcStreamModeKeysOnly
+                  Format: ECdcStreamFormatProto
+                }
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TString collectionSettingsWithCDC = R"(
+                Name: ")" DEFAULT_NAME_1 R"("
+                ExplicitEntryList {
+                    Entries {
+                        Type: ETypeTable
+                        Path: "/MyRoot/Table1"
+                    }
+                }
+                Cluster: {}
+                IncrementalBackupConfig: {}
+            )";
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", collectionSettingsWithCDC);
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePrivatePath(runtime, "/MyRoot/Table1/Stream1"), {NLs::PathExist});
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropCollectionRollbackOnFailure) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+                Name: "Table1"
+                Columns { Name: "key" Type: "Uint32" }
+                Columns { Name: "value" Type: "Utf8" }
+                KeyColumnNames: ["key"]
+            )");
+            env.TestWaitNotification(runtime, txId);
+
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+                "Name: \"NonExistentCollection\"",
+                {NKikimrScheme::StatusPathDoesNotExist});
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropLargeBackupCollection) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TString largeCollectionSettings = R"(
+                Name: ")" DEFAULT_NAME_1 R"("
+                ExplicitEntryList {)";
+
+            for (int i = 1; i <= 5; ++i) {
+                largeCollectionSettings += TStringBuilder() <<
+                    R"(
+                    Entries {
+                        Type: ETypeTable
+                        Path: "/MyRoot/Table)" << i << R"("
+                    })";
+            }
+            largeCollectionSettings += R"(
+                }
+                Cluster: {}
+            )";
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", largeCollectionSettings);
+            env.TestWaitNotification(runtime, txId);
+
+            for (int i = 1; i <= 5; ++i) {
+                TestCreateTable(runtime, ++txId, "/MyRoot", TStringBuilder() << R"(
+                    Name: "Table)" << i << R"("
+                    Columns { Name: "key" Type: "Uint32" }
+                    Columns { Name: "value" Type: "Utf8" }
+                    KeyColumnNames: ["key"]
+                )");
+                env.TestWaitNotification(runtime, txId);
+            }
+
+            for (int i = 0; i < 3; ++i) {
+                if (i > 0) {
+                    runtime.AdvanceCurrentTime(TDuration::Seconds(1));
+                }
+                
+                TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                    R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+                env.TestWaitNotification(runtime, txId);
+            }
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
+                NLs::PathExist,
+                NLs::IsBackupCollection,
+            });
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                              {NLs::PathNotExist});
+        }
+
+        Y_UNIT_TEST(DropCollectionValidationCases) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+                "Name: \"\"", 
+                {NKikimrScheme::StatusInvalidParameter});
+            env.TestWaitNotification(runtime, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/NonExistent/path", 
+                "Name: \"test\"", 
+                {NKikimrScheme::StatusPathDoesNotExist});
+            env.TestWaitNotification(runtime, txId);
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot", 
+                "Name: \"test\"", 
+                {NKikimrScheme::StatusSchemeError});
+            env.TestWaitNotification(runtime, txId);
+        }
+
+        Y_UNIT_TEST(DropSpecificCollectionAmongMultiple) {
+            TTestBasicRuntime runtime;
+            TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
+            ui64 txId = 100;
+
+            SetupLogging(runtime);
+            PrepareDirs(runtime, env, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+                DefaultCollectionSettingsWithName("Collection1"));
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+                DefaultCollectionSettingsWithName("Collection2"));
+            env.TestWaitNotification(runtime, txId);
+
+            TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+                DefaultCollectionSettingsWithName("Collection3"));
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection1"), {NLs::PathExist});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection2"), {NLs::PathExist});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection3"), {NLs::PathExist});
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"Collection2\"");
+            env.TestWaitNotification(runtime, txId);
+
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection1"), {NLs::PathExist});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection2"), {NLs::PathNotExist});
+            TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/Collection3"), {NLs::PathExist});
+
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"Collection1\"");
+            env.TestWaitNotification(runtime, txId);
+            TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"Collection3\"");
+            env.TestWaitNotification(runtime, txId);
+        }
+
+    
+    Y_UNIT_TEST(DropCollectionVerifyLocalDatabaseCleanup) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
+        SetupLogging(runtime);
         PrepareDirs(runtime, env, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+        TString localDbCollectionSettings = R"(
+            Name: "LocalDbTestCollection"
+
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/LocalDbTestTable"
+                }
+            }
+            Cluster: {}
+        )";
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+            localDbCollectionSettings);
         env.TestWaitNotification(runtime, txId);
 
-        AsyncDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
-        AsyncDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
-        TestModificationResult(runtime, txId - 1);
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "LocalDbTestTable"
+            Columns { Name: "key"   Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )");
+        env.TestWaitNotification(runtime, txId);
 
-        auto ev = runtime.GrabEdgeEvent<TEvSchemeShard::TEvModifySchemeTransactionResult>();
-        UNIT_ASSERT(ev);
+        TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+            R"(Name: ".backups/collections/LocalDbTestCollection")");
+        env.TestWaitNotification(runtime, txId);        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"LocalDbTestCollection\"");
+        env.TestWaitNotification(runtime, txId);
 
-        const auto& record = ev->Record;
-        UNIT_ASSERT_VALUES_EQUAL(record.GetTxId(), txId);
-        UNIT_ASSERT_VALUES_EQUAL(record.GetStatus(), NKikimrScheme::StatusMultipleModifications);
-        UNIT_ASSERT_VALUES_EQUAL(record.GetPathDropTxId(), txId - 1);
+        RebootTablet(runtime, TTestTxConfig::SchemeShard, runtime.AllocateEdgeActor());
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/LocalDbTestCollection"), 
+                          {NLs::PathNotExist});
+
+        ui64 schemeshardTabletId = TTestTxConfig::SchemeShard;
+        
+        bool backupCollectionTableClean = true;
+        try {
+            auto result = LocalMiniKQL(runtime, schemeshardTabletId, R"(
+                (
+                    (let key '('('OwnerPathId (Uint64 '0)) '('LocalPathId (Uint64 '0))))
+                    (let select '('OwnerPathId 'LocalPathId))
+                    (let row (SelectRow 'BackupCollection key select))
+                    (return (AsList
+                        (SetResult 'Result row)
+                    ))
+                )
+            )");
+            
+            auto& value = result.GetValue();
+            if (value.GetStruct(0).GetOptional().HasOptional()) {
+                backupCollectionTableClean = false;
+                Cerr << "ERROR: BackupCollection table still has entries after DROP" << Endl;
+            }
+        } catch (...) {
+            backupCollectionTableClean = false;
+            Cerr << "ERROR: Failed to query BackupCollection table" << Endl;
+        }
+        
+        UNIT_ASSERT_C(backupCollectionTableClean, "BackupCollection table not properly cleaned up");
+
+        bool incrementalRestoreOperationsClean = true;
+        try {
+            auto result = LocalMiniKQL(runtime, schemeshardTabletId, R"(
+                (
+                    (let key '('('Id (Uint64 '0))))
+                    (let select '('Id))
+                    (let row (SelectRow 'IncrementalRestoreOperations key select))
+                    (return (AsList
+                        (SetResult 'Result row)
+                    ))
+                )
+            )");
+            
+            auto& value = result.GetValue();
+            if (value.GetStruct(0).GetOptional().HasOptional()) {
+                incrementalRestoreOperationsClean = false;
+                Cerr << "ERROR: IncrementalRestoreOperations table still has entries after DROP" << Endl;
+            }
+        } catch (...) {
+            incrementalRestoreOperationsClean = false;
+            Cerr << "ERROR: Failed to query IncrementalRestoreOperations table" << Endl;
+        }
+        
+        UNIT_ASSERT_C(incrementalRestoreOperationsClean, "IncrementalRestoreOperations table not properly cleaned up");
+
+        bool incrementalRestoreStateClean = true;
+        try {
+            auto result = LocalMiniKQL(runtime, schemeshardTabletId, R"(
+                (
+                    (let key '('('OperationId (Uint64 '0))))
+                    (let select '('OperationId))
+                    (let row (SelectRow 'IncrementalRestoreState key select))
+                    (return (AsList
+                        (SetResult 'Result row)
+                    ))
+                )
+            )");
+            
+            auto& value = result.GetValue();
+            if (value.GetStruct(0).GetOptional().HasOptional()) {
+                incrementalRestoreStateClean = false;
+                Cerr << "ERROR: IncrementalRestoreState table still has entries after DROP" << Endl;
+            }
+        } catch (...) {
+            incrementalRestoreStateClean = false;
+            Cerr << "ERROR: Failed to query IncrementalRestoreState table" << Endl;
+        }
+        
+        UNIT_ASSERT_C(incrementalRestoreStateClean, "IncrementalRestoreState table not properly cleaned up");
+
+        bool incrementalRestoreShardProgressClean = true;
+        try {
+            auto result = LocalMiniKQL(runtime, schemeshardTabletId, R"(
+                (
+                    (let key '('('OperationId (Uint64 '0)) '('ShardIdx (Uint64 '0))))
+                    (let select '('OperationId))
+                    (let row (SelectRow 'IncrementalRestoreShardProgress key select))
+                    (return (AsList
+                        (SetResult 'Result row)
+                    ))
+                )
+            )");
+            
+            auto& value = result.GetValue();
+            if (value.GetStruct(0).GetOptional().HasOptional()) {
+                incrementalRestoreShardProgressClean = false;
+                Cerr << "ERROR: IncrementalRestoreShardProgress table still has entries after DROP" << Endl;
+            }
+        } catch (...) {
+            incrementalRestoreShardProgressClean = false;
+            Cerr << "ERROR: Failed to query IncrementalRestoreShardProgress table" << Endl;
+        }
+        
+        UNIT_ASSERT_C(incrementalRestoreShardProgressClean, "IncrementalRestoreShardProgress table not properly cleaned up");
 
-        env.TestWaitNotification(runtime, txId - 1);
-        TestLs(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1, false, NLs::PathNotExist);
-    }
+        Cerr << "SUCCESS: All LocalDB tables properly cleaned up after DROP BACKUP COLLECTION" << Endl;
 
-    Y_UNIT_TEST(TableWithSystemColumns) {
+        TString recreateCollectionSettings = R"(
+            Name: "LocalDbTestCollection"
+
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/LocalDbTestTable"
+                }
+            }
+            Cluster: {}
+        )";
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+            recreateCollectionSettings);
+        env.TestWaitNotification(runtime, txId);
+    }
+    Y_UNIT_TEST(DropCollectionDuringActiveOperation) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
         SetupLogging(runtime);
-
         PrepareDirs(runtime, env, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+        TString activeOpCollectionSettings = R"(
+            Name: "ActiveOpTestCollection"
 
-        env.TestWaitNotification(runtime, txId);
-
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-        });
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/ActiveOpTestTable"
+                }
+            }
+            Cluster: {}
+        )";
 
-        TestCreateTable(runtime, ++txId, "/MyRoot/.backups/collections", R"(
-            Name: "Table1"
-            Columns { Name: "key" Type: "Utf8" }
-            Columns { Name: "__ydb_system_column" Type: "Utf8" }
-            KeyColumnNames: ["key"]
-        )");
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", activeOpCollectionSettings);
         env.TestWaitNotification(runtime, txId);
 
         TestCreateTable(runtime, ++txId, "/MyRoot", R"(
-            Name: ".backups/collections/Table2"
-            Columns { Name: "key" Type: "Utf8" }
-            Columns { Name: "__ydb_system_column" Type: "Utf8" }
+            Name: "ActiveOpTestTable"
+            Columns { Name: "key"   Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
             KeyColumnNames: ["key"]
         )");
         env.TestWaitNotification(runtime, txId);
 
-        TestCreateTable(runtime, ++txId, "/MyRoot/.backups/collections", R"(
-            Name: "somepath/Table3"
-            Columns { Name: "key" Type: "Utf8" }
-            Columns { Name: "__ydb_system_column" Type: "Utf8" }
-            KeyColumnNames: ["key"]
-        )");
+        // Start a backup operation (async, don't wait for completion)
+        AsyncBackupBackupCollection(runtime, ++txId, "/MyRoot",
+            R"(Name: ".backups/collections/ActiveOpTestCollection")");
+        ui64 backupTxId = txId;
+
+        // This shows that active operation protection IS implemented
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"ActiveOpTestCollection\"", 
+            {NKikimrScheme::StatusPreconditionFailed}); // CORRECT: System properly rejects this
+        env.TestWaitNotification(runtime, txId);
+
+        env.TestWaitNotification(runtime, backupTxId);
+
+        // VERIFICATION: Collection should still exist since drop was properly rejected
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/ActiveOpTestCollection"), 
+            {NLs::PathExist});
+
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"ActiveOpTestCollection\"");
         env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/ActiveOpTestCollection"), 
+            {NLs::PathNotExist});
+
     }
 
-    Y_UNIT_TEST(BackupAbsentCollection) {
+    Y_UNIT_TEST(VerifyCdcStreamCleanupInIncrementalBackup) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
-        SetupLogging(runtime);
-
         PrepareDirs(runtime, env, txId);
 
+        // Create backup collection that supports incremental backups
+        TString collectionSettingsWithIncremental = R"(
+            Name: ")" DEFAULT_NAME_1 R"("
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/TestTable"
+                }
+            }
+            Cluster: {}
+            IncrementalBackupConfig: {}
+        )";
+
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", 
+            collectionSettingsWithIncremental);
+        env.TestWaitNotification(runtime, txId);
+
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "TestTable"
+            Columns { Name: "key" Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )");
+        env.TestWaitNotification(runtime, txId);
+
         TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
-            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
-            {NKikimrScheme::EStatus::StatusPathDoesNotExist});
+            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+        env.TestWaitNotification(runtime, txId);
+
+        runtime.AdvanceCurrentTime(TDuration::Seconds(1));
+        TestBackupIncrementalBackupCollection(runtime, ++txId, "/MyRoot",
+            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+        env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/TestTable"),
+                          {NLs::PathExist, NLs::IsTable});
+        
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathExist, NLs::IsBackupCollection});
+
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"" DEFAULT_NAME_1 "\"");
         env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathNotExist});
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/TestTable"),
+                          {NLs::PathExist, NLs::IsTable});
+
+        // TODO: Add specific CDC stream cleanup verification
+        // This requires understanding the CDC stream naming and location patterns
+        // Current test verifies basic incremental backup drop functionality
     }
 
-    Y_UNIT_TEST(BackupDroppedCollection) {
+    Y_UNIT_TEST(VerifyCdcStreamCleanupInIncrementalDrop) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
-        SetupLogging(runtime);
-
         PrepareDirs(runtime, env, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+        // Create backup collection with incremental support
+        TString collectionSettings = R"(
+            Name: ")" DEFAULT_NAME_1 R"("
+            ExplicitEntryList {
+                Entries {
+                    Type: ETypeTable
+                    Path: "/MyRoot/Table1"
+                }
+            }
+            Cluster: {}
+            IncrementalBackupConfig: {}
+        )";
+
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", collectionSettings);
         env.TestWaitNotification(runtime, txId);
 
-        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", "Name: \"" DEFAULT_NAME_1 "\"");
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "Table1"
+            Columns { Name: "key" Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )");
         env.TestWaitNotification(runtime, txId);
 
         TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
-            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
-            {NKikimrScheme::EStatus::StatusPathDoesNotExist});
+            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+        env.TestWaitNotification(runtime, txId);
+
+        runtime.AdvanceCurrentTime(TDuration::Seconds(1));
+        TestBackupIncrementalBackupCollection(runtime, ++txId, "/MyRoot",
+            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
         env.TestWaitNotification(runtime, txId);
+
+        auto describeResult = DescribePath(runtime, "/MyRoot/Table1", true, true);
+        TVector<TString> cdcStreamNames;
+        
+        // Check table description for CDC streams (this is where they are actually stored)
+        if (describeResult.GetPathDescription().HasTable()) {
+            const auto& tableDesc = describeResult.GetPathDescription().GetTable();
+            if (tableDesc.CdcStreamsSize() > 0) {
+                Cerr << "Table has " << tableDesc.CdcStreamsSize() << " CDC streams in description" << Endl;
+                for (size_t i = 0; i < tableDesc.CdcStreamsSize(); ++i) {
+                    const auto& cdcStream = tableDesc.GetCdcStreams(i);
+                    if (cdcStream.GetName().EndsWith("_continuousBackupImpl")) {
+                        cdcStreamNames.push_back(cdcStream.GetName());
+                        Cerr << "Found incremental backup CDC stream: " << cdcStream.GetName() << Endl;
+                    }
+                }
+            }
+        }
+        
+        UNIT_ASSERT_C(!cdcStreamNames.empty(), "Expected to find CDC streams with '_continuousBackupImpl' suffix after incremental backup");
+        
+        for (const auto& streamName : cdcStreamNames) {
+            UNIT_ASSERT_C(streamName.size() >= 15 + TString("_continuousBackupImpl").size(), 
+                "CDC stream name should have timestamp prefix: " + streamName);
+            
+            TString prefix = streamName.substr(0, streamName.size() - TString("_continuousBackupImpl").size());
+            UNIT_ASSERT_C(prefix.EndsWith("Z"), "CDC stream timestamp should end with 'Z': " + prefix);
+        }
+
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"" DEFAULT_NAME_1 "\"");
+        env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathNotExist});
+
+        auto describeAfter = DescribePath(runtime, "/MyRoot/Table1", true, true);
+        TVector<TString> remainingCdcStreams;
+        
+        // Check table description for remaining CDC streams
+        if (describeAfter.GetPathDescription().HasTable()) {
+            const auto& tableDesc = describeAfter.GetPathDescription().GetTable();
+            if (tableDesc.CdcStreamsSize() > 0) {
+                Cerr << "Table still has " << tableDesc.CdcStreamsSize() << " CDC streams after drop" << Endl;
+                for (size_t i = 0; i < tableDesc.CdcStreamsSize(); ++i) {
+                    const auto& cdcStream = tableDesc.GetCdcStreams(i);
+                    if (cdcStream.GetName().EndsWith("_continuousBackupImpl")) {
+                        remainingCdcStreams.push_back(cdcStream.GetName());
+                        Cerr << "Incremental backup CDC stream still exists after drop: " << cdcStream.GetName() << Endl;
+                    }
+                }
+            }
+        }
+        
+        UNIT_ASSERT_C(remainingCdcStreams.empty(), 
+            "Incremental backup CDC streams with '_continuousBackupImpl' suffix should be cleaned up after dropping backup collection");
+        // During incremental backup, CDC streams are created under the source table
+        // They should be properly cleaned up when the backup collection is dropped
+        
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/Table1"),
+                          {NLs::PathExist, NLs::IsTable});
+
+        // Restart SchemeShard to verify persistent cleanup
+        TActorId sender = runtime.AllocateEdgeActor();
+        RebootTablet(runtime, TTestTxConfig::SchemeShard, sender);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathNotExist});
+        
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/Table1"),
+                          {NLs::PathExist, NLs::IsTable});
+
+        auto describeAfterReboot = DescribePath(runtime, "/MyRoot/Table1", true, true);
+        TVector<TString> cdcStreamsAfterReboot;
+        
+        if (describeAfterReboot.GetPathDescription().HasTable()) {
+            const auto& tableDesc = describeAfterReboot.GetPathDescription().GetTable();
+            if (tableDesc.CdcStreamsSize() > 0) {
+                Cerr << "Table still has " << tableDesc.CdcStreamsSize() << " CDC streams after restart" << Endl;
+                for (size_t i = 0; i < tableDesc.CdcStreamsSize(); ++i) {
+                    const auto& cdcStream = tableDesc.GetCdcStreams(i);
+                    if (cdcStream.GetName().EndsWith("_continuousBackupImpl")) {
+                        cdcStreamsAfterReboot.push_back(cdcStream.GetName());
+                        Cerr << "Incremental backup CDC stream still exists after restart: " << cdcStream.GetName() << Endl;
+                    }
+                }
+            }
+        }
+        
+        UNIT_ASSERT_C(cdcStreamsAfterReboot.empty(), 
+            "Incremental backup CDC streams with '_continuousBackupImpl' suffix should remain cleaned up after restart");
+
+        // The implementation properly handles CDC stream cleanup during backup collection drop
     }
 
-    Y_UNIT_TEST(BackupAbsentDirs) {
+    Y_UNIT_TEST(DropErrorRecoveryTest) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
-        SetupLogging(runtime);
-
         PrepareDirs(runtime, env, txId);
 
-        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", DefaultCollectionSettings());
+        // Create backup collection
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
+        env.TestWaitNotification(runtime, txId);
+
+        TestCreateTable(runtime, ++txId, "/MyRoot", R"(
+            Name: "Table1"
+            Columns { Name: "key" Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )");
         env.TestWaitNotification(runtime, txId);
 
-        TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
-            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
-            {NKikimrScheme::EStatus::StatusPathDoesNotExist});
+        // Create multiple backups
+        for (int i = 0; i < 3; ++i) {
+            runtime.AdvanceCurrentTime(TDuration::Seconds(1));
+            TestBackupBackupCollection(runtime, ++txId, "/MyRoot",
+                R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
+            env.TestWaitNotification(runtime, txId);
+        }
+
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"" DEFAULT_NAME_1 "\"");
+        env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathNotExist});
+
+        TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
         env.TestWaitNotification(runtime, txId);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathExist, NLs::IsBackupCollection});
     }
 
-    Y_UNIT_TEST(BackupNonIncrementalCollection) {
+    Y_UNIT_TEST(ConcurrentDropProtectionTest) {
         TTestBasicRuntime runtime;
         TTestEnv env(runtime, TTestEnvOptions().EnableBackupService(true));
         ui64 txId = 100;
 
-        SetupLogging(runtime);
-
         PrepareDirs(runtime, env, txId);
 
+        // Create backup collection
         TestCreateBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections/", DefaultCollectionSettings());
-
         env.TestWaitNotification(runtime, txId);
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-        });
-
         TestCreateTable(runtime, ++txId, "/MyRoot", R"(
             Name: "Table1"
-            Columns { Name: "key" Type: "Utf8" }
+            Columns { Name: "key" Type: "Uint32" }
+            Columns { Name: "value" Type: "Utf8" }
             KeyColumnNames: ["key"]
         )");
         env.TestWaitNotification(runtime, txId);
@@ -372,16 +1400,21 @@ Y_UNIT_TEST_SUITE(TBackupCollectionTests) {
             R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")");
         env.TestWaitNotification(runtime, txId);
 
-        TestBackupIncrementalBackupCollection(runtime, ++txId, "/MyRoot",
-            R"(Name: ".backups/collections/)" DEFAULT_NAME_1 R"(")",
-            {NKikimrScheme::EStatus::StatusInvalidParameter});
-        env.TestWaitNotification(runtime, txId);
+        // Start first drop operation asynchronously
+        AsyncDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"" DEFAULT_NAME_1 "\"");
 
-        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1), {
-            NLs::PathExist,
-            NLs::IsBackupCollection,
-            NLs::ChildrenCount(1),
-            NLs::Finished,
-        });
+        // Immediately try second drop operation (should fail)
+        TestDropBackupCollection(runtime, ++txId, "/MyRoot/.backups/collections", 
+            "Name: \"" DEFAULT_NAME_1 "\"", 
+            {NKikimrScheme::StatusMultipleModifications}); // Expect concurrent operation error
+
+        env.TestWaitNotification(runtime, txId - 1);
+
+        TestDescribeResult(DescribePath(runtime, "/MyRoot/.backups/collections/" DEFAULT_NAME_1),
+                          {NLs::PathNotExist});
     }
-} // TBackupCollectionTests
+
+    // TODO: DropCollectionWithIncrementalRestoreStateCleanup
+
+} // TBackupCollectionTests
\ No newline at end of file
diff --git a/ydb/core/ydb_convert/table_description.cpp b/ydb/core/ydb_convert/table_description.cpp
index 8f5ca9e86fc2..1b981012ebca 100644
--- a/ydb/core/ydb_convert/table_description.cpp
+++ b/ydb/core/ydb_convert/table_description.cpp
@@ -242,7 +242,7 @@ bool BuildAlterTableModifyScheme(const TString& path, const Ydb::Table::AlterTab
     for (const auto& drop : req->drop_changefeeds()) {
         modifyScheme->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpDropCdcStream);
         auto op = modifyScheme->MutableDropCdcStream();
-        op->SetStreamName(drop);
+        op->AddStreamName(drop);
         op->SetTableName(name);
     }