Empty blobs for compaction (#26451)

Author: Alek5andr-Kotov

.github/last_commit.txt

@@ -1 +1 @@
-8ca586a7fefca64794ea9ae26aed6c720d216d32
+4c550f72db64962251932dab343839815916c555

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());
+    }
+}
+
 }
 
 }