stable-25-3-1: Empty blobs for compaction (#26519)

Author: ydbot

ydb/core/persqueue/pqtablet/partition/partition_compaction.cpp

@@ -178,41 +178,28 @@ void TPartition::TryRunCompaction()
     const ui64 blobsKeyCountLimit = GetBodyKeysCountLimit();
     const ui64 compactedBlobSizeLowerBound = GetCompactedBlobSizeLowerBound();
 
-    if (BlobEncoder.DataKeysBody.size() >= blobsKeyCountLimit) {
-        CompactionInProgress = true;
-        Send(SelfId(), new TEvPQ::TEvRunCompaction(BlobEncoder.DataKeysBody.size()));
+    if ((BlobEncoder.DataKeysBody.size() < blobsKeyCountLimit) && (BlobEncoder.GetSize() < GetCumulativeSizeLimit())) {
+        LOG_D("No data for blobs compaction");
         return;
     }
 
-    size_t blobsCount = 0, blobsSize = 0, totalSize = 0;
+    size_t blobsCount = 0, blobsSize = 0;
     for (; blobsCount < BlobEncoder.DataKeysBody.size(); ++blobsCount) {
         const auto& k = BlobEncoder.DataKeysBody[blobsCount];
         if (k.Size < compactedBlobSizeLowerBound) {
             // неполный блоб. можно дописать
             blobsSize += k.Size;
-            totalSize += k.Size;
             if (blobsSize > 2 * MaxBlobSize) {
                 // KV не может отдать много
                 blobsSize -= k.Size;
-                totalSize -= k.Size;
                 break;
             }
             LOG_D("Blob key for append " << k.Key.ToString());
         } else {
-            totalSize += k.Size;
             LOG_D("Blob key for rename " << k.Key.ToString());
         }
     }
-    LOG_D(blobsCount << " keys were taken away. Let's read " << blobsSize << " bytes (" << totalSize << ")");
-
-    if (totalSize < GetCumulativeSizeLimit()) {
-        LOG_D("Need more data for compaction. " <<
-                 "Blobs " << BlobEncoder.DataKeysBody.size() <<
-                 ", size " << totalSize << " (" << GetCumulativeSizeLimit() << ")");
-        return;
-    }
-
-    LOG_D("Run compaction for " << blobsCount << " blobs");
+    LOG_D(blobsCount << " keys were taken away. Let's read " << blobsSize << " bytes");
 
     CompactionInProgress = true;
 

ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp

@@ -3358,6 +3358,64 @@ Y_UNIT_TEST_F(Write_And_Read_Gigant_Messages_2, TFixtureNoClient)
     TestWriteAndReadMessages(4, 61'000'000, true);
 }
 
+Y_UNIT_TEST_F(Write_50k_100times_50tx, TFixtureTable)
+{
+    // 100 transactions. Write 100 50KB messages in each folder. Call the commit at the same time.
+    // As a result, there will be a lot of small blobs in the FastWrite zone of the main batch,
+    // which will be picked up by a compact. The scenario is similar to the work of Ya.Metrika.
+
+    const std::size_t PARTITIONS_COUNT = 2;
+    const std::size_t TXS_COUNT = 50;
+
+    auto makeSourceId = [](unsigned txId, unsigned partitionId) {
+        std::string sourceId = TEST_MESSAGE_GROUP_ID;
+        sourceId += "_";
+        sourceId += ToString(txId);
+        sourceId += "_";
+        sourceId += ToString(partitionId);
+        return sourceId;
+    };
+
+    CreateTopic("topic_A", TEST_CONSUMER, PARTITIONS_COUNT);
+
+    SetPartitionWriteSpeed("topic_A", 50'000'000);
+
+    std::vector<std::unique_ptr<TFixture::ISession>> sessions;
+    std::vector<std::unique_ptr<TTransactionBase>> transactions;
+
+    for (std::size_t i = 0; i < TXS_COUNT; ++i) {
+        sessions.push_back(CreateSession());
+        auto& session = sessions.back();
+
+        transactions.push_back(session->BeginTx());
+        auto& tx = transactions.back();
+
+        auto sourceId = makeSourceId(i, 0);
+        for (size_t j = 0; j < 100; ++j) {
+            WriteToTopic("topic_A", sourceId, std::string(50'000, 'x'), tx.get(), 0);
+        }
+        WaitForAcks("topic_A", sourceId);
+
+        sourceId = makeSourceId(i, 1);
+        WriteToTopic("topic_A", sourceId, std::string(50'000, 'x'), tx.get(), 1);
+        WaitForAcks("topic_A", sourceId);
+    }
+
+    // We are doing an asynchronous commit of transactions. They will be executed simultaneously.
+    std::vector<TAsyncStatus> futures;
+
+    for (std::size_t i = 0; i < TXS_COUNT; ++i) {
+        futures.push_back(sessions[i]->AsyncCommitTx(*transactions[i]));
+    }
+
+    // All transactions must be completed successfully.
+    for (std::size_t i = 0; i < TXS_COUNT; ++i) {
+        futures[i].Wait();
+        const auto& result = futures[i].GetValueSync();
+        UNIT_ASSERT_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+    }
+}
+
 }
 
 }