[-] левая граница диапазона ключей

Author: Alek5andr-Kotov

ydb/core/persqueue/key.cpp

@@ -1,9 +1,12 @@
 #include "key.h"
+#include <ydb/library/dbgtrace/debug_trace.h>
 
 namespace NKikimr::NPQ {
 
 std::pair<TKeyPrefix, TKeyPrefix> MakeKeyPrefixRange(TKeyPrefix::EType type, const TPartitionId& partition)
 {
+    DBGTRACE("MakeKeyPrefixRange");
+    DBGTRACE_LOG("type=" << (char)type << ", partition=" << partition);
     TKeyPrefix from(type, partition);
     TKeyPrefix to(type, TPartitionId(partition.OriginalPartitionId, partition.WriteId, partition.InternalPartitionId + 1));
 

ydb/core/persqueue/partition.cpp

@@ -19,6 +19,7 @@
 #include <util/folder/path.h>
 #include <util/string/escape.h>
 #include <util/system/byteorder.h>
+#include <ydb/library/dbgtrace/debug_trace.h>
 
 namespace {
 
@@ -364,6 +365,7 @@ void TPartition::HandleWakeup(const TActorContext& ctx) {
 }
 
 void TPartition::AddMetaKey(TEvKeyValue::TEvRequest* request) {
+    DBGTRACE("TPartition::AddMetaKey");
     //Set Start Offset
     auto write = request->Record.AddCmdWrite();
     TKeyPrefix ikey(TKeyPrefix::TypeMeta, Partition);
@@ -978,6 +980,7 @@ void TPartition::Handle(TEvPQ::TEvTxCalcPredicate::TPtr& ev, const TActorContext
 
 void TPartition::Handle(TEvPQ::TEvTxCommit::TPtr& ev, const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::Handle(TEvPQ::TEvTxCommit)");
     EndTransaction(*ev->Get(), ctx);
 
     TxInProgress = false;
@@ -987,6 +990,7 @@ void TPartition::Handle(TEvPQ::TEvTxCommit::TPtr& ev, const TActorContext& ctx)
 
 void TPartition::Handle(TEvPQ::TEvTxRollback::TPtr& ev, const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::Handle(TEvPQ::TEvTxRollback)");
     EndTransaction(*ev->Get(), ctx);
 
     TxInProgress = false;
@@ -996,6 +1000,7 @@ void TPartition::Handle(TEvPQ::TEvTxRollback::TPtr& ev, const TActorContext& ctx
 
 void TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse::TPtr& ev, const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse)");
     LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE,
                 "TEvGetWriteInfoResponse Cookie: " << ev->Get()->Cookie);
 
@@ -1017,6 +1022,7 @@ void TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse::TPtr& ev, const TActorCo
     WriteInfoResponse = ev->Release();
 
     if (auto* t = TryGetCurrentImmediateTransaction()) {
+        DBGTRACE_LOG("immediate tx");
         if (!WriteInfoResponse->BodyKeys.empty()) {
             predicate = false;
         }
@@ -1053,6 +1059,7 @@ void TPartition::Handle(TEvPQ::TEvGetWriteInfoResponse::TPtr& ev, const TActorCo
 
 void TPartition::Handle(TEvPQ::TEvGetWriteInfoError::TPtr& ev, const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::Handle(TEvPQ::TEvGetWriteInfoError)");
     LOG_DEBUG_S(ctx, NKikimrServices::PERSQUEUE,
                 "TEvGetWriteInfoError Cookie: " << ev->Get()->Cookie);
 
@@ -1061,6 +1068,7 @@ void TPartition::Handle(TEvPQ::TEvGetWriteInfoError::TPtr& ev, const TActorConte
     WriteInfoResponse = nullptr;
 
     if (auto* t = TryGetCurrentImmediateTransaction()) {
+        DBGTRACE_LOG("immediate tx");
         ScheduleReplyPropose(t->Record,
                              NKikimrPQ::TEvProposeTransactionResult::ABORTED,
                              NKikimrPQ::TError::BAD_REQUEST,
@@ -1648,7 +1656,9 @@ size_t TPartition::GetUserActCount(const TString& consumer) const
 
 void TPartition::ProcessTxsAndUserActs(const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::ProcessTxsAndUserActs");
     if (KVWriteInProgress || TxInProgress) {
+        DBGTRACE_LOG("skip");
         return;
     }
 
@@ -1661,12 +1671,16 @@ void TPartition::ProcessTxsAndUserActs(const TActorContext& ctx)
 
 void TPartition::ContinueProcessTxsAndUserActs(const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::ContinueProcessTxsAndUserActs");
+    DBGTRACE_LOG("Partition=" << Partition);
+    DBGTRACE_LOG("DeletePartitionState=" << (int)DeletePartitionState);
     Y_ABORT_UNLESS(!KVWriteInProgress);
     Y_ABORT_UNLESS(!TxInProgress);
 
     THolder<TEvKeyValue::TEvRequest> request(new TEvKeyValue::TEvRequest);
 
     if (DeletePartitionState == DELETION_INITED) {
+        DBGTRACE_LOG("deletion inited");
         ScheduleNegativeReplies();
         ScheduleDeletePartitionDone();
 
@@ -1682,6 +1696,7 @@ void TPartition::ContinueProcessTxsAndUserActs(const TActorContext& ctx)
     HaveWriteMsg = false;
 
     if (UserActionAndTransactionEvents.empty()) {
+        DBGTRACE_LOG("empty queue");
         const auto now = ctx.Now();
 
         if (ManageWriteTimestampEstimate) {
@@ -1697,6 +1712,7 @@ void TPartition::ContinueProcessTxsAndUserActs(const TActorContext& ctx)
         ProcessReserveRequests(ctx);
 
         if (haveChanges || TxIdHasChanged || !AffectedUsers.empty() || ChangeConfig) {
+            DBGTRACE_LOG("begin write");
             AddCmdWriteTxMeta(request->Record);
             AddCmdWriteUserInfos(request->Record);
             AddCmdWriteConfig(request->Record);
@@ -1711,6 +1727,7 @@ void TPartition::ContinueProcessTxsAndUserActs(const TActorContext& ctx)
             KVWriteInProgress = true;
             HaveWriteMsg = true;
         } else {
+            DBGTRACE_LOG("reply");
             AnswerCurrentWrites(ctx);
             AnswerCurrentReplies(ctx);
         }
@@ -1746,6 +1763,7 @@ void TPartition::ContinueProcessTxsAndUserActs(const TActorContext& ctx)
     }
 
     if (HaveWriteMsg) {
+        DBGTRACE_LOG("have write operation");
         if (!DiskIsFull) {
             EndAppendHeadWithNewWrites(request.Get(), ctx);
             EndProcessWrites(request.Get(), ctx);
@@ -3075,6 +3093,7 @@ void TPartition::ScheduleNegativeReplies()
 
 void TPartition::AddCmdDeleteRangeForAllKeys(TEvKeyValue::TEvRequest& request)
 {
+    DBGTRACE("TPartition::AddCmdDeleteRangeForAllKeys");
     NPQ::AddCmdDeleteRange(request, TKeyPrefix::TypeInfo, Partition);
     NPQ::AddCmdDeleteRange(request, TKeyPrefix::TypeData, Partition);
     NPQ::AddCmdDeleteRange(request, TKeyPrefix::TypeTmpData, Partition);

ydb/core/persqueue/partition_init.cpp

@@ -1,6 +1,7 @@
 #include "partition.h"
 #include "partition_util.h"
 #include <memory>
+#include <ydb/library/dbgtrace/debug_trace.h>
 
 namespace NKikimr::NPQ {
 
@@ -1009,6 +1010,8 @@ bool DiskIsFull(TEvKeyValue::TEvResponse::TPtr& ev) {
 
 void AddCmdDeleteRange(TEvKeyValue::TEvRequest& request, TKeyPrefix::EType c, const TPartitionId& partitionId)
 {
+    DBGTRACE("AddCmdDeleteRange");
+    DBGTRACE_LOG("partitionId=" << partitionId);
     auto keyPrefixes = MakeKeyPrefixRange(c, partitionId);
     const TKeyPrefix& from = keyPrefixes.first;
     const TKeyPrefix& to = keyPrefixes.second;
@@ -1017,7 +1020,12 @@ void AddCmdDeleteRange(TEvKeyValue::TEvRequest& request, TKeyPrefix::EType c, co
     auto range = del->MutableRange();
 
     range->SetFrom(from.Data(), from.Size());
+    range->SetIncludeFrom(true);
     range->SetTo(to.Data(), to.Size());
+    range->SetIncludeTo(false);
+
+    DBGTRACE_LOG("from=" << TString(from.Data(), from.Size()));
+    DBGTRACE_LOG("to=" << TString(to.Data(), to.Size()));
 }
 
 static void RequestRange(const TActorContext& ctx, const TActorId& dst, const TPartitionId& partition,

ydb/core/persqueue/partition_write.cpp

@@ -22,6 +22,7 @@
 #include <util/folder/path.h>
 #include <util/string/escape.h>
 #include <util/system/byteorder.h>
+#include <ydb/library/dbgtrace/debug_trace.h>
 
 namespace NKikimr::NPQ {
 
@@ -1444,6 +1445,7 @@ void TPartition::BeginHandleRequests(TEvKeyValue::TEvRequest* request, const TAc
 
 void TPartition::EndHandleRequests(TEvKeyValue::TEvRequest* request, const TActorContext& ctx)
 {
+    DBGTRACE("TPartition::EndHandleRequests");
     HaveDrop = CleanUp(request, ctx);
 
     ProcessReserveRequests(ctx);

ydb/core/persqueue/ya.make

@@ -74,6 +74,7 @@ PEERDIR(
     ydb/library/protobuf_printer
     ydb/public/lib/base
     ydb/public/sdk/cpp/client/ydb_persqueue_public
+    ydb/library/dbgtrace
 )
 
 END()

ydb/public/sdk/cpp/client/ydb_topic/ut/topic_to_table_ut.cpp

@@ -3,10 +3,12 @@
 #include <ydb/public/sdk/cpp/client/ydb_topic/topic.h>
 #include <ydb/public/sdk/cpp/client/ydb_table/table.h>
 #include <ydb/public/sdk/cpp/client/ydb_persqueue_public/ut/ut_utils/ut_utils.h>
+#include <ydb/core/keyvalue/keyvalue_events.h>
 
 #include <library/cpp/logger/stream.h>
 
 #include <library/cpp/testing/unittest/registar.h>
+#include <ydb/library/dbgtrace/debug_trace.h>
 
 namespace NYdb::NTopic::NTests {
 
@@ -77,6 +79,13 @@ class TFixture : public NUnitTest::TBaseFixture {
     void WaitForAcks(const TString& topicPath,
                      const TString& messageGroupId);
 
+    ui64 GetTopicTabletId(const TActorId& actorId,
+                          const TString& topicPath,
+                          ui32 partition);
+    void PrintTopicTabletId(const TString& topicPath);
+    THashSet<TString> GetTabletKeys(const TActorId& actorId,
+                                    ui64 tabletId);
+
 protected:
     const TDriver& GetDriver() const;
 
@@ -569,6 +578,102 @@ void TFixture::WaitForAcks(const TString& topicPath, const TString& messageGroup
     UNIT_ASSERT(context.AckCount == context.WriteCount);
 }
 
+ui64 TFixture::GetTopicTabletId(const TActorId& actorId, const TString& topicPath, ui32 partition)
+{
+    auto navigate = std::make_unique<NSchemeCache::TSchemeCacheNavigate>();
+    navigate->DatabaseName = "/Root";
+
+    NSchemeCache::TSchemeCacheNavigate::TEntry entry;
+    entry.Path = SplitPath(topicPath);
+    entry.SyncVersion = true;
+    entry.ShowPrivatePath = true;
+    entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
+
+    navigate->ResultSet.push_back(std::move(entry));
+    //navigate->UserToken = "root@builtin";
+    navigate->Cookie = 12345;
+
+    auto& runtime = Setup->GetRuntime();
+
+    runtime.Send(MakeSchemeCacheID(), actorId,
+                 new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()),
+                 0,
+                 true);
+    auto response = runtime.GrabEdgeEvent<TEvTxProxySchemeCache::TEvNavigateKeySetResult>();
+
+    UNIT_ASSERT_VALUES_EQUAL(response->Request->Cookie, 12345);
+    UNIT_ASSERT_VALUES_EQUAL(response->Request->ErrorCount, 0);
+
+    auto& front = response->Request->ResultSet.front();
+    UNIT_ASSERT(front.PQGroupInfo);
+    UNIT_ASSERT_GT(front.PQGroupInfo->Description.PartitionsSize(), 0);
+    UNIT_ASSERT_LT(partition, front.PQGroupInfo->Description.PartitionsSize());
+
+    for (size_t i = 0; i < front.PQGroupInfo->Description.PartitionsSize(); ++i) {
+        auto& p = front.PQGroupInfo->Description.GetPartitions(partition);
+        if (p.GetPartitionId() == partition) {
+            return p.GetTabletId();
+        }
+    }
+
+    UNIT_FAIL("unknown partition");
+
+    return Max<ui64>();
+}
+
+THashSet<TString> TFixture::GetTabletKeys(const TActorId& actorId, ui64 tabletId)
+{
+    using TEvKeyValue = NKikimr::TEvKeyValue;
+
+    auto request = std::make_unique<TEvKeyValue::TEvRequest>();
+    request->Record.SetCookie(12345);
+
+    auto cmd = request->Record.AddCmdReadRange();
+    TString from(1, '\x00');
+    TString to(1, '\xFF');
+    auto range = cmd->MutableRange();
+    range->SetFrom(from);
+    range->SetIncludeFrom(true);
+    range->SetTo(to);
+    range->SetIncludeTo(true);
+
+    auto& runtime = Setup->GetRuntime();
+
+    runtime.SendToPipe(tabletId, actorId, request.release());
+    auto response = runtime.GrabEdgeEvent<TEvKeyValue::TEvResponse>();
+
+    UNIT_ASSERT(response->Record.HasCookie());
+    UNIT_ASSERT_VALUES_EQUAL(response->Record.GetCookie(), 12345);
+    UNIT_ASSERT_VALUES_EQUAL(response->Record.ReadRangeResultSize(), 1);
+
+    THashSet<TString> keys;
+
+    auto& result = response->Record.GetReadRangeResult(0);
+    for (size_t i = 0; i < result.PairSize(); ++i) {
+        auto& kv = result.GetPair(i);
+        keys.insert(kv.GetKey());
+    }
+
+    return keys;
+}
+
+void TFixture::PrintTopicTabletId(const TString& topicPath)
+{
+    DBGTRACE("TFixture::PrintTopicTabletId");
+    auto& runtime = Setup->GetRuntime();
+    TActorId edge = runtime.AllocateEdgeActor();
+
+    ui64 tabletId = GetTopicTabletId(edge, topicPath, 0);
+    DBGTRACE_LOG("tabletId=" << tabletId);
+
+    auto keys = GetTabletKeys(edge, tabletId);
+    DBGTRACE_LOG("keys.size=" << keys.size());
+
+    for (auto& key : keys) {
+        DBGTRACE_LOG(key);
+    }
+}
+
 Y_UNIT_TEST_F(WriteToTopic_Demo_1, TFixture)
 {
     CreateTopic("topic_A");
@@ -966,6 +1071,21 @@ Y_UNIT_TEST_F(WriteToTopic_Demo_10, TFixture)
     }
 }
 
+Y_UNIT_TEST_F(WriteToTopic_Demo_11, TFixture)
+{
+    DBGTRACE("WriteToTopic_Demo_11");
+    CreateTopic("topic_A");
+    PrintTopicTabletId("/Root/topic_A");
+    NTable::TSession tableSession = CreateTableSession();
+    NTable::TTransaction tx_1 = BeginTx(tableSession);
+    WriteToTopic("topic_A", TEST_MESSAGE_GROUP_ID, "message #1", &tx_1);
+    WaitForAcks("topic_A", TEST_MESSAGE_GROUP_ID);
+    PrintTopicTabletId("/Root/topic_A");
+    CommitTx(tx_1, EStatus::SUCCESS);
+    Sleep(TDuration::Seconds(5));
+    PrintTopicTabletId("/Root/topic_A");
+}
+
 }
 
 }

ydb/public/sdk/cpp/client/ydb_topic/ut/ya.make

@@ -29,6 +29,7 @@ PEERDIR(
 
     ydb/core/tx/schemeshard/ut_helpers
     ydb/core/persqueue/ut/common
+    ydb/library/dbgtrace
 )
 
 YQL_LAST_ABI_VERSION()