EnableChangefeedsOnIndexTables feature flag (#7085)

CyberROFL · CyberROFL · commit b353499d6399 · 2024-07-26T16:50:48.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
@@ -4076,7 +4076,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
@@ -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];
 }
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/replication/ut_helpers/test_env.h b/ydb/core/tx/replication/ut_helpers/test_env.h
@@ -10,6 +10,9 @@
 
 namespace NKikimr::NReplication::NTestHelpers {
 
+class TFeatureFlags: public TTestFeatureFlagsHolder<TFeatureFlags> {
+};
+
 template <bool UseDatabase = true>
 class TEnv {
     static constexpr char DomainName[] = "Root";
@@ -63,7 +66,7 @@ class TEnv {
     TEnv(const TFeatureFlags& featureFlags, bool init = true)
         : Settings(Tests::TServerSettings(PortManager.GetPort(), {}, MakePqConfig())
             .SetDomainName(DomainName)
-            .SetFeatureFlags(featureFlags)
+            .SetFeatureFlags(featureFlags.FeatureFlags)
         )
         , Server(Settings)
         , Client(Settings)
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_create_cdc_stream.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_create_cdc_stream.cpp
@@ -803,8 +803,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
@@ -541,6 +541,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
`@@ -4076,7 +4076,9 @@ Y_UNIT_TEST_SUITE(KqpScheme) {`
`4076`	`4076`	`}`
`4077`	`4077`
`4078`	`4078`	`Y_UNIT_TEST(ChangefeedOnIndexTable) {`
`4079`		`- TKikimrRunner kikimr(TKikimrSettings().SetPQConfig(DefaultPQConfig()));`
	`4079`	`+ TKikimrRunner kikimr(TKikimrSettings()`
	`4080`	`+ .SetPQConfig(DefaultPQConfig())`
	`4081`	`+ .SetEnableChangefeedsOnIndexTables(true));`
`4080`	`4082`	`auto db = kikimr.GetTableClient();`
`4081`	`4083`	`auto session = db.CreateSession().GetValueSync().GetSession();`
`4082`	`4084`
Original file line number	Diff line number	Diff line change
`@@ -145,4 +145,5 @@ message TFeatureFlags {`
`145`	`145`	`optional bool EnableColumnStatistics = 130 [default = false];`
`146`	`146`	`optional bool EnableSingleCompositeActionGroup = 131 [default = false];`
`147`	`147`	`optional bool EnableResourcePoolsOnServerless = 132 [default = false];`
	`148`	`+ optional bool EnableChangefeedsOnIndexTables = 134 [default = false];`
`148`	`149`	`}`
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{`