Skip to content

Commit c024042

Browse files
stanislav-shchetininstanislav-shchetinin
authored
Merge 4a0dabe into a1750c6 
2 parents a1750c6 + 4a0dabe commit c024042

File tree

9 files changed

+178
-4
lines changed

9 files changed

+178
-4
lines changed

‎ydb/core/tx/schemeshard/schemeshard_import__create.cpp‎

Lines changed: 64 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -541,6 +541,21 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
541541
Send(Self->SelfId(), CreateChangefeedPropose(Self, txId, item));
542542
}
543543

544+
void CreateConsumers(TImportInfo::TPtr importInfo, ui32 itemIdx, TTxId txId) {
545+
Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
546+
auto& item = importInfo->Items.at(itemIdx);
547+
item.SubState = ESubState::Proposed;
548+
549+
LOG_I("TImport::TTxProgress: CreateConsumers propose"
550+
<< ": info# " << importInfo->ToString()
551+
<< ", item# " << item.ToString(itemIdx)
552+
<< ", txId# " << txId);
553+
554+
Y_ABORT_UNLESS(item.WaitTxId == InvalidTxId);
555+
556+
Send(Self->SelfId(), CreateConsumersPropose(Self, txId, item));
557+
}
558+
544559
void AllocateTxId(TImportInfo::TPtr importInfo, ui32 itemIdx) {
545560
Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
546561
auto& item = importInfo->Items.at(itemIdx);
@@ -622,6 +637,26 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
622637
return path->LastTxId;
623638
}
624639

640+
TTxId GetActiveCreateConsumerTxId(TImportInfo::TPtr importInfo, ui32 itemIdx) {
641+
Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
642+
const auto& item = importInfo->Items.at(itemIdx);
643+
644+
Y_ABORT_UNLESS(item.State == EState::CreateChangefeed);
645+
Y_ABORT_UNLESS(item.DstPathId);
646+
Y_ABORT_UNLESS(item.StreamImplPath);
647+
648+
if (!Self->PathsById.contains(item.StreamImplPath)) {
649+
return InvalidTxId;
650+
}
651+
652+
auto path = Self->PathsById.at(item.StreamImplPath);
653+
if (path->PathState != NKikimrSchemeOp::EPathStateAlter) {
654+
return InvalidTxId;
655+
}
656+
657+
return path->LastTxId;
658+
}
659+
625660
static TString MakeIndexBuildUid(TImportInfo::TPtr importInfo, ui32 itemIdx) {
626661
Y_ABORT_UNLESS(itemIdx < importInfo->Items.size());
627662
const auto& item = importInfo->Items.at(itemIdx);
@@ -1045,7 +1080,11 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
10451080
break;
10461081

10471082
case EState::CreateChangefeed:
1048-
CreateChangefeed(importInfo, i, txId);
1083+
if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
1084+
CreateChangefeed(importInfo, i, txId);
1085+
} else {
1086+
CreateConsumers(importInfo, i, txId);
1087+
}
10491088
itemIdx = i;
10501089
break;
10511090

@@ -1109,8 +1148,26 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
11091148
} else if (item.State == EState::Transferring) {
11101149
txId = GetActiveRestoreTxId(importInfo, itemIdx);
11111150
} else if (item.State == EState::CreateChangefeed) {
1112-
txId = GetActiveCreateChangefeedTxId(importInfo, itemIdx);
1151+
if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
1152+
txId = GetActiveCreateChangefeedTxId(importInfo, itemIdx);
1153+
} else {
1154+
txId = GetActiveCreateConsumerTxId(importInfo, itemIdx);
1155+
}
1156+
1157+
}
1158+
}
1159+
1160+
if (record.GetStatus() == NKikimrScheme::StatusAlreadyExists && item.State == EState::CreateChangefeed) {
1161+
if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
1162+
item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateConsumers;
1163+
AllocateTxId(importInfo, itemIdx);
1164+
} else if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
1165+
item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateChangefeed;
1166+
AllocateTxId(importInfo, itemIdx);
1167+
} else {
1168+
item.State = EState::Done;
11131169
}
1170+
return;
11141171
}
11151172

11161173
if (txId == InvalidTxId) {
@@ -1290,7 +1347,11 @@ struct TSchemeShard::TImport::TTxProgress: public TSchemeShard::TXxport::TTxBase
12901347
break;
12911348

12921349
case EState::CreateChangefeed:
1293-
if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
1350+
if (item.ChangefeedState == TImportInfo::TItem::EChangefeedState::CreateChangefeed) {
1351+
item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateConsumers;
1352+
AllocateTxId(importInfo, itemIdx);
1353+
} else if (++item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size()) {
1354+
item.ChangefeedState = TImportInfo::TItem::EChangefeedState::CreateChangefeed;
12941355
AllocateTxId(importInfo, itemIdx);
12951356
} else {
12961357
item.State = EState::Done;

‎ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.cpp‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
#include <ydb/core/base/path.h>
55
#include <ydb/core/ydb_convert/table_description.h>
66
#include <ydb/core/ydb_convert/ydb_convert.h>
7+
#include <ydb/services/lib/actors/pq_schema_actor.h>
78

89
namespace NKikimr {
910
namespace NSchemeShard {
@@ -291,5 +292,51 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
291292
return propose;
292293
}
293294

295+
THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateConsumersPropose(
296+
TSchemeShard* ss,
297+
TTxId txId,
298+
TImportInfo::TItem& item
299+
) {
300+
Y_ABORT_UNLESS(item.NextChangefeedIdx < item.Changefeeds.GetChangefeeds().size());
301+
302+
const auto& importChangefeedTopic = item.Changefeeds.GetChangefeeds()[item.NextChangefeedIdx];
303+
const auto& topic = importChangefeedTopic.GetTopic();
304+
305+
auto propose = MakeHolder<TEvSchemeShard::TEvModifySchemeTransaction>(ui64(txId), ss->TabletID());
306+
auto& record = propose->Record;
307+
auto& modifyScheme = *record.AddTransaction();
308+
modifyScheme.SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpAlterPersQueueGroup);
309+
auto& pqGroup = *modifyScheme.MutableAlterPersQueueGroup();
310+
311+
const TPath dstPath = TPath::Init(item.DstPathId, ss);
312+
const TString changefeedPath = dstPath.PathString() + "/" + importChangefeedTopic.GetChangefeed().name();
313+
modifyScheme.SetWorkingDir(changefeedPath);
314+
modifyScheme.SetInternal(true);
315+
316+
pqGroup.SetName("streamImpl");
317+
318+
auto describeSchemeResult = DescribePath(ss, TlsActivationContext->AsActorContext(),changefeedPath + "/streamImpl");
319+
320+
const auto& response = describeSchemeResult->GetRecord().GetPathDescription();
321+
item.StreamImplPath = {response.GetSelf().GetSchemeshardId(), response.GetSelf().GetPathId()};
322+
pqGroup.CopyFrom(response.GetPersQueueGroup());
323+
324+
pqGroup.ClearTotalGroupCount();
325+
pqGroup.MutablePQTabletConfig()->ClearPartitionKeySchema();
326+
327+
auto* tabletConfig = pqGroup.MutablePQTabletConfig();
328+
const auto& pqConfig = AppData()->PQConfig;
329+
auto serviceTypes = NGRpcProxy::V1::GetSupportedClientServiceTypes(pqConfig);
330+
331+
for (const auto& consumer : topic.consumers()) {
332+
auto rule = ::Ydb::PersQueue::V1::TopicSettings_ReadRule();
333+
rule.set_consumer_name(consumer.name());
334+
rule.set_important(consumer.important());
335+
AddReadRuleToConfig(tabletConfig, rule, serviceTypes, pqConfig);
336+
}
337+
338+
return propose;
339+
}
340+
294341
} // NSchemeShard
295342
} // NKikimr

‎ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,5 +52,11 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateChangefeedPropose(
5252
const TImportInfo::TItem& item
5353
);
5454

55+
THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateConsumersPropose(
56+
TSchemeShard* ss,
57+
TTxId txId,
58+
TImportInfo::TItem& item
59+
);
60+
5561
} // NSchemeShard
5662
} // NKikimr

‎ydb/core/tx/schemeshard/schemeshard_info_types.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2845,6 +2845,11 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
28452845
Subscribed,
28462846
};
28472847

2848+
enum class EChangefeedState: ui8 {
2849+
CreateChangefeed = 0,
2850+
CreateConsumers,
2851+
};
2852+
28482853
TString DstPathName;
28492854
TPathId DstPathId;
28502855
Ydb::Table::CreateTableRequest Scheme;
@@ -2856,13 +2861,15 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
28562861

28572862
EState State = EState::GetScheme;
28582863
ESubState SubState = ESubState::AllocateTxId;
2864+
EChangefeedState ChangefeedState = EChangefeedState::CreateChangefeed;
28592865
TTxId WaitTxId = InvalidTxId;
28602866
TActorId SchemeGetter;
28612867
TActorId SchemeQueryExecutor;
28622868
int NextIndexIdx = 0;
28632869
int NextChangefeedIdx = 0;
28642870
TString Issue;
28652871
int ViewCreationRetries = 0;
2872+
NKikimr::TPathId StreamImplPath;
28662873

28672874
TItem() = default;
28682875

‎ydb/core/tx/schemeshard/schemeshard_path_describer.cpp‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1299,6 +1299,29 @@ THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
12991299
return DescribePath(self, ctx, pathId, options);
13001300
}
13011301

1302+
THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
1303+
TSchemeShard* self,
1304+
const TActorContext& ctx,
1305+
const TString& path,
1306+
const NKikimrSchemeOp::TDescribeOptions& opts
1307+
) {
1308+
NKikimrSchemeOp::TDescribePath params;
1309+
params.SetPath(path);
1310+
params.MutableOptions()->CopyFrom(opts);
1311+
1312+
return TPathDescriber(self, std::move(params)).Describe(ctx);
1313+
}
1314+
1315+
THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
1316+
TSchemeShard* self,
1317+
const TActorContext& ctx,
1318+
const TString& path
1319+
) {
1320+
NKikimrSchemeOp::TDescribeOptions options;
1321+
options.SetShowPrivateTable(true);
1322+
return DescribePath(self, ctx, path, options);
1323+
}
1324+
13021325
void TSchemeShard::DescribeTable(
13031326
const TTableInfo& tableInfo,
13041327
const NScheme::TTypeRegistry* typeRegistry,

‎ydb/core/tx/schemeshard/schemeshard_path_describer.h‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,5 +83,18 @@ THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
8383
TPathId pathId
8484
);
8585

86+
THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
87+
TSchemeShard* self,
88+
const TActorContext& ctx,
89+
const TString& path,
90+
const NKikimrSchemeOp::TDescribeOptions& opts
91+
);
92+
93+
THolder<TEvSchemeShard::TEvDescribeSchemeResultBuilder> DescribePath(
94+
TSchemeShard* self,
95+
const TActorContext& ctx,
96+
const TString& path
97+
);
98+
8699
} // NSchemeShard
87100
} // NKikimr

‎ydb/core/tx/schemeshard/ut_helpers/ls_checks.cpp‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -968,6 +968,19 @@ TCheckFunc RetentionPeriod(const TDuration& value) {
968968
};
969969
}
970970

971+
TCheckFunc ConsumerExist(const TString& name) {
972+
return [=] (const NKikimrScheme::TEvDescribeSchemeResult& record) {
973+
bool isExist = false;
974+
for (const auto& consumer : record.GetPathDescription().GetPersQueueGroup().GetPQTabletConfig().GetConsumers()) {
975+
if (consumer.GetName() == name) {
976+
isExist = true;
977+
break;
978+
}
979+
}
980+
UNIT_ASSERT(isExist);
981+
};
982+
}
983+
971984
void NoChildren(const NKikimrScheme::TEvDescribeSchemeResult& record) {
972985
ChildrenCount(0)(record);
973986
}

‎ydb/core/tx/schemeshard/ut_helpers/ls_checks.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,7 @@ namespace NLs {
169169
TCheckFunc StreamAwsRegion(const TString& value);
170170
TCheckFunc StreamInitialScanProgress(ui32 total, ui32 completed);
171171
TCheckFunc RetentionPeriod(const TDuration& value);
172+
TCheckFunc ConsumerExist(const TString& name);
172173

173174
TCheckFunc HasBackupInFly(ui64 txId);
174175
void NoBackupInFly(const NKikimrScheme::TEvDescribeSchemeResult& record);

‎ydb/core/tx/schemeshard/ut_restore/ut_restore.cpp‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4897,7 +4897,10 @@ Y_UNIT_TEST_SUITE(TImportTests) {
48974897
{changefeedPath, GenerateTestData({EPathTypeCdcStream, changefeedDesc, std::move(attr)})},
48984898
[changefeedPath = TString(changefeedPath)](TTestBasicRuntime& runtime) {
48994899
TestDescribeResult(DescribePath(runtime, "/MyRoot/Table" + changefeedPath, false, false, true), {
4900-
NLs::PathExist
4900+
NLs::PathExist,
4901+
});
4902+
TestDescribeResult(DescribePath(runtime, "/MyRoot/Table" + changefeedPath + "/streamImpl", false, false, true), {
4903+
NLs::ConsumerExist("my_consumer")
49014904
});
49024905
}
49034906
};

0 commit comments

Comments
 (0)