save the WriteId + PartitionId (#1037)

Author: Alek5andr-Kotov

ydb/core/persqueue/events/internal.h

@@ -1029,7 +1029,7 @@ struct TEvPQ {
         {
         }
 
-        ui32 Cookie; // ShadowPartitionId
+        ui32 Cookie; // InternalPartitionId
     };
 
     struct TEvGetWriteInfoResponse : public TEventLocal<TEvGetWriteInfoResponse, EvGetWriteInfoResponse> {
@@ -1044,14 +1044,14 @@ struct TEvPQ {
         {
         }
 
-        ui32 Cookie; // ShadowPartitionId
+        ui32 Cookie; // InternalPartitionId
         THashMap<TString, NPQ::TSeqNoRange> SeqNo; // SourceId -> (MinSeqNo, MaxSeqNo)
         std::deque<NPQ::TDataKey> BodyKeys;
         TVector<NPQ::TClientBlob> Head;
     };
 
     struct TEvGetWriteInfoError : public TEventLocal<TEvGetWriteInfoError, EvGetWriteInfoError> {
-        ui32 Cookie; // ShadowPartitionId
+        ui32 Cookie; // InternalPartitionId
         TString Message;
 
         TEvGetWriteInfoError(ui32 cookie, TString message) :

ydb/core/persqueue/partition_id.h

@@ -13,8 +13,10 @@ namespace NKikimr::NPQ {
 class TPartitionId {
 public:
     TPartitionId() = default;
+
     explicit TPartitionId(ui32 partition) :
-        TPartitionId(partition, Nothing(), partition)
+        OriginalPartitionId(partition),
+        InternalPartitionId(partition)
     {
     }
 
@@ -46,12 +48,9 @@ class TPartitionId {
         }
     }
 
-    //
-    // FIXME: используется в RequestRange
-    //
-    TPartitionId NextInternalPartitionId() const
+    bool IsSupportivePartition() const
     {
-        return {OriginalPartitionId, WriteId, InternalPartitionId + 1};
+        return WriteId.Defined();
     }
 
     ui32 OriginalPartitionId = 0;

ydb/core/persqueue/partition_init.cpp

@@ -990,32 +990,46 @@ bool DiskIsFull(TEvKeyValue::TEvResponse::TPtr& ev) {
     return !diskIsOk;
 }
 
+static std::pair<TKeyPrefix, TKeyPrefix> MakeKeyPrefixRange(TKeyPrefix::EType type, const TPartitionId& partition)
+{
+    TKeyPrefix from(type, partition);
+    TKeyPrefix to(type, TPartitionId(partition.OriginalPartitionId, partition.WriteId, partition.InternalPartitionId + 1));
+
+    return {std::move(from), std::move(to)};
+}
+
 static void RequestRange(const TActorContext& ctx, const TActorId& dst, const TPartitionId& partition,
                          TKeyPrefix::EType c, bool includeData = false, const TString& key = "", bool dropTmp = false) {
     THolder<TEvKeyValue::TEvRequest> request(new TEvKeyValue::TEvRequest);
-    auto read = request->Record.AddCmdReadRange();
-    auto range = read->MutableRange();
-    TKeyPrefix from(c, partition);
+
+    auto keyPrefixes = MakeKeyPrefixRange(c, partition);
+    TKeyPrefix& from = keyPrefixes.first;
+    const TKeyPrefix& to = keyPrefixes.second;
+
     if (!key.empty()) {
         Y_ABORT_UNLESS(key.StartsWith(TStringBuf(from.Data(), from.Size())));
         from.Clear();
         from.Append(key.data(), key.size());
     }
-    range->SetFrom(from.Data(), from.Size());
 
-    TKeyPrefix to(c, partition.NextInternalPartitionId());
+    auto read = request->Record.AddCmdReadRange();
+    auto range = read->MutableRange();
+
+    range->SetFrom(from.Data(), from.Size());
     range->SetTo(to.Data(), to.Size());
 
-    if(includeData)
+    if (includeData)
         read->SetIncludeData(true);
 
     if (dropTmp) {
+        keyPrefixes = MakeKeyPrefixRange(TKeyPrefix::TypeTmpData, partition);
+        const TKeyPrefix& from = keyPrefixes.first;
+        const TKeyPrefix& to = keyPrefixes.second;
+
         auto del = request->Record.AddCmdDeleteRange();
         auto range = del->MutableRange();
-        TKeyPrefix from(TKeyPrefix::TypeTmpData, partition);
-        range->SetFrom(from.Data(), from.Size());
 
-        TKeyPrefix to(TKeyPrefix::TypeTmpData, partition.NextInternalPartitionId());
+        range->SetFrom(from.Data(), from.Size());
         range->SetTo(to.Data(), to.Size());
     }