24-3: Allow streams on index tables, replicate index tables

ydb/core/base/path.h

@@ -37,4 +37,12 @@ inline TVector<TString> ChildPath(const TVector<TString>& parentPath, const TStr
     return path;
 }
 
+inline TVector<TString> ChildPath(const TVector<TString>& parentPath, const TVector<TString>& childPath) {
+    auto path = parentPath;
+    for (const auto& childName : childPath) {
+        path.push_back(childName);
+    }
+    return path;
+}
+
 }

ydb/core/grpc_services/rpc_alter_table.cpp

@@ -109,12 +109,12 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
             break;
 
         case EOp::Attribute:
-            PrepareAlterUserAttrubutes();
+        case EOp::AddChangefeed:
+        case EOp::DropChangefeed:
+            GetProxyServices();
             break;
 
-        case EOp::AddChangefeed:
         case EOp::DropIndex:
-        case EOp::DropChangefeed:
         case EOp::RenameIndex:
             AlterTable(ctx);
             break;
@@ -197,7 +197,7 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
         Navigate(msg->Services.SchemeCache, ctx);
     }
 
-    void PrepareAlterUserAttrubutes() {
+    void GetProxyServices() {
         using namespace NTxProxy;
         Send(MakeTxProxyID(), new TEvTxUserProxy::TEvGetProxyServicesRequest);
     }
@@ -222,13 +222,38 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
         auto ev = CreateNavigateForPath(DatabaseName);
         {
             auto& entry = static_cast<TEvTxProxySchemeCache::TEvNavigateKeySet*>(ev)->Request->ResultSet.emplace_back();
-            entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpTable;
             entry.Path = paths;
         }
 
         Send(schemeCache, ev);
     }
 
+    void Navigate(const TTableId& pathId) {
+        DatabaseName = Request_->GetDatabaseName()
+            .GetOrElse(DatabaseFromDomain(AppData()));
+
+        auto ev = CreateNavigateForPath(DatabaseName);
+        {
+            auto& entry = static_cast<TEvTxProxySchemeCache::TEvNavigateKeySet*>(ev)->Request->ResultSet.emplace_back();
+            entry.RequestType = NSchemeCache::TSchemeCacheNavigate::TEntry::ERequestType::ByTableId;
+            entry.TableId = pathId;
+            entry.ShowPrivatePath = true;
+            entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
+        }
+
+        Send(MakeSchemeCacheID(), ev);
+    }
+
+    static bool IsChangefeedOperation(EOp type) {
+        switch (type) {
+        case EOp::AddChangefeed:
+        case EOp::DropChangefeed:
+            return true;
+        default:
+            return false;
+        }
+    }
+
     void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
         TXLOG_D("Handle TEvTxProxySchemeCache::TEvNavigateKeySetResult"
                     << ", errors# " << ev->Get()->Request.Get()->ErrorCount);
@@ -251,13 +276,48 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
             return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
         }
 
+        Y_ABORT_UNLESS(!resp->ResultSet.empty());
+        const auto& entry = resp->ResultSet.back();
+
+        switch (entry.Kind) {
+        case NSchemeCache::TSchemeCacheNavigate::KindTable:
+        case NSchemeCache::TSchemeCacheNavigate::KindColumnTable:
+        case NSchemeCache::TSchemeCacheNavigate::KindExternalTable:
+        case NSchemeCache::TSchemeCacheNavigate::KindExternalDataSource:
+        case NSchemeCache::TSchemeCacheNavigate::KindView:
+            break; // table
+        case NSchemeCache::TSchemeCacheNavigate::KindIndex:
+            if (IsChangefeedOperation(OpType)) {
+                break;
+            }
+            [[fallthrough]];
+        default:
+            Request_->RaiseIssue(MakeIssue(NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR, TStringBuilder()
+                << "Unable to nagivate: " << JoinPath(entry.Path) << " status: PathNotTable"));
+            return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
+        }
+
         switch (OpType) {
         case EOp::AddIndex:
             return AlterTableAddIndexOp(resp, ctx);
         case EOp::Attribute:
-            Y_ABORT_UNLESS(!resp->ResultSet.empty());
             ResolvedPathId = resp->ResultSet.back().TableId.PathId;
             return AlterTable(ctx);
+        case EOp::AddChangefeed:
+        case EOp::DropChangefeed:
+            if (entry.Kind != NSchemeCache::TSchemeCacheNavigate::KindIndex) {
+                AlterTable(ctx);
+            } else if (auto list = entry.ListNodeEntry) {
+                if (list->Children.size() != 1) {
+                    return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
+                }
+
+                const auto& child = list->Children.at(0);
+                AlterTable(ctx, CanonizePath(ChildPath(NKikimr::SplitPath(GetProtoRequest()->path()), child.Name)));
+            } else {
+                Navigate(entry.TableId);
+            }
+            break;
         default:
             TXLOG_E("Got unexpected cache response");
             return Reply(Ydb::StatusIds::INTERNAL_ERROR, ctx);
@@ -351,13 +411,14 @@ class TAlterTableRPC : public TRpcSchemeRequestActor<TAlterTableRPC, TEvAlterTab
         Die(ctx);
     }
 
-    void AlterTable(const TActorContext &ctx) { 
+    void AlterTable(const TActorContext &ctx, const TMaybe<TString>& overridePath = {}) { 
         const auto req = GetProtoRequest();
         std::unique_ptr<TEvTxUserProxy::TEvProposeTransaction> proposeRequest = CreateProposeTransaction();
         auto modifyScheme = proposeRequest->Record.MutableTransaction()->MutableModifyScheme();
+        modifyScheme->SetAllowAccessToPrivatePaths(overridePath.Defined());
         Ydb::StatusIds::StatusCode code;
         TString error;
-        if (!BuildAlterTableModifyScheme(req, modifyScheme, Profiles, ResolvedPathId, code, error)) {
+        if (!BuildAlterTableModifyScheme(overridePath.GetOrElse(req->path()), req, modifyScheme, Profiles, ResolvedPathId, code, error)) {
             NYql::TIssues issues;
             issues.AddIssue(NYql::TIssue(error));
             return Reply(code, issues, ctx);

ydb/core/grpc_services/rpc_describe_table.cpp

@@ -1,11 +1,11 @@
-#include "service_table.h"
-#include <ydb/core/grpc_services/base/base.h>
-
 #include "rpc_calls.h"
 #include "rpc_scheme_base.h"
-
 #include "service_table.h"
-#include "rpc_common/rpc_common.h"
+
+#include <ydb/core/base/path.h>
+#include <ydb/core/base/table_index.h>
+#include <ydb/core/grpc_services/base/base.h>
+#include <ydb/core/grpc_services/rpc_common/rpc_common.h>
 #include <ydb/core/tx/schemeshard/schemeshard.h>
 #include <ydb/core/ydb_convert/table_description.h>
 #include <ydb/core/ydb_convert/ydb_convert.h>
@@ -22,25 +22,84 @@ using TEvDescribeTableRequest = TGrpcRequestOperationCall<Ydb::Table::DescribeTa
 class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDescribeTableRequest> {
     using TBase = TRpcSchemeRequestActor<TDescribeTableRPC, TEvDescribeTableRequest>;
 
+    TString OverrideName;
+
+    static bool ShowPrivatePath(const TString& path) {
+        if (AppData()->AllowPrivateTableDescribeForTest) {
+           return true;
+        }
+
+        if (path.EndsWith("/indexImplTable")) {
+            return true;
+        }
+
+        return false;
+    }
+
 public:
     TDescribeTableRPC(IRequestOpCtx* msg)
         : TBase(msg) {}
 
     void Bootstrap(const TActorContext &ctx) {
         TBase::Bootstrap(ctx);
 
-        SendProposeRequest(ctx);
+        const auto& path = GetProtoRequest()->path();
+        const auto paths = NKikimr::SplitPath(path);
+        if (paths.empty()) {
+            Request_->RaiseIssue(NYql::TIssue("Invalid path"));
+            return Reply(Ydb::StatusIds::BAD_REQUEST, ctx);
+        }
+
+        auto navigate = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
+        navigate->DatabaseName = CanonizePath(Request_->GetDatabaseName().GetOrElse(""));
+        auto& entry = navigate->ResultSet.emplace_back();
+        entry.Path = paths;
+        entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpList;
+        entry.SyncVersion = true;
+        entry.ShowPrivatePath = ShowPrivatePath(path);
+
+        Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate));
         Become(&TDescribeTableRPC::StateWork);
     }
 
 private:
     void StateWork(TAutoPtr<IEventHandle>& ev) {
         switch (ev->GetTypeRewrite()) {
+            HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
             HFunc(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult, Handle);
             default: TBase::StateWork(ev);
         }
     }
 
+    void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
+        auto* navigate = ev->Get()->Request.Get();
+
+        Y_ABORT_UNLESS(navigate->ResultSet.size() == 1);
+        const auto& entry = navigate->ResultSet.front();
+
+        if (navigate->ErrorCount > 0) {
+            switch (entry.Status) {
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown:
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::RootUnknown:
+                return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
+            default:
+                return Reply(Ydb::StatusIds::UNAVAILABLE, ctx);
+            }
+        }
+
+        if (entry.Kind == NSchemeCache::TSchemeCacheNavigate::KindIndex) {
+            auto list = entry.ListNodeEntry;
+            if (!list || list->Children.size() != 1) {
+                return Reply(Ydb::StatusIds::SCHEME_ERROR, ctx);
+            }
+
+            OverrideName = entry.Path.back();
+            SendProposeRequest(CanonizePath(ChildPath(entry.Path, list->Children.at(0).Name)), ctx);
+        } else {
+            SendProposeRequest(GetProtoRequest()->path(), ctx);
+        }
+    }
+
     void Handle(NSchemeShard::TEvSchemeShard::TEvDescribeSchemeResult::TPtr& ev, const TActorContext& ctx) {
         const auto& record = ev->Get()->GetRecord();
         const auto status = record.GetStatus();
@@ -53,9 +112,10 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
             case NKikimrScheme::StatusSuccess: {
                 const auto& pathDescription = record.GetPathDescription();
                 Ydb::Scheme::Entry* selfEntry = describeTableResult.mutable_self();
-                selfEntry->set_name(pathDescription.GetSelf().GetName());
-                selfEntry->set_type(static_cast<Ydb::Scheme::Entry::Type>(pathDescription.GetSelf().GetPathType()));
                 ConvertDirectoryEntry(pathDescription.GetSelf(), selfEntry, true);
+                if (OverrideName) {
+                    selfEntry->set_name(OverrideName);
+                }
 
                 if (pathDescription.HasColumnTableDescription()) {
                     const auto& tableDescription = pathDescription.GetColumnTableDescription();
@@ -136,9 +196,8 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
         }
     }
 
-    void SendProposeRequest(const TActorContext &ctx) {
+    void SendProposeRequest(const TString& path, const TActorContext& ctx) {
         const auto req = GetProtoRequest();
-        const TString path = req->path();
 
         std::unique_ptr<TEvTxUserProxy::TEvNavigate> navigateRequest(new TEvTxUserProxy::TEvNavigate());
         SetAuthToken(navigateRequest, *Request_);
@@ -153,9 +212,7 @@ class TDescribeTableRPC : public TRpcSchemeRequestActor<TDescribeTableRPC, TEvDe
             record->MutableOptions()->SetReturnPartitionStats(true);
         }
 
-        if (AppData(ctx)->AllowPrivateTableDescribeForTest || path.EndsWith("/indexImplTable")) {
-            record->MutableOptions()->SetShowPrivateTable(true);
-        }
+        record->MutableOptions()->SetShowPrivateTable(ShowPrivatePath(path));
 
         ctx.Send(MakeTxProxyID(), navigateRequest.release());
     }

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

@@ -4075,6 +4075,70 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
         }
     }
 
+    Y_UNIT_TEST(ChangefeedOnIndexTable) {
+        TKikimrRunner kikimr(TKikimrSettings()
+            .SetPQConfig(DefaultPQConfig())
+            .SetEnableChangefeedsOnIndexTables(true));
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        {
+            auto query = R"(
+                --!syntax_v1
+                CREATE TABLE `/Root/table` (
+                    Key Uint64,
+                    Value String,
+                    PRIMARY KEY (Key),
+                    INDEX SyncIndex GLOBAL SYNC ON (`Value`),
+                    INDEX AsyncIndex GLOBAL ASYNC ON (`Value`)
+                );
+            )";
+
+            auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+
+        const auto changefeed = TChangefeedDescription("feed", EChangefeedMode::KeysOnly, EChangefeedFormat::Json);
+        {
+            auto result = session.AlterTable("/Root/table/AsyncIndex", TAlterTableSettings()
+                .AppendAddChangefeeds(changefeed)
+            ).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::PRECONDITION_FAILED, result.GetIssues().ToString());
+        }
+        {
+            auto result = session.AlterTable("/Root/table/SyncIndex", TAlterTableSettings()
+                .AppendAddChangefeeds(changefeed)
+            ).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+    }
+
+    Y_UNIT_TEST(DescribeIndexTable) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        {
+            auto query = R"(
+                --!syntax_v1
+                CREATE TABLE `/Root/table` (
+                    Key Uint64,
+                    Value String,
+                    PRIMARY KEY (Key),
+                    INDEX SyncIndex GLOBAL SYNC ON (`Value`)
+                );
+            )";
+
+            auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            auto desc = session.DescribeTable("/Root/table/SyncIndex").ExtractValueSync();
+            UNIT_ASSERT_C(desc.IsSuccess(), desc.GetIssues().ToString());
+            UNIT_ASSERT_VALUES_EQUAL(desc.GetEntry().Name, "SyncIndex");
+        }
+    }
+
     Y_UNIT_TEST(CreatedAt) {
         TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));
         auto scheme = NYdb::NScheme::TSchemeClient(kikimr.GetDriver(), TCommonClientSettings().Database("/Root"));

ydb/core/protos/feature_flags.proto

@@ -145,4 +145,5 @@ message TFeatureFlags {
     optional bool EnableColumnStatistics = 130 [default = false];
     optional bool EnableSingleCompositeActionGroup = 131 [default = false];
     optional bool EnableResourcePoolsOnServerless = 132 [default = false];
+    optional bool EnableChangefeedsOnIndexTables = 134 [default = false];
 }

ydb/core/protos/flat_scheme_op.proto

@@ -913,10 +913,6 @@ message TCreateCdcStream {
     optional TCdcStreamDescription StreamDescription = 2;
     optional uint64 RetentionPeriodSeconds = 3 [default = 86400]; // 1d by default
     optional uint32 TopicPartitions = 4;
-    oneof IndexMode {
-        google.protobuf.Empty AllIndexes = 5; // Create topic per each index
-        string IndexName = 6;
-    }
 }
 
 message TAlterCdcStream {
@@ -1631,7 +1627,6 @@ message TIndexBuildControl {
 
 message TLockConfig {
     optional string Name = 1;
-    optional bool AllowIndexImplLock = 2;
 }
 
 message TLockGuard {

ydb/core/testlib/basics/feature_flags.h

@@ -61,6 +61,7 @@ class TTestFeatureFlagsHolder {
     FEATURE_FLAG_SETTER(EnableCMSRequestPriorities)
     FEATURE_FLAG_SETTER(EnableTableDatetime64)
     FEATURE_FLAG_SETTER(EnableResourcePools)
+    FEATURE_FLAG_SETTER(EnableChangefeedsOnIndexTables)
 
     #undef FEATURE_FLAG_SETTER
 };

ydb/core/tx/replication/controller/dst_alterer.cpp

@@ -41,6 +41,7 @@ class TDstAlterer: public TActorBootstrapped<TDstAlterer> {
 
         switch (Kind) {
         case TReplication::ETargetKind::Table:
+        case TReplication::ETargetKind::IndexTable:
             tx.SetOperationType(NKikimrSchemeOp::ESchemeOpAlterTable);
             PathIdFromPathId(DstPathId, tx.MutableAlterTable()->MutablePathId());
             tx.MutableAlterTable()->MutableReplicationConfig()->SetMode(