Merge 'Ignore no such column family in repair' from Aleksandra Martyniuk

While repair requested by user is performed, some tables
may be dropped. When the repair proceeds to these tables,
it should skip them and continue with others.

When no_such_column_family is thrown during user requested
repair, it is logged and swallowed. Then the repair continues with
the remaining tables.

Fixes: #13045

Closes #13068

* github.com:scylladb/scylladb:
  repair: fix indentation
  repair: continue user requested repair if no_such_column_family is thrown
  repair: add find_column_family_if_exists function

repair/repair.cc

@@ -144,6 +144,29 @@ static std::vector<sstring> list_column_families(const replica::database& db, co
     return ret;
 }
 
+static const replica::column_family* find_column_family_if_exists(const replica::database& db, std::string_view ks_name, std::string_view cf_name, bool warn = true) {
+    try {
+        auto uuid = db.find_uuid(std::move(ks_name), std::move(cf_name));
+        return &db.find_column_family(uuid);
+    } catch (replica::no_such_column_family&) {
+        if (warn) {
+            rlogger.warn("{}", std::current_exception());
+        }
+        return nullptr;
+    }
+}
+
+static const replica::column_family* find_column_family_if_exists(const replica::database& db, const table_id& uuid, bool warn = true) {
+    try {
+        return &db.find_column_family(uuid);
+    } catch (...) {
+        if (warn) {
+            rlogger.warn("{}", std::current_exception());
+        }
+        return nullptr;
+    }
+}
+
 std::ostream& operator<<(std::ostream& os, const repair_uniq_id& x) {
     return os << format("[id={}, uuid={}]", x.id, x.uuid());
 }
@@ -154,7 +177,11 @@ static std::vector<table_id> get_table_ids(const replica::database& db, const ss
     table_ids.reserve(tables.size());
     for (auto& table : tables) {
         thread::maybe_yield();
-        table_ids.push_back(db.find_uuid(keyspace, table));
+        try {
+            table_ids.push_back(db.find_uuid(keyspace, table));
+        } catch (replica::no_such_column_family&) {
+            rlogger.warn("Column family {} does not exist in keyspace {}", table, keyspace);
+        }
     }
     return table_ids;
 }
@@ -163,7 +190,8 @@ static std::vector<sstring> get_table_names(const replica::database& db, const s
     std::vector<sstring> table_names;
     table_names.reserve(table_ids.size());
     for (auto& table_id : table_ids) {
-        table_names.push_back(db.find_column_family(table_id).schema()->cf_name());
+        auto* cf = find_column_family_if_exists(db, table_id);
+        table_names.push_back(cf ? cf->schema()->cf_name() : "");
     }
     return table_names;
 }
@@ -503,10 +531,10 @@ get_sharder_for_tables(seastar::sharded<replica::database>& db, const sstring& k
     schema_ptr last_s;
     for (size_t idx = 0 ; idx < table_ids.size(); idx++) {
         schema_ptr s;
-        try {
-            s = db.local().find_column_family(table_ids[idx]).schema();
-        } catch(...) {
-            throw replica::no_such_column_family(keyspace, table_ids[idx]);
+        if (const auto* cf = find_column_family_if_exists(db.local(), table_ids[idx])) {
+            s = cf->schema();
+        } else {
+            continue;
         }
         if (last_s && last_s->get_sharder() != s->get_sharder()) {
             throw std::runtime_error(
@@ -1094,10 +1122,12 @@ future<> user_requested_repair_task_impl::run() {
         bool needs_flush_before_repair = false;
         if (db.features().tombstone_gc_options) {
             for (auto& table: cfs) {
-                auto s = db.find_column_family(keyspace, table).schema();
-                const auto& options = s->tombstone_gc_options();
-                if (options.mode() == tombstone_gc_mode::repair) {
-                    needs_flush_before_repair = true;
+                if (const auto* cf = find_column_family_if_exists(db, keyspace, table)) {
+                    auto s = cf->schema();
+                    const auto& options = s->tombstone_gc_options();
+                    if (options.mode() == tombstone_gc_mode::repair) {
+                        needs_flush_before_repair = true;
+                    }
                 }
             }
         }