Skip to content

Commit 81ddfd5

Browse files
CyberROFLCyberROFL
authored
24-2: Allow streams on index tables (#7154)
1 parent 546620e commit 81ddfd5

24 files changed

+803
-201
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) {
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);
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/grpc_services/rpc_describe_table.cpp

Lines changed: 70 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,11 @@
1-
#include "service_table.h"
2-
#include <ydb/core/grpc_services/base/base.h>
3-
41
#include "rpc_calls.h"
52
#include "rpc_scheme_base.h"
6-
73
#include "service_table.h"
8-
#include "rpc_common/rpc_common.h"
4+
5+
#include <ydb/core/base/path.h>
6+
#include <ydb/core/base/table_index.h>
7+
#include <ydb/core/grpc_services/base/base.h>
8+
#include <ydb/core/grpc_services/rpc_common/rpc_common.h>
99
#include <ydb/core/tx/schemeshard/schemeshard.h>
1010
#include <ydb/core/ydb_convert/table_description.h>
1111
#include <ydb/core/ydb_convert/ydb_convert.h>
@@ -22,25 +22,84 @@ using TEvDescribeTableRequest = TGrpcRequestOperationCall<Ydb::Table::DescribeTa
2222
class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDescribeTableRequest> {
2323
using TBase = TRpcSchemeRequestActor<TDescribeTableRPC, TEvDescribeTableRequest>;
2424

25+
TString OverrideName;
26+
27+
static bool ShowPrivatePath(const TString& path) {
28+
if (AppData()->AllowPrivateTableDescribeForTest) {
29+
return true;
30+
}
31+
32+
if (path.EndsWith("/indexImplTable")) {
33+
return true;
34+
}
35+
36+
return false;
37+
}
38+
2539
public:
2640
TDescribeTableRPC(IRequestOpCtx* msg)
2741
: TBase(msg) {}
2842

2943
void Bootstrap(const TActorContext &ctx) {
3044
TBase::Bootstrap(ctx);
3145

32-
SendProposeRequest(ctx);
46+
const auto& path = GetProtoRequest()->path();
47+
const auto paths = NKikimr::SplitPath(path);
48+
if (paths.empty()) {
49+
Request_->RaiseIssue(NYql::TIssue("Invalid path"));
50+
return Reply(Ydb::StatusIds::BAD_REQUEST, ctx);
51+
}
52+
53+
auto navigate = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
54+
navigate->DatabaseName = CanonizePath(Request_->GetDatabaseName().GetOrElse(""));
55+
auto& entry = navigate->ResultSet.emplace_back();
56+
entry.Path = paths;
57+
entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
58+
entry.SyncVersion = true;
59+
entry.ShowPrivatePath = ShowPrivatePath(path);
60+
61+
Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate));
3362
Become(&TDescribeTableRPC::StateWork);
3463
}
3564

3665
private:
3766
void StateWork(TAutoPtr<IEventHandle>& ev) {
3867
switch (ev->GetTypeRewrite()) {
68+
HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
3969
HFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle);
4070
default: TBase::StateWork(ev);
4171
}
4272
}
4373

74+
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
75+
auto* navigate = ev->Get()->Request.Get();
76+
77+
Y_ABORT_UNLESS(navigate->ResultSet.size() == 1);
78+
const auto& entry = navigate->ResultSet.front();
79+
80+
if (navigate->ErrorCount > 0) {
81+
switch (entry.Status) {
82+
case NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown:
83+
case NSchemeCache::TSchemeCacheNavigate::EStatus::RootUnknown:
84+
return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
85+
default:
86+
return Reply(Ydb::StatusIds::UNAVAILABLE, ctx);
87+
}
88+
}
89+
90+
if (entry.Kind == NSchemeCache::TSchemeCacheNavigate::KindIndex) {
91+
auto list = entry.ListNodeEntry;
92+
if (!list || list->Children.size() != 1) {
93+
return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
94+
}
95+
96+
OverrideName = entry.Path.back();
97+
SendProposeRequest(CanonizePath(ChildPath(entry.Path, list->Children.at(0).Name)), ctx);
98+
} else {
99+
SendProposeRequest(GetProtoRequest()->path(), ctx);
100+
}
101+
}
102+
44103
void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev, const TActorContext& ctx) {
45104
const auto& record = ev->Get()->GetRecord();
46105
const auto status = record.GetStatus();
@@ -53,9 +112,10 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
53112
case NKikimrScheme::StatusSuccess: {
54113
const auto& pathDescription = record.GetPathDescription();
55114
Ydb::Scheme::Entry* selfEntry = describeTableResult.mutable_self();
56-
selfEntry->set_name(pathDescription.GetSelf().GetName());
57-
selfEntry->set_type(static_cast<Ydb::Scheme::Entry::Type>(pathDescription.GetSelf().GetPathType()));
58115
ConvertDirectoryEntry(pathDescription.GetSelf(), selfEntry, true);
116+
if (OverrideName) {
117+
selfEntry->set_name(OverrideName);
118+
}
59119

60120
if (pathDescription.HasColumnTableDescription()) {
61121
const auto& tableDescription = pathDescription.GetColumnTableDescription();
@@ -129,9 +189,8 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
129189
}
130190
}
131191

132-
void SendProposeRequest(const TActorContext &ctx) {
192+
void SendProposeRequest(const TString& path, const TActorContext& ctx) {
133193
const auto req = GetProtoRequest();
134-
const TString path = req->path();
135194

136195
std::unique_ptr<TEvTxUserProxy::TEvNavigate> navigateRequest(new TEvTxUserProxy::TEvNavigate());
137196
SetAuthToken(navigateRequest, *Request_);
@@ -146,9 +205,7 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
146205
record->MutableOptions()->SetReturnPartitionStats(true);
147206
}
148207

149-
if (AppData(ctx)->AllowPrivateTableDescribeForTest || path.EndsWith("/indexImplTable")) {
150-
record->MutableOptions()->SetShowPrivateTable(true);
151-
}
208+
record->MutableOptions()->SetShowPrivateTable(ShowPrivatePath(path));
152209

153210
ctx.Send(MakeTxProxyID(), navigateRequest.release());
154211
}

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

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3929,6 +3929,70 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
39293929
}
39303930
}
39313931

3932+
Y_UNIT_TEST(ChangefeedOnIndexTable) {
3933+
TKikimrRunner kikimr(TKikimrSettings()
3934+
.SetPQConfig(DefaultPQConfig())
3935+
.SetEnableChangefeedsOnIndexTables(true));
3936+
auto db = kikimr.GetTableClient();
3937+
auto session = db.CreateSession().GetValueSync().GetSession();
3938+
3939+
{
3940+
auto query = R"(
3941+
--!syntax_v1
3942+
CREATE TABLE `/Root/table` (
3943+
Key Uint64,
3944+
Value String,
3945+
PRIMARY KEY (Key),
3946+
INDEX SyncIndex GLOBAL SYNC ON (`Value`),
3947+
INDEX AsyncIndex GLOBAL ASYNC ON (`Value`)
3948+
);
3949+
)";
3950+
3951+
auto result = session.ExecuteSchemeQuery(query).GetValueSync();
3952+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
3953+
}
3954+
3955+
const auto changefeed = TChangefeedDescription("feed", EChangefeedMode::KeysOnly, EChangefeedFormat::Json);
3956+
{
3957+
auto result = session.AlterTable("/Root/table/AsyncIndex", TAlterTableSettings()
3958+
.AppendAddChangefeeds(changefeed)
3959+
).ExtractValueSync();
3960+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
3961+
}
3962+
{
3963+
auto result = session.AlterTable("/Root/table/SyncIndex", TAlterTableSettings()
3964+
.AppendAddChangefeeds(changefeed)
3965+
).ExtractValueSync();
3966+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
3967+
}
3968+
}
3969+
3970+
Y_UNIT_TEST(DescribeIndexTable) {
3971+
TKikimrRunner kikimr;
3972+
auto db = kikimr.GetTableClient();
3973+
auto session = db.CreateSession().GetValueSync().GetSession();
3974+
3975+
{
3976+
auto query = R"(
3977+
--!syntax_v1
3978+
CREATE TABLE `/Root/table` (
3979+
Key Uint64,
3980+
Value String,
3981+
PRIMARY KEY (Key),
3982+
INDEX SyncIndex GLOBAL SYNC ON (`Value`)
3983+
);
3984+
)";
3985+
3986+
auto result = session.ExecuteSchemeQuery(query).GetValueSync();
3987+
UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
3988+
}
3989+
{
3990+
auto desc = session.DescribeTable("/Root/table/SyncIndex").ExtractValueSync();
3991+
UNIT_ASSERT_C(desc.IsSuccess(), desc.GetIssues().ToString());
3992+
UNIT_ASSERT_VALUES_EQUAL(desc.GetEntry().Name, "SyncIndex");
3993+
}
3994+
}
3995+
39323996
Y_UNIT_TEST(CreatedAt) {
39333997
TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));
39343998
auto scheme = NYdb::NScheme::TSchemeClient(kikimr.GetDriver(), TCommonClientSettings().Database("/Root"));

‎ydb/core/protos/feature_flags.proto

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,4 +132,5 @@ message TFeatureFlags {
132132
optional bool EnableCMSRequestPriorities = 117 [default = false];
133133
optional bool EnableStableNodeNames = 122 [default = false];
134134
optional bool EnableVolatileTransactionArbiters = 124 [default = false];
135+
optional bool EnableChangefeedsOnIndexTables = 134 [default = false];
135136
}

‎ydb/core/testlib/basics/feature_flags.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ class TTestFeatureFlagsHolder {
5959
FEATURE_FLAG_SETTER(EnableAddColumsWithDefaults)
6060
FEATURE_FLAG_SETTER(EnableReplaceIfExistsForExternalEntities)
6161
FEATURE_FLAG_SETTER(EnableCMSRequestPriorities)
62+
FEATURE_FLAG_SETTER(EnableChangefeedsOnIndexTables)
6263

6364
#undef FEATURE_FLAG_SETTER
6465
};

‎ydb/core/tx/scheme_board/cache.cpp

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -897,8 +897,6 @@ class TSchemeCache: public TMonitorableActor<TSchemeCache> {
897897
default:
898898
return false;
899899
}
900-
case NKikimrSchemeOp::EPathTypeTableIndex:
901-
return true;
902900
default:
903901
return false;
904902
}
@@ -2561,14 +2559,12 @@ class TSchemeCache: public TMonitorableActor<TSchemeCache> {
25612559

25622560
if (entry.RequestType == TNavigate::TEntry::ERequestType::ByPath) {
25632561
auto pathExtractor = [this](TNavigate::TEntry& entry) {
2562+
NSysView::ISystemViewResolver::TSystemViewPath sysViewPath;
25642563
if (AppData()->FeatureFlags.GetEnableSystemViews()
2565-
&& (entry.Operation == TNavigate::OpPath || entry.Operation == TNavigate::OpTable))
2564+
&& SystemViewResolver->IsSystemViewPath(entry.Path, sysViewPath))
25662565
{
2567-
NSysView::ISystemViewResolver::TSystemViewPath sysViewPath;
2568-
if (SystemViewResolver->IsSystemViewPath(entry.Path, sysViewPath)) {
2569-
entry.TableId.SysViewInfo = sysViewPath.ViewName;
2570-
return CanonizePath(sysViewPath.Parent);
2571-
}
2566+
entry.TableId.SysViewInfo = sysViewPath.ViewName;
2567+
return CanonizePath(sysViewPath.Parent);
25722568
}
25732569

25742570
TString path = CanonizePath(entry.Path);

0 commit comments

Comments
 (0)