clean configuration for splitting

ivanmorozov333 · ivanmorozov333 · commit e2e0636b299c · 2024-01-02T13:09:15.000+03:00
diff --git a/ydb/core/tx/columnshard/engines/changes/compaction.cpp b/ydb/core/tx/columnshard/engines/changes/compaction.cpp
@@ -31,13 +31,6 @@ bool TCompactColumnEngineChanges::DoApplyChanges(TColumnEngineForLogs& self, TAp
     return TBase::DoApplyChanges(self, context);
 }
 
-ui32 TCompactColumnEngineChanges::NumSplitInto(const ui32 srcRows) const {
-    Y_ABORT_UNLESS(srcRows > 1);
-    const ui64 totalBytes = TotalBlobsSize();
-    const ui32 numSplitInto = (totalBytes / Limits.GranuleSizeForOverloadPrevent) + 1;
-    return std::max<ui32>(2, numSplitInto);
-}
-
 void TCompactColumnEngineChanges::DoWriteIndex(NColumnShard::TColumnShard& self, TWriteIndexContext& context) {
     TBase::DoWriteIndex(self, context);
 }
@@ -75,11 +68,9 @@ void TCompactColumnEngineChanges::DoOnFinish(NColumnShard::TColumnShard& self, T
     NeedGranuleStatusProvide = false;
 }
 
-TCompactColumnEngineChanges::TCompactColumnEngineChanges(const TCompactionLimits& limits, std::shared_ptr<TGranuleMeta> granule, const std::vector<std::shared_ptr<TPortionInfo>>& portions, const TSaverContext& saverContext)
-    : TBase(limits.GetSplitSettings(), saverContext, StaticTypeName())
-    , Limits(limits)
-    , GranuleMeta(granule)
-{
+TCompactColumnEngineChanges::TCompactColumnEngineChanges(const TSplitSettings& splitSettings, std::shared_ptr<TGranuleMeta> granule, const std::vector<std::shared_ptr<TPortionInfo>>& portions, const TSaverContext& saverContext)
+    : TBase(splitSettings, saverContext, StaticTypeName())
+    , GranuleMeta(granule) {
     Y_ABORT_UNLESS(GranuleMeta);
 
     SwitchedPortions.reserve(portions.size());
diff --git a/ydb/core/tx/columnshard/engines/changes/compaction.h b/ydb/core/tx/columnshard/engines/changes/compaction.h
@@ -12,7 +12,6 @@ class TCompactColumnEngineChanges: public TChangesWithAppend {
     using TBase = TChangesWithAppend;
     bool NeedGranuleStatusProvide = false;
 protected:
-    const TCompactionLimits Limits;
     std::shared_ptr<TGranuleMeta> GranuleMeta;
 
     virtual void DoStart(NColumnShard::TColumnShard& self) override;
@@ -31,14 +30,12 @@ class TCompactColumnEngineChanges: public TChangesWithAppend {
 
     virtual THashSet<TPortionAddress> GetTouchedPortions() const override;
 
-    TCompactColumnEngineChanges(const TCompactionLimits& limits, std::shared_ptr<TGranuleMeta> granule, const std::vector<std::shared_ptr<TPortionInfo>>& portions, const TSaverContext& saverContext);
+    TCompactColumnEngineChanges(const TSplitSettings& splitSettings, std::shared_ptr<TGranuleMeta> granule, const std::vector<std::shared_ptr<TPortionInfo>>& portions, const TSaverContext& saverContext);
     ~TCompactColumnEngineChanges();
 
     static TString StaticTypeName() {
         return "CS::GENERAL";
     }
-
-    ui32 NumSplitInto(const ui32 srcRows) const;
 };
 
 }
diff --git a/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp b/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp
@@ -112,8 +112,8 @@ TConclusionStatus TGeneralCompactColumnEngineChanges::DoConstructBlobs(TConstruc
         std::map<std::string, std::vector<TColumnPortionResult>> columnChunks;
         ui32 batchIdx = 0;
         for (auto&& batchResult : batchResults) {
-            const ui32 portionRecordsCountLimit = batchResult->num_rows() / (batchResult->num_rows() / 10000 + 1) + 1;
-            TColumnMergeContext context(columnId, resultSchema, portionRecordsCountLimit, 50 * 1024 * 1024, columnInfo, SaverContext);
+            const ui32 portionRecordsCountLimit = batchResult->num_rows() / (batchResult->num_rows() / GetSplitSettings().GetExpectedRecordsCountOnPage() + 1) + 1;
+            TColumnMergeContext context(columnId, resultSchema, portionRecordsCountLimit, GetSplitSettings().GetExpectedUnpackColumnChunkRawSize(), columnInfo, SaverContext);
             TMergedColumn mColumn(context);
 
             auto columnPortionIdx = batchResult->GetColumnByName(portionIdFieldName);
@@ -177,8 +177,7 @@ TConclusionStatus TGeneralCompactColumnEngineChanges::DoConstructBlobs(TConstruc
             }
             batchSlices.emplace_back(portionColumns, schemaDetails, context.Counters.SplitterCounters, GetSplitSettings());
         }
-
-        TSimilarSlicer slicer(4 * 1024 * 1024);
+        TSimilarSlicer slicer(GetSplitSettings().GetExpectedPortionSize());
         auto packs = slicer.Split(batchSlices);
 
         ui32 recordIdx = 0;
diff --git a/ydb/core/tx/columnshard/engines/changes/with_appended.h b/ydb/core/tx/columnshard/engines/changes/with_appended.h
@@ -9,8 +9,9 @@ namespace NKikimr::NOlap {
 class TChangesWithAppend: public TColumnEngineChanges {
 private:
     using TBase = TColumnEngineChanges;
-    TSplitSettings SplitSettings;
+
 protected:
+    TSplitSettings SplitSettings;
     TSaverContext SaverContext;
     virtual void DoDebugString(TStringOutput& out) const override;
     virtual void DoCompile(TFinalizationContext& context) override;
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/intervals/blob_size.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/intervals/blob_size.cpp
@@ -30,7 +30,7 @@ std::shared_ptr<NKikimr::NOlap::TColumnEngineChanges> TBlobsWithSizeLimit::Build
         if (currentSum > SizeLimitToMerge || PortionsCount > CountLimitToMerge) {
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule_with_small")("portions", portions.size())("current_sum", currentSum);
             TSaverContext saverContext(StoragesManager->GetOperator(tierName.value_or(IStoragesManager::DefaultStorageId)), StoragesManager);
-            return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits, granule, portions, saverContext);
+            return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits.GetSplitSettings(), granule, portions, saverContext);
         } else {
             AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule_with_small")("skip", "not_enough_data");
         }
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/intervals/optimizer.cpp b/ydb/core/tx/columnshard/engines/storage/optimizer/intervals/optimizer.cpp
@@ -65,7 +65,7 @@ std::shared_ptr<TColumnEngineChanges> TIntervalsOptimizerPlanner::DoGetOptimizat
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule")("features", features.DebugJson().GetStringRobust())("count", features.GetPortionsCount());
 
     TSaverContext saverContext(StoragesManager->GetOperator(tierName.value_or(IStoragesManager::DefaultStorageId)), StoragesManager);
-    return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits, granule, portions, saverContext);
+    return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits.GetSplitSettings(), granule, portions, saverContext);
 }
 
 void TIntervalsOptimizerPlanner::RemovePortion(const std::shared_ptr<TPortionInfo>& info) {
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/lbuckets/optimizer.h b/ydb/core/tx/columnshard/engines/storage/optimizer/lbuckets/optimizer.h
@@ -747,7 +747,7 @@ class TPortionsBucket: public TMoveOnly {
         AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("stop_instant", stopInstant.value_or(TInstant::Zero()))("size", size)("next", NextBorder ? NextBorder->DebugString() : "")
             ("count", portions.size())("info", Others.DebugString())("event", "start_optimization")("stop_point", stopPoint ? stopPoint->DebugString() : "");
         TSaverContext saverContext(storagesManager->GetOperator(IStoragesManager::DefaultStorageId), storagesManager);
-        auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits, granule, portions, saverContext);
+        auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits.GetSplitSettings(), granule, portions, saverContext);
         if (MainPortion) {
             NIndexedReader::TSortableBatchPosition pos(MainPortion->IndexKeyStart().ToBatch(primaryKeysSchema), 0, primaryKeysSchema->field_names(), {}, false);
             result->AddCheckPoint(pos, true, false);
diff --git a/ydb/core/tx/columnshard/engines/storage/optimizer/levels/optimizer.h b/ydb/core/tx/columnshard/engines/storage/optimizer/levels/optimizer.h
@@ -450,7 +450,7 @@ class TLevel {
             positions.emplace_back(*position);
         }
         TSaverContext saverContext(StoragesManager->GetOperator(IStoragesManager::DefaultStorageId), StoragesManager);
-        auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(CompactionLimits, granule, portions, saverContext);
+        auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(CompactionLimits.GetSplitSettings(), granule, portions, saverContext);
         for (auto&& i : positions) {
             result->AddCheckPoint(i);
         }
diff --git a/ydb/core/tx/columnshard/normalizer/portion/normalizer.cpp b/ydb/core/tx/columnshard/normalizer/portion/normalizer.cpp
@@ -1,4 +1,5 @@
 #include "normalizer.h"
 
 namespace NKikimr::NOlap {
+    
 }
diff --git a/ydb/core/tx/columnshard/splitter/settings.h b/ydb/core/tx/columnshard/splitter/settings.h
@@ -1,5 +1,7 @@
 #pragma once
+
 #include <ydb/library/accessor/accessor.h>
+
 #include <util/system/types.h>
 
 namespace NKikimr::NOlap {
@@ -9,11 +11,22 @@ class TSplitSettings {
     static const inline i64 DefaultMaxBlobSize = 8 * 1024 * 1024;
     static const inline i64 DefaultMinBlobSize = 4 * 1024 * 1024;
     static const inline i64 DefaultMinRecordsCount = 10000;
-    static const inline i64 DefaultMaxPortionSize = 4 * DefaultMaxBlobSize;
+    static const inline i64 DefaultMaxPortionSize = 6 * DefaultMaxBlobSize;
     YDB_ACCESSOR(i64, MaxBlobSize, DefaultMaxBlobSize);
     YDB_ACCESSOR(i64, MinBlobSize, DefaultMinBlobSize);
     YDB_ACCESSOR(i64, MinRecordsCount, DefaultMinRecordsCount);
     YDB_ACCESSOR(i64, MaxPortionSize, DefaultMaxPortionSize);
 public:
+    ui64 GetExpectedRecordsCountOnPage() const {
+        return 1.5 * MinRecordsCount;
+    }
+
+    ui64 GetExpectedUnpackColumnChunkRawSize() const {
+        return (ui64)50 * 1024 * 1024;
+    }
+
+    ui64 GetExpectedPortionSize() const {
+        return MaxPortionSize;
+    }
 };
 }

Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ std::shared_ptr<NKikimr::NOlap::TColumnEngineChanges> TBlobsWithSizeLimit::Build`
`30`	`30`	`if (currentSum > SizeLimitToMerge \|\| PortionsCount > CountLimitToMerge) {`
`31`	`31`	`AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule_with_small")("portions", portions.size())("current_sum", currentSum);`
`32`	`32`	`TSaverContext saverContext(StoragesManager->GetOperator(tierName.value_or(IStoragesManager::DefaultStorageId)), StoragesManager);`
`33`		`- return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits, granule, portions, saverContext);`
	`33`	`+ return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits.GetSplitSettings(), granule, portions, saverContext);`
`34`	`34`	`} else {`
`35`	`35`	`AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule_with_small")("skip", "not_enough_data");`
`36`	`36`	`}`
Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ std::shared_ptr<TColumnEngineChanges> TIntervalsOptimizerPlanner::DoGetOptimizat`
`65`	`65`	`AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "take_granule")("features", features.DebugJson().GetStringRobust())("count", features.GetPortionsCount());`
`66`	`66`
`67`	`67`	`TSaverContext saverContext(StoragesManager->GetOperator(tierName.value_or(IStoragesManager::DefaultStorageId)), StoragesManager);`
`68`		`- return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits, granule, portions, saverContext);`
	`68`	`+ return std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(limits.GetSplitSettings(), granule, portions, saverContext);`
`69`	`69`	`}`
`70`	`70`
`71`	`71`	`void TIntervalsOptimizerPlanner::RemovePortion(const std::shared_ptr<TPortionInfo>& info) {`
Original file line number	Diff line number	Diff line change
`@@ -450,7 +450,7 @@ class TLevel {`
`450`	`450`	`positions.emplace_back(*position);`
`451`	`451`	`}`
`452`	`452`	`TSaverContext saverContext(StoragesManager->GetOperator(IStoragesManager::DefaultStorageId), StoragesManager);`
`453`		`- auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(CompactionLimits, granule, portions, saverContext);`
	`453`	`+ auto result = std::make_shared<NCompaction::TGeneralCompactColumnEngineChanges>(CompactionLimits.GetSplitSettings(), granule, portions, saverContext);`
`454`	`454`	`for (auto&& i : positions) {`
`455`	`455`	`result->AddCheckPoint(i);`
`456`	`456`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`#include "normalizer.h"`
`2`	`2`
`3`	`3`	`namespace NKikimr::NOlap {`
	`4`	`+`
`4`	`5`	`}`