multi layers compaction configuration (#18305)

ivanmorozov333 · ivanmorozov333 · web-flow · commit 805752b4efad · 2025-05-15T12:43:40.000+03:00
Co-authored-by: ivanmorozov333 &lt;imorozov333@ya.ru&gt;
diff --git a/ydb/core/protos/flat_scheme_op.proto b/ydb/core/protos/flat_scheme_op.proto
@@ -526,11 +526,13 @@ message TCompactionLevelConstructorContainer {
         optional uint64 ExpectedBlobsSize = 2;
         optional uint64 PortionsCountAvailable = 3;
         optional uint64 PortionsCountLimit = 4;
+        optional uint64 PortionsSizeLimit = 5;
     }
 
     message TOneLayer {
         optional uint32 ExpectedPortionSize = 1;
         optional float BytesLimitFraction = 2;
+        optional uint64 SizeLimitGuarantee = 3;
     }
 
     oneof Implementation {
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/one_layer.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/one_layer.cpp
@@ -8,6 +8,13 @@ TConclusionStatus TOneLayerConstructor::DoDeserializeFromJson(const NJson::TJson
     if (!json.IsMap()) {
         return TConclusionStatus::Fail("incorrect level description");
     }
+    if (json.Has("size_limit_guarantee")) {
+        const auto& jsonValue = json["size_limit_guarantee"];
+        if (!jsonValue.IsUInteger()) {
+            return TConclusionStatus::Fail("incorrect size_limit_guarantee value (have to be unsigned integer)");
+        }
+        SizeLimitGuarantee = jsonValue.GetUInteger();
+    }
     if (json.Has("bytes_limit_fraction")) {
         const auto& jsonValue = json["bytes_limit_fraction"];
         if (!jsonValue.IsDouble()) {
@@ -38,6 +45,9 @@ bool TOneLayerConstructor::DoDeserializeFromProto(const NKikimrSchemeOp::TCompac
     if (proto.GetOneLayer().HasBytesLimitFraction()) {
         BytesLimitFraction = proto.GetOneLayer().GetBytesLimitFraction();
     }
+    if (proto.GetOneLayer().HasSizeLimitGuarantee()) {
+        SizeLimitGuarantee = proto.GetOneLayer().GetSizeLimitGuarantee();
+    }
     return true;
 }
 
@@ -48,13 +58,16 @@ void TOneLayerConstructor::DoSerializeToProto(NKikimrSchemeOp::TCompactionLevelC
     if (BytesLimitFraction) {
         proto.MutableOneLayer()->SetBytesLimitFraction(*BytesLimitFraction);
     }
+    if (SizeLimitGuarantee) {
+        proto.MutableOneLayer()->SetSizeLimitGuarantee(SizeLimitGuarantee);
+    }
 }
 
 std::shared_ptr<NStorageOptimizer::NLCBuckets::IPortionsLevel> TOneLayerConstructor::DoBuildLevel(
     const std::shared_ptr<IPortionsLevel>& nextLevel, const ui32 indexLevel, const std::shared_ptr<TSimplePortionsGroupInfo>& portionsInfo,
     const TLevelCounters& counters) const {
-    return std::make_shared<TOneLayerPortions>(
-        indexLevel, BytesLimitFraction.value_or(1), ExpectedPortionSize.value_or(2 << 20), nextLevel, portionsInfo, counters);
+    return std::make_shared<TOneLayerPortions>(indexLevel, BytesLimitFraction.value_or(1), ExpectedPortionSize.value_or(2 << 20), nextLevel,
+        portionsInfo, counters, SizeLimitGuarantee);
 }
 
 }   // namespace NKikimr::NOlap::NStorageOptimizer::NLCBuckets
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/one_layer.h b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/one_layer.h
@@ -12,6 +12,7 @@ class TOneLayerConstructor: public ILevelConstructor {
 private:
     std::optional<double> BytesLimitFraction;
     std::optional<ui64> ExpectedPortionSize;
+    ui64 SizeLimitGuarantee = 0;
 
     virtual std::shared_ptr<IPortionsLevel> DoBuildLevel(const std::shared_ptr<IPortionsLevel>& nextLevel, const ui32 indexLevel,
         const std::shared_ptr<TSimplePortionsGroupInfo>& portionsInfo, const TLevelCounters& counters) const override;
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/zero_level.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/zero_level.cpp
@@ -40,6 +40,13 @@ TConclusionStatus TZeroLevelConstructor::DoDeserializeFromJson(const NJson::TJso
         }
         PortionsCountLimit = jsonValue.GetUInteger();
     }
+    if (json.Has("portions_size_limit")) {
+        const auto& jsonValue = json["portions_size_limit"];
+        if (!jsonValue.IsUInteger()) {
+            return TConclusionStatus::Fail("incorrect portions_size_limit value (have to be unsigned int)");
+        }
+        PortionsSizeLimit = jsonValue.GetUInteger();
+    }
     return TConclusionStatus::Success();
 }
 
@@ -59,6 +66,9 @@ bool TZeroLevelConstructor::DoDeserializeFromProto(const NKikimrSchemeOp::TCompa
     if (proto.GetZeroLevel().HasPortionsCountLimit()) {
         PortionsCountLimit = proto.GetZeroLevel().GetPortionsCountLimit();
     }
+    if (proto.GetZeroLevel().HasPortionsSizeLimit()) {
+        PortionsSizeLimit = proto.GetZeroLevel().GetPortionsSizeLimit();
+    }
     return true;
 }
 
@@ -75,13 +85,16 @@ void TZeroLevelConstructor::DoSerializeToProto(NKikimrSchemeOp::TCompactionLevel
     if (PortionsCountLimit) {
         proto.MutableZeroLevel()->SetPortionsCountLimit(*PortionsCountLimit);
     }
+    if (PortionsSizeLimit) {
+        proto.MutableZeroLevel()->SetPortionsSizeLimit(*PortionsSizeLimit);
+    }
 }
 
 std::shared_ptr<NKikimr::NOlap::NStorageOptimizer::NLCBuckets::IPortionsLevel> TZeroLevelConstructor::DoBuildLevel(
     const std::shared_ptr<IPortionsLevel>& nextLevel, const ui32 indexLevel, const std::shared_ptr<TSimplePortionsGroupInfo>& /*portionsInfo*/,
     const TLevelCounters& counters) const {
     return std::make_shared<TZeroLevelPortions>(indexLevel, nextLevel, counters, PortionsLiveDuration.value_or(TDuration::Max()),
-        ExpectedBlobsSize.value_or((ui64)1 << 20), PortionsCountAvailable.value_or(10), PortionsCountLimit);
+        ExpectedBlobsSize.value_or((ui64)1 << 20), PortionsCountAvailable.value_or(10), PortionsCountLimit, PortionsSizeLimit);
 }
 
 }   // namespace NKikimr::NOlap::NStorageOptimizer::NLCBuckets
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/zero_level.h b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/constructor/zero_level.h
@@ -14,6 +14,7 @@ class TZeroLevelConstructor: public ILevelConstructor {
     std::optional<ui64> ExpectedBlobsSize;
     std::optional<ui64> PortionsCountAvailable;
     std::optional<ui64> PortionsCountLimit;
+    std::optional<ui64> PortionsSizeLimit;
 
     virtual std::shared_ptr<IPortionsLevel> DoBuildLevel(const std::shared_ptr<IPortionsLevel>& nextLevel, const ui32 indexLevel,
         const std::shared_ptr<TSimplePortionsGroupInfo>& portionsInfo, const TLevelCounters& counters) const override;
@@ -23,7 +24,8 @@ class TZeroLevelConstructor: public ILevelConstructor {
     virtual bool IsEqualToSameClass(const ILevelConstructor& item) const override {
         const auto& itemCast = dynamic_cast<const TZeroLevelConstructor&>(item);
         return PortionsLiveDuration == itemCast.PortionsLiveDuration && ExpectedBlobsSize == itemCast.ExpectedBlobsSize &&
-               PortionsCountAvailable == itemCast.PortionsCountAvailable && PortionsCountLimit == itemCast.PortionsCountLimit;
+               PortionsCountAvailable == itemCast.PortionsCountAvailable && PortionsCountLimit == itemCast.PortionsCountLimit &&
+               PortionsSizeLimit == itemCast.PortionsSizeLimit;
     }
 
     static const inline TFactory::TRegistrator<TZeroLevelConstructor> Registrator =
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/common_level.h b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/common_level.h
@@ -12,12 +12,12 @@ class TOneLayerPortions: public IPortionsLevel {
     const TLevelCounters LevelCounters;
     const double BytesLimitFraction = 1;
     const ui64 ExpectedPortionSize = (1 << 20);
+    const ui64 SizeLimitGuarantee = 0;
     const bool StrictOneLayer = true;
     std::shared_ptr<TSimplePortionsGroupInfo> SummaryPortionsInfo;
 
     ui64 GetLevelBlobBytesLimit() const {
-        const ui32 discrete = SummaryPortionsInfo->GetBlobBytes() / (150 << 20);
-        return (discrete + 1) * (150 << 20) * BytesLimitFraction;
+        return std::max<ui64>(SizeLimitGuarantee, SummaryPortionsInfo->GetBlobBytes() * BytesLimitFraction);
     }
 
     virtual NJson::TJsonValue DoSerializeToJson() const override {
@@ -26,6 +26,7 @@ class TOneLayerPortions: public IPortionsLevel {
         result.InsertValue("bytes_limit", GetLevelBlobBytesLimit());
         result.InsertValue("total_bytes", SummaryPortionsInfo->GetBlobBytes());
         result.InsertValue("fraction", BytesLimitFraction);
+        result.InsertValue("size_limit_guarantee", SizeLimitGuarantee);
         return result;
     }
 
@@ -74,11 +75,12 @@ class TOneLayerPortions: public IPortionsLevel {
 public:
     TOneLayerPortions(const ui64 levelId, const double bytesLimitFraction, const ui64 expectedPortionSize,
         const std::shared_ptr<IPortionsLevel>& nextLevel, const std::shared_ptr<TSimplePortionsGroupInfo>& summaryPortionsInfo,
-        const TLevelCounters& levelCounters, const bool strictOneLayer = true)
+        const TLevelCounters& levelCounters, const ui64 sizeLimitGuarantee, const bool strictOneLayer = true)
         : TBase(levelId, nextLevel)
         , LevelCounters(levelCounters)
         , BytesLimitFraction(bytesLimitFraction)
         , ExpectedPortionSize(expectedPortionSize)
+        , SizeLimitGuarantee(sizeLimitGuarantee)
         , StrictOneLayer(strictOneLayer)
         , SummaryPortionsInfo(summaryPortionsInfo)
     {
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/optimizer.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/optimizer.cpp
@@ -31,11 +31,12 @@ TOptimizerPlanner::TOptimizerPlanner(const TInternalPathId pathId, const std::sh
         }
     } else {
         Levels.emplace_back(std::make_shared<TZeroLevelPortions>(
-            2, nullptr, Counters->GetLevelCounters(2), TDuration::Max(), 1 << 20, 10, std::nullopt));
+            2, nullptr, Counters->GetLevelCounters(2), TDuration::Max(), 1 << 20, 10, std::nullopt, std::nullopt));
+        Levels.emplace_back(
+            std::make_shared<TZeroLevelPortions>(
+            1, Levels.back(), Counters->GetLevelCounters(1), TDuration::Max(), 1 << 20, 10, std::nullopt, std::nullopt));
         Levels.emplace_back(std::make_shared<TZeroLevelPortions>(
-            1, Levels.back(), Counters->GetLevelCounters(1), TDuration::Max(), 1 << 20, 10, std::nullopt));
-        Levels.emplace_back(std::make_shared<TZeroLevelPortions>(
-            0, Levels.back(), Counters->GetLevelCounters(0), TDuration::Seconds(180), 1 << 20, 10, std::nullopt));
+            0, Levels.back(), Counters->GetLevelCounters(0), TDuration::Seconds(180), 1 << 20, 10, std::nullopt, std::nullopt));
     }
     std::reverse(Levels.begin(), Levels.end());
     RefreshWeights();
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/zero_level.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/zero_level.cpp
@@ -9,7 +9,7 @@ TCompactionTaskData TZeroLevelPortions::DoGetOptimizationTask() const {
         result.AddCurrentLevelPortion(
             i.GetPortion(), NextLevel->GetAffectedPortions(i.GetPortion()->IndexKeyStart(), i.GetPortion()->IndexKeyEnd()), true);
         if (!result.CanTakeMore()) {
-//            result.SetStopSeparation(i.GetPortion()->IndexKeyStart());
+            //            result.SetStopSeparation(i.GetPortion()->IndexKeyStart());
             break;
         }
     }
@@ -70,13 +70,14 @@ TInstant TZeroLevelPortions::DoGetWeightExpirationInstant() const {
 
 TZeroLevelPortions::TZeroLevelPortions(const ui32 levelIdx, const std::shared_ptr<IPortionsLevel>& nextLevel,
     const TLevelCounters& levelCounters, const TDuration durationToDrop, const ui64 expectedBlobsSize, const ui64 portionsCountAvailable,
-    const std::optional<ui64> portionsCountLimit)
+    const std::optional<ui64> portionsCountLimit, const std::optional<ui64> portionsSizeLimit)
     : TBase(levelIdx, nextLevel)
     , LevelCounters(levelCounters)
     , DurationToDrop(durationToDrop)
     , ExpectedBlobsSize(expectedBlobsSize)
     , PortionsCountAvailable(portionsCountAvailable)
-    , PortionsCountLimit(portionsCountLimit) {
+    , PortionsCountLimit(portionsCountLimit)
+    , PortionsSizeLimit(portionsSizeLimit) {
     if (DurationToDrop != TDuration::Max() && PredOptimization) {
         *PredOptimization -= TDuration::Seconds(RandomNumber<ui32>(DurationToDrop.Seconds()));
     }
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/zero_level.h b/ydb/core/tx/columnshard/engines/storage/optimizer/lcbuckets/planner/zero_level.h
@@ -12,6 +12,7 @@ class TZeroLevelPortions: public IPortionsLevel {
     const ui64 ExpectedBlobsSize;
     const ui64 PortionsCountAvailable;
     const std::optional<ui64> PortionsCountLimit;
+    const std::optional<ui64> PortionsSizeLimit;
     class TOrderedPortion {
     private:
         YDB_READONLY_DEF(TPortionInfo::TConstPtr, Portion);
@@ -41,10 +42,13 @@ class TZeroLevelPortions: public IPortionsLevel {
     std::set<TOrderedPortion> Portions;
 
     virtual bool IsOverloaded() const override {
-        if (!PortionsCountLimit) {
-            return false;
+        if (PortionsCountLimit && Portions.size() > *PortionsCountLimit) {
+            return true;
         }
-        return Portions.size() > *PortionsCountLimit;
+        if (PortionsSizeLimit && (ui64)PortionsInfo.GetBlobBytes() > (ui64)*PortionsSizeLimit) {
+            return true;
+        }
+        return false;
     }
 
     virtual NArrow::NMerger::TIntervalPositions DoGetBucketPositions(const std::shared_ptr<arrow::Schema>& /*pkSchema*/) const override {
@@ -103,7 +107,8 @@ class TZeroLevelPortions: public IPortionsLevel {
 
 public:
     TZeroLevelPortions(const ui32 levelIdx, const std::shared_ptr<IPortionsLevel>& nextLevel, const TLevelCounters& levelCounters,
-        const TDuration durationToDrop, const ui64 expectedBlobsSize, const ui64 portionsCountAvailable, const std::optional<ui64> portionsCountLimit);
+        const TDuration durationToDrop, const ui64 expectedBlobsSize, const ui64 portionsCountAvailable,
+        const std::optional<ui64> portionsCountLimit, const std::optional<ui64> portionsSizeLimit);
 };
 
 }   // namespace NKikimr::NOlap::NStorageOptimizer::NLCBuckets

Original file line number	Diff line number	Diff line change
`@@ -526,11 +526,13 @@ message TCompactionLevelConstructorContainer {`
`526`	`526`	`optional uint64 ExpectedBlobsSize = 2;`
`527`	`527`	`optional uint64 PortionsCountAvailable = 3;`
`528`	`528`	`optional uint64 PortionsCountLimit = 4;`
	`529`	`+ optional uint64 PortionsSizeLimit = 5;`
`529`	`530`	`}`
`530`	`531`
`531`	`532`	`message TOneLayer {`
`532`	`533`	`optional uint32 ExpectedPortionSize = 1;`
`533`	`534`	`optional float BytesLimitFraction = 2;`
	`535`	`+ optional uint64 SizeLimitGuarantee = 3;`
`534`	`536`	`}`
`535`	`537`
`536`	`538`	`oneof Implementation {`
Original file line number	Diff line number	Diff line change
`@@ -40,6 +40,13 @@ TConclusionStatus TZeroLevelConstructor::DoDeserializeFromJson(const NJson::TJso`
`40`	`40`	`}`
`41`	`41`	`PortionsCountLimit = jsonValue.GetUInteger();`
`42`	`42`	`}`
	`43`	`+ if (json.Has("portions_size_limit")) {`
	`44`	`+ const auto& jsonValue = json["portions_size_limit"];`
	`45`	`+ if (!jsonValue.IsUInteger()) {`
	`46`	`+ return TConclusionStatus::Fail("incorrect portions_size_limit value (have to be unsigned int)");`
	`47`	`+ }`
	`48`	`+ PortionsSizeLimit = jsonValue.GetUInteger();`
	`49`	`+ }`
`43`	`50`	`return TConclusionStatus::Success();`
`44`	`51`	`}`
`45`	`52`
`@@ -59,6 +66,9 @@ bool TZeroLevelConstructor::DoDeserializeFromProto(const NKikimrSchemeOp::TCompa`
`59`	`66`	`if (proto.GetZeroLevel().HasPortionsCountLimit()) {`
`60`	`67`	`PortionsCountLimit = proto.GetZeroLevel().GetPortionsCountLimit();`
`61`	`68`	`}`
	`69`	`+ if (proto.GetZeroLevel().HasPortionsSizeLimit()) {`
	`70`	`+ PortionsSizeLimit = proto.GetZeroLevel().GetPortionsSizeLimit();`
	`71`	`+ }`
`62`	`72`	`return true;`
`63`	`73`	`}`
`64`	`74`
`@@ -75,13 +85,16 @@ void TZeroLevelConstructor::DoSerializeToProto(NKikimrSchemeOp::TCompactionLevel`
`75`	`85`	`if (PortionsCountLimit) {`
`76`	`86`	`proto.MutableZeroLevel()->SetPortionsCountLimit(*PortionsCountLimit);`
`77`	`87`	`}`
	`88`	`+ if (PortionsSizeLimit) {`
	`89`	`+ proto.MutableZeroLevel()->SetPortionsSizeLimit(*PortionsSizeLimit);`
	`90`	`+ }`
`78`	`91`	`}`
`79`	`92`
`80`	`93`	`std::shared_ptr<NKikimr::NOlap::NStorageOptimizer::NLCBuckets::IPortionsLevel> TZeroLevelConstructor::DoBuildLevel(`
`81`	`94`	`const std::shared_ptr<IPortionsLevel>& nextLevel, const ui32 indexLevel, const std::shared_ptr<TSimplePortionsGroupInfo>& /portionsInfo/,`
`82`	`95`	`const TLevelCounters& counters) const {`
`83`	`96`	`return std::make_shared<TZeroLevelPortions>(indexLevel, nextLevel, counters, PortionsLiveDuration.value_or(TDuration::Max()),`
`84`		`- ExpectedBlobsSize.value_or((ui64)1 << 20), PortionsCountAvailable.value_or(10), PortionsCountLimit);`
	`97`	`+ ExpectedBlobsSize.value_or((ui64)1 << 20), PortionsCountAvailable.value_or(10), PortionsCountLimit, PortionsSizeLimit);`
`85`	`98`	`}`
`86`	`99`
`87`	`100`	`} // namespace NKikimr::NOlap::NStorageOptimizer::NLCBuckets`