[+] the WriteId field in the TEvProposeTransaction message

Author: Alek5andr-Kotov

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -2173,7 +2173,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
 
     using TDatashardTxs = THashMap<ui64, NKikimrTxDataShard::TKqpTransaction*>;
     using TEvWriteTxs = THashMap<ui64, NKikimrDataEvents::TEvWrite*>;
-    using TTopicTabletTxs = THashMap<ui64, NKikimrPQ::TDataTransaction>;
+    using TTopicTabletTxs = NTopic::TTopicOperationTransactions;
 
     void ContinueExecute() {
         if (Stats) {
@@ -2430,10 +2430,10 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
                     }
                 }
 
-                for (auto& [_, tx] : topicTxs) {
-                    tx.SetOp(NKikimrPQ::TDataTransaction::Commit);
-                    *tx.MutableSendingShards() = sendingShards;
-                    *tx.MutableReceivingShards() = receivingShards;
+                for (auto& [_, t] : topicTxs) {
+                    t.tx.SetOp(NKikimrPQ::TDataTransaction::Commit);
+                    *t.tx.MutableSendingShards() = sendingShards;
+                    *t.tx.MutableReceivingShards() = receivingShards;
                     YQL_ENSURE(!arbiter);
                 }
             }
@@ -2595,13 +2595,12 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
             writeId = Request.TopicOperations.GetWriteId();
         }
 
-        for (auto& tx : topicTxs) {
-            auto tabletId = tx.first;
-            auto& transaction = tx.second;
+        for (auto& [tabletId, t] : topicTxs) {
+            auto& transaction = t.tx;
 
             auto ev = std::make_unique<TEvPersQueue::TEvProposeTransaction>();
 
-            if (writeId.Defined()) {
+            if (t.hasWrite && writeId.Defined()) {
                 auto* w = transaction.MutableWriteId();
                 w->SetNodeId(SelfId().NodeId());
                 w->SetKeyId(*writeId);

ydb/core/kqp/topics/kqp_topics.cpp

@@ -105,15 +105,15 @@ void TTopicPartitionOperations::AddOperation(const TString& topic, ui32 partitio
     HasWriteOperations_ = true;
 }
 
-void TTopicPartitionOperations::BuildTopicTxs(THashMap<ui64, NKikimrPQ::TDataTransaction> &txs)
+void TTopicPartitionOperations::BuildTopicTxs(TTopicOperationTransactions& txs)
 {
     Y_ABORT_UNLESS(TabletId_.Defined());
     Y_ABORT_UNLESS(Partition_.Defined());
 
-    auto& tx = txs[*TabletId_];
+    auto& t = txs[*TabletId_];
 
     for (auto& [consumer, operations] : Operations_) {
-        NKikimrPQ::TPartitionOperation* o = tx.MutableOperations()->Add();
+        NKikimrPQ::TPartitionOperation* o = t.tx.MutableOperations()->Add();
         o->SetPartitionId(*Partition_);
         auto [begin, end] = operations.GetRange();
         o->SetBegin(begin);
@@ -123,12 +123,13 @@ void TTopicPartitionOperations::BuildTopicTxs(THashMap<ui64, NKikimrPQ::TDataTra
     }
 
     if (HasWriteOperations_) {
-        NKikimrPQ::TPartitionOperation* o = tx.MutableOperations()->Add();
+        NKikimrPQ::TPartitionOperation* o = t.tx.MutableOperations()->Add();
         o->SetPartitionId(*Partition_);
         o->SetPath(*Topic_);
         if (SupportivePartition_.Defined()) {
             o->SetSupportivePartition(*SupportivePartition_);
         }
+        t.hasWrite = true;
     }
 }
 
@@ -355,7 +356,7 @@ bool TTopicOperations::ProcessSchemeCacheNavigate(const NSchemeCache::TSchemeCac
     return true;
 }
 
-void TTopicOperations::BuildTopicTxs(THashMap<ui64, NKikimrPQ::TDataTransaction> &txs)
+void TTopicOperations::BuildTopicTxs(TTopicOperationTransactions& txs)
 {
     for (auto& [_, operations] : Operations_) {
         operations.BuildTopicTxs(txs);

ydb/core/kqp/topics/kqp_topics.h

@@ -42,6 +42,13 @@ class TConsumerOperations {
     TDisjointIntervalTree<ui64> Offsets_;
 };
 
+struct TTopicOperationTransaction {
+    NKikimrPQ::TDataTransaction tx;
+    bool hasWrite = false;
+};
+
+using TTopicOperationTransactions = THashMap<ui64, TTopicOperationTransaction>;
+
 class TTopicPartitionOperations {
 public:
     bool IsValid() const;
@@ -52,7 +59,7 @@ class TTopicPartitionOperations {
     void AddOperation(const TString& topic, ui32 partition,
                       TMaybe<ui32> supportivePartition);
 
-    void BuildTopicTxs(THashMap<ui64, NKikimrPQ::TDataTransaction> &txs);
+    void BuildTopicTxs(TTopicOperationTransactions &txs);
 
     void Merge(const TTopicPartitionOperations& rhs);
 
@@ -109,7 +116,7 @@ class TTopicOperations {
                                     Ydb::StatusIds_StatusCode& status,
                                     TString& message);
 
-    void BuildTopicTxs(THashMap<ui64, NKikimrPQ::TDataTransaction> &txs);
+    void BuildTopicTxs(TTopicOperationTransactions &txs);
 
     void Merge(const TTopicOperations& rhs);