Fill change exchange split & activation lists just before commit; unify checks (#6281)

Author: CyberROFL

ydb/core/tx/datashard/datashard_change_sending.cpp

@@ -286,7 +286,7 @@ class TDataShard::TTxRemoveChangeRecords: public TTransactionBase<TDataShard> {
                 ChangeExchangeSplit = true;
             } else {
                 for (const auto dstTabletId : Self->ChangeSenderActivator.GetDstSet()) {
-                    if (Self->SplitSrcSnapshotSender.Acked(dstTabletId)) {
+                    if (Self->SplitSrcSnapshotSender.Acked(dstTabletId) && !Self->ChangeSenderActivator.Acked(dstTabletId)) {
                         ActivationList.insert(dstTabletId);
                     }
                 }
@@ -340,13 +340,13 @@ class TDataShard::TTxRemoveChangeRecords: public TTransactionBase<TDataShard> {
             Self->RemoveChangeRecordsInFly = false;
         }
 
-        if (ChangeExchangeSplit) {
-            Self->KillChangeSender(ctx);
-            Self->ChangeExchangeSplitter.DoSplit(ctx);
-        }
+        if (!Self->ChangesQueue) { // double check queue
+            if (ChangeExchangeSplit) {
+                Self->KillChangeSender(ctx);
+                Self->ChangeExchangeSplitter.DoSplit(ctx);
+            }
 
-        for (const auto dstTabletId : ActivationList) {
-            if (!Self->ChangeSenderActivator.Acked(dstTabletId)) {
+            for (const auto dstTabletId : ActivationList) {
                 Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
             }
         }
@@ -383,7 +383,7 @@ class TDataShard::TTxChangeExchangeSplitAck: public TTransactionBase<TDataShard>
         Y_ABORT_UNLESS(Self->ChangeExchangeSplitter.Done());
 
         for (const auto dstTabletId : Self->ChangeSenderActivator.GetDstSet()) {
-            if (Self->SplitSrcSnapshotSender.Acked(dstTabletId)) {
+            if (Self->SplitSrcSnapshotSender.Acked(dstTabletId) && !Self->ChangeSenderActivator.Acked(dstTabletId)) {
                 ActivationList.insert(dstTabletId);
             }
         }
@@ -396,9 +396,7 @@ class TDataShard::TTxChangeExchangeSplitAck: public TTransactionBase<TDataShard>
             << ", at tablet# " << Self->TabletID());
 
         for (const auto dstTabletId : ActivationList) {
-            if (!Self->ChangeSenderActivator.Acked(dstTabletId)) {
-                Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
-            }
+            Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
         }
     }
 

ydb/core/tx/datashard/datashard_split_src.cpp

@@ -244,6 +244,8 @@ class TDataShard::TTxSplitSnapshotComplete : public NTabletFlatExecutor::TTransa
 private:
     TIntrusivePtr<TSplitSnapshotContext> SnapContext;
     bool ChangeExchangeSplit;
+    THashSet<ui64> ActivationList;
+    THashSet<ui64> SplitList;
 
 public:
     TTxSplitSnapshotComplete(TDataShard* ds, TIntrusivePtr<TSplitSnapshotContext> snapContext)
@@ -378,13 +380,11 @@ class TDataShard::TTxSplitSnapshotComplete : public NTabletFlatExecutor::TTransa
                     proto->SetTimeoutMs(kv.second.Timeout.MilliSeconds());
                 }
 
-                if (Self->ChangesQueue || tableInfo.HasCdcStreams()) {
+                if (tableInfo.HasAsyncIndexes() || tableInfo.HasCdcStreams()) {
                     snapshot->SetWaitForActivation(true);
-                    Self->ChangeSenderActivator.AddDst(dstTablet);
-                    db.Table<Schema::SrcChangeSenderActivations>().Key(dstTablet).Update();
-
+                    ActivationList.insert(dstTablet);
                     if (tableInfo.HasCdcStreams()) {
-                        Self->ChangeExchangeSplitter.AddDst(dstTablet);
+                        SplitList.insert(dstTablet);
                     }
                 }
 
@@ -403,14 +403,23 @@ class TDataShard::TTxSplitSnapshotComplete : public NTabletFlatExecutor::TTransa
             }
         }
 
-        ChangeExchangeSplit = !Self->ChangesQueue && !Self->ChangeExchangeSplitter.Done();
-
         if (needToReadPages) {
             LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, Self->TabletID() << " BorrowSnapshot is restarting for split OpId " << opId);
             return false;
         } else {
             txc.Env.DropSnapshot(SnapContext);
 
+            for (ui64 dstTabletId : ActivationList) {
+                Self->ChangeSenderActivator.AddDst(dstTabletId);
+                db.Table<Schema::SrcChangeSenderActivations>().Key(dstTabletId).Update();
+            }
+
+            for (ui64 dstTabletId : SplitList) {
+                Self->ChangeExchangeSplitter.AddDst(dstTabletId);
+            }
+
+            ChangeExchangeSplit = !Self->ChangesQueue && !Self->ChangeExchangeSplitter.Done();
+
             Self->State = TShardState::SplitSrcSendingSnapshot;
             Self->PersistSys(db, Schema::Sys_State, Self->State);
 
@@ -421,7 +430,7 @@ class TDataShard::TTxSplitSnapshotComplete : public NTabletFlatExecutor::TTransa
     void Complete(const TActorContext &ctx) override {
         LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, Self->TabletID() << " Sending snapshots from src for split OpId " << Self->SrcSplitOpId);
         Self->SplitSrcSnapshotSender.DoSend(ctx);
-        if (ChangeExchangeSplit) {
+        if (ChangeExchangeSplit && !Self->ChangesQueue) { // double check queue
             Self->KillChangeSender(ctx);
             Self->ChangeExchangeSplitter.DoSplit(ctx);
         }
@@ -438,14 +447,14 @@ class TDataShard::TTxSplitTransferSnapshotAck : public NTabletFlatExecutor::TTra
 private:
     TEvDataShard::TEvSplitTransferSnapshotAck::TPtr Ev;
     bool AllDstAcksReceived;
-    bool Activate;
+    ui64 ActivateTabletId;
 
 public:
     TTxSplitTransferSnapshotAck(TDataShard* ds, TEvDataShard::TEvSplitTransferSnapshotAck::TPtr& ev)
         : NTabletFlatExecutor::TTransactionBase<TDataShard>(ds)
         , Ev(ev)
         , AllDstAcksReceived(false)
-        , Activate(false)
+        , ActivateTabletId(0)
     {}
 
     TTxType GetTxType() const override { return TXTYPE_SPLIT_TRANSFER_SNAPSHOT_ACK; }
@@ -469,8 +478,8 @@ class TDataShard::TTxSplitTransferSnapshotAck : public NTabletFlatExecutor::TTra
         // Remove the row for acked snapshot
         db.Table<Schema::SplitSrcSnapshots>().Key(dstTabletId).Delete();
 
-        if (!Self->ChangesQueue && Self->ChangeExchangeSplitter.Done()) {
-            Activate = !Self->ChangeSenderActivator.Acked(dstTabletId);
+        if (!Self->ChangesQueue && Self->ChangeExchangeSplitter.Done() && !Self->ChangeSenderActivator.Acked(dstTabletId)) {
+            ActivateTabletId = dstTabletId;
         }
 
         return true;
@@ -485,11 +494,8 @@ class TDataShard::TTxSplitTransferSnapshotAck : public NTabletFlatExecutor::TTra
             }
         }
 
-        if (Activate) {
-            const ui64 dstTabletId = Ev->Get()->Record.GetTabletId();
-            if (!Self->ChangeSenderActivator.Acked(dstTabletId)) {
-                Self->ChangeSenderActivator.DoSend(dstTabletId, ctx);
-            }
+        if (ActivateTabletId && !Self->ChangesQueue) { // double check queue
+            Self->ChangeSenderActivator.DoSend(ActivateTabletId, ctx);
         }
     }
 };