Skip to content

Commit a8d7567

Browse files
shnikdshnikd
authored
Support sequences in SHOW CREATE TABLE (#17488)
1 parent 0c2eebe commit a8d7567

File tree

5 files changed

+351
-15
lines changed

5 files changed

+351
-15
lines changed

‎ydb/core/sys_view/show_create/create_table_formatter.cpp

Lines changed: 51 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -245,8 +245,9 @@ class TStringStreamWrapper {
245245
TStringStream& Stream;
246246
};
247247

248-
TFormatResult TCreateTableFormatter::Format(const TString& tablePath, const NKikimrSchemeOp::TTableDescription& tableDesc,
249-
bool temporary, const THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>>& persQueues) {
248+
TFormatResult TCreateTableFormatter::Format(const TString& tablePath, const TString& fullPath, const NKikimrSchemeOp::TTableDescription& tableDesc,
249+
bool temporary, const THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>>& persQueues,
250+
const THashMap<TPathId, THolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>>& sequences) {
250251
Stream.Clear();
251252

252253
TStringStreamWrapper wrapper(Stream);
@@ -411,8 +412,7 @@ TFormatResult TCreateTableFormatter::Format(const TString& tablePath, const NKik
411412
Y_ENSURE((ui32)tableDesc.GetCdcStreams().size() == persQueues.size());
412413
auto firstColumnTypeId = columns[tableDesc.GetKeyColumnIds(0)]->GetTypeId();
413414
try {
414-
Format(tablePath, tableDesc.GetCdcStreams(0), persQueues, firstColumnTypeId);
415-
for (int i = 1; i < tableDesc.GetCdcStreams().size(); i++) {
415+
for (int i = 0; i < tableDesc.GetCdcStreams().size(); i++) {
416416
Format(tablePath, tableDesc.GetCdcStreams(i), persQueues, firstColumnTypeId);
417417
}
418418
} catch (const TFormatFail& ex) {
@@ -422,6 +422,18 @@ TFormatResult TCreateTableFormatter::Format(const TString& tablePath, const NKik
422422
}
423423
}
424424

425+
if (!tableDesc.GetSequences().empty()) {
426+
try {
427+
for (int i = 0; i < tableDesc.GetSequences().size(); i++) {
428+
Format(fullPath, tableDesc.GetSequences(i), sequences);
429+
}
430+
} catch (const TFormatFail& ex) {
431+
return TFormatResult(ex.Status, ex.Error);
432+
} catch (const yexception& e) {
433+
return TFormatResult(Ydb::StatusIds::INTERNAL_ERROR, e.what());
434+
}
435+
}
436+
425437
TString statement = Stream.Str();
426438
TString formattedStatement;
427439
NYql::TIssues issues;
@@ -1001,6 +1013,41 @@ void TCreateTableFormatter::Format(const TString& tablePath, const NKikimrScheme
10011013
Stream << ");";
10021014
}
10031015

1016+
void TCreateTableFormatter::Format(const TString& tablePath, const NKikimrSchemeOp::TSequenceDescription& sequence, const THashMap<TPathId, THolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>>& sequences) {
1017+
auto it = sequences.find(TPathId::FromProto(sequence.GetPathId()));
1018+
if (it == sequences.end() || !it->second) {
1019+
ythrow TFormatFail(Ydb::StatusIds::INTERNAL_ERROR, "Unexpected sequence path id");
1020+
}
1021+
const auto& getSequenceResult = *it->second;
1022+
1023+
if (getSequenceResult.StartValue == 1 && getSequenceResult.Increment == 1
1024+
&& getSequenceResult.NextValue == 1) {
1025+
return;
1026+
}
1027+
1028+
Stream << "ALTER SEQUENCE ";
1029+
auto sequencePath = JoinPath({tablePath, sequence.GetName()});
1030+
EscapeName(sequencePath, Stream);
1031+
1032+
if (getSequenceResult.StartValue != 1) {
1033+
Stream << " START WITH " << getSequenceResult.StartValue;
1034+
}
1035+
1036+
if (getSequenceResult.Increment != 1) {
1037+
Stream << " INCREMENT BY " << getSequenceResult.Increment;
1038+
}
1039+
1040+
if (getSequenceResult.NextValue != 1) {
1041+
if (getSequenceResult.NextValue == getSequenceResult.StartValue) {
1042+
Stream << " RESTART";
1043+
} else {
1044+
Stream << " RESTART WITH " << getSequenceResult.NextValue;
1045+
}
1046+
}
1047+
1048+
Stream << ";";
1049+
}
1050+
10041051

10051052
TFormatResult TCreateTableFormatter::Format(const TString& tablePath, const TColumnTableDescription& tableDesc, bool temporary) {
10061053
Stream.Clear();

‎ydb/core/sys_view/show_create/create_table_formatter.h

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,13 @@
22

33
#include "formatters_common.h"
44

5-
#include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/value/value.h>
6-
75
#include <ydb/core/protos/flat_scheme_op.pb.h>
8-
6+
#include <ydb/core/scheme/scheme_pathid.h>
97
#include <ydb/core/tx/columnshard/engines/scheme/defaults/protos/data.pb.h>
8+
#include <ydb/core/tx/sequenceproxy/public/events.h>
109

1110
#include <ydb/public/api/protos/ydb_table.pb.h>
11+
#include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/value/value.h>
1212

1313
#include <yql/essentials/minikql/mkql_alloc.h>
1414

@@ -30,8 +30,9 @@ class TCreateTableFormatter {
3030
Alloc.Acquire();
3131
}
3232

33-
TFormatResult Format(const TString& tablePath, const NKikimrSchemeOp::TTableDescription& tableDesc, bool temporary,
34-
const THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>>& persQueues);
33+
TFormatResult Format(const TString& tablePath, const TString& fullPath, const NKikimrSchemeOp::TTableDescription& tableDesc, bool temporary,
34+
const THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>>& persQueues,
35+
const THashMap<TPathId, THolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>>& sequences);
3536
TFormatResult Format(const TString& tablePath, const NKikimrSchemeOp::TColumnTableDescription& tableDesc, bool temporary);
3637

3738
private:
@@ -41,6 +42,7 @@ class TCreateTableFormatter {
4142

4243
void Format(const TString& tablePath, const NKikimrSchemeOp::TCdcStreamDescription& cdcStream,
4344
const THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>>& persQueues, ui32 firstColumnTypeId);
45+
void Format(const TString& fullTablePath, const NKikimrSchemeOp::TSequenceDescription& sequence, const THashMap<TPathId, THolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>>& sequences);
4446

4547
void Format(const Ydb::Table::TableIndex& index);
4648
bool Format(const Ydb::Table::ExplicitPartitions& explicitPartitions, TString& del, bool needWith);

‎ydb/core/sys_view/show_create/show_create.cpp

Lines changed: 79 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,12 @@
33
#include "show_create.h"
44

55
#include <ydb/core/base/tablet_pipe.h>
6-
#include <ydb/core/base/tablet_pipe.h>
6+
#include <ydb/core/scheme/scheme_pathid.h>
77
#include <ydb/core/sys_view/common/scan_actor_base_impl.h>
88
#include <ydb/core/sys_view/common/schema.h>
99
#include <ydb/core/tx/scheme_cache/scheme_cache.h>
1010
#include <ydb/core/tx/schemeshard/schemeshard.h>
11+
#include <ydb/core/tx/sequenceproxy/public/events.h>
1112
#include <ydb/core/tx/tx_proxy/proxy.h>
1213

1314
#include <ydb/library/actors/core/hfunc.h>
@@ -79,6 +80,7 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
7980
STFUNC(StateCollectTableSettings) {
8081
switch (ev->GetTypeRewrite()) {
8182
hFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, HandleCollectTableSettings);
83+
hFunc(NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult, Handle);
8284
default:
8385
LOG_CRIT(*TlsActivationContext, NKikimrServices::SYSTEM_VIEWS,
8486
"NSysView::TScanActorBase: unexpected event 0x%08" PRIx32, ev->GetTypeRewrite());
@@ -169,6 +171,10 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
169171
}
170172
}
171173

174+
bool NeedToCollectTableSettings(const NKikimrSchemeOp::TTableDescription& tableDesc) {
175+
return !tableDesc.GetCdcStreams().empty() || !tableDesc.GetSequences().empty();
176+
}
177+
172178
void StartCollectTableSettings(const TString& tablePath, const NKikimrSchemeOp::TTableDescription& tableDesc, bool temporary) {
173179
CollectTableSettingsState = MakeHolder<TCollectTableSettingsState>();
174180
CollectTableSettingsState->TablePath = tablePath;
@@ -196,6 +202,15 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
196202

197203
Send(MakeTxProxyID(), navigateRequest.release());
198204
}
205+
206+
for (const auto& sequence: tableDesc.GetSequences()) {
207+
auto sequencePathId = TPathId::FromProto(sequence.GetPathId());
208+
CollectTableSettingsState->Sequences[sequencePathId] = nullptr;
209+
210+
Send(NSequenceProxy::MakeSequenceProxyServiceID(),
211+
new NSequenceProxy::TEvSequenceProxy::TEvGetSequence(Database, sequencePathId)
212+
);
213+
}
199214
}
200215

201216
void FillBatch(NKqp::TEvKqpCompute::TEvScanData& batch, const TString& path, const TString& statement) {
@@ -263,14 +278,14 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
263278
temporary = true;
264279
}
265280

266-
if (!tableDesc.GetCdcStreams().empty()) {
281+
if (NeedToCollectTableSettings(tableDesc)) {
267282
StartCollectTableSettings(tablePath, tableDesc, temporary);
268283
Become(&TShowCreate::StateCollectTableSettings);
269284
return;
270285
}
271286

272287
TCreateTableFormatter formatter;
273-
auto formatterResult = formatter.Format(tablePath, tableDesc, temporary, {});
288+
auto formatterResult = formatter.Format(tablePath, Path, tableDesc, temporary, {}, {});
274289
if (formatterResult.IsSuccess()) {
275290
path = tablePath;
276291
statement = formatterResult.ExtractOut();
@@ -385,16 +400,18 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
385400
it->second = MakeHolder<NKikimrSchemeOp::TPersQueueGroupDescription>(description);
386401
CollectTableSettingsState->CurrentPersQueuesNumber++;
387402

388-
if (CollectTableSettingsState->CurrentPersQueuesNumber != CollectTableSettingsState->PersQueues.size()) {
403+
if (!CollectTableSettingsState->IsReady()) {
389404
return;
390405
}
391406

392407
TCreateTableFormatter formatter;
393408
auto formatterResult = formatter.Format(
394409
CollectTableSettingsState->TablePath,
410+
Path,
395411
CollectTableSettingsState->TableDescription,
396412
CollectTableSettingsState->Temporary,
397-
CollectTableSettingsState->PersQueues
413+
CollectTableSettingsState->PersQueues,
414+
CollectTableSettingsState->Sequences
398415
);
399416
if (formatterResult.IsSuccess()) {
400417
path = CollectTableSettingsState->TablePath;
@@ -441,6 +458,57 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
441458
SendBatch(std::move(batch));
442459
}
443460

461+
void Handle(NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult::TPtr& ev) {
462+
if (ev->Get()->Status != Ydb::StatusIds::SUCCESS) {
463+
ReplyErrorAndDie(ev->Get()->Status, ev->Get()->Issues.ToString());
464+
return;
465+
}
466+
467+
auto* msg = ev->Get();
468+
469+
auto it = CollectTableSettingsState->Sequences.find(msg->PathId);
470+
if (it == CollectTableSettingsState->Sequences.end()) {
471+
return ReplyErrorAndDie(Ydb::StatusIds::INTERNAL_ERROR, TStringBuilder() << "Unknown sequence path id: " << msg->PathId);
472+
}
473+
if (it->second) {
474+
return ReplyErrorAndDie(Ydb::StatusIds::INTERNAL_ERROR, TStringBuilder() << "Found duplicate sequence path id: " << msg->PathId);
475+
}
476+
it->second = MakeHolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>(*msg);
477+
CollectTableSettingsState->CurrentSequencesNumber++;
478+
479+
if (!CollectTableSettingsState->IsReady()) {
480+
return;
481+
}
482+
483+
TCreateTableFormatter formatter;
484+
auto formatterResult = formatter.Format(
485+
CollectTableSettingsState->TablePath,
486+
Path,
487+
CollectTableSettingsState->TableDescription,
488+
CollectTableSettingsState->Temporary,
489+
CollectTableSettingsState->PersQueues,
490+
CollectTableSettingsState->Sequences
491+
);
492+
std::optional<TString> path;
493+
std::optional<TString> statement;
494+
if (formatterResult.IsSuccess()) {
495+
path = CollectTableSettingsState->TablePath;
496+
statement = formatterResult.ExtractOut();
497+
} else {
498+
ReplyErrorAndDie(formatterResult.GetStatus(), formatterResult.GetError());
499+
return;
500+
}
501+
502+
Y_ENSURE(path.has_value());
503+
Y_ENSURE(statement.has_value());
504+
505+
auto batch = MakeHolder<NKqp::TEvKqpCompute::TEvScanData>(ScanId);
506+
507+
FillBatch(*batch, path.value(), statement.value());
508+
509+
SendBatch(std::move(batch));
510+
}
511+
444512
private:
445513
TString Database;
446514
TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
@@ -453,6 +521,12 @@ class TShowCreate : public TScanActorBase<TShowCreate> {
453521
bool Temporary;
454522
THashMap<TString, THolder<NKikimrSchemeOp::TPersQueueGroupDescription>> PersQueues;
455523
ui32 CurrentPersQueuesNumber = 0;
524+
THashMap<TPathId, THolder<NSequenceProxy::TEvSequenceProxy::TEvGetSequenceResult>> Sequences;
525+
ui32 CurrentSequencesNumber = 0;
526+
527+
bool IsReady() const {
528+
return CurrentPersQueuesNumber == PersQueues.size() && CurrentSequencesNumber == Sequences.size();
529+
}
456530
};
457531
THolder<TCollectTableSettingsState> CollectTableSettingsState;
458532
};

‎ydb/core/sys_view/show_create/ya.make

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,10 @@ PEERDIR(
1111
ydb/core/base
1212
ydb/core/kqp/runtime
1313
ydb/core/protos
14-
ydb/core/tx/columnshard/engines/scheme/defaults/protos
1514
ydb/core/sys_view/common
15+
ydb/core/tx/columnshard/engines/scheme/defaults/protos
1616
ydb/core/tx/schemeshard
17+
ydb/core/tx/sequenceproxy
1718
ydb/core/tx/tx_proxy
1819
ydb/core/ydb_convert
1920
ydb/library/actors/core

0 commit comments

Comments
 (0)