Messages are written to the partition in two stages (#18118)

Author: Alek5andr-Kotov

ydb/core/http_proxy/ut/internal_counters.json

@@ -611,7 +611,7 @@
         "folder_id": "folder4",
         "database": "/Root"
         },
-        "value":243,
+        "value":245,
         "kind":"RATE"
     },
     {
@@ -768,6 +768,45 @@
       }
       ,"value": 0,
       "kind":"RATE"
-     }
+     },
+    {
+        "kind":"GAUGE",
+        "labels":
+        {
+            "cloud_id":"cloud4",
+            "database":"/Root",
+            "database_id":"database4",
+            "folder_id":"folder4",
+            "name":"topic.compaction.unprocessed_count_max",
+            "topic":"test-counters-stream"
+        },
+        "value":0
+    },
+    {
+        "kind":"GAUGE",
+        "labels":
+        {
+            "cloud_id":"cloud4",
+            "database":"/Root",
+            "database_id":"database4",
+            "folder_id":"folder4",
+            "name":"topic.compaction.unprocessed_bytes_max",
+            "topic":"test-counters-stream"
+        },
+        "value":0
+    },
+    {
+        "kind":"GAUGE",
+        "labels":
+        {
+            "cloud_id":"cloud4",
+            "database":"/Root",
+            "database_id":"database4",
+            "folder_id":"folder4",
+            "name":"topic.compaction.lag_milliseconds_max",
+            "topic":"test-counters-stream"
+        },
+        "value":0
+    }
   ]
 }

ydb/core/keyvalue/keyvalue_state.cpp

@@ -2122,7 +2122,7 @@ bool TKeyValueState::PrepareCmdRead(const TActorContext &ctx, NKikimrClient::TKe
         auto it = Index.find(request.GetKey());
         if (it == Index.end()) {
             response.Status = NKikimrProto::NODATA;
-            response.Message = "No such key Marker# KV48";
+            response.Message = "No such key Marker# KV48 (" + request.GetKey() + ")";
         } else {
             bool isOverrun = PrepareOneRead<NKikimrClient::TKeyValueRequest>(it->first, it->second, offset, size,
                     priority, 0, intermediate, response, outIsInlineOnly);

ydb/core/persqueue/blob.cpp

@@ -16,7 +16,9 @@ TBlobIterator::TBlobIterator(const TKey& key, const TString& blob)
     , Count(0)
     , InternalPartsCount(0)
 {
-    Y_ABORT_UNLESS(Data != End);
+    Y_ABORT_UNLESS(Data != End,
+                   "Key=%s, blob.size=%" PRISZT,
+                   Key.ToString().data(), blob.size());
     ParseBatch();
     Y_ABORT_UNLESS(Header.GetPartNo() == Key.GetPartNo());
 }
@@ -666,7 +668,9 @@ ui32 THead::GetCount() const
         return 0;
 
     //how much offsets before last batch and how much offsets in last batch
-    Y_ABORT_UNLESS(Batches.front().GetOffset() == Offset);
+    Y_ABORT_UNLESS(Batches.front().GetOffset() == Offset,
+                   "front.Offset=%" PRIu64 ", offset=%" PRIu64,
+                   Batches.front().GetOffset(), Offset);
 
     return Batches.back().GetOffset() - Offset + Batches.back().GetCount();
 }
@@ -778,6 +782,7 @@ TPartitionedBlob& TPartitionedBlob::operator=(const TPartitionedBlob& x)
     GlueNewHead = x.GlueNewHead;
     NeedCompactHead = x.NeedCompactHead;
     MaxBlobSize = x.MaxBlobSize;
+    FastWrite = x.FastWrite;
     return *this;
 }
 
@@ -803,11 +808,12 @@ TPartitionedBlob::TPartitionedBlob(const TPartitionedBlob& x)
     , GlueNewHead(x.GlueNewHead)
     , NeedCompactHead(x.NeedCompactHead)
     , MaxBlobSize(x.MaxBlobSize)
+    , FastWrite(x.FastWrite)
 {}
 
 TPartitionedBlob::TPartitionedBlob(const TPartitionId& partition, const ui64 offset, const TString& sourceId, const ui64 seqNo, const ui16 totalParts,
                                     const ui32 totalSize, THead& head, THead& newHead, bool headCleared, bool needCompactHead, const ui32 maxBlobSize,
-                                    const ui16 nextPartNo)
+                                    const ui16 nextPartNo, const bool fastWrite)
     : Partition(partition)
     , Offset(offset)
     , InternalPartsCount(0)
@@ -827,6 +833,7 @@ TPartitionedBlob::TPartitionedBlob(const TPartitionId& partition, const ui64 off
     , GlueNewHead(true)
     , NeedCompactHead(needCompactHead)
     , MaxBlobSize(maxBlobSize)
+    , FastWrite(fastWrite)
 {
     Y_ABORT_UNLESS(NewHead.Offset == Head.GetNextOffset() && NewHead.PartNo == 0 || headCleared || needCompactHead || Head.PackedSize == 0); // if head not cleared, then NewHead is going after Head
     if (!headCleared) {
@@ -883,8 +890,15 @@ auto TPartitionedBlob::CreateFormedBlob(ui32 size, bool useRename) -> std::optio
 
     Y_ABORT_UNLESS(NewHead.GetNextOffset() >= (GlueHead ? Head.Offset : NewHead.Offset));
 
-    auto tmpKey = TKey::ForBody(TKeyPrefix::TypeTmpData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
-    auto dataKey = TKey::ForBody(TKeyPrefix::TypeData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
+    TKey tmpKey, dataKey;
+
+    if (FastWrite) {
+        tmpKey = TKey::ForFastWrite(TKeyPrefix::TypeTmpData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
+        dataKey = TKey::ForFastWrite(TKeyPrefix::TypeData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
+    } else {
+        tmpKey = TKey::ForBody(TKeyPrefix::TypeTmpData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
+        dataKey = TKey::ForBody(TKeyPrefix::TypeData, Partition, StartOffset, StartPartNo, count, InternalPartsCount);
+    }
 
     StartOffset = Offset;
     StartPartNo = NextPartNo;
@@ -915,7 +929,9 @@ auto TPartitionedBlob::CreateFormedBlob(ui32 size, bool useRename) -> std::optio
 
 auto TPartitionedBlob::Add(TClientBlob&& blob) -> std::optional<TFormedBlobInfo>
 {
-    Y_ABORT_UNLESS(NewHead.Offset >= Head.Offset);
+    Y_ABORT_UNLESS(NewHead.Offset >= Head.Offset,
+                   "Head.Offset=%" PRIu64 ", NewHead.Offset=%" PRIu64,
+                   Head.Offset, NewHead.Offset);
     ui32 size = blob.GetBlobSize();
     Y_ABORT_UNLESS(InternalPartsCount < 1000); //just check for future packing
     if (HeadSize + BlobsSize + size + GetMaxHeaderSize() > MaxBlobSize) {
@@ -956,6 +972,7 @@ auto TPartitionedBlob::Add(const TKey& oldKey, ui32 size) -> std::optional<TForm
     }
 
     auto newKey = TKey::FromKey(oldKey, TKeyPrefix::TypeData, Partition, StartOffset);
+    newKey.SetFastWrite();
 
     FormedBlobs.emplace_back(oldKey, newKey, size);
 

ydb/core/persqueue/blob.h

@@ -323,7 +323,7 @@ class TPartitionedBlob {
 
     TPartitionedBlob(const TPartitionId& partition, const ui64 offset, const TString& sourceId, const ui64 seqNo,
                      const ui16 totalParts, const ui32 totalSize, THead& head, THead& newHead, bool headCleared, bool needCompactHead, const ui32 maxBlobSize,
-                     ui16 nextPartNo = 0);
+                     ui16 nextPartNo = 0, bool fastWrite = true);
 
     struct TFormedBlobInfo {
         TKey Key;
@@ -382,6 +382,7 @@ class TPartitionedBlob {
     bool GlueNewHead;
     bool NeedCompactHead;
     ui32 MaxBlobSize;
+    bool FastWrite = true;
 };
 
 }// NPQ

ydb/core/persqueue/cache_eviction.h

@@ -14,27 +14,29 @@ namespace NKikimr::NPQ {
         ui16 PartNo;
         ui32 Count; // have to be unique for {Partition, Offset, partNo}
         ui16 InternalPartsCount; // have to be unique for {Partition, Offset, partNo}
+        char Suffix;
 
-        TBlobId(const TPartitionId& partition, ui64 offset, ui16 partNo, ui32 count, ui16 internalPartsCount)
+        TBlobId(const TPartitionId& partition, ui64 offset, ui16 partNo, ui32 count, ui16 internalPartsCount, TMaybe<char> suffix)
             : Partition(partition)
             , Offset(offset)
             , PartNo(partNo)
             , Count(count)
             , InternalPartsCount(internalPartsCount)
+            , Suffix(suffix ? *suffix : '\0')
         {
         }
 
         bool operator<(const TBlobId& r) const {
             auto makeTuple = [](const TBlobId& v) {
-                return std::make_tuple(v.Partition, v.Offset, v.PartNo, v.Count, v.InternalPartsCount);
+                return std::make_tuple(v.Partition, v.Offset, v.PartNo, v.Count, v.InternalPartsCount, v.Suffix);
             };
 
             return makeTuple(*this) < makeTuple(r);
         }
 
         bool operator==(const TBlobId& r) const {
             auto makeTuple = [](const TBlobId& v) {
-                return std::make_tuple(v.Partition, v.Offset, v.PartNo, v.Count, v.InternalPartsCount);
+                return std::make_tuple(v.Partition, v.Offset, v.PartNo, v.Count, v.InternalPartsCount, v.Suffix);
             };
 
             return makeTuple(*this) == makeTuple(r);
@@ -44,6 +46,7 @@ namespace NKikimr::NPQ {
             ui64 hash = Hash128to32((ui64(Partition.InternalPartitionId) << 17) + (Partition.IsSupportivePartition() ? 0 : (1 << 16)) + PartNo, Offset);
             hash = Hash128to32(hash, Count);
             hash = Hash128to32(hash, InternalPartsCount);
+            hash = Hash128to32(hash, Suffix);
             return hash;
         }
     };
@@ -58,7 +61,7 @@ struct THash<NKikimr::NPQ::TBlobId> {
 
 namespace NKikimr::NPQ {
     inline TBlobId MakeBlobId(const TPartitionId& partitionId, const TRequestedBlob& blob) {
-        return {partitionId, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount};
+        return {partitionId, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, blob.Key.GetSuffix()};
     }
 
     struct TKvRequest {
@@ -106,7 +109,7 @@ namespace NKikimr::NPQ {
             for (auto& blob : Blobs) {
                 if (blob.Value.empty()) {
                     // add reading command
-                    auto key = TKey::ForBody(TKeyPrefix::TypeData, Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount);
+                    const auto& key = blob.Key;
                     auto read = request->Record.AddCmdRead();
                     read->SetKey(key.Data(), key.Size());
                 }
@@ -142,9 +145,10 @@ namespace NKikimr::NPQ {
         }
 
         void Verify(const TRequestedBlob& blob) const {
-            auto key = TKey::ForBody(TKeyPrefix::TypeData, TPartitionId(0), blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount);
-            Y_ABORT_UNLESS(blob.Value.size() == blob.Size);
-            TClientBlob::CheckBlob(key, blob.Value);
+            Y_ABORT_UNLESS(blob.Value.size() == blob.Size,
+                           "\nblob.Value.size=%" PRISZT ", blob.Size=%" PRIu64 "\nblob.Key=%s",
+                           blob.Value.size(), blob.Size, blob.Key.ToString().data());
+            TClientBlob::CheckBlob(blob.Key, blob.Value);
         }
     };
 
@@ -271,7 +275,7 @@ namespace NKikimr::NPQ {
             for (const auto& blob : kvReq.Blobs) {
                 // Touching blobs in L2. We don't need data here
                 auto& blobs = blob.Cached ? reqData->RequestedBlobs : reqData->MissedBlobs;
-                blobs.emplace_back(kvReq.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, nullptr);
+                blobs.emplace_back(kvReq.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, blob.Key.GetSuffix(), nullptr);
             }
 
             auto l2Request = MakeHolder<TEvPqCache::TEvCacheL2Request>(reqData.Release());
@@ -298,7 +302,7 @@ namespace NKikimr::NPQ {
 
                 // there could be a new blob with same id (for big messages)
                 if (RemoveExists(ctx, blob)) {
-                    reqData.RemovedBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, reqBlob.Count, reqBlob.InternalPartsCount, nullptr);
+                    reqData.RemovedBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, reqBlob.Count, reqBlob.InternalPartsCount, reqBlob.Key.GetSuffix(), nullptr);
                 }
 
                 auto cached = std::make_shared<TCacheValue>(reqBlob.Value, ctx.SelfID, TAppData::TimeProvider->Now());
@@ -308,7 +312,7 @@ namespace NKikimr::NPQ {
                 if (L1Strategy)
                     L1Strategy->SaveHeadBlob(blob);
 
-                reqData.StoredBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, blob.Count, blob.InternalPartsCount, cached);
+                reqData.StoredBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, blob.Count, blob.InternalPartsCount, blob.Suffix, cached);
 
                 LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Caching head blob in L1. Partition "
                     << blob.Partition << " offset " << blob.Offset << " count " << blob.Count
@@ -322,10 +326,10 @@ namespace NKikimr::NPQ {
                 TPartitionId partitionId;
                 partitionId.OriginalPartitionId = FromString<ui32>(s.data() + 1, 10);
                 partitionId.InternalPartitionId = partitionId.OriginalPartitionId;
-                return {partitionId, 0, 0, 0, 0};
+                return {partitionId, 0, 0, 0, 0, Nothing()};
             } else {
                 auto key = TKey::FromString(s);
-                return {key.GetPartition(), key.GetOffset(), key.GetPartNo(), key.GetCount(), key.GetInternalPartsCount()};
+                return {key.GetPartition(), key.GetOffset(), key.GetPartNo(), key.GetCount(), key.GetInternalPartsCount(), key.GetSuffix()};
             }
         }
 
@@ -336,8 +340,8 @@ namespace NKikimr::NPQ {
                 TBlobId newBlob = MakeBlobId(newKey);
                 if (RenameExists(ctx, oldBlob, newBlob)) {
                     reqData.RenamedBlobs.emplace_back(std::piecewise_construct,
-                                                      std::make_tuple(oldBlob.Partition, oldBlob.Offset, oldBlob.PartNo, oldBlob.Count, oldBlob.InternalPartsCount, nullptr),
-                                                      std::make_tuple(newBlob.Partition, newBlob.Offset, newBlob.PartNo, newBlob.Count, newBlob.InternalPartsCount, nullptr));
+                                                      std::make_tuple(oldBlob.Partition, oldBlob.Offset, oldBlob.PartNo, oldBlob.Count, oldBlob.InternalPartsCount, oldBlob.Suffix, nullptr),
+                                                      std::make_tuple(newBlob.Partition, newBlob.Offset, newBlob.PartNo, newBlob.Count, newBlob.InternalPartsCount, newBlob.Suffix, nullptr));
 
                     LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Renaming head blob in L1. Old partition "
                                 << oldBlob.Partition << " old offset " << oldBlob.Offset << " old count " << oldBlob.Count
@@ -357,7 +361,7 @@ namespace NKikimr::NPQ {
                 for (auto i = lowerBound; i != upperBound; ++i) {
                     const auto& [blob, value] = *i;
 
-                    reqData.RemovedBlobs.emplace_back(blob.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, nullptr);
+                    reqData.RemovedBlobs.emplace_back(blob.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, blob.Suffix, nullptr);
                     Counters.Dec(value);
 
                     LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Deleting head blob in L1. Partition "
@@ -395,7 +399,7 @@ namespace NKikimr::NPQ {
                 Cache[blob] = valL1; // weak
                 Counters.Inc(valL1);
 
-                reqData->StoredBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, reqBlob.Count, reqBlob.InternalPartsCount, cached);
+                reqData->StoredBlobs.emplace_back(kvReq.Partition, reqBlob.Offset, reqBlob.PartNo, reqBlob.Count, reqBlob.InternalPartsCount, reqBlob.Key.GetSuffix(), cached);
 
                 LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Prefetched blob in L1. Partition "
                     << blob.Partition << " offset " << blob.Offset << " count " << blob.Count
@@ -450,7 +454,7 @@ namespace NKikimr::NPQ {
         void PrepareTouch(const TActorContext& ctx, THolder<TCacheL2Request>& reqData, const TDeque<TBlobId>& used)
         {
             for (auto& blob : used) {
-                reqData->ExpectedBlobs.emplace_back(blob.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, nullptr);
+                reqData->ExpectedBlobs.emplace_back(blob.Partition, blob.Offset, blob.PartNo, blob.Count, blob.InternalPartsCount, blob.Suffix, nullptr);
 
                 LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Touching blob. Partition "
                     << blob.Partition << " offset " << blob.Offset << " count " << blob.Count);
@@ -462,14 +466,18 @@ namespace NKikimr::NPQ {
             const auto it = Cache.find(blobId);
             if (it == Cache.end()) {
                 LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "No blob in L1. Partition "
-                    << blobId.Partition << " offset " << blobId.Offset << " actorID " << ctx.SelfID);
+                    << blobId.Partition << " offset " << blobId.Offset <<
+                    " partno " << blobId.PartNo << " count " << blobId.Count << " parts_count " << blobId.InternalPartsCount <<
+                    " actorID " << ctx.SelfID);
                 return nullptr;
             }
 
             TCacheValue::TPtr data = it->second.GetBlob();
             if (!data) {
                 LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Evicted blob in L1. Partition "
-                    << blobId.Partition << " offset " << blobId.Offset << " actorID " << ctx.SelfID);
+                    << blobId.Partition << " offset " << blobId.Offset <<
+                    " partno " << blobId.PartNo << " count " << blobId.Count << " parts_count " << blobId.InternalPartsCount <<
+                    " actorID " << ctx.SelfID);
                 RemoveBlob(it);
                 return nullptr;
             }
@@ -478,7 +486,8 @@ namespace NKikimr::NPQ {
 
             const TBlobId& blob = it->first;
             LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE, "Got data from cache. Partition "
-                << blob.Partition << " offset " << blob.Offset << " count " << blob.Count
+                << blob.Partition << " offset " << blob.Offset <<
+                " partno " << blob.PartNo << " count " << blob.Count << " parts_count " << blob.InternalPartsCount
                 << " source " << (ui32)it->second.Source << " size " << data->DataSize()
                 << " accessed " << data->GetAccessCount() << " times before, last time " << data->GetAccessTime());