Split local db transaction in partition tablat for batch compaction: handle eash range in separate transaction

cloud/blockstore/config/storage.proto

@@ -1419,4 +1419,7 @@ message TStorageServiceConfig
 
     // Timeout for attach/detach path requests (in milliseconds).
     optional uint32 AttachDetachPathRequestTimeout = 480;
+
+    // Execute separate compaction transaction for each range of the batch.
+    optional bool SplitTxInBatchCompactionEnabled = 481;
 }

cloud/blockstore/libs/storage/core/config.cpp

@@ -209,6 +209,7 @@ NProto::TLinkedDiskFillBandwidth GetBandwidth(
     xxx(ForcedCompactionRangeCountPerRun,       ui32,   1                     )\
     xxx(CompactionCountPerRunChangingPeriod,    TDuration, Seconds(60)        )\
     xxx(BatchCompactionEnabled,                 bool,   false                 )\
+    xxx(SplitTxInBatchCompactionEnabled,        bool,   false                 )\
     xxx(BlobPatchingEnabled,                    bool,   false                 )\
     /* If threshold is not 0, use it */                                        \
     xxx(MaxDiffPercentageForBlobPatching,       ui32,   0                     )\
@@ -689,6 +690,7 @@ BLOCKSTORE_STORAGE_CONFIG(BLOCKSTORE_STORAGE_DECLARE_CONFIG)
     xxx(FreshChannelWriteRequests)                                             \
     xxx(MixedIndexCacheV1)                                                     \
     xxx(BatchCompaction)                                                       \
+    xxx(SplitTxInBatchCompaction)                                              \
     xxx(BlobPatching)                                                          \
     xxx(UseRdma)                                                               \
     xxx(ChangeThrottlingPolicy)                                                \

cloud/blockstore/libs/storage/core/config.h

@@ -108,6 +108,7 @@ class TStorageConfig
     ui32 GetForcedCompactionRangeCountPerRun() const;
     TDuration GetCompactionCountPerRunChangingPeriod() const;
     bool GetBatchCompactionEnabled() const;
+    bool GetSplitTxInBatchCompactionEnabled() const;
     bool GetBlobPatchingEnabled() const;
     ui32 GetMaxDiffPercentageForBlobPatching() const;
 
@@ -343,6 +344,10 @@ class TStorageConfig
         const TString& cloudId,
         const TString& folderId,
         const TString& diskId) const;
+    bool IsSplitTxInBatchCompactionFeatureEnabled(
+        const TString& cloudId,
+        const TString& folderId,
+        const TString& diskId) const;
     bool IsBlobPatchingFeatureEnabled(
         const TString& cloudId,
         const TString& folderId,

cloud/blockstore/libs/storage/partition/model/affected.cpp

@@ -0,0 +1 @@
+#include "affected.h"

cloud/blockstore/libs/storage/partition/model/affected.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include <cloud/blockstore/libs/storage/partition/model/block_mask.h>
+#include <cloud/blockstore/libs/storage/protos/part.pb.h>
+
+#include <cloud/storage/core/libs/tablet/model/partial_blob_id.h>
+
+#include <util/generic/hash.h>
+#include <util/generic/maybe.h>
+#include <util/generic/vector.h>
+
+namespace NCloud::NBlockStore::NStorage::NPartition {
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TAffectedBlob
+{
+    TVector<ui16> Offsets;
+    TMaybe<TBlockMask> BlockMask;
+    TVector<ui32> AffectedBlockIndices;
+
+    // Filled only if a flag is set. BlobMeta is needed only to do some extra
+    // consistency checks.
+    TMaybe<NProto::TBlobMeta> BlobMeta;
+};
+
+using TAffectedBlobs = THashMap<TPartialBlobId, TAffectedBlob, TPartialBlobIdHash>;
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TAffectedBlock
+{
+    ui32 BlockIndex = 0;
+    ui64 CommitId = 0;
+};
+
+using TAffectedBlocks = TVector<TAffectedBlock>;
+
+}   // namespace NCloud::NBlockStore::NStorage::NPartition

cloud/blockstore/libs/storage/partition/model/commit_queue.cpp

@@ -7,7 +7,7 @@ namespace NCloud::NBlockStore::NStorage::NPartition {
 void TCommitQueue::Enqueue(TTxPtr tx, ui64 commitId)
 {
     if (Items) {
-        Y_ABORT_UNLESS(Items.back().CommitId < commitId);
+        Y_ABORT_UNLESS(Items.back().CommitId <= commitId);
     }
     Items.emplace_back(commitId, std::move(tx));
 }

cloud/blockstore/libs/storage/partition/part_actor.cpp

@@ -1043,6 +1043,7 @@ STFUNC(TPartitionActor::StateWork)
         IgnoreFunc(TEvPartitionPrivate::TEvCleanupResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvCollectGarbageResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvCompactionResponse);
+        IgnoreFunc(TEvPartitionPrivate::TEvCompactionTxResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvMetadataRebuildUsedBlocksResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvMetadataRebuildBlockCountResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvFlushResponse);
@@ -1106,6 +1107,7 @@ STFUNC(TPartitionActor::StateZombie)
         IgnoreFunc(TEvPartitionPrivate::TEvCleanupResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvCollectGarbageResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvCompactionResponse);
+        IgnoreFunc(TEvPartitionPrivate::TEvCompactionTxResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvMetadataRebuildUsedBlocksResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvMetadataRebuildBlockCountResponse);
         IgnoreFunc(TEvPartitionPrivate::TEvFlushResponse);

cloud/blockstore/libs/storage/partition/part_actor.h

@@ -2,6 +2,7 @@
 
 #include "public.h"
 
+#include "part_compaction.h"
 #include "part_counters.h"
 #include "part_events_private.h"
 #include "part_state.h"