Skip to content

Commit 2ddd853

Browse files
GrigoriyPAGrigoriyPA
authored
Merge 7d5ba91 into 1159d25 
2 parents 1159d25 + 7d5ba91 commit 2ddd853

8 files changed

Lines changed: 112 additions & 10 deletions

File tree

‎ydb/core/kqp/gateway/behaviour/resource_pool_classifier/checker.cpp‎

Lines changed: 33 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "checker.h"
2+
#include "fetcher.h"
23

34
#include <ydb/core/base/path.h>
45
#include <ydb/core/cms/console/configs_dispatcher.h>
@@ -153,7 +154,7 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
153154
return;
154155
}
155156

156-
if (ev->Get()->NumberClassifiers >= CLASSIFIER_COUNT_LIMIT) {
157+
if (Context.GetActivityType() == NMetadata::NModifications::IOperationsManager::EActivityType::Create && ev->Get()->NumberClassifiers >= CLASSIFIER_COUNT_LIMIT) {
157158
FailAndPassAway(TStringBuilder() << "Number of resource pool classifiers reached limit in " << CLASSIFIER_COUNT_LIMIT);
158159
return;
159160
}
@@ -204,11 +205,28 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
204205
CheckFeatureFlag(ev->Get()->Config->GetFeatureFlags());
205206
}
206207

208+
void Handle(NMetadata::NProvider::TEvRefreshSubscriberData::TPtr& ev) {
209+
const auto& snapshot = ev->Get()->GetSnapshotAs<TResourcePoolClassifierSnapshot>();
210+
for (const auto& objectRecord : AlterContext.GetRestoreObjectIds().GetTableRecords()) {
211+
TResourcePoolClassifierConfig object;
212+
TResourcePoolClassifierConfig::TDecoder::DeserializeFromRecord(object, objectRecord);
213+
214+
if (!snapshot->GetClassifierConfig(CanonizePath(object.GetDatabase()), object.GetName())) {
215+
FailAndPassAway(TStringBuilder() << "Classifier with name " << object.GetName() << " not found in database " << object.GetDatabase());
216+
return;
217+
}
218+
}
219+
220+
ExistenceChecked = true;
221+
TryFinish();
222+
}
223+
207224
STRICT_STFUNC(StateFunc,
208225
hFunc(TEvPrivate::TEvRanksCheckerResponse, Handle);
209226
hFunc(TEvPrivate::TEvFetchDatabaseResponse, Handle);
210227
hFunc(TEvents::TEvUndelivered, Handle);
211228
hFunc(NConsole::TEvConfigsDispatcher::TEvGetConfigResponse, Handle);
229+
hFunc(NMetadata::NProvider::TEvRefreshSubscriberData, Handle)
212230
)
213231

214232
private:
@@ -237,14 +255,14 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
237255
}
238256

239257
Register(new TQueryRetryActor<TRanksCheckerActor, TEvPrivate::TEvRanksCheckerResponse, TString, TString, TString, std::unordered_map<i64, TString>>(
240-
SelfId(), Context.GetExternalData().GetDatabase(), AlterContext.SessionId, AlterContext.TransactionId, ranksToNames
258+
SelfId(), Context.GetExternalData().GetDatabase(), AlterContext.GetSessionId(), AlterContext.GetTransactionId(), ranksToNames
241259
));
242260
}
243261

244262
void CheckFeatureFlag(const NKikimrConfig::TFeatureFlags& featureFlags) {
245263
if (Context.GetActivityType() == NMetadata::NModifications::IOperationsManager::EActivityType::Drop) {
246264
FeatureFlagChecked = true;
247-
TryFinish();
265+
ValidateExistence();
248266
return;
249267
}
250268

@@ -258,6 +276,16 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
258276
}
259277

260278
FeatureFlagChecked = true;
279+
ValidateExistence();
280+
}
281+
282+
void ValidateExistence() {
283+
if (Context.GetActivityType() != NMetadata::NModifications::IOperationsManager::EActivityType::Create && NMetadata::NProvider::TServiceOperator::IsEnabled()) {
284+
Send(NMetadata::NProvider::MakeServiceId(SelfId().NodeId()), new NMetadata::NProvider::TEvAskSnapshot(std::make_shared<TResourcePoolClassifierSnapshotsFetcher>()));
285+
return;
286+
}
287+
288+
ExistenceChecked = true;
261289
TryFinish();
262290
}
263291

@@ -271,7 +299,7 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
271299
}
272300

273301
void TryFinish() {
274-
if (!FeatureFlagChecked || !RanksChecked) {
302+
if (!FeatureFlagChecked || !RanksChecked || !ExistenceChecked) {
275303
return;
276304
}
277305

@@ -286,6 +314,7 @@ class TResourcePoolClassifierPreparationActor : public TActorBootstrapped<TResou
286314
bool Serverless = false;
287315
bool FeatureFlagChecked = false;
288316
bool RanksChecked = false;
317+
bool ExistenceChecked = false;
289318

290319
NMetadata::NModifications::IAlterPreparationController<TResourcePoolClassifierConfig>::TPtr Controller;
291320
std::vector<TResourcePoolClassifierConfig> PatchedObjects;

‎ydb/core/kqp/gateway/behaviour/resource_pool_classifier/snapshot.cpp‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,4 +22,16 @@ TString TResourcePoolClassifierSnapshot::DoSerializeToString() const {
2222
return result.GetStringRobust();
2323
}
2424

25+
std::optional<TResourcePoolClassifierConfig> TResourcePoolClassifierSnapshot::GetClassifierConfig(const TString& database, const TString& name) const {
26+
const auto databaseIt = ResourcePoolClassifierConfigs.find(database);
27+
if (databaseIt == ResourcePoolClassifierConfigs.end()) {
28+
return std::nullopt;
29+
}
30+
const auto configIt = databaseIt->second.find(name);
31+
if (configIt == databaseIt->second.end()) {
32+
return std::nullopt;
33+
}
34+
return configIt->second;
35+
}
36+
2537
} // namespace NKikimr::NKqp

‎ydb/core/kqp/gateway/behaviour/resource_pool_classifier/snapshot.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ class TResourcePoolClassifierSnapshot : public NMetadata::NFetcher::ISnapshot {
1919

2020
public:
2121
using TBase::TBase;
22+
23+
std::optional<TResourcePoolClassifierConfig> GetClassifierConfig(const TString& database, const TString& name) const;
2224
};
2325

2426
} // namespace NKikimr::NKqp

‎ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp‎

Lines changed: 40 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6802,7 +6802,8 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
68026802

68036803
// DROP RESOURCE POOL CLASSIFIER
68046804
checkQuery("DROP RESOURCE POOL CLASSIFIER MyResourcePoolClassifier;",
6805-
EStatus::SUCCESS);
6805+
EStatus::GENERIC_ERROR,
6806+
"Classifier with name MyResourcePoolClassifier not found in database /Root");
68066807
}
68076808

68086809
Y_UNIT_TEST(ResourcePoolClassifiersValidation) {
@@ -7027,6 +7028,27 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
70277028
}
70287029
}
70297030

7031+
Y_UNIT_TEST(AlterNonExistingResourcePoolClassifier) {
7032+
NKikimrConfig::TAppConfig config;
7033+
config.MutableFeatureFlags()->SetEnableResourcePools(true);
7034+
7035+
TKikimrRunner kikimr(NKqp::TKikimrSettings()
7036+
.SetAppConfig(config)
7037+
.SetEnableResourcePools(true));
7038+
7039+
auto db = kikimr.GetTableClient();
7040+
auto session = db.CreateSession().GetValueSync().GetSession();
7041+
7042+
auto query = R"(
7043+
ALTER RESOURCE POOL CLASSIFIER MyResourcePoolClassifier
7044+
SET (MEMBERNAME = "test@user", RANK = 100),
7045+
RESET (RESOURCE_POOL);
7046+
)";
7047+
auto result = session.ExecuteSchemeQuery(query).GetValueSync();
7048+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::GENERIC_ERROR, result.GetIssues().ToString());
7049+
UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(), "Classifier with name MyResourcePoolClassifier not found in database /Root");
7050+
}
7051+
70307052
Y_UNIT_TEST(DropResourcePoolClassifier) {
70317053
NKikimrConfig::TAppConfig config;
70327054
config.MutableFeatureFlags()->SetEnableResourcePools(true);
@@ -7055,6 +7077,23 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
70557077
UNIT_ASSERT_VALUES_EQUAL(FetchResourcePoolClassifiers(kikimr), "{\"resource_pool_classifiers\":[]}");
70567078
}
70577079
}
7080+
7081+
Y_UNIT_TEST(DropNonExistingResourcePoolClassifier) {
7082+
NKikimrConfig::TAppConfig config;
7083+
config.MutableFeatureFlags()->SetEnableResourcePools(true);
7084+
7085+
TKikimrRunner kikimr(NKqp::TKikimrSettings()
7086+
.SetAppConfig(config)
7087+
.SetEnableResourcePools(true));
7088+
7089+
auto db = kikimr.GetTableClient();
7090+
auto session = db.CreateSession().GetValueSync().GetSession();
7091+
7092+
auto query = "DROP RESOURCE POOL CLASSIFIER MyResourcePoolClassifier;";
7093+
auto result = session.ExecuteSchemeQuery(query).GetValueSync();
7094+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::GENERIC_ERROR, result.GetIssues().ToString());
7095+
UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(), "Classifier with name MyResourcePoolClassifier not found in database /Root");
7096+
}
70587097
}
70597098

70607099
Y_UNIT_TEST_SUITE(KqpOlapScheme) {

‎ydb/services/metadata/manager/abstract.h‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,17 @@ namespace NKikimr::NMetadata::NModifications {
1919

2020
using TOperationParsingResult = TConclusion<NInternal::TTableRecord>;
2121

22-
struct TAlterOperationContext {
23-
TString SessionId;
24-
TString TransactionId;
22+
class TAlterOperationContext {
23+
private:
24+
YDB_READONLY_DEF(TString, SessionId);
25+
YDB_READONLY_DEF(TString, TransactionId);
26+
YDB_READONLY_DEF(NInternal::TTableRecords, RestoreObjectIds);
27+
public:
28+
TAlterOperationContext(const TString& sessionId, const TString& transactionId, const NInternal::TTableRecords& RestoreObjectIds)
29+
: SessionId(sessionId)
30+
, TransactionId(transactionId)
31+
, RestoreObjectIds(RestoreObjectIds) {
32+
}
2533
};
2634

2735
class TColumnInfo {

‎ydb/services/metadata/manager/alter_impl.h‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -148,8 +148,7 @@ class TModificationActorImpl: public NActors::TActorBootstrapped<TModificationAc
148148
if (!PrepareRestoredObjects(objects)) {
149149
TBase::PassAway();
150150
} else {
151-
const TAlterOperationContext alterContext = {.SessionId = SessionId, .TransactionId = TransactionId};
152-
Manager->PrepareObjectsBeforeModification(std::move(objects), InternalController, Context, alterContext);
151+
Manager->PrepareObjectsBeforeModification(std::move(objects), InternalController, Context, TAlterOperationContext(SessionId, TransactionId, RestoreObjectIds));
153152
}
154153
}
155154

‎ydb/services/metadata/manager/table_record.cpp‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -288,4 +288,16 @@ TTableRecords TTableRecords::SelectColumns(const std::vector<TString>& columnIds
288288
return result;
289289
}
290290

291+
std::vector<TTableRecord> TTableRecords::GetTableRecords() const {
292+
std::vector<TTableRecord> result;
293+
result.reserve(Records.size());
294+
for (const Ydb::Value& record : Records) {
295+
result.emplace_back();
296+
for (size_t i = 0; i < std::min(Columns.size(), static_cast<size_t>(record.items_size())); ++i) {
297+
result.back().SetColumn(Columns[i].name(), record.items(i));
298+
}
299+
}
300+
return result;
301+
}
302+
291303
}

‎ydb/services/metadata/manager/table_record.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ class TTableRecords {
4545

4646
public:
4747
TTableRecords SelectColumns(const std::vector<TString>& columnIds) const;
48+
std::vector<TTableRecord> GetTableRecords() const;
4849

4950
Ydb::Table::ExecuteDataQueryRequest BuildInsertQuery(const TString& tablePath) const;
5051
Ydb::Table::ExecuteDataQueryRequest BuildUpsertQuery(const TString& tablePath) const;

0 commit comments

Comments
 (0)