-
Notifications
You must be signed in to change notification settings - Fork 610
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
19 changed files
with
337 additions
and
189 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| #include "changes.h" | ||
|
|
||
| namespace NKikimr::NOlap { | ||
|
|
||
| } |
37 changes: 37 additions & 0 deletions
37
ydb/core/tx/columnshard/engines/changes/abstract/changes.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| #pragma once | ||
| #include "abstract.h" | ||
|
|
||
| #include <ydb/core/tx/columnshard/data_locks/locks/abstract.h> | ||
|
|
||
| namespace NKikimr::NColumnShard { | ||
| class TColumnShard; | ||
| } | ||
|
|
||
| namespace NKikimr::NOlap { | ||
|
|
||
| class IChangeAction { | ||
| private: | ||
| virtual std::shared_ptr<NDataLocks::ILock> DoBuildDataLock(const TString& id, const NDataLocks::ELockCategory lockCategory) const = 0; | ||
| virtual void DoApplyOnExecute( | ||
| NColumnShard::TColumnShard* self, TWriteIndexContext& context, const TDataAccessorsResult& fetchedDataAccessors) = 0; | ||
| virtual void DoApplyOnComplete( | ||
| NColumnShard::TColumnShard* self, TWriteIndexCompleteContext& context, const TDataAccessorsResult& fetchedDataAccessors) = 0; | ||
|
|
||
| public: | ||
| IChangeAction() = default; | ||
|
|
||
| std::shared_ptr<NDataLocks::ILock> BuildDataLock(const TString& id, const NDataLocks::ELockCategory lockCategory) const { | ||
| return DoBuildDataLock(id, lockCategory); | ||
| } | ||
|
|
||
| void ApplyOnExecute(NColumnShard::TColumnShard* self, TWriteIndexContext& context, const TDataAccessorsResult& fetchedDataAccessors) { | ||
| return DoApplyOnExecute(self, context, fetchedDataAccessors); | ||
| } | ||
|
|
||
| void ApplyOnComplete( | ||
| NColumnShard::TColumnShard* self, TWriteIndexCompleteContext& context, const TDataAccessorsResult& fetchedDataAccessors) { | ||
| return DoApplyOnComplete(self, context, fetchedDataAccessors); | ||
| } | ||
| }; | ||
|
|
||
| } // namespace NKikimr::NOlap |
48 changes: 48 additions & 0 deletions
48
ydb/core/tx/columnshard/engines/changes/abstract/move_portions.cpp
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| #include "move_portions.h" | ||
|
|
||
| #include <ydb/core/tx/columnshard/counters/portions.h> | ||
| #include <ydb/core/tx/columnshard/engines/changes/counters/general.h> | ||
| #include <ydb/core/tx/columnshard/engines/column_engine_logs.h> | ||
|
|
||
| namespace NKikimr::NOlap { | ||
|
|
||
| std::shared_ptr<NDataLocks::ILock> TMovePortionsChange::DoBuildDataLock(const TString& id, const NDataLocks::ELockCategory lockCategory) const { | ||
| THashSet<TPortionAddress> portions; | ||
| for (auto&& i : Portions) { | ||
| AFL_VERIFY(portions.emplace(i.first).second); | ||
| } | ||
| return std::make_shared<NDataLocks::TListPortionsLock>(id, portions, lockCategory); | ||
| } | ||
|
|
||
| void TMovePortionsChange::DoApplyOnExecute( | ||
| NColumnShard::TColumnShard* self, TWriteIndexContext& context, const TDataAccessorsResult& fetchedDataAccessor) { | ||
| auto schemaPtr = context.EngineLogs.GetVersionedIndex().GetLastSchema(); | ||
| for (auto&& [_, i] : Portions) { | ||
| const auto pred = [&](TPortionInfo& portionCopy) { | ||
| portionCopy.MutableMeta().ResetCompactionLevel(TargetCompactionLevel.value_or(0)); | ||
| }; | ||
| context.EngineLogs.GetGranuleVerified(i->GetPathId()) | ||
| .ModifyPortionOnExecute(context.DBWrapper, fetchedDataAccessor.GetPortionAccessorVerified(i->GetPortionId()), pred, | ||
| schemaPtr->GetIndexInfo().GetPKFirstColumnId()); | ||
| } | ||
| } | ||
|
|
||
| void TMovePortionsChange::DoApplyOnComplete( | ||
| NColumnShard::TColumnShard* self, TWriteIndexCompleteContext& context, const TDataAccessorsResult& /*fetchedDataAccessor*/) { | ||
| if (!Portions.size()) { | ||
| return; | ||
| } | ||
| THashMap<ui32, TSimplePortionsGroupInfo> portionGroups; | ||
| for (auto&& [_, i] : Portions) { | ||
| portionGroups[i->GetMeta().GetCompactionLevel()].AddPortion(i); | ||
| } | ||
| NChanges::TGeneralCompactionCounters::OnMovePortionsByLevel(portionGroups, TargetCompactionLevel.value_or(0)); | ||
| for (auto&& [_, i] : Portions) { | ||
| const auto pred = [&](const std::shared_ptr<TPortionInfo>& portion) { | ||
| portion->MutableMeta().ResetCompactionLevel(TargetCompactionLevel.value_or(0)); | ||
| }; | ||
| context.EngineLogs.ModifyPortionOnComplete(i, pred); | ||
| } | ||
| } | ||
|
|
||
| } // namespace NKikimr::NOlap |
44 changes: 44 additions & 0 deletions
44
ydb/core/tx/columnshard/engines/changes/abstract/move_portions.h
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| #pragma once | ||
| #include "changes.h" | ||
|
|
||
| #include <ydb/core/tx/columnshard/engines/portions/portion_info.h> | ||
|
|
||
| namespace NKikimr::NOlap { | ||
|
|
||
| class TMovePortionsChange: public IChangeAction { | ||
| private: | ||
| THashMap<TPortionAddress, std::shared_ptr<const TPortionInfo>> Portions; | ||
| YDB_READONLY_DEF(std::optional<ui64>, TargetCompactionLevel); | ||
|
|
||
| virtual std::shared_ptr<NDataLocks::ILock> DoBuildDataLock(const TString& id, const NDataLocks::ELockCategory lockCategory) const override; | ||
| virtual void DoApplyOnExecute( | ||
| NColumnShard::TColumnShard* self, TWriteIndexContext& context, const TDataAccessorsResult& fetchedDataAccessor) override; | ||
| virtual void DoApplyOnComplete( | ||
| NColumnShard::TColumnShard* self, TWriteIndexCompleteContext& context, const TDataAccessorsResult& fetchedDataAccessor) override; | ||
|
|
||
| public: | ||
| const THashMap<TPortionAddress, TPortionInfo::TConstPtr>& GetPortionsToRemove() const { | ||
| return Portions; | ||
| } | ||
|
|
||
| ui32 GetSize() const { | ||
| return Portions.size(); | ||
| } | ||
|
|
||
| bool HasPortions() const { | ||
| return Portions.size(); | ||
| } | ||
|
|
||
| void AddPortions(const std::vector<std::shared_ptr<TPortionInfo>>& portions) { | ||
| for (auto&& i : portions) { | ||
| AFL_VERIFY(i); | ||
| AFL_VERIFY(Portions.emplace(i->GetAddress(), i).second)("portion_id", i->GetPortionId()); | ||
| } | ||
| } | ||
|
|
||
| void SetTargetCompactionLevel(const ui64 level) { | ||
| TargetCompactionLevel = level; | ||
| } | ||
| }; | ||
|
|
||
| } // namespace NKikimr::NOlap |
Oops, something went wrong.