Merge 40a5e60 into a328b30

Author: nshestakov

ydb/core/persqueue/blob.cpp

@@ -399,6 +399,10 @@ void TBatch::Pack() {
         Header.SetPayloadSize(PackedData.size());
     }
 
+    for (auto& b : Blobs) {
+        EndWriteTimestamp = std::max(EndWriteTimestamp, b.WriteTimestamp);
+    }
+
 
     TVector<TClientBlob> tmp;
     Blobs.swap(tmp);
@@ -414,11 +418,14 @@ void TBatch::Unpack() {
     UnpackTo(&Blobs);
     Y_ABORT_UNLESS(InternalPartsPos.empty());
     for (ui32 i = 0; i < Blobs.size(); ++i) {
-        if (!Blobs[i].IsLastPart())
+        auto& b = Blobs[i];
+        if (!b.IsLastPart()) {
             InternalPartsPos.push_back(i);
+        }
+        EndWriteTimestamp = std::max(EndWriteTimestamp, b.WriteTimestamp);
     }
     Y_ABORT_UNLESS(InternalPartsPos.size() == GetInternalPartsCount());
-    
+
     PackedData.Clear();
 }
 
@@ -978,4 +985,3 @@ bool TPartitionedBlob::IsNextPart(const TString& sourceId, const ui64 seqNo, con
 
 }// NPQ
 }// NKikimr
-

ydb/core/persqueue/blob.h

@@ -101,7 +101,7 @@ struct TClientBlob {
     void SerializeTo(TBuffer& buffer) const;
     static TClientBlob Deserialize(const char *data, ui32 size);
 
-    static void CheckBlob(const TKey& key, const TString& blob); 
+    static void CheckBlob(const TKey& key, const TString& blob);
 };
 
 static constexpr const ui32 MAX_BLOB_SIZE = 8_MB;
@@ -121,6 +121,7 @@ struct TBatch {
     TVector<ui32> InternalPartsPos;
     NKikimrPQ::TBatchHeader Header;
     TBuffer PackedData;
+    TInstant EndWriteTimestamp;
 
     TBatch()
         : Packed(false)
@@ -162,27 +163,42 @@ struct TBatch {
         Header.SetUnpackedSize(unpackedSize);
         Header.SetCount(count);
         Header.SetInternalPartsCount(InternalPartsPos.size());
+
+        EndWriteTimestamp = std::max(EndWriteTimestamp, b.WriteTimestamp);
     }
 
     ui64 GetOffset() const {
         return Header.GetOffset();
     }
+
     ui16 GetPartNo() const {
         return Header.GetPartNo();
     }
+
     ui32 GetUnpackedSize() const {
         return Header.GetUnpackedSize();
     }
+
     ui32 GetCount() const {
         return Header.GetCount();
     }
+
     ui16 GetInternalPartsCount() const {
         return Header.GetInternalPartsCount();
     }
+
     bool IsGreaterThan(ui64 offset, ui16 partNo) const {
         return GetOffset() > offset || GetOffset() == offset && GetPartNo() > partNo;
     }
 
+    bool Empty() const {
+        return Blobs.empty();
+    }
+
+    TInstant GetEndWriteTimestamp() const {
+        return EndWriteTimestamp;
+    }
+
     TBatch(const NKikimrPQ::TBatchHeader &header, const char* data)
         : Packed(true)
         , Header(header)
@@ -239,7 +255,7 @@ struct THead {
 private:
     std::deque<TBatch> Batches;
     ui16 InternalPartsCount = 0;
-    
+
     friend class TPartitionedBlob;
 
     class TBatchAccessor {

ydb/core/persqueue/partition.cpp

@@ -313,6 +313,14 @@ ui64 TPartition::ImportantClientsMinOffset() const {
     return minOffset;
 }
 
+TInstant TPartition::GetEndWriteTimestamp() const {
+    return EndWriteTimestamp;
+}
+
+THead& TPartition::GetHead() {
+    return Head;
+}
+
 void TPartition::HandleWakeup(const TActorContext& ctx) {
     FilterDeadlinedWrites(ctx);
 
@@ -356,6 +364,7 @@ void TPartition::AddMetaKey(TEvKeyValue::TEvRequest* request) {
     meta.SetStartOffset(StartOffset);
     meta.SetEndOffset(Max(NewHead.GetNextOffset(), EndOffset));
     meta.SetSubDomainOutOfSpace(SubDomainOutOfSpace);
+    meta.SetEndWriteTimestamp(PendingWriteTimestamp.MilliSeconds());
 
     if (IsSupportive()) {
         auto* counterData = meta.MutableCounterData();

ydb/core/persqueue/partition.h

@@ -116,6 +116,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
     friend TInitInfoRangeStep;
     friend TInitDataRangeStep;
     friend TInitDataStep;
+    friend TInitEndWriteTimestampStep;
 
     friend TPartitionSourceManager;
 
@@ -440,6 +441,7 @@ class TPartition : public TActorBootstrapped<TPartition> {
     void HandleOnInit(TEvPQ::TEvDeletePartition::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvPQ::TEvDeletePartition::TPtr& ev, const TActorContext& ctx);
 
+    ui64 GetReadOffset(ui64 offset, TMaybe<TInstant> readTimestamp) const;
 
 public:
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -470,6 +472,8 @@ class TPartition : public TActorBootstrapped<TPartition> {
     // Minimal offset, the data from which cannot be deleted, because it is required by an important consumer
     ui64 ImportantClientsMinOffset() const;
 
+    TInstant GetEndWriteTimestamp() const; // For tests only
+    THead& GetHead(); // For tests only
 
     //Bootstrap sends kvRead
     //Become StateInit
@@ -666,6 +670,8 @@ class TPartition : public TActorBootstrapped<TPartition> {
 //                          [DataKeysBody  ][DataKeysHead                      ]
     ui64 StartOffset;
     ui64 EndOffset;
+    TInstant EndWriteTimestamp;
+    TInstant PendingWriteTimestamp;
 
     ui64 WriteInflightSize;
     TActorId Tablet;

ydb/core/persqueue/partition_init.cpp

@@ -30,6 +30,7 @@ TInitializer::TInitializer(TPartition* partition)
     Steps.push_back(MakeHolder<TInitInfoRangeStep>(this));
     Steps.push_back(MakeHolder<TInitDataRangeStep>(this));
     Steps.push_back(MakeHolder<TInitDataStep>(this));
+    Steps.push_back(MakeHolder<TInitEndWriteTimestampStep>(this));
 
     CurrentStep = Steps.begin();
 }
@@ -316,6 +317,8 @@ void TInitMetaStep::LoadMeta(const NKikimrClient::TResponse& kvResponse, const T
            }
            */
         Partition()->SubDomainOutOfSpace = meta.GetSubDomainOutOfSpace();
+        Partition()->EndWriteTimestamp = TInstant::MilliSeconds(meta.GetEndWriteTimestamp());
+        Partition()->PendingWriteTimestamp = Partition()->EndWriteTimestamp;
         if (Partition()->IsSupportive()) {
             const auto& counterData = meta.GetCounterData();
             Partition()->BytesWrittenGrpc.SetSavedValue(counterData.GetBytesWrittenGrpc());
@@ -495,7 +498,7 @@ void TInitDataRangeStep::FillBlobsMetaData(const NKikimrClient::TKeyValueRespons
             if (k.GetPartNo() > 0) ++startOffset;
             head.PartNo = 0;
         } else {
-            Y_ABORT_UNLESS(endOffset <= k.GetOffset(), "%s", pair.GetKey().c_str());
+            Y_ABORT_UNLESS(endOffset <= k.GetOffset(), "%d <= %d %s", endOffset, k.GetOffset(), pair.GetKey().c_str());
             if (endOffset < k.GetOffset()) {
                 gapOffsets.push_back(std::make_pair(endOffset, k.GetOffset()));
                 gapSize += k.GetOffset() - endOffset;
@@ -619,7 +622,7 @@ void TInitDataStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActorConte
 
                 Y_ABORT_UNLESS(offset + 1 >= Partition()->StartOffset);
                 Y_ABORT_UNLESS(offset < Partition()->EndOffset);
-                Y_ABORT_UNLESS(size == read.GetValue().size());
+                Y_ABORT_UNLESS(size == read.GetValue().size(), "size=%d == read.GetValue().size() = %d", size, read.GetValue().size());
 
                 for (TBlobIterator it(key, read.GetValue()); it.IsValid(); it.Next()) {
                     head.AddBatch(it.GetBatch());
@@ -653,6 +656,117 @@ void TInitDataStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActorConte
 }
 
 
+//
+// TInitEndWriteTimestampStep
+//
+
+TInitEndWriteTimestampStep::TInitEndWriteTimestampStep(TInitializer* initializer)
+    : TBaseKVStep(initializer, "TInitEndWriteTimestampStep", true) {
+}
+
+void TInitEndWriteTimestampStep::Execute(const TActorContext &ctx) {
+    if (Partition()->EndWriteTimestamp != TInstant::Zero() || (Partition()->Head.GetBatches().empty() && Partition()->DataKeysBody.empty())) {
+        PQ_LOG_D("Initializing EndWriteTimestamp of the topic '" << Partition()->TopicName()
+            << "' partition " << Partition()->Partition
+            << " skiped because already initialized.");
+        return Done(ctx);
+    }
+
+    PQ_LOG_D("Initializing EndWriteTimestamp of the topic '" << Partition()->TopicName()
+            << "' partition " << Partition()->Partition
+            << ".");
+
+    auto& head = Partition()->Head;
+    if (!head.GetBatches().empty()) {
+        auto& batch = head.GetLastBatch();
+        Y_VERIFY (batch.Packed);
+
+        TVector<TClientBlob> result;
+        batch.UnpackTo(&result);
+        Y_VERIFY(!result.empty());
+
+        Partition()->EndWriteTimestamp = result.back().WriteTimestamp;
+        Partition()->PendingWriteTimestamp = Partition()->EndWriteTimestamp;
+
+        PQ_LOG_I("Initializing EndWriteTimestamp of the topic '" << Partition()->TopicName()
+            << "' partition " << Partition()->Partition
+            << " from head completed. Value " << Partition()->EndWriteTimestamp);
+
+        return Done(ctx);
+    }
+
+    if (Partition()->DataKeysBody.empty()) {
+        PQ_LOG_I("Initializing EndWriteTimestamp of the topic '" << Partition()->TopicName()
+            << "' partition " << Partition()->Partition
+            << " skiped because DataKeys is empty.");
+
+        return Done(ctx);
+    }
+
+    auto& p = Partition()->DataKeysBody.back();
+
+    THolder<TEvKeyValue::TEvRequest> request(new TEvKeyValue::TEvRequest);
+    auto read = request->Record.AddCmdRead();
+    read->SetKey({p.Key.Data(), p.Key.Size()});
+
+    ctx.Send(Partition()->Tablet, request.Release());
+}
+
+void TInitEndWriteTimestampStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActorContext &ctx) {
+    if (!ValidateResponse(*this, ev, ctx)) {
+        PoisonPill(ctx);
+        return;
+    }
+
+    auto& response = ev->Get()->Record;
+    Y_ABORT_UNLESS(1 == response.ReadResultSize());
+
+    auto& read = response.GetReadResult(0);
+    Y_ABORT_UNLESS(read.HasStatus());
+    switch(read.GetStatus()) {
+        case NKikimrProto::OK: {
+            auto& key = Partition()->DataKeysBody.back().Key;
+
+            for (TBlobIterator it(key, read.GetValue()); it.IsValid(); it.Next()) {
+                auto b = it.GetBatch();
+                b.Unpack();
+                if (!b.Empty()) {
+                    Partition()->EndWriteTimestamp = b.GetEndWriteTimestamp();
+                }
+            }
+
+            PQ_LOG_I("Initializing EndWriteTimestamp of the topic '" << Partition()->TopicName()
+                << "' partition " << Partition()->Partition
+                << " from last blob completed. Value " << Partition()->EndWriteTimestamp);
+
+            Partition()->PendingWriteTimestamp = Partition()->EndWriteTimestamp;
+
+            break;
+            }
+        case NKikimrProto::OVERRUN:
+            Y_ABORT("implement overrun in readresult!!");
+            return;
+        case NKikimrProto::NODATA:
+            Y_ABORT("NODATA can't be here");
+            return;
+        case NKikimrProto::ERROR:
+            PQ_LOG_ERROR("tablet " << Partition()->TabletID << " HandleOnInit topic '" << TopicName()
+                    << "' partition " << PartitionId()
+                    << "  status NKikimrProto::ERROR result message: \"" << read.GetMessage()
+                    << " \" errorReason: \"" << response.GetErrorReason() << "\""
+            );
+            PoisonPill(ctx);
+            return;
+        default:
+            Cerr << "ERROR " << read.GetStatus() << " message: \"" << read.GetMessage() << "\"\n";
+            Y_ABORT("bad status");
+
+    };
+
+    Done(ctx);
+}
+
+
 //
 // TPartition
 //

ydb/core/persqueue/partition_init.h

@@ -152,4 +152,13 @@ class TInitDataStep: public TBaseKVStep {
     void Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext& ctx) override;
 };
 
+class TInitEndWriteTimestampStep: public TBaseKVStep {
+    friend class TPartitionTestWrapper;
+public:
+    TInitEndWriteTimestampStep(TInitializer* initializer);
+
+    void Execute(const TActorContext& ctx) override;
+    void Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext& ctx) override;
+};
+
 } // NKikimr::NPQ

ydb/core/persqueue/partition_monitoring.cpp

@@ -59,6 +59,7 @@ void TPartition::HandleMonitoring(TEvPQ::TEvMonRequest::TPtr& ev, const TActorCo
                         PROPERTY("StartOffset", StartOffset);
                         PROPERTY("EndOffset", EndOffset);
                         PROPERTY("LastOffset", Head.GetNextOffset());
+                        PROPERTY("Last message WriteTimestamp", EndWriteTimestamp.ToRfc822String());
                         PROPERTY("HeadOffset", Head.Offset << ", count: " << Head.GetCount());
                     }
                 }