Skip to content

Commit

Permalink
[BACKPORT 2024.2][#23978] xCluster: set up sequences_data stream(s) o…
Browse files Browse the repository at this point in the history 
…n target universe

Summary:
Original commit: 64ac031 / D38052
No conflicts at all.

This is part of the new sequences replication feature.

Here we do the following when setting up sequences replication on the target universe:
  * create the sequences_data table if it does not already exist
  * connect to the corresponding stream for it from the source universe for each namespace we are replicating

Checking that xCluster safe time is computed correctly will be done in a later diff.

This feature is gated by new flag, `--TEST_xcluster_enable_sequence_replication` for now as well as automatic mode replication being on.

This feature is only usable with DB-scoped replication.

More detail on changes:
  * in order to deal with the fact that the code assumes that table IDs are unique across namespaces, I introduce the idea of a *sequences_data alias*.  This is a table ID that looks like `0000ffff00003000800000000000ffff.sequences_data_for.00004001000030008000000000000000`
    * this table ID denotes the sequences_data table, of which there is one per universe, to code outside of xCluster
    * xCluster, however, can distinguish the aliases for different name spaces (the namespace is the last part in the alias) and extract the namespace that alias corresponds to
    * in this way, we can pretend, at least for xCluster, that sequences_data is really a series of tables, one per namespace

  * added code for creating aliases of sequences_data TableId's
  * modified selected catalog manager routines to accept such aliases, operating on the actual sequences_data table
  * refactor `GetTablesEligibleForXClusterReplication` so it just returns information about the tables' naming instead of a full-blown COW entity that is hard to modify
  * alter now sets the field `automatic_ddl_mode` of `SetupUniverseReplicationRequestPB` to convey to the target universe that we are in automatic mode
  * modify inbound replication group setup task so for alter cases the existing automatic DDL mode and is_db_scope settings rather than the incoming set up request
    * discrepancies for automatic DDL mode return status errors
    * fixed code that used to test `is_db_scoped_` to now instead check `!source_namespace_ids_.empty()`, which is what the check actually did before I fixed is_db_scoped_ to mean what it name implies
      * (that is what is_db_scoped_ was previously incorrectly set to)
  * To allow tests using ybadmin to work even though we don't have a way of turning automatic mode using ybadmin parameters, introduced a temporary new flag `--TEST_force_automatic_ddl_replication_mode` that causes ybadmin or YBA to get automatic replication mode when it asks for (semiautomatic) DB-scoped mode.

Fixes #23978
Jira: DB-12879

Test Plan:
The xCluster integration tests that test DB-scoped replication include the following.  I have modified these to test both automatic and semi-automatic modes as follows:
  * xcluster_outbound_replication_group-itest
    * most parameterized by mode; some use default mode
  * xcluster_db_scoped-test
    * a few parameterized by mode; some use default mode
  * xcluster_ddl_replication-test
    * always use automatic mode
  * xcluster_ddl_replication_pgregress-test
    * always use automatic mode
  * xcluster_ysql_index-test
    * a couple of tests use default mode; others don't use DB-scoped replication

Default mode is currently false to match production, but I have tested that everything passes with it switched to true.  Julian, you may want to parameterize more or different tests as you add the extension to automatic mode.

There are various changes to the tests besides allowing automatic versus semi automatic mode testing:
  * in automatic mode for xcluster_db_scoped-test, I ensure that the source and target universes have different namespace IDs
    * (this makes the sequence aliases for the two universes different, making sure the code doesn't get the wrong universe namespace ID when making an alias)
  * I fixed a bunch of bugs where we were using the incorrect universe's namespace or client
  * I verify that the sequences_data table gets created on the target

I am not verifying that the sequences_data streams are connected to each other correctly or that data can flow through them in this diff.  I will do that in the diff where I fix the pollers, which will allow me to flow data through these steams correctly.

Testing the ybadmin sensing commands.  I used the following setup:
```
~/code/yugabyte-db/bin/yb-ctl start --data_dir ~/yugabyte-data1 --ip_start 10 --master_flags "v=2,enable_xcluster_api_v2=true,allowed_preview_flags_csv=enable_xcluster_api_v2,TEST_xcluster_enable_sequence_replication=true,TEST_xcluster_enable_ddl_replication=true,TEST_force_automatic_ddl_replication_mode=true"
~/code/yugabyte-db/bin/yb-ctl start --data_dir ~/yugabyte-data2 --ip_start 20 --master_flags "v=2,enable_xcluster_api_v2=true,allowed_preview_flags_csv=enable_xcluster_api_v2,TEST_xcluster_enable_sequence_replication=true,TEST_xcluster_enable_ddl_replication=true"

~/code/yugabyte-db/bin/ysqlsh -h 127.0.0.10 -c 'CREATE SEQUENCE mdl_sequence INCREMENT 42 START 13;'
~/code/yugabyte-db/bin/ysqlsh -h 127.0.0.20 -c 'CREATE SEQUENCE mdl_sequence INCREMENT 42 START 13;'

# yugabyte database and mdl_sequence have the same OID's on both sides at this point

~/code/yugabyte-db/bin/ysqlsh -h 127.0.0.10 <<EOF
CREATE TABLE my_table (
    id integer DEFAULT nextval('mdl_sequence'),
    value integer
);
EOF

~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.10:7100 create_xcluster_checkpoint yugabyte_replication yugabyte

~/code/yugabyte-db/bin/ysqlsh -h 127.0.0.20 <<EOF
CREATE TABLE my_table (
    id integer DEFAULT nextval('mdl_sequence'),
    value integer
);
EOF

~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.10:7100 setup_xcluster_replication yugabyte_replication 127.0.0.20:7100
```
Which gave me the following results:
```
mdbridge@ml-win-DIB1N [target_side] [558]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.20:7100 get_replication_status
statuses {
  table_id: "0000ffff00003000800000000000ffff.sequences_data_for.000033c9000030008000000000000000"
  stream_id: "917b00fe5e14a39f3444b6207b1cfe33"
}
statuses {
  table_id: "000033c9000030008000000000004002"
  stream_id: "bf7a5e3c9bb80cb74840b249a82c9adb"
}

mdbridge@ml-win-DIB1N [target_side] [559]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.20:7100 list_universe_replications
1 Universe Replication Groups found:
[yugabyte_replication]

mdbridge@ml-win-DIB1N [target_side] [560]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.10:7100 list_xcluster_outbound_replication_groups
1 Outbound Replication Groups found:
[yugabyte_replication]

mdbridge@ml-win-DIB1N [target_side] [562]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.20:7100 get_xcluster_info|jq .
{
  "version": 0,
  "xcluster_producer_registry": {},
  "consumer_registry": {
    "producerMap": {
      "yugabyte_replication": {
        "streamMap": {
          "917b00fe5e14a39f3444b6207b1cfe33": {
            "consumerProducerTabletMap": {
              "67cc19bfc7d145b09a52c7116b79c757": {
                "tablets": [
                  "666a84c16e4d44d4b3e53889c8b32989"
                ],
                "startKey": [
                  ""
                ],
                "endKey": [
                  "gAA="
                ]
              },
              "034668733e524119abdeded62cf40cee": {
                "tablets": [
                  "7ec84a49dd0a48edb573755447e28259"
                ],
                "startKey": [
                  "gAA="
                ],
                "endKey": [
                  ""
                ]
              }
            },
            "consumerTableId": "0000ffff00003000800000000000ffff.sequences_data_for.000033c9000030008000000000000000",
            "producerTableId": "0000ffff00003000800000000000ffff.sequences_data_for.000033c9000030008000000000000000",
            "localTserverOptimized": true,
            "producerSchema": {},
            "schemaVersions": {}
          },
          "bf7a5e3c9bb80cb74840b249a82c9adb": {
            "consumerProducerTabletMap": {
              "ee1ca64e02a04a7cbf01f7ecb4f3428f": {
                "tablets": [
                  "7b424567322d4887b83ee053269a7017"
                ],
                "startKey": [
                  ""
                ],
                "endKey": [
                  "gAA="
                ]
              },
              "8f929ce0cd8e46cebf0fc628a98a5bf1": {
                "tablets": [
                  "af76272a3ac4430a94490a50428aaa97"
                ],
                "startKey": [
                  "gAA="
                ],
                "endKey": [
                  ""
                ]
              }
            },
            "consumerTableId": "000033c9000030008000000000004002",
            "producerTableId": "000033c9000030008000000000004002",
            "localTserverOptimized": true,
            "producerSchema": {},
            "schemaVersions": {}
          }
        },
        "masterAddrs": [
          {
            "host": "127.0.0.10",
            "port": 7100
          }
        ],
        "compatibleAutoFlagConfigVersion": 1,
        "validatedAutoFlagsConfigVersion": 1
      }
    }
  }
}

mdbridge@ml-win-DIB1N [target_side] [563]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.10:7100 list_cdc_streams
CDC Streams:
streams {
  stream_id: "917b00fe5e14a39f3444b6207b1cfe33"
  table_id: "0000ffff00003000800000000000ffff.sequences_data_for.000033c9000030008000000000000000"
  options {
    key: "source_type"
    value: "XCLUSTER"
  }
  options {
    key: "record_format"
    value: "WAL"
  }
  options {
    key: "record_type"
    value: "CHANGE"
  }
  options {
    key: "checkpoint_type"
    value: "IMPLICIT"
  }
  options {
    key: "state"
    value: "ACTIVE"
  }
}
streams {
  stream_id: "bf7a5e3c9bb80cb74840b249a82c9adb"
  table_id: "000033c9000030008000000000004002"
  options {
    key: "source_type"
    value: "XCLUSTER"
  }
  options {
    key: "record_format"
    value: "WAL"
  }
  options {
    key: "record_type"
    value: "CHANGE"
  }
  options {
    key: "checkpoint_type"
    value: "IMPLICIT"
  }
  options {
    key: "state"
    value: "ACTIVE"
  }
}

mdbridge@ml-win-DIB1N [target_side] [564]$ ~/code/yugabyte-db/build/latest/bin/yb-admin -master_addresses 127.0.0.20:7100 get_xcluster_safe_time include_lag_and_skew
[
    {
        "namespace_id": "000033c9000030008000000000000000",
        "namespace_name": "yugabyte",
        "safe_time": "2024-09-18 14:24:15.303713",
        "safe_time_epoch": "1726694655303713",
        "safe_time_lag_sec": "0.61",
        "safe_time_skew_sec": "0.00"
    },
    {
        "namespace_id": "0000ffff000030008000000000000000",
        "namespace_name": "system_postgres",
        "safe_time": "2024-09-18 14:24:15.321099",
        "safe_time_epoch": "1726694655321099",
        "safe_time_lag_sec": "0.59",
        "safe_time_skew_sec": "0.00"
    }
]
```
All of that looks good except the safe time part, which I will fix in a later diff.

Reviewers: jhe, xCluster, hsunder

Reviewed By: jhe

Subscribers: ybase, ycdcxcluster

Differential Revision: https://phorge.dev.yugabyte.com/D38229
  • Loading branch information
@mdbridge
mdbridge committed Sep 23, 2024
1 parent 975462e commit 8af7f97
Show file tree
Hide file tree
Showing 31 changed files with 461 additions and 184 deletions.
9 changes: 6 additions & 3 deletions src/yb/cdc/cdc_service.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,10 +43,11 @@
#include "yb/client/table_handle.h"
#include "yb/client/yb_table_name.h"

#include "yb/common/colocated_util.h"
#include "yb/common/pg_system_attr.h"
#include "yb/common/schema.h"
#include "yb/common/wire_protocol.h"
#include "yb/common/colocated_util.h"
#include "yb/common/xcluster_util.h"

#include "yb/consensus/log.h"
#include "yb/consensus/log_reader.h"
Expand Down Expand Up @@ -1301,7 +1302,8 @@ void CDCServiceImpl::GetTabletListToPollForCDC(

client::YBTableName table_name;
google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
table_name.set_table_id(table_id);
auto stripped_table_id = xcluster::StripSequencesDataAliasIfPresent(table_id);
table_name.set_table_id(stripped_table_id);
RPC_STATUS_RETURN_ERROR(
client()->GetTablets(
table_name, 0, &tablets, /* partition_list_version =*/nullptr,
Expand Down Expand Up @@ -1483,7 +1485,8 @@ Result<google::protobuf::RepeatedPtrField<master::TabletLocationsPB>> CDCService

for (const auto& table_id : table_ids) {
google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
table_name.set_table_id(table_id);
auto stripped_table_id = xcluster::StripSequencesDataAliasIfPresent(table_id);
table_name.set_table_id(stripped_table_id);
Status s = client()->GetTablets(
table_name, 0, &tablets, /* partition_list_version =*/nullptr,
RequireTabletsRunning::kFalse, master::IncludeInactive::kTrue);
Expand Down
1 change: 1 addition & 0 deletions src/yb/common/entity_ids_types.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ using NamespaceName = std::string;
using TableName = std::string;
using UDTypeName = std::string;
using RoleName = std::string;
using PgSchemaName = std::string;

using NamespaceId = std::string;
using ObjectId = std::string;
Expand Down
2 changes: 0 additions & 2 deletions src/yb/common/schema.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -516,8 +516,6 @@ class TableProperties {
std::optional<PgReplicaIdentity> ysql_replica_identity_;
};

using PgSchemaName = std::string;

// Provides missing, i.e. default, value for specified column if present.
class MissingValueProvider {
public:
Expand Down
31 changes: 31 additions & 0 deletions src/yb/common/xcluster_util.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,20 @@

#include "yb/common/xcluster_util.h"

#include "yb/common/entity_ids.h"
#include "yb/common/ysql_utils.h"
#include "yb/gutil/strings/util.h"
#include "yb/util/result.h"

namespace yb::xcluster {

namespace {
constexpr char kAlterReplicationGroupSuffix[] = ".ALTER";

constexpr char kSequencesDataAliasTableIdMid[] = ".sequences_data_for.";

// How many characters a normal TableId (e.g., no suffixes) takes up.
constexpr int kTableIdSize = 32;
} // namespace

ReplicationGroupId GetAlterReplicationGroupId(const ReplicationGroupId& replication_group_id) {
Expand All @@ -44,4 +52,27 @@ std::string ShortReplicationType(XClusterReplicationType type) {
/*replace_all=*/false);
}

TableId GetSequencesDataAliasForNamespace(const NamespaceId& namespace_id) {
DCHECK(kPgSequencesDataTableId.size() == kTableIdSize);
return kPgSequencesDataTableId + kSequencesDataAliasTableIdMid + namespace_id;
}

bool IsSequencesDataAlias(const TableId& table_id) {
return table_id.find(kSequencesDataAliasTableIdMid) == kTableIdSize;
}

TableId StripSequencesDataAliasIfPresent(const TableId& table_id) {
if (!IsSequencesDataAlias(table_id)) {
return table_id;
}
return kPgSequencesDataTableId;
}

Result<NamespaceId> GetReplicationNamespaceBelongsTo(const TableId& table_id) {
if (!IsSequencesDataAlias(table_id)) {
return GetNamespaceIdFromYsqlTableId(table_id);
}
return table_id.substr(kTableIdSize + strlen(kSequencesDataAliasTableIdMid));
}

} // namespace yb::xcluster
11 changes: 11 additions & 0 deletions src/yb/common/xcluster_util.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,4 +25,15 @@ ReplicationGroupId GetOriginalReplicationGroupId(const ReplicationGroupId& repli

std::string ShortReplicationType(XClusterReplicationType type);


TableId GetSequencesDataAliasForNamespace(const NamespaceId& namespace_id);

bool IsSequencesDataAlias(const TableId& table_id);

TableId StripSequencesDataAliasIfPresent(const TableId& table_id);

// Returns the namespace a sequences_data alias is for; if table_id is not a sequences_data alias,
// instead returns the ID's namespace as usual.
Result<NamespaceId> GetReplicationNamespaceBelongsTo(const TableId& table_id);

} // namespace yb::xcluster
Loading

0 comments on commit 8af7f97

Please sign in to comment.