fix compaction intervals construction (#7176)

ydb-platform · Jul 29, 2024 · 123a335 · 123a335
1 parent bd147a9
commit 123a335
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 9 deletions.
diff --git a/ydb/core/formats/arrow/reader/position.h b/ydb/core/formats/arrow/reader/position.h
@@ -406,7 +406,7 @@ class TSortableBatchPosition {
 class TIntervalPosition {
 private:
     TSortableBatchPosition Position;
-    const bool LeftIntervalInclude;
+    bool LeftIntervalInclude;
 public:
     const TSortableBatchPosition& GetPosition() const {
         return Position;
@@ -458,21 +458,41 @@ class TIntervalPositions {
         return Positions.end();
     }
 
-    void AddPosition(TSortableBatchPosition&& position, const bool includeLeftInterval) {
-        TIntervalPosition intervalPosition(std::move(position), includeLeftInterval);
-        if (Positions.size()) {
-            AFL_VERIFY(Positions.back() < intervalPosition)("back", Positions.back().DebugJson())("pos", intervalPosition.DebugJson());
-        }
+    void InsertPosition(TIntervalPosition&& intervalPosition) {
         Positions.emplace_back(std::move(intervalPosition));
+        ui32 index = Positions.size() - 1;
+        while (index >= 1 && Positions[index] < Positions[index - 1]) {
+            std::swap(Positions[index], Positions[index - 1]);
+            index = index - 1;
+        }
+    }
+
+    void InsertPosition(TSortableBatchPosition&& position, const bool includePositionToLeftInterval) {
+        TIntervalPosition intervalPosition(std::move(position), includePositionToLeftInterval);
+        InsertPosition(std::move(intervalPosition));
     }
 
-    void AddPosition(const TSortableBatchPosition& position, const bool includeLeftInterval) {
-        TIntervalPosition intervalPosition(position, includeLeftInterval);
+    void InsertPosition(const TSortableBatchPosition& position, const bool includePositionToLeftInterval) {
+        TIntervalPosition intervalPosition(position, includePositionToLeftInterval);
+        InsertPosition(std::move(intervalPosition));
+    }
+
+    void AddPosition(TIntervalPosition&& intervalPosition) {
         if (Positions.size()) {
             AFL_VERIFY(Positions.back() < intervalPosition)("back", Positions.back().DebugJson())("pos", intervalPosition.DebugJson());
         }
         Positions.emplace_back(std::move(intervalPosition));
     }
+
+    void AddPosition(TSortableBatchPosition&& position, const bool includePositionToLeftInterval) {
+        TIntervalPosition intervalPosition(std::move(position), includePositionToLeftInterval);
+        AddPosition(std::move(intervalPosition));
+        }
+
+    void AddPosition(const TSortableBatchPosition& position, const bool includePositionToLeftInterval) {
+        TIntervalPosition intervalPosition(position, includePositionToLeftInterval);
+        AddPosition(std::move(intervalPosition));
+    }
 };
 
 class TRWSortableBatchPosition: public TSortableBatchPosition, public TMoveOnly {

diff --git a/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp b/ydb/core/tx/columnshard/engines/changes/general_compaction.cpp
@@ -212,7 +212,7 @@ NColumnShard::ECumulativeCounters TGeneralCompactColumnEngineChanges::GetCounter
 
 void TGeneralCompactColumnEngineChanges::AddCheckPoint(
     const NArrow::NMerger::TSortableBatchPosition& position, const bool include) {
-    CheckPoints.AddPosition(position, include);
+    CheckPoints.InsertPosition(position, include);
 }
 
 std::shared_ptr<TGeneralCompactColumnEngineChanges::IMemoryPredictor> TGeneralCompactColumnEngineChanges::BuildMemoryPredictor() {