Skip to content

Commit cf78169

Browse files
authored
Merge 2fd1bea into 893f325
2 parents 893f325 + 2fd1bea commit cf78169

File tree

5 files changed

+117
-13
lines changed

5 files changed

+117
-13
lines changed

ydb/core/grpc_services/rpc_alter_table.cpp

Lines changed: 69 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -109,12 +109,12 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
109109
break;
110110

111111
case EOp::Attribute:
112-
PrepareAlterUserAttrubutes();
112+
case EOp::AddChangefeed:
113+
case EOp::DropChangefeed:
114+
GetProxyServices();
113115
break;
114116

115-
case EOp::AddChangefeed:
116117
case EOp::DropIndex:
117-
case EOp::DropChangefeed:
118118
case EOp::RenameIndex:
119119
AlterTable(ctx);
120120
break;
@@ -197,7 +197,7 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
197197
Navigate(msg->Services.SchemeCache, ctx);
198198
}
199199

200-
void PrepareAlterUserAttrubutes() {
200+
void GetProxyServices() {
201201
using namespace NTxProxy;
202202
Send(MakeTxProxyID(), new TEvTxUserProxy::TEvGetProxyServicesRequest);
203203
}
@@ -222,13 +222,38 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
222222
auto ev = CreateNavigateForPath(DatabaseName);
223223
{
224224
auto& entry = static_cast<TEvTxProxySchemeCache::TEvNavigateKeySet*>(ev)->Request->ResultSet.emplace_back();
225-
entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpTable;
226225
entry.Path = paths;
227226
}
228227

229228
Send(schemeCache, ev);
230229
}
231230

231+
void Navigate(const TTableId& pathId, const TActorContext& ctx) {
232+
DatabaseName = Request_->GetDatabaseName()
233+
.GetOrElse(DatabaseFromDomain(AppData()));
234+
235+
auto ev = CreateNavigateForPath(DatabaseName);
236+
{
237+
auto& entry = static_cast<TEvTxProxySchemeCache::TEvNavigateKeySet*>(ev)->Request->ResultSet.emplace_back();
238+
entry.RequestType = NSchemeCache::TSchemeCacheNavigate::TEntry::ERequestType::ByTableId;
239+
entry.TableId = pathId;
240+
entry.ShowPrivatePath = true;
241+
entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
242+
}
243+
244+
Send(MakeSchemeCacheID(), ev);
245+
}
246+
247+
static bool IsChangefeedOperation(EOp type) {
248+
switch (type) {
249+
case EOp::AddChangefeed:
250+
case EOp::DropChangefeed:
251+
return true;
252+
default:
253+
return false;
254+
}
255+
}
256+
232257
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
233258
TXLOG_D("Handle TEvTxProxySchemeCache::TEvNavigateKeySetResult"
234259
<< ", errors# " << ev->Get()->Request.Get()->ErrorCount);
@@ -251,13 +276,48 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
251276
return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
252277
}
253278

279+
Y_ABORT_UNLESS(!resp->ResultSet.empty());
280+
const auto& entry = resp->ResultSet.back();
281+
282+
switch (entry.Kind) {
283+
case NSchemeCache::TSchemeCacheNavigate::KindTable:
284+
case NSchemeCache::TSchemeCacheNavigate::KindColumnTable:
285+
case NSchemeCache::TSchemeCacheNavigate::KindExternalTable:
286+
case NSchemeCache::TSchemeCacheNavigate::KindExternalDataSource:
287+
case NSchemeCache::TSchemeCacheNavigate::KindView:
288+
break; // table
289+
case NSchemeCache::TSchemeCacheNavigate::KindIndex:
290+
if (IsChangefeedOperation(OpType)) {
291+
break;
292+
}
293+
[[fallthrough]];
294+
default:
295+
Request_->RaiseIssue(MakeIssue(NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR, TStringBuilder()
296+
<< "Unable to nagivate: " << JoinPath(entry.Path) << " status: PathNotTable"));
297+
return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
298+
}
299+
254300
switch (OpType) {
255301
case EOp::AddIndex:
256302
return AlterTableAddIndexOp(resp, ctx);
257303
case EOp::Attribute:
258-
Y_ABORT_UNLESS(!resp->ResultSet.empty());
259304
ResolvedPathId = resp->ResultSet.back().TableId.PathId;
260305
return AlterTable(ctx);
306+
case EOp::AddChangefeed:
307+
case EOp::DropChangefeed:
308+
if (entry.Kind != NSchemeCache::TSchemeCacheNavigate::KindIndex) {
309+
AlterTable(ctx);
310+
} else if (auto list = entry.ListNodeEntry) {
311+
if (list->Children.size() != 1) {
312+
return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
313+
}
314+
315+
const auto& child = list->Children.at(0);
316+
AlterTable(ctx, CanonizePath(ChildPath(NKikimr::SplitPath(GetProtoRequest()->path()), child.Name)));
317+
} else {
318+
Navigate(entry.TableId, ctx);
319+
}
320+
break;
261321
default:
262322
TXLOG_E("Got unexpected cache response");
263323
return Reply(Ydb::StatusIds::INTERNAL_ERROR, ctx);
@@ -351,13 +411,14 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
351411
Die(ctx);
352412
}
353413

354-
void AlterTable(const TActorContext &ctx) {
414+
void AlterTable(const TActorContext &ctx, const TMaybe<TString>& overridePath = {}) {
355415
const auto req = GetProtoRequest();
356416
std::unique_ptr<TEvTxUserProxy::TEvProposeTransaction> proposeRequest = CreateProposeTransaction();
357417
auto modifyScheme = proposeRequest->Record.MutableTransaction()->MutableModifyScheme();
418+
modifyScheme->SetAllowAccessToPrivatePaths(overridePath.Defined());
358419
Ydb::StatusIds::StatusCode code;
359420
TString error;
360-
if (!BuildAlterTableModifyScheme(req, modifyScheme, Profiles, ResolvedPathId, code, error)) {
421+
if (!BuildAlterTableModifyScheme(overridePath.GetOrElse(req->path()), req, modifyScheme, Profiles, ResolvedPathId, code, error)) {
361422
NYql::TIssues issues;
362423
issues.AddIssue(NYql::TIssue(error));
363424
return Reply(code, issues, ctx);

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

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4157,6 +4157,42 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
41574157
}
41584158
}
41594159

4160+
Y_UNIT_TEST(ChangefeedOnIndexTable) {
4161+
TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));
4162+
auto db = kikimr.GetTableClient();
4163+
auto session = db.CreateSession().GetValueSync().GetSession();
4164+
4165+
{
4166+
auto query = R"(
4167+
--!syntax_v1
4168+
CREATE TABLE `/Root/table` (
4169+
Key Uint64,
4170+
Value String,
4171+
PRIMARY KEY (Key),
4172+
INDEX SyncIndex GLOBAL SYNC ON (`Value`),
4173+
INDEX AsyncIndex GLOBAL ASYNC ON (`Value`)
4174+
);
4175+
)";
4176+
4177+
auto result = session.ExecuteSchemeQuery(query).GetValueSync();
4178+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
4179+
}
4180+
4181+
const auto changefeed = TChangefeedDescription("feed", EChangefeedMode::KeysOnly, EChangefeedFormat::Json);
4182+
{
4183+
auto result = session.AlterTable("/Root/table/AsyncIndex", TAlterTableSettings()
4184+
.AppendAddChangefeeds(changefeed)
4185+
).ExtractValueSync();
4186+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
4187+
}
4188+
{
4189+
auto result = session.AlterTable("/Root/table/SyncIndex", TAlterTableSettings()
4190+
.AppendAddChangefeeds(changefeed)
4191+
).ExtractValueSync();
4192+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
4193+
}
4194+
}
4195+
41604196
Y_UNIT_TEST(CreatedAt) {
41614197
TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));
41624198
auto scheme = NYdb::NScheme::TSchemeClient(kikimr.GetDriver(), TCommonClientSettings().Database("/Root"));

ydb/core/tx/scheme_board/cache.cpp

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -901,8 +901,6 @@ class TSchemeCache: public TMonitorableActor<TSchemeCache> {
901901
default:
902902
return false;
903903
}
904-
case NKikimrSchemeOp::EPathTypeTableIndex:
905-
return true;
906904
default:
907905
return false;
908906
}

ydb/core/ydb_convert/table_description.cpp

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,8 @@ bool BuildAlterTableAddIndexRequest(const Ydb::Table::AlterTableRequest* req, NK
167167
return true;
168168
}
169169

170-
bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKikimrSchemeOp::TModifyScheme* modifyScheme, const TTableProfiles& profiles,
170+
bool BuildAlterTableModifyScheme(const TString& path, const Ydb::Table::AlterTableRequest* req,
171+
NKikimrSchemeOp::TModifyScheme* modifyScheme, const TTableProfiles& profiles,
171172
const TPathId& resolvedPathId,
172173
Ydb::StatusIds::StatusCode& code, TString& error)
173174
{
@@ -188,7 +189,7 @@ bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKiki
188189
const auto OpType = *ops.begin();
189190

190191
try {
191-
pathPair = SplitPathIntoWorkingDirAndName(req->path());
192+
pathPair = SplitPathIntoWorkingDirAndName(path);
192193
} catch (const std::exception&) {
193194
code = Ydb::StatusIds::BAD_REQUEST;
194195
return false;
@@ -231,7 +232,7 @@ bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKiki
231232
for(const auto& rename: req->rename_indexes()) {
232233
modifyScheme->SetOperationType(NKikimrSchemeOp::EOperationType::ESchemeOpMoveIndex);
233234
auto& alter = *modifyScheme->MutableMoveIndex();
234-
alter.SetTablePath(req->path());
235+
alter.SetTablePath(path);
235236
alter.SetSrcPath(rename.source_name());
236237
alter.SetDstPath(rename.destination_name());
237238
alter.SetAllowOverwrite(rename.replace_destination());
@@ -376,6 +377,11 @@ bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKiki
376377
return true;
377378
}
378379

380+
bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKikimrSchemeOp::TModifyScheme* modifyScheme,
381+
const TTableProfiles& profiles, const TPathId& resolvedPathId, Ydb::StatusIds::StatusCode& code, TString& error)
382+
{
383+
return BuildAlterTableModifyScheme(req->path(), req, modifyScheme, profiles, resolvedPathId, code, error);
384+
}
379385

380386
template <typename TColumn>
381387
static Ydb::Type* AddColumn(Ydb::Table::ColumnMeta* newColumn, const TColumn& column) {

ydb/core/ydb_convert/table_description.h

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,9 @@ struct TPathId;
3131

3232

3333
THashSet<EAlterOperationKind> GetAlterOperationKinds(const Ydb::Table::AlterTableRequest* req);
34+
bool BuildAlterTableModifyScheme(const TString& path, const Ydb::Table::AlterTableRequest* req, NKikimrSchemeOp::TModifyScheme* modifyScheme,
35+
const TTableProfiles& profiles, const TPathId& resolvedPathId,
36+
Ydb::StatusIds::StatusCode& status, TString& error);
3437
bool BuildAlterTableModifyScheme(const Ydb::Table::AlterTableRequest* req, NKikimrSchemeOp::TModifyScheme* modifyScheme,
3538
const TTableProfiles& profiles, const TPathId& resolvedPathId,
3639
Ydb::StatusIds::StatusCode& status, TString& error);

0 commit comments

Comments
 (0)