reading from a topic and writing to a topic in a transaction (#5841)

Alek5andr-Kotov · web-flow · commit b15d3d72d1a2 · 2024-06-24T14:32:17.000+03:00
diff --git a/ydb/core/persqueue/partition.cpp b/ydb/core/persqueue/partition.cpp
@@ -3,6 +3,7 @@
 #include "offload_actor.h"
 #include "partition_util.h"
 #include "partition.h"
+#include "partition_log.h"
 
 #include <ydb/core/base/appdata.h>
 #include <ydb/core/base/blobstorage.h>
@@ -81,7 +82,7 @@ TString TPartition::LogPrefix() const {
     } else {
         state = "Unknown";
     }
-    return TStringBuilder() << "" << SelfId() << " " << state << " Partition: " << Partition << " ";
+    return TStringBuilder() << "[Partition:" << Partition << ", State:" << state << "] ";
 }
 
 bool TPartition::IsActive() const {
@@ -1091,11 +1092,13 @@ TPartition::EProcessResult TPartition::ApplyWriteInfoResponse(TTransaction& tx)
 }
 
 void TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse::TPtr& ev, const TActorContext& ctx) {
+    PQ_LOG_D("Handle TEvPQ::TEvGetWriteInfoResponse");
     WriteInfoResponseHandler(ev->Sender, ev->Release(), ctx);
 }
 
 
 void TPartition::Handle(TEvPQ::TEvGetWriteInfoError::TPtr& ev, const TActorContext& ctx) {
+    PQ_LOG_D("Handle TEvPQ::TEvGetWriteInfoError");
     WriteInfoResponseHandler(ev->Sender, ev->Release(), ctx);
 }
 
diff --git a/ydb/core/persqueue/pq_impl.cpp b/ydb/core/persqueue/pq_impl.cpp
@@ -2675,6 +2675,7 @@ void TPersQueue::HandleEventForSupportivePartition(const ui64 responseCookie,
         //
         TTxWriteInfo& writeInfo = TxWrites[writeId];
         TPartitionId partitionId(originalPartitionId, writeId, NextSupportivePartitionId++);
+        PQ_LOG_D("partition " << partitionId << " for WriteId " << writeId);
 
         writeInfo.Partitions.emplace(originalPartitionId, partitionId);
         TxWritesChanged = true;
@@ -3109,7 +3110,7 @@ void TPersQueue::Handle(TEvPersQueue::TEvCancelTransactionProposal::TPtr& ev, co
 
 void TPersQueue::Handle(TEvPersQueue::TEvProposeTransaction::TPtr& ev, const TActorContext& ctx)
 {
-    LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Tablet " << TabletID() << " Handle TEvPersQueue::TEvProposeTransaction");
+    PQ_LOG_D("Handle TEvPersQueue::TEvProposeTransaction " << ev->Get()->Record.DebugString());
 
     NKikimrPQ::TEvProposeTransaction& event = ev->Get()->Record;
     switch (event.GetTxBodyCase()) {
@@ -3122,6 +3123,8 @@ void TPersQueue::Handle(TEvPersQueue::TEvProposeTransaction::TPtr& ev, const TAc
     case NKikimrPQ::TEvProposeTransaction::TXBODY_NOT_SET:
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::ERROR,
+                                    "missing TxBody",
                                     ctx);
         break;
     }
@@ -3134,6 +3137,7 @@ bool TPersQueue::CheckTxWriteOperation(const NKikimrPQ::TPartitionOperation& ope
     TPartitionId partitionId(operation.GetPartitionId(),
                              writeId,
                              operation.GetSupportivePartition());
+    PQ_LOG_D("partitionId=" << partitionId);
     return Partitions.contains(partitionId);
 }
 
@@ -3144,6 +3148,7 @@ bool TPersQueue::CheckTxWriteOperations(const NKikimrPQ::TDataTransaction& txBod
     }
 
     ui64 writeId = txBody.GetWriteId();
+    PQ_LOG_D("writeId=" << writeId);
 
     for (auto& operation : txBody.GetOperations()) {
         auto isWrite = [](const NKikimrPQ::TPartitionOperation& o) {
@@ -3169,8 +3174,11 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
     const NKikimrPQ::TDataTransaction& txBody = event.GetData();
 
     if (TabletState != NKikimrPQ::ENormal) {
+        PQ_LOG_D("invalid PQ tablet state (" << NKikimrPQ::ETabletState_Name(TabletState) << ")");
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::ERROR,
+                                    "invalid PQ tablet state",
                                     ctx);
         return;
     }
@@ -3180,36 +3188,49 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
     //
 
     if (txBody.OperationsSize() <= 0) {
+        PQ_LOG_D("empty list of operations");
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::BAD_REQUEST,
+                                    "empty list of operations",
                                     ctx);
         return;
     }
 
     if (!CheckTxWriteOperations(txBody)) {
+        PQ_LOG_D("invalid WriteId " << txBody.GetWriteId());
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::BAD_REQUEST,
+                                    "invalid WriteId",
                                     ctx);
         return;
     }
 
     TMaybe<TPartitionId> partitionId = FindPartitionId(txBody);
     if (!partitionId.Defined()) {
+        PQ_LOG_D("unknown partition for WriteId " << txBody.GetWriteId());
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::INTERNAL,
+                                    "unknown supportive partition",
                                     ctx);
         return;
     }
 
     if (!Partitions.contains(*partitionId)) {
+        PQ_LOG_D("unknown partition " << *partitionId);
         SendProposeTransactionAbort(ActorIdFromProto(event.GetSourceActor()),
                                     event.GetTxId(),
+                                    NKikimrPQ::TError::INTERNAL,
+                                    "unknown supportive partition",
                                     ctx);
         return;
     }
 
 
     if (txBody.GetImmediate()) {
+        PQ_LOG_D("immediate transaction");
         TPartitionId originalPartitionId(txBody.GetOperations(0).GetPartitionId());
         const TPartitionInfo& partition = Partitions.at(originalPartitionId);
 
@@ -3221,6 +3242,7 @@ void TPersQueue::HandleDataTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransact
 
         ctx.Send(partition.Actor, ev.Release());
     } else {
+        PQ_LOG_D("distributed transaction");
         EvProposeTransactionQueue.emplace_back(ev.Release());
 
         TryWriteTxs(ctx);
@@ -3697,6 +3719,7 @@ void TPersQueue::SaveTxWrites(NKikimrPQ::TTabletTxInfo& info)
 
 void TPersQueue::ScheduleProposeTransactionResult(const TDistributedTransaction& tx)
 {
+    PQ_LOG_D("schedule TEvProposeTransactionResult(PREPARED)");
     auto event = std::make_unique<TEvPersQueue::TEvProposeTransactionResult>();
 
     event->Record.SetOrigin(TabletID());
@@ -3766,16 +3789,27 @@ void TPersQueue::SendEvReadSetAckToSenders(const TActorContext& ctx,
 
 TMaybe<TPartitionId> TPersQueue::FindPartitionId(const NKikimrPQ::TDataTransaction& txBody) const
 {
+    auto hasWriteOperation = [](const auto& txBody) {
+        for (const auto& o : txBody.GetOperations()) {
+            if (!o.HasBegin()) {
+                return true;
+            }
+        }
+        return false;
+    };
+
     ui32 partitionId = txBody.GetOperations(0).GetPartitionId();
 
-    if (txBody.HasWriteId()) {
+    if (txBody.HasWriteId() && hasWriteOperation(txBody)) {
         ui64 writeId = txBody.GetWriteId();
         if (!TxWrites.contains(writeId)) {
+            PQ_LOG_D("unknown WriteId " << writeId);
             return Nothing();
         }
 
         const TTxWriteInfo& writeInfo = TxWrites.at(writeId);
         if (!writeInfo.Partitions.contains(partitionId)) {
+            PQ_LOG_D("unknown partition " << partitionId << " for WriteId " << writeId);
             return Nothing();
         }
 
@@ -3882,6 +3916,9 @@ void TPersQueue::SendEvProposeTransactionResult(const TActorContext& ctx,
     result->Record.SetTxId(tx.TxId);
     result->Record.SetStep(tx.Step);
 
+    PQ_LOG_D("send TEvPersQueue::TEvProposeTransactionResult(" <<
+             NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(result->Record.GetStatus()) <<
+             ")");
     ctx.Send(tx.SourceActor, std::move(result));
 }
 
@@ -4204,6 +4241,8 @@ bool TPersQueue::AllTransactionsHaveBeenProcessed() const
 
 void TPersQueue::SendProposeTransactionAbort(const TActorId& target,
                                              ui64 txId,
+                                             NKikimrPQ::TError::EKind kind,
+                                             const TString& reason,
                                              const TActorContext& ctx)
 {
     auto event = std::make_unique<TEvPersQueue::TEvProposeTransactionResult>();
@@ -4212,6 +4251,15 @@ void TPersQueue::SendProposeTransactionAbort(const TActorId& target,
     event->Record.SetStatus(NKikimrPQ::TEvProposeTransactionResult::ABORTED);
     event->Record.SetTxId(txId);
 
+    if (kind != NKikimrPQ::TError::OK) {
+        auto* error = event->Record.MutableErrors()->Add();
+        error->SetKind(kind);
+        error->SetReason(reason);
+    }
+
+    PQ_LOG_D("send TEvPersQueue::TEvProposeTransactionResult(" <<
+             NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(event->Record.GetStatus()) <<
+             ")");
     ctx.Send(target, std::move(event));
 }
 
@@ -4509,6 +4557,7 @@ void TPersQueue::Handle(TEvPQ::TEvDeletePartitionDone::TPtr& ev, const TActorCon
                 DeleteTx(*tx);
             }
         }
+        PQ_LOG_D("delete WriteId " << writeId);
         TxWrites.erase(writeId);
     }
     TxWritesChanged = true;
@@ -4562,7 +4611,7 @@ void TPersQueue::BeginDeletePartitions(TTxWriteInfo& writeInfo)
 }
 
 TString TPersQueue::LogPrefix() const {
-    return TStringBuilder() << SelfId() << " ";
+    return TStringBuilder() << "[PQ: " << TabletID() << "] ";
 }
 
 ui64 TPersQueue::GetGeneration() {
diff --git a/ydb/core/persqueue/pq_impl.h b/ydb/core/persqueue/pq_impl.h
@@ -344,6 +344,8 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
 
     void SendProposeTransactionAbort(const TActorId& target,
                                      ui64 txId,
+                                     NKikimrPQ::TError::EKind kind,
+                                     const TString& reason,
                                      const TActorContext& ctx);
 
     void Handle(TEvPQ::TEvProposePartitionConfigResult::TPtr& ev, const TActorContext& ctx);
diff --git a/ydb/core/persqueue/transaction.cpp b/ydb/core/persqueue/transaction.cpp
@@ -60,6 +60,10 @@ void TDistributedTransaction::InitPartitions(const google::protobuf::RepeatedPtr
     Partitions.clear();
 
     for (auto& o : operations) {
+        if (!o.HasBegin()) {
+            HasWriteOperations = true;
+        }
+
         Operations.push_back(o);
         Partitions.insert(o.GetPartitionId());
     }
@@ -133,7 +137,7 @@ void TDistributedTransaction::OnProposeTransaction(const NKikimrPQ::TDataTransac
 
     InitPartitions(txBody.GetOperations());
 
-    if (txBody.HasWriteId()) {
+    if (txBody.HasWriteId() && HasWriteOperations) {
         WriteId = txBody.GetWriteId();
     } else {
         WriteId = Nothing();
diff --git a/ydb/core/persqueue/transaction.h b/ydb/core/persqueue/transaction.h
@@ -109,6 +109,8 @@ struct TDistributedTransaction {
     void BindMsgToPipe(ui64 tabletId, const IEventBase& event);
     void UnbindMsgsFromPipe(ui64 tabletId);
     const TVector<TSerializedMessage>& GetBindedMsgs(ui64 tabletId);
+
+    bool HasWriteOperations = false;
 };
 
 }
diff --git a/ydb/core/persqueue/ut/pqtablet_ut.cpp b/ydb/core/persqueue/ut/pqtablet_ut.cpp
@@ -29,10 +29,11 @@ namespace NHelpers {
 
 struct TTxOperation {
     ui32 Partition;
-    TString Consumer;
-    ui64 Begin = 0;
-    ui64 End = 0;
+    TMaybe<TString> Consumer;
+    TMaybe<ui64> Begin;
+    TMaybe<ui64> End;
     TString Path;
+    TMaybe<ui32> SupportivePartition;
 };
 
 struct TConfigParams {
@@ -307,10 +308,15 @@ void TPQTabletFixture::SendProposeTransactionRequest(const TProposeTransactionPa
         for (auto& txOp : params.TxOps) {
             auto* operation = body->MutableOperations()->Add();
             operation->SetPartitionId(txOp.Partition);
-            operation->SetBegin(txOp.Begin);
-            operation->SetEnd(txOp.End);
-            operation->SetConsumer(txOp.Consumer);
+            if (txOp.Begin.Defined()) {
+                operation->SetBegin(*txOp.Begin);
+                operation->SetEnd(*txOp.End);
+                operation->SetConsumer(*txOp.Consumer);
+            }
             operation->SetPath(txOp.Path);
+            if (txOp.SupportivePartition.Defined()) {
+                operation->SetSupportivePartition(*txOp.SupportivePartition);
+            }
 
             partitions.insert(txOp.Partition);
         }
@@ -342,7 +348,9 @@ void TPQTabletFixture::WaitProposeTransactionResponse(const TProposeTransactionR
 
     if (matcher.Status) {
         UNIT_ASSERT(event->Record.HasStatus());
-        UNIT_ASSERT_EQUAL(*matcher.Status, event->Record.GetStatus());
+        UNIT_ASSERT_EQUAL_C(*matcher.Status, event->Record.GetStatus(),
+                            "expected: " << NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(*matcher.Status) <<
+                            ", received " << NKikimrPQ::TEvProposeTransactionResult_EStatus_Name(event->Record.GetStatus()));
     }
 }
 
@@ -1292,7 +1300,7 @@ Y_UNIT_TEST_F(ProposeTx_Command_After_Propose, TPQTabletFixture)
                      .Status=NMsgBusProxy::MSTATUS_OK});
 
     SendProposeTransactionRequest({.TxId=txId,
-                                  .TxOps={{.Partition=partitionId, .Path="/topic"}},
+                                  .TxOps={{.Partition=partitionId, .Path="/topic", .SupportivePartition=100'000}},
                                   .WriteId=writeId});
     WaitProposeTransactionResponse({.TxId=txId,
                                    .Status=NKikimrPQ::TEvProposeTransactionResult::PREPARED});
diff --git a/ydb/public/sdk/cpp/client/ydb_topic/ut/topic_to_table_ut.cpp b/ydb/public/sdk/cpp/client/ydb_topic/ut/topic_to_table_ut.cpp

Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@`
`3`	`3`	`#include "offload_actor.h"`
`4`	`4`	`#include "partition_util.h"`
`5`	`5`	`#include "partition.h"`
	`6`	`+#include "partition_log.h"`
`6`	`7`
`7`	`8`	`#include <ydb/core/base/appdata.h>`
`8`	`9`	`#include <ydb/core/base/blobstorage.h>`
`@@ -81,7 +82,7 @@ TString TPartition::LogPrefix() const {`
`81`	`82`	`} else {`
`82`	`83`	`state = "Unknown";`
`83`	`84`	`}`
`84`		`- return TStringBuilder() << "" << SelfId() << " " << state << " Partition: " << Partition << " ";`
	`85`	`+ return TStringBuilder() << "[Partition:" << Partition << ", State:" << state << "] ";`
`85`	`86`	`}`
`86`	`87`
`87`	`88`	`bool TPartition::IsActive() const {`
`@@ -1091,11 +1092,13 @@ TPartition::EProcessResult TPartition::ApplyWriteInfoResponse(TTransaction& tx)`
`1091`	`1092`	`}`
`1092`	`1093`
`1093`	`1094`	`void TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse::TPtr& ev, const TActorContext& ctx) {`
	`1095`	`+ PQ_LOG_D("Handle TEvPQ::TEvGetWriteInfoResponse");`
`1094`	`1096`	`WriteInfoResponseHandler(ev->Sender, ev->Release(), ctx);`
`1095`	`1097`	`}`
`1096`	`1098`
`1097`	`1099`
`1098`	`1100`	`void TPartition::Handle(TEvPQ::TEvGetWriteInfoError::TPtr& ev, const TActorContext& ctx) {`
	`1101`	`+ PQ_LOG_D("Handle TEvPQ::TEvGetWriteInfoError");`
`1099`	`1102`	`WriteInfoResponseHandler(ev->Sender, ev->Release(), ctx);`
`1100`	`1103`	`}`
`1101`	`1104`
Original file line number	Diff line number	Diff line change
`@@ -109,6 +109,8 @@ struct TDistributedTransaction {`
`109`	`109`	`void BindMsgToPipe(ui64 tabletId, const IEventBase& event);`
`110`	`110`	`void UnbindMsgsFromPipe(ui64 tabletId);`
`111`	`111`	`const TVector<TSerializedMessage>& GetBindedMsgs(ui64 tabletId);`
	`112`	`+`
	`113`	`+ bool HasWriteOperations = false;`
`112`	`114`	`};`
`113`	`115`
`114`	`116`	`}`