From 8517eecc2883945dc56d2fbb49d9f8244ca875bc Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@scylladb.com>
Date: Thu, 12 Sep 2019 11:57:19 +0200
Subject: [PATCH] Revert "Simplify db::cql_type_parser::parse"

This reverts commit 7f64a6ec4bda96b53dc2f7fbc60b7ad5fd900c60.

Fixes #5011

The reverted commit exposes #3760 for all schemas, not only those
which have UDTs.

The problem is that table schema deserialization now requires keyspace
to be present. If the replica hasn't received schema changes which
introduce the keyspace yet, the write will fail.
---
 db/cql_type_parser.cc | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/db/cql_type_parser.cc b/db/cql_type_parser.cc
index 685c5cf658..d665fb4a87 100644
--- a/db/cql_type_parser.cc
+++ b/db/cql_type_parser.cc
@@ -57,9 +57,30 @@ static ::shared_ptr<cql3::cql3_type::raw> parse_raw(const sstring& str) {
 }
 
 data_type db::cql_type_parser::parse(const sstring& keyspace, const sstring& str, lw_shared_ptr<user_types_metadata> user_types) {
+    static const thread_local std::unordered_map<sstring, cql3::cql3_type> native_types = []{
+        std::unordered_map<sstring, cql3::cql3_type> res;
+        for (auto& nt : cql3::cql3_type::values()) {
+            res.emplace(nt.to_string(), nt);
+        }
+        return res;
+    }();
+
+    auto i = native_types.find(str);
+    if (i != native_types.end()) {
+        return i->second.get_type();
+    }
+
     if (!user_types && service::get_storage_proxy().local_is_initialized()) {
         user_types = service::get_storage_proxy().local().get_db().local().find_keyspace(keyspace).metadata()->user_types();
     }
+    // special-case top-level UDTs
+    if (user_types) {
+        auto& map = user_types->get_all_types();
+        auto i = map.find(utf8_type->decompose(str));
+        if (i != map.end()) {
+            return i->second;
+        }
+    }
 
     auto raw = parse_raw(str);
     auto cql = raw->prepare_internal(keyspace, user_types);