tool: check for existence of keyspace before getting it

in general, user should save output of `DESC foo.bar` to a file,
and pass the path to the file as the argument of `--schema-file`
option of `scylla sstable` commands. the CQL statement generated
from `DESC` command always include the keyspace name of the table.
but in case user create the CQL statement manually and misses
the keyspace name. he/she would have following assertion failure
```
scylla: cql3/statements/cf_statement.cc:49: virtual const sstring &cql3::statements::raw::cf_statement::keyspace() const: Assertion `_cf_name->has_keyspace()' failed.
```
this is not a great user experience.

so, in this change, we check for the existence of keyspace before
looking it up. and throw a runtime error with a better error mesage.
so when the CQL statement does not have the keyspace name, the new
error message would look like:
```
error processing arguments: could not load schema via schema-file: std::runtime_error (tools::do_load_schemas(): CQL statement does not have keyspace specified)
```

since this check is only performed by `do_load_schemas()` which
care about the existence of keyspace, and it only expects the
CQL statement to create table/keyspace/type, we just override the
new `has_keyspace()` method of the corresponding types derived
from `cf_statement`.

Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>

Closes scylladb/scylladb#16981

cql3/statements/alter_keyspace_statement.hh

@@ -29,6 +29,9 @@ class alter_keyspace_statement : public schema_altering_statement {
 public:
     alter_keyspace_statement(sstring name, ::shared_ptr<ks_prop_defs> attrs);
 
+    bool has_keyspace() const override {
+        return true;
+    }
     const sstring& keyspace() const override;
 
     future<> check_access(query_processor& qp, const service::client_state& state) const override;

cql3/statements/cf_statement.cc

@@ -39,6 +39,11 @@ void cf_statement::prepare_keyspace(std::string_view keyspace)
     }
 }
 
+bool cf_statement::has_keyspace() const {
+    assert(_cf_name.has_value());
+    return _cf_name->has_keyspace();
+}
+
 const sstring& cf_statement::keyspace() const
 {
     assert(_cf_name->has_keyspace()); // "The statement hasn't be prepared correctly";

cql3/statements/create_keyspace_statement.hh

@@ -51,6 +51,9 @@ public:
      */
     create_keyspace_statement(const sstring& name, shared_ptr<ks_prop_defs> attrs, bool if_not_exists);
 
+    virtual bool has_keyspace() const override {
+        return true;
+    }
     virtual const sstring& keyspace() const override;
 
     virtual future<> check_access(query_processor& qp, const service::client_state& state) const override;

cql3/statements/create_type_statement.hh

@@ -35,6 +35,10 @@ public:
 
     virtual void validate(query_processor&, const service::client_state& state) const override;
 
+    virtual bool has_keyspace() const override {
+        return true;
+    }
+
     virtual const sstring& keyspace() const override;
 
     future<std::tuple<::shared_ptr<cql_transport::event::schema_change>, std::vector<mutation>, cql3::cql_warnings_vec>> prepare_schema_mutations(query_processor& qp, api::timestamp_type) const override;

cql3/statements/raw/cf_statement.hh

@@ -38,6 +38,8 @@ public:
     // Only for internal calls, use the version with ClientState for user queries
     void prepare_keyspace(std::string_view keyspace);
 
+    virtual bool has_keyspace() const;
+
     virtual const sstring& keyspace() const;
 
     virtual const sstring& column_family() const;

tools/schema_loader.cc

@@ -281,6 +281,9 @@ std::vector<schema_ptr> do_load_schemas(const db::config& cfg, std::string_view
         if (!cf_statement) {
             continue; // we don't support any non-cf statements here
         }
+        if (!cf_statement->has_keyspace()) {
+            throw std::runtime_error("tools::do_load_schemas(): CQL statement does not have keyspace specified");
+        }
         auto ks = find_or_create_keyspace(cf_statement->keyspace());
         auto prepared_statement = cf_statement->prepare(db, cql_stats);
         auto* statement = prepared_statement->statement.get();