Merge c9d95d6 into f36b8b5

CyberROFL · web-flow · commit 6b4dcccc270b · 2024-07-25T12:52:05.000+03:00
diff --git a/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp b/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
@@ -4385,7 +4385,9 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
     }
 
     Y_UNIT_TEST(ChangefeedOnIndexTable) {
-        TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));
+        TKikimrRunner kikimr(TKikimrSettings()
+            .SetPQConfig(DefaultPQConfig())
+            .SetEnableChangefeedsOnIndexTables(true));
         auto db = kikimr.GetTableClient();
         auto session = db.CreateSession().GetValueSync().GetSession();
 
diff --git a/ydb/core/protos/feature_flags.proto b/ydb/core/protos/feature_flags.proto
@@ -146,4 +146,5 @@ message TFeatureFlags {
     optional bool EnableSingleCompositeActionGroup = 131 [default = false];
     optional bool EnableResourcePoolsOnServerless = 132 [default = false];
     optional bool EnableVectorIndex = 133 [default = false];
+    optional bool EnableChangefeedsOnIndexTables = 134 [default = false];
 }
diff --git a/ydb/core/testlib/basics/feature_flags.h b/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
 };
diff --git a/ydb/core/tx/replication/controller/dst_creator_ut.cpp b/ydb/core/tx/replication/controller/dst_creator_ut.cpp
@@ -71,7 +71,7 @@ Y_UNIT_TEST_SUITE(DstCreator) {
     }
 
     void WithIndex(const TString& replicatedPath, NKikimrSchemeOp::EIndexType indexType) {
-        TEnv env;
+        TEnv env(TFeatureFlags().SetEnableChangefeedsOnIndexTables(true));
         env.GetRuntime().SetLogPriority(NKikimrServices::REPLICATION_CONTROLLER, NLog::PRI_TRACE);
 
         const auto tableDesc = TTestTableDescription{
diff --git a/ydb/core/tx/replication/service/table_writer_ut.cpp b/ydb/core/tx/replication/service/table_writer_ut.cpp
@@ -39,10 +39,7 @@ Y_UNIT_TEST_SUITE(LocalTableWriter) {
     }
 
     Y_UNIT_TEST(SupportedTypes) {
-        auto featureFlags = TFeatureFlags();
-        featureFlags.SetEnableTableDatetime64(true);
-
-        TEnv env(featureFlags);
+        TEnv env(TFeatureFlags().SetEnableTableDatetime64(true));
         env.GetRuntime().SetLogPriority(NKikimrServices::REPLICATION_SERVICE, NLog::PRI_DEBUG);
 
         env.CreateTable("/Root", *MakeTableDescription(TTestTableDescription{
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_create_cdc_stream.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_create_cdc_stream.cpp
@@ -808,8 +808,15 @@ ISubOperation::TPtr RejectOnTablePathChecks(const TOperationId& opId, const TPat
     if (checks) {
         if (!tablePath.IsInsideTableIndexPath()) {
             checks.IsCommonSensePath();
-        } else if (!tablePath.Parent().IsTableIndex(NKikimrSchemeOp::EIndexTypeGlobal)) {
-            return CreateReject(opId, NKikimrScheme::StatusPreconditionFailed, "Cannot add changefeed to index table");
+        } else {
+            if (!tablePath.Parent().IsTableIndex(NKikimrSchemeOp::EIndexTypeGlobal)) {
+                return CreateReject(opId, NKikimrScheme::StatusPreconditionFailed,
+                    "Cannot add changefeed to index table");
+            }
+            if (!AppData()->FeatureFlags.GetEnableChangefeedsOnIndexTables()) {
+                return CreateReject(opId, NKikimrScheme::StatusPreconditionFailed,
+                    "Changefeed on index table is not supported yet");
+            }
         }
     }
 
diff --git a/ydb/core/tx/schemeshard/ut_cdc_stream/ut_cdc_stream.cpp b/ydb/core/tx/schemeshard/ut_cdc_stream/ut_cdc_stream.cpp
@@ -1200,9 +1200,38 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
         }
     }
 
+    Y_UNIT_TEST(StreamOnIndexTableNegative) {
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime, TTestEnvOptions().EnableChangefeedsOnIndexTables(false));
+        ui64 txId = 100;
+
+        TestCreateIndexedTable(runtime, ++txId, "/MyRoot", R"(
+            TableDescription {
+              Name: "Table"
+              Columns { Name: "key" Type: "Uint64" }
+              Columns { Name: "indexed" Type: "Uint64" }
+              KeyColumnNames: ["key"]
+            }
+            IndexDescription {
+              Name: "Index"
+              KeyColumnNames: ["indexed"]
+            }
+        )");
+        env.TestWaitNotification(runtime, txId);
+
+        TestCreateCdcStream(runtime, ++txId, "/MyRoot/Table/Index", R"(
+            TableName: "indexImplTable"
+            StreamDescription {
+              Name: "Stream"
+              Mode: ECdcStreamModeKeysOnly
+              Format: ECdcStreamFormatProto
+            }
+        )", {NKikimrScheme::StatusPreconditionFailed});
+    }
+
     Y_UNIT_TEST(StreamOnIndexTable) {
         TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        TTestEnv env(runtime, TTestEnvOptions().EnableChangefeedsOnIndexTables(true));
         ui64 txId = 100;
 
         TestCreateIndexedTable(runtime, ++txId, "/MyRoot", R"(
@@ -1298,7 +1327,7 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
 
     Y_UNIT_TEST(StreamOnBuildingIndexTable) {
         TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        TTestEnv env(runtime, TTestEnvOptions().EnableChangefeedsOnIndexTables(true));
         ui64 txId = 100;
 
         TestCreateTable(runtime, ++txId, "/MyRoot", R"(
@@ -1350,7 +1379,7 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
 
     Y_UNIT_TEST(DropIndexWithStream) {
         TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        TTestEnv env(runtime, TTestEnvOptions().EnableChangefeedsOnIndexTables(true));
         ui64 txId = 100;
 
         TestCreateIndexedTable(runtime, ++txId, "/MyRoot", R"(
@@ -1393,7 +1422,7 @@ Y_UNIT_TEST_SUITE(TCdcStreamTests) {
 
     Y_UNIT_TEST(DropTableWithIndexWithStream) {
         TTestBasicRuntime runtime;
-        TTestEnv env(runtime, TTestEnvOptions().EnableProtoSourceIdInfo(true));
+        TTestEnv env(runtime, TTestEnvOptions().EnableChangefeedsOnIndexTables(true));
         ui64 txId = 100;
 
         TestCreateIndexedTable(runtime, ++txId, "/MyRoot", R"(
diff --git a/ydb/core/tx/schemeshard/ut_cdc_stream_reboots/ut_cdc_stream_reboots.cpp b/ydb/core/tx/schemeshard/ut_cdc_stream_reboots/ut_cdc_stream_reboots.cpp
@@ -10,7 +10,9 @@ Y_UNIT_TEST_SUITE(TCdcStreamWithRebootsTests) {
     template <typename T>
     void CreateStream(const TMaybe<NKikimrSchemeOp::ECdcStreamState>& state = Nothing(), bool vt = false, bool onIndex = false) {
         T t;
-        t.GetTestEnvOptions().EnableChangefeedInitialScan(true);
+        t.GetTestEnvOptions()
+            .EnableChangefeedInitialScan(true)
+            .EnableChangefeedsOnIndexTables(true);
 
         t.Run([&](TTestActorRuntime& runtime, bool& activeZone) {
             {
@@ -290,7 +292,9 @@ Y_UNIT_TEST_SUITE(TCdcStreamWithRebootsTests) {
     template <typename T>
     void DropStream(const TMaybe<NKikimrSchemeOp::ECdcStreamState>& state = Nothing(), bool onIndex = false) {
         T t;
-        t.GetTestEnvOptions().EnableChangefeedInitialScan(true);
+        t.GetTestEnvOptions()
+            .EnableChangefeedInitialScan(true)
+            .EnableChangefeedsOnIndexTables(true);
 
         t.Run([&](TTestActorRuntime& runtime, bool& activeZone) {
             const TString path = !onIndex ? "/MyRoot" : "/MyRoot/Table/Index";
diff --git a/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp b/ydb/core/tx/schemeshard/ut_helpers/test_env.cpp
@@ -542,6 +542,7 @@ NSchemeShardUT_Private::TTestEnv::TTestEnv(TTestActorRuntime& runtime, const TTe
     app.SetEnableServerlessExclusiveDynamicNodes(opts.EnableServerlessExclusiveDynamicNodes_);
     app.SetEnableAddColumsWithDefaults(opts.EnableAddColumsWithDefaults_);
     app.SetEnableReplaceIfExistsForExternalEntities(opts.EnableReplaceIfExistsForExternalEntities_);
+    app.SetEnableChangefeedsOnIndexTables(opts.EnableChangefeedsOnIndexTables_);
 
     app.ColumnShardConfig.SetDisabledOnSchemeShard(false);
 
diff --git a/ydb/core/tx/schemeshard/ut_helpers/test_env.h b/ydb/core/tx/schemeshard/ut_helpers/test_env.h
@@ -65,6 +65,7 @@ namespace NSchemeShardUT_Private {
         OPTION(std::optional<bool>, EnableAddColumsWithDefaults, std::nullopt);
         OPTION(std::optional<bool>, EnableReplaceIfExistsForExternalEntities, std::nullopt);
         OPTION(std::optional<TString>, GraphBackendType, std::nullopt);
+        OPTION(std::optional<bool>, EnableChangefeedsOnIndexTables, std::nullopt);
 
         #undef OPTION
     };

Original file line number	Diff line number	Diff line change
`@@ -4385,7 +4385,9 @@ Y_UNIT_TEST_SUITE(KqpScheme) {`
`4385`	`4385`	`}`
`4386`	`4386`
`4387`	`4387`	`Y_UNIT_TEST(ChangefeedOnIndexTable) {`
`4388`		`- TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));`
	`4388`	`+ TKikimrRunner kikimr(TKikimrSettings()`
	`4389`	`+ .SetPQConfig(DefaultPQConfig())`
	`4390`	`+ .SetEnableChangefeedsOnIndexTables(true));`
`4389`	`4391`	`auto db = kikimr.GetTableClient();`
`4390`	`4392`	`auto session = db.CreateSession().GetValueSync().GetSession();`
`4391`	`4393`
Original file line number	Diff line number	Diff line change
`@@ -146,4 +146,5 @@ message TFeatureFlags {`
`146`	`146`	`optional bool EnableSingleCompositeActionGroup = 131 [default = false];`
`147`	`147`	`optional bool EnableResourcePoolsOnServerless = 132 [default = false];`
`148`	`148`	`optional bool EnableVectorIndex = 133 [default = false];`
	`149`	`+ optional bool EnableChangefeedsOnIndexTables = 134 [default = false];`
`149`	`150`	`}`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ Y_UNIT_TEST_SUITE(DstCreator) {`
`71`	`71`	`}`
`72`	`72`
`73`	`73`	`void WithIndex(const TString& replicatedPath, NKikimrSchemeOp::EIndexType indexType) {`
`74`		`- TEnv env;`
	`74`	`+ TEnv env(TFeatureFlags().SetEnableChangefeedsOnIndexTables(true));`
`75`	`75`	`env.GetRuntime().SetLogPriority(NKikimrServices::REPLICATION_CONTROLLER, NLog::PRI_TRACE);`
`76`	`76`
`77`	`77`	`const auto tableDesc = TTestTableDescription{`