From ff26b2ba3f0c4dfa27dddd595bb4e988e483c86d Mon Sep 17 00:00:00 2001
From: Aleksandra Martyniuk <aleksandra.martyniuk@scylladb.com>
Date: Wed, 19 Jul 2023 11:50:16 +0200
Subject: [PATCH] replica: add methods to check if given table exists

---
 replica/database.cc | 20 ++++++++++++++------
 replica/database.hh |  2 ++
 2 files changed, 16 insertions(+), 6 deletions(-)

diff --git a/replica/database.cc b/replica/database.cc
index b5b5989948..28cb49a62d 100644
--- a/replica/database.cc
+++ b/replica/database.cc
@@ -863,7 +863,7 @@ database::init_commitlog() {
     return db::commitlog::create_commitlog(db::commitlog::config::from_db_config(_cfg, _dbcfg.commitlog_scheduling_group, _dbcfg.available_memory)).then([this](db::commitlog&& log) {
         _commitlog = std::make_unique<db::commitlog>(std::move(log));
         _commitlog->add_flush_handler([this](db::cf_id_type id, db::replay_position pos) {
-            if (!_tables_metadata._column_families.contains(id)) {
+            if (!_tables_metadata.contains(id)) {
                 // the CF has been removed.
                 _commitlog->discard_completed_segments(id);
                 return;
@@ -959,7 +959,7 @@ void database::maybe_init_schema_commitlog() {
 
     _schema_commitlog = std::make_unique<db::commitlog>(db::commitlog::create_commitlog(std::move(c)).get0());
     _schema_commitlog->add_flush_handler([this] (db::cf_id_type id, db::replay_position pos) {
-        if (!_tables_metadata._column_families.contains(id)) {
+        if (!_tables_metadata.contains(id)) {
             // the CF has been removed.
             _schema_commitlog->discard_completed_segments(id);
             return;
@@ -1017,11 +1017,11 @@ future<> database::add_column_family(keyspace& ks, schema_ptr schema, column_fam
     cf->set_durable_writes(ks.metadata()->durable_writes());
 
     auto uuid = schema->id();
-    if (_tables_metadata._column_families.contains(uuid)) {
+    if (_tables_metadata.contains(uuid)) {
         throw std::invalid_argument("UUID " + uuid.to_sstring() + " already mapped");
     }
     auto kscf = std::make_pair(schema->ks_name(), schema->cf_name());
-    if (_tables_metadata._ks_cf_to_uuid.contains(kscf)) {
+    if (_tables_metadata.contains(kscf)) {
         throw std::invalid_argument("Column family " + schema->cf_name() + " exists");
     }
     ks.add_or_update_column_family(schema);
@@ -1283,7 +1283,7 @@ const column_family& database::find_column_family(const table_id& uuid) const {
 }
 
 bool database::column_family_exists(const table_id& uuid) const {
-    return _tables_metadata._column_families.contains(uuid);
+    return _tables_metadata.contains(uuid);
 }
 
 future<>
@@ -1407,7 +1407,7 @@ schema_ptr database::find_schema(const table_id& uuid) const {
 }
 
 bool database::has_schema(std::string_view ks_name, std::string_view cf_name) const {
-    return _tables_metadata._ks_cf_to_uuid.contains(std::make_pair(ks_name, cf_name));
+    return _tables_metadata.contains(std::make_pair(ks_name, cf_name));
 }
 
 std::vector<view_ptr> database::get_views() const {
@@ -2905,6 +2905,14 @@ table_id database::tables_metadata::get_table_id_if_exists(const std::pair<std::
     return table_id::create_null_id();
 }
 
+bool database::tables_metadata::contains(table_id id) const {
+    return _column_families.contains(id);
+}
+
+bool database::tables_metadata::contains(std::pair<std::string_view, std::string_view> kscf) const {
+    return _ks_cf_to_uuid.contains(kscf);
+}
+
 void database::tables_metadata::for_each_table(std::function<void(table_id, lw_shared_ptr<table>)> f) const {
     for (auto& [id, table]: _column_families) {
         f(id, table);
diff --git a/replica/database.hh b/replica/database.hh
index b66945c5ad..929d2fd4c8 100644
--- a/replica/database.hh
+++ b/replica/database.hh
@@ -1317,6 +1317,8 @@ public:
         table_id get_table_id(const std::pair<std::string_view, std::string_view>& kscf) const;
         lw_shared_ptr<table> get_table_if_exists(table_id id) const;
         table_id get_table_id_if_exists(const std::pair<std::string_view, std::string_view>& kscf) const;
+        bool contains(table_id id) const;
+        bool contains(std::pair<std::string_view, std::string_view> kscf) const;
         void for_each_table(std::function<void(table_id, lw_shared_ptr<table>)> f) const;
         void for_each_table_id(std::function<void(const ks_cf_t&, table_id)> f) const;
         future<> for_each_table_gently(std::function<future<>(table_id, lw_shared_ptr<table>)> f);