few fixes

Author: niksaveliev

ydb/core/persqueue/partition.cpp

@@ -164,7 +164,6 @@ NKikimrClient::TKeyValueRequest::EStorageChannel GetChannel(ui32 i) {
     return NKikimrClient::TKeyValueRequest::EStorageChannel(NKikimrClient::TKeyValueRequest::MAIN + i);
 }
 
-
 void AddCheckDiskRequest(TEvKeyValue::TEvRequest *request, ui32 numChannels) {
     for (ui32 i = 0; i < numChannels; ++i) {
         request->Record.AddCmdGetStatus()->SetStorageChannel(GetChannel(i));
@@ -2101,8 +2100,8 @@ TPartition::EProcessResult TPartition::BeginTransaction(const TEvPQ::TEvTxCalcPr
         bool isAffectedConsumer = AffectedUsers.contains(consumer); // savnik check
         TUserInfoBase& userInfo = GetOrCreatePendingUser(consumer);
 
-        if (operation.HasOnlyCheckCommitedToFinish() && operation.GetOnlyCheckCommitedToFinish()) {
-            if (static_cast<ui64>(userInfo.Offset) != EndOffset) { // savnik что если откатят коммит пока выполняется транзакция
+        if (operation.HasOnlyCheckCommitedToFinish() && operation.GetOnlyCheckCommitedToFinish() && !IsActive()) {
+            if (static_cast<ui64>(userInfo.Offset) != EndOffset) {
                ok = false;
             }
         } else {

ydb/core/persqueue/ut/pqtablet_ut.cpp

@@ -320,8 +320,8 @@ void TPQTabletFixture::SendProposeTransactionRequest(const TProposeTransactionPa
             auto* operation = body->MutableOperations()->Add();
             operation->SetPartitionId(txOp.Partition);
             if (txOp.Begin.Defined()) {
-                operation->SetBegin(*txOp.Begin);
-                operation->SetEnd(*txOp.End);
+                operation->SetCommitOffsetsBegin(*txOp.Begin);
+                operation->SetCommitOffsetsEnd(*txOp.End);
                 operation->SetConsumer(*txOp.Consumer);
             }
             operation->SetPath(txOp.Path);

ydb/core/persqueue/ut/user_action_processor_ut.cpp

@@ -679,8 +679,8 @@ void TUserActionProcessorFixture::SendProposeTransactionRequest(const TProposeTr
     for (auto& txOp : params.TxOps) {
         auto* operation = body->MutableOperations()->Add();
         operation->SetPartitionId(txOp.Partition);
-        operation->SetBegin(txOp.Begin);
-        operation->SetEnd(txOp.End);
+        operation->SetCommitOffsetsBegin(txOp.Begin);
+        operation->SetCommitOffsetsEnd(txOp.End);
         operation->SetConsumer(txOp.Consumer);
         operation->SetPath(txOp.Path);
     }

ydb/core/persqueue/utils.cpp

@@ -266,7 +266,8 @@ std::unordered_map<ui32, TPartitionGraph::Node> BuildGraph(const TCollection& pa
     }
 
     std::deque<TPartitionGraph::Node*> queue;
-    for(const auto& p : partitions) {
+
+    for (const auto& p : partitions) {
         auto& node = result[GetPartitionId(p)];
 
         node.Children.reserve(p.ChildPartitionIdsSize());
@@ -284,27 +285,42 @@ std::unordered_map<ui32, TPartitionGraph::Node> BuildGraph(const TCollection& pa
         }
     }
 
-    while(!queue.empty()) {
+    while (!queue.empty()) {
         auto* n = queue.front();
         queue.pop_front();
 
         bool allCompleted = true;
-        for(auto* c : n->Parents) {
+        for (auto* c : n->Parents) {
             if (c->HierarhicalParents.empty() && !c->Parents.empty()) {
                 allCompleted = false;
                 break;
             }
         }
 
         if (allCompleted) {
-            for(auto* c : n->Parents) {
+            for (auto* c : n->Parents) {
                 n->HierarhicalParents.insert(c->HierarhicalParents.begin(), c->HierarhicalParents.end());
                 n->HierarhicalParents.insert(c);
             }
             queue.insert(queue.end(), n->Children.begin(), n->Children.end());
         }
     }
 
+    for (auto& [_, node] : result) {
+        queue.push_back(&node);
+
+        while (!queue.empty()) {
+            auto* current = queue.front();
+            queue.pop_front();
+
+            for (auto* child : current->Children) {
+                if (node.HierarhicalChildren.insert(child).second) {
+                    queue.push_back(child);
+                }
+            }
+        }
+    }
+
     return result;
 }
 

ydb/core/persqueue/utils.h

@@ -48,7 +48,9 @@ class TPartitionGraph {
         // Direct children of this node
         std::vector<Node*> Children;
         // All parents include parents of parents and so on
-        std::set<Node*> HierarhicalParents;
+        std::set<Node*> HierarhicalParents; // savnik rename to Ancestors?
+        // All children include children of children and so on
+        std::set<Node*> HierarhicalChildren;
 
         bool IsRoot() const;
         bool IsParent(ui32 partitionId) const;

ydb/services/persqueue_v1/actors/commit_offset_actor.cpp

@@ -115,7 +115,7 @@ void TCommitOffsetActor::Handle(TEvPQProxy::TEvAuthResultOk::TPtr& ev, const TAc
             commits.push_back(commit);
         }
 
-        for (auto& child: partitionNode->Children) {
+        for (auto& child: partitionNode->HierarhicalChildren) {
             TKqpHelper::TCommitInfo commit {.PartitionId = child->Id, .Offset = 0};
             commits.push_back(commit);
         }