[BACKPORT 2024.1][yugabyte#23482] CDCSDK: Add an option to disable dynamic table addition while creating a stream

Summary:
####Backport Description
No merge conflicts were encountered.

####Original Description
This diff adds an option in the yb-admin command `create_change_data_stream` to disable dynamic tables addition on the newly created streams. Dynamic table addition cannot be re-enabled on such streams. By default all the newly created streams will have dynamic tables addition enabled.

In order to create a stream with dynamic tables addition disabled, the user needs to specify the `[<dynamic_tables_option>]` which can take the following values:

  - DYNAMIC_TABLES_ENABLED
  - DYNAMIC_TABLES_DISABLED

Example to create a stream with dynamic tables disabled:
```
./yb-admin --master_addresses <master-addresses> create_change_data_stream ysql.yugabyte EXPLICIT CHANGE USE_SNAPSHOT DYNAMIC_TABLES_DISABLED
```

For a stream created with dynamic table addition disabled, the field `cdcsdk_disable_dynamic_table_addition` in stream metadata will be set to true.

Dynamic table addition will not be disabled by this mechanism in streams associated with replication slots.

**Upgrade / Rollback safety **
The diff introduces a new message `CDCSDKStreamCreateOptionsPB` in common.proto. This message contains the newly introduced `CDCSDKDynamicTablesOption` as its only field. The message `CDCSDKStreamCreateOptionsPB` is used only in `CreateCDCStreamRequestPB` and should be used for adding more options if required in future.

These changes do not need to be protected by an autoflag for the following reasons:

  # The proto changes do not get written to the disk.
  # The newly introduced field is optional.
  # The proto changes are for yb-admin command related rpcs.
Jira: DB-12399

Original commit: d0dfe63 / D37534

Test Plan:
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestDisablingDynamicTableAdditionAtStreamCreationTime
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestDynamicTablesShouldBeEnabledForStreamsWithSlotName

Manual testing

Reviewers: skumar, asrinivasan, siddharth.shah, xCluster, hsunder

Reviewed By: siddharth.shah

Subscribers: ycdcxcluster, ybase, yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D37732

Author: Sumukh-Phalgaonkar

src/yb/cdc/cdc_service.cc

@@ -1092,10 +1092,18 @@ Status CDCServiceImpl::CreateCDCStreamForNamespace(
      snapshot_option = req->cdcsdk_consistent_snapshot_option();
   }
 
+  // Dynamic Tables option
+  CDCSDKDynamicTablesOption dynamic_tables_option =
+      CDCSDKDynamicTablesOption::DYNAMIC_TABLES_ENABLED;
+  if (req->has_cdcsdk_stream_create_options() &&
+      req->cdcsdk_stream_create_options().has_cdcsdk_dynamic_tables_option()) {
+     dynamic_tables_option = req->cdcsdk_stream_create_options().cdcsdk_dynamic_tables_option();
+  }
+
   xrepl::StreamId db_stream_id = VERIFY_RESULT_OR_SET_CODE(
       client()->CreateCDCSDKStreamForNamespace(
           ns_id, options, populate_namespace_id_as_table_id, ReplicationSlotName(""), std::nullopt,
-          snapshot_option, deadline),
+          snapshot_option, deadline, dynamic_tables_option),
       CDCError(CDCErrorPB::INTERNAL_ERROR));
   resp->set_db_stream_id(db_stream_id.ToString());
   return Status::OK();

src/yb/cdc/cdc_service.proto

@@ -230,6 +230,9 @@ message CreateCDCStreamRequestPB {
   optional bool transactional = 7; // [default = false]
   optional YQLDatabase db_type = 8 [default = YQL_DATABASE_PGSQL] ;
   optional CDCSDKSnapshotOption cdcsdk_consistent_snapshot_option = 9;
+  // cdcsdk_stream_create_options field should be used for adding more stream creation options
+  // related to CDCSDK in future.
+  optional CDCSDKStreamCreateOptionsPB cdcsdk_stream_create_options = 10;
 }
 
 message CreateCDCStreamResponsePB {

src/yb/client/client.cc

@@ -1497,7 +1497,8 @@ Result<xrepl::StreamId> YBClient::CreateCDCSDKStreamForNamespace(
     const std::optional<std::string>& replication_slot_plugin_name,
     const std::optional<CDCSDKSnapshotOption>& consistent_snapshot_option,
     CoarseTimePoint deadline,
-    uint64_t *consistent_snapshot_time) {
+    const CDCSDKDynamicTablesOption& dynamic_tables_option,
+    uint64_t *consistent_snapshot_time_out) {
   CreateCDCStreamRequestPB req;
 
   if (populate_namespace_id_as_table_id) {
@@ -1521,13 +1522,15 @@ Result<xrepl::StreamId> YBClient::CreateCDCSDKStreamForNamespace(
   if (replication_slot_plugin_name.has_value()) {
     req.set_cdcsdk_ysql_replication_slot_plugin_name(*replication_slot_plugin_name);
   }
+  req.mutable_cdcsdk_stream_create_options()->set_cdcsdk_dynamic_tables_option(
+      dynamic_tables_option);
 
   CreateCDCStreamResponsePB resp;
   deadline = PatchAdminDeadline(deadline);
   CALL_SYNC_LEADER_MASTER_RPC_WITH_DEADLINE(Replication, req, resp, deadline, CreateCDCStream);
 
-  if (consistent_snapshot_time && resp.has_cdcsdk_consistent_snapshot_time()) {
-    *consistent_snapshot_time = resp.cdcsdk_consistent_snapshot_time();
+  if (consistent_snapshot_time_out && resp.has_cdcsdk_consistent_snapshot_time()) {
+    *consistent_snapshot_time_out = resp.cdcsdk_consistent_snapshot_time();
   }
   return xrepl::StreamId::FromString(resp.stream_id());
 }

src/yb/client/client.h

@@ -611,7 +611,9 @@ class YBClient {
       const std::optional<std::string>& replication_slot_plugin_name = std::nullopt,
       const std::optional<CDCSDKSnapshotOption>& consistent_snapshot_option = std::nullopt,
       CoarseTimePoint deadline = CoarseTimePoint(),
-      uint64_t *consistent_snapshot_time = nullptr);
+      const CDCSDKDynamicTablesOption& dynamic_tables_option =
+          CDCSDKDynamicTablesOption::DYNAMIC_TABLES_ENABLED,
+      uint64_t* consistent_snapshot_time_out = nullptr);
 
   // Delete multiple CDC streams.
   Status DeleteCDCStream(

src/yb/common/common.proto

@@ -671,3 +671,9 @@ message WaitStateInfoPB {
   // A string representation of wait_status_code for easy human consumption.
   optional string wait_status_code_as_string = 4;
 }
+
+// This message should be used for specifying stream creation options for CDCSDK streams.
+message CDCSDKStreamCreateOptionsPB {
+  optional
+      CDCSDKDynamicTablesOption cdcsdk_dynamic_tables_option = 1 [default = DYNAMIC_TABLES_ENABLED];
+}

src/yb/common/common_types.proto

@@ -180,6 +180,12 @@ enum CDCSDKSnapshotOption {
   USE_SNAPSHOT = 2; // Client wishes to consume snapshot from the source universe
 }
 
+enum CDCSDKDynamicTablesOption {
+  DYNAMIC_TABLES_UNDEFINED = 0;
+  DYNAMIC_TABLES_ENABLED = 1;
+  DYNAMIC_TABLES_DISABLED = 2;
+}
+
 // Available replica identity modes for use in CDC
 enum PgReplicaIdentity {
   // Entire updated row as new image, only key as old image for DELETE

src/yb/integration-tests/cdcsdk_test_base.cc

@@ -407,24 +407,29 @@ void CDCSDKTestBase::InitCreateStreamRequest(
     CreateCDCStreamRequestPB* create_req,
     const CDCCheckpointType& checkpoint_type,
     const CDCRecordType& record_type,
-    const std::string& namespace_name) {
+    const std::string& namespace_name,
+    CDCSDKDynamicTablesOption dynamic_tables_option) {
   create_req->set_namespace_name(namespace_name);
   create_req->set_checkpoint_type(checkpoint_type);
   create_req->set_record_type(record_type);
   create_req->set_record_format(CDCRecordFormat::PROTO);
   create_req->set_source_type(CDCSDK);
+  create_req->mutable_cdcsdk_stream_create_options()->set_cdcsdk_dynamic_tables_option(
+      dynamic_tables_option);
 }
 
 // This creates a DB stream on the database kNamespaceName by default.
 Result<xrepl::StreamId> CDCSDKTestBase::CreateDBStream(
-    CDCCheckpointType checkpoint_type, CDCRecordType record_type, std::string namespace_name) {
+    CDCCheckpointType checkpoint_type, CDCRecordType record_type, std::string namespace_name,
+    CDCSDKDynamicTablesOption dynamic_tables_option) {
   CreateCDCStreamRequestPB req;
   CreateCDCStreamResponsePB resp;
 
   rpc::RpcController rpc;
   rpc.set_timeout(MonoDelta::FromMilliseconds(FLAGS_cdc_write_rpc_timeout_ms));
 
-  InitCreateStreamRequest(&req, checkpoint_type, record_type, namespace_name);
+  InitCreateStreamRequest(
+      &req, checkpoint_type, record_type, namespace_name, dynamic_tables_option);
 
   RETURN_NOT_OK(cdc_proxy_->CreateCDCStream(req, &resp, &rpc));
   if (resp.has_error()) {

src/yb/integration-tests/cdcsdk_test_base.h

@@ -215,12 +215,16 @@ class CDCSDKTestBase : public YBTest {
       CreateCDCStreamRequestPB* create_req,
       const CDCCheckpointType& checkpoint_type = CDCCheckpointType::EXPLICIT,
       const CDCRecordType& record_type = CDCRecordType::CHANGE,
-      const std::string& namespace_name = kNamespaceName);
+      const std::string& namespace_name = kNamespaceName,
+      CDCSDKDynamicTablesOption dynamic_tables_option =
+          CDCSDKDynamicTablesOption::DYNAMIC_TABLES_ENABLED);
 
   Result<xrepl::StreamId> CreateDBStream(
       CDCCheckpointType checkpoint_type = CDCCheckpointType::EXPLICIT,
       CDCRecordType record_type = CDCRecordType::CHANGE,
-      std::string namespace_name = kNamespaceName);
+      std::string namespace_name = kNamespaceName,
+      CDCSDKDynamicTablesOption dynamic_tables_option =
+          CDCSDKDynamicTablesOption::DYNAMIC_TABLES_ENABLED);
 
   // Creates a DB stream on the database kNamespaceName using the Replication Slot syntax.
   // Only supports the CDCCheckpointType::EXPLICIT and CDCRecordType::CHANGE.

src/yb/integration-tests/cdcsdk_ysql-test.cc

@@ -9985,5 +9985,70 @@ TEST_F(CDCSDKYsqlTest, TestCleanupOfUnqualifiedTableOnDrop) {
       "Waiting for cdc_state table to be in sync after table drop");
 }
 
+TEST_F(CDCSDKYsqlTest, TestDisablingDynamicTableAdditionAtStreamCreationTime) {
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_yb_enable_cdc_consistent_snapshot_streams) = true;
+
+  // Setup cluster.
+  ASSERT_OK(SetUpWithParams(3, 3, false));
+
+  auto table_1 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_1"));
+  auto table_2 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_2"));
+
+  // Create a yb-admin stream with dynamic tables enabled and one with dynamic tables disabled.
+  auto stream_id_1 = ASSERT_RESULT(CreateDBStream(
+      CDCCheckpointType::EXPLICIT, CDCRecordType::CHANGE, kNamespaceName,
+      CDCSDKDynamicTablesOption::DYNAMIC_TABLES_ENABLED));
+  auto stream_id_2 = ASSERT_RESULT(CreateDBStream(
+      CDCCheckpointType::EXPLICIT, CDCRecordType::CHANGE, kNamespaceName,
+      CDCSDKDynamicTablesOption::DYNAMIC_TABLES_DISABLED));
+
+  auto stream_info = ASSERT_RESULT(GetDBStreamInfo(stream_id_1));
+  ASSERT_EQ(stream_info.table_info_size(), 2);
+
+  stream_info = ASSERT_RESULT(GetDBStreamInfo(stream_id_2));
+  ASSERT_EQ(stream_info.table_info_size(), 2);
+
+  // Create a dynamic table.
+  auto table_3 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_3"));
+
+  // Sleep for 10 seconds for master bg thread to run.
+  SleepFor(MonoDelta::FromSeconds(10 * kTimeMultiplier));
+
+  // Stream_id_1 will have the dynamic table added.
+  stream_info = ASSERT_RESULT(GetDBStreamInfo(stream_id_1));
+  ASSERT_EQ(stream_info.table_info_size(), 3);
+
+  // Stream_id_2 will not have the dynamic table added.
+  stream_info = ASSERT_RESULT(GetDBStreamInfo(stream_id_2));
+  ASSERT_EQ(stream_info.table_info_size(), 2);
+}
+
+TEST_F(CDCSDKYsqlTest, TestDynamicTablesShouldBeEnabledForStreamsWithSlotName) {
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_yb_enable_cdc_consistent_snapshot_streams) = true;
+
+  // Setup cluster.
+  ASSERT_OK(SetUpWithParams(3, 3, false));
+
+  auto table_1 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_1"));
+  auto table_2 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_2"));
+
+  // Create a slot.
+  auto slot_stream_id = ASSERT_RESULT(CreateDBStreamWithReplicationSlot());
+
+  auto stream_info = ASSERT_RESULT(GetDBStreamInfo(slot_stream_id));
+  ASSERT_EQ(stream_info.table_info_size(), 2);
+
+  // Create a dynamic table.
+  auto table_3 = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, "table_3"));
+
+  // Sleep for 10 seconds for master bg thread to run.
+  SleepFor(MonoDelta::FromSeconds(10 * kTimeMultiplier));
+
+  // Since dynamic table addition is always enabled in streams associated with slots, table_3 will
+  // get added to its stream metadata.
+  stream_info = ASSERT_RESULT(GetDBStreamInfo(slot_stream_id));
+  ASSERT_EQ(stream_info.table_info_size(), 3);
+}
+
 }  // namespace cdc
 }  // namespace yb

src/yb/master/master_replication.proto

@@ -76,6 +76,9 @@ message CreateCDCStreamRequestPB {
   optional string cdcsdk_ysql_replication_slot_name = 8;
   optional CDCSDKSnapshotOption cdcsdk_consistent_snapshot_option = 9;
   optional string cdcsdk_ysql_replication_slot_plugin_name = 10;
+  // cdcsdk_stream_create_options field should be used for adding more stream creation options
+  // related to CDCSDK in future.
+  optional CDCSDKStreamCreateOptionsPB cdcsdk_stream_create_options = 11;
 }
 
 message CreateCDCStreamResponsePB {