-
Notifications
You must be signed in to change notification settings - Fork 606
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
provide bg tasks flow into schemeshard
- Loading branch information
1 parent
04c59b3
commit 903a699
Showing
38 changed files
with
825 additions
and
92 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,33 @@ | ||
| #include "control.h" | ||
| #include <ydb/core/tx/columnshard/bg_tasks/protos/data.pb.h> | ||
|
|
||
| namespace NKikimr::NOlap::NBackground { | ||
|
|
||
| NKikimrTxBackgroundProto::TSessionControlContainer TSessionControlContainer::SerializeToProto() const { | ||
| NKikimrTxBackgroundProto::TSessionControlContainer result; | ||
| result.SetSessionClassName(SessionClassName); | ||
| result.SetSessionIdentifier(SessionIdentifier); | ||
| result.SetStatusChannelContainer(ChannelContainer.SerializeToString()); | ||
| result.SetLogicControlContainer(LogicControlContainer.SerializeToString()); | ||
| return result; | ||
| } | ||
|
|
||
| NKikimr::TConclusionStatus TSessionControlContainer::DeserializeFromProto(const NKikimrTxBackgroundProto::TSessionControlContainer& proto) { | ||
| SessionClassName = proto.GetSessionClassName(); | ||
| SessionIdentifier = proto.GetSessionIdentifier(); | ||
| if (!SessionClassName) { | ||
| return TConclusionStatus::Fail("incorrect session class name for bg_task"); | ||
| } | ||
| if (!SessionIdentifier) { | ||
| return TConclusionStatus::Fail("incorrect session id for bg_task"); | ||
| } | ||
| if (!ChannelContainer.DeserializeFromString(proto.GetStatusChannelContainer())) { | ||
| return TConclusionStatus::Fail("cannot parse channel from proto"); | ||
| } | ||
| if (!LogicControlContainer.DeserializeFromString(proto.GetLogicControlContainer())) { | ||
| return TConclusionStatus::Fail("cannot parse logic from proto"); | ||
| } | ||
| return TConclusionStatus::Success(); | ||
| } | ||
|
|
||
| } |
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 |
|---|---|---|
| @@ -1,5 +1,28 @@ | ||
| #include "task.h" | ||
| #include <ydb/core/tx/columnshard/bg_tasks/protos/data.pb.h> | ||
|
|
||
| namespace NKikimr::NOlap::NBackground { | ||
|
|
||
| NKikimrTxBackgroundProto::TTaskContainer TTask::SerializeToProto() const { | ||
| NKikimrTxBackgroundProto::TTaskContainer result; | ||
| result.SetIdentifier(Identifier); | ||
| result.SetStatusChannelContainer(ChannelContainer.SerializeToString()); | ||
| result.SetTaskDescriptionContainer(DescriptionContainer.SerializeToString()); | ||
| return result; | ||
| } | ||
|
|
||
| NKikimr::TConclusionStatus TTask::DeserializeFromProto(const NKikimrTxBackgroundProto::TTaskContainer& proto) { | ||
| Identifier = proto.GetIdentifier(); | ||
| if (!Identifier) { | ||
| return TConclusionStatus::Fail("empty identifier is not correct for bg_task"); | ||
| } | ||
| if (!ChannelContainer.DeserializeFromString(proto.GetStatusChannelContainer())) { | ||
| return TConclusionStatus::Fail("cannot parse status channel from proto"); | ||
| } | ||
| if (!DescriptionContainer.DeserializeFromString(proto.GetTaskDescriptionContainer())) { | ||
| return TConclusionStatus::Fail("cannot parse task description from proto"); | ||
| } | ||
| return TConclusionStatus::Success(); | ||
| } | ||
|
|
||
| } |
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
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,65 @@ | ||
| #include "adapter.h" | ||
| #include <ydb/core/tx/schemeshard/schemeshard_schema.h> | ||
|
|
||
| namespace NKikimr::NSchemeShard::NBackground { | ||
|
|
||
| bool TAdapter::DoLoadSessionsFromLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, std::deque<NKikimr::NOlap::NBackground::TSessionRecord>& records) { | ||
| NIceDb::TNiceDb db(txc.DB); | ||
| using BackgroundSessions = NSchemeShard::Schema::BackgroundSessions; | ||
| auto rowset = db.Table<BackgroundSessions>().Select(); | ||
| if (!rowset.IsReady()) { | ||
| return false; | ||
| } | ||
|
|
||
| std::deque<NOlap::NBackground::TSessionRecord> result; | ||
| while (!rowset.EndOfSet()) { | ||
| NOlap::NBackground::TSessionRecord sRecord; | ||
| sRecord.SetClassName(rowset.GetValue<BackgroundSessions::ClassName>()); | ||
| sRecord.SetIdentifier(rowset.GetValue<BackgroundSessions::Identifier>()); | ||
| sRecord.SetLogicDescription(rowset.GetValue<BackgroundSessions::LogicDescription>()); | ||
| sRecord.SetStatusChannel(rowset.GetValue<BackgroundSessions::StatusChannel>()); | ||
| sRecord.SetProgress(rowset.GetValue<BackgroundSessions::Progress>()); | ||
| sRecord.SetState(rowset.GetValue<BackgroundSessions::State>()); | ||
| result.emplace_back(std::move(sRecord)); | ||
| if (!rowset.Next()) { | ||
| return false; | ||
| } | ||
| } | ||
| std::swap(result, records); | ||
| return true; | ||
| } | ||
|
|
||
| void TAdapter::DoSaveProgressToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) { | ||
| NIceDb::TNiceDb db(txc.DB); | ||
| using BackgroundSessions = NSchemeShard::Schema::BackgroundSessions; | ||
| db.Table<BackgroundSessions>().Key(container.GetClassName(), container.GetIdentifier()).Update( | ||
| NIceDb::TUpdate<BackgroundSessions::Progress>(container.GetProgress()) | ||
| ); | ||
| } | ||
|
|
||
| void TAdapter::DoSaveStateToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) { | ||
| NIceDb::TNiceDb db(txc.DB); | ||
| using BackgroundSessions = NSchemeShard::Schema::BackgroundSessions; | ||
| db.Table<BackgroundSessions>().Key(container.GetClassName(), container.GetIdentifier()).Update( | ||
| NIceDb::TUpdate<BackgroundSessions::State>(container.GetState()) | ||
| ); | ||
| } | ||
|
|
||
| void TAdapter::DoSaveSessionToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) { | ||
| NIceDb::TNiceDb db(txc.DB); | ||
| using BackgroundSessions = NSchemeShard::Schema::BackgroundSessions; | ||
| db.Table<BackgroundSessions>().Key(container.GetClassName(), container.GetIdentifier()).Update( | ||
| NIceDb::TUpdate<BackgroundSessions::LogicDescription>(container.GetLogicDescription()), | ||
| NIceDb::TUpdate<BackgroundSessions::StatusChannel>(container.GetStatusChannel()), | ||
| NIceDb::TUpdate<BackgroundSessions::Progress>(container.GetProgress()), | ||
| NIceDb::TUpdate<BackgroundSessions::State>(container.GetState()) | ||
| ); | ||
| } | ||
|
|
||
| void TAdapter::DoRemoveSessionFromLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const TString& className, const TString& identifier) { | ||
| NIceDb::TNiceDb db(txc.DB); | ||
| using BackgroundSessions = NSchemeShard::Schema::BackgroundSessions; | ||
| db.Table<BackgroundSessions>().Key(className, identifier).Delete(); | ||
| } | ||
|
|
||
| } |
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,14 @@ | ||
| #pragma once | ||
| #include <ydb/core/tx/columnshard/bg_tasks/abstract/adapter.h> | ||
|
|
||
| namespace NKikimr::NSchemeShard::NBackground { | ||
|
|
||
| class TAdapter: public NKikimr::NOlap::NBackground::ITabletAdapter { | ||
| protected: | ||
| virtual bool DoLoadSessionsFromLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, std::deque<NKikimr::NOlap::NBackground::TSessionRecord>& records) override; | ||
| virtual void DoSaveProgressToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) override; | ||
| virtual void DoSaveStateToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) override; | ||
| virtual void DoSaveSessionToLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const NKikimr::NOlap::NBackground::TSessionRecord& container) override; | ||
| virtual void DoRemoveSessionFromLocalDatabase(NTabletFlatExecutor::TTransactionContext& txc, const TString& className, const TString& identifier) override;; | ||
| }; | ||
| } |
Oops, something went wrong.