Change store to unparsed_schema and remove upsert validation

Author: IoannisRP

src/v/pandaproxy/schema_registry/avro.cc

@@ -534,32 +534,32 @@ class collected_schema {
     bool contains(const ss::sstring& name) const {
         return _names.contains(name);
     }
-    bool insert(ss::sstring name, canonical_schema_definition def) {
+    bool insert(ss::sstring name, unparsed_schema_definition def) {
         bool inserted = _names.insert(std::move(name)).second;
         if (inserted) {
             _schemas.push_back(std::move(def).raw());
         }
         return inserted;
     }
-    canonical_schema_definition::raw_string flatten() && {
+    unparsed_schema_definition::raw_string flatten() && {
         iobuf out;
         for (auto& s : _schemas) {
             out.append(std::move(s));
             out.append("\n", 1);
         }
-        return canonical_schema_definition::raw_string{std::move(out)};
+        return unparsed_schema_definition::raw_string{std::move(out)};
     }
 
 private:
     absl::flat_hash_set<ss::sstring> _names;
-    std::vector<canonical_schema_definition::raw_string> _schemas;
+    std::vector<unparsed_schema_definition::raw_string> _schemas;
 };
 
 ss::future<collected_schema> collect_schema(
   schema_getter& store,
   collected_schema collected,
   ss::sstring name,
-  canonical_schema schema) {
+  unparsed_schema schema) {
     for (const auto& ref : schema.def().refs()) {
         if (!collected.contains(ref.name)) {
             auto ss = co_await store.get_subject_schema(
@@ -573,7 +573,7 @@ ss::future<collected_schema> collect_schema(
 }
 
 ss::future<avro_schema_definition>
-make_avro_schema_definition(schema_getter& store, canonical_schema schema) {
+make_avro_schema_definition(schema_getter& store, unparsed_schema schema) {
     std::optional<avro::Exception> ex;
     try {
         auto name = schema.sub()();

src/v/pandaproxy/schema_registry/avro.h

@@ -18,7 +18,7 @@
 namespace pandaproxy::schema_registry {
 
 ss::future<avro_schema_definition>
-make_avro_schema_definition(schema_getter& store, canonical_schema schema);
+make_avro_schema_definition(schema_getter& store, unparsed_schema schema);
 
 result<canonical_schema_definition>
 sanitize_avro_schema_definition(unparsed_schema_definition def);

src/v/pandaproxy/schema_registry/errors.h

@@ -145,7 +145,7 @@ inline error_info invalid_subject_schema(const subject& sub) {
       fmt::format("Error while looking up schema under subject {}", sub())};
 }
 
-inline error_info invalid_schema(const canonical_schema& schema) {
+inline error_info invalid_schema(const unparsed_schema& schema) {
     return {
       error_code::schema_invalid, fmt::format("Invalid schema {}", schema)};
 }

src/v/pandaproxy/schema_registry/handlers.cc

@@ -421,12 +421,14 @@ post_subject(server::request_t rq, server::reply_t rp) {
     // Force 40401 if no subject
     co_await rq.service().schema_store().get_versions(sub, inc_del);
 
-    canonical_schema schema;
+    unparsed_schema validated;
     try {
         auto unparsed = co_await ppj::rjson_parse(
           std::move(rq.req), post_subject_versions_request_handler<>{sub});
-        schema = co_await rq.service().schema_store().make_canonical_schema(
-          std::move(unparsed.def), norm);
+        canonical_schema canonical
+          = co_await rq.service().schema_store().make_canonical_schema(
+            std::move(unparsed.def), norm);
+        validated = to_unparsed_schema(std::move(canonical));
     } catch (const exception& e) {
         if (e.code() == error_code::schema_empty) {
             throw as_exception(invalid_subject_schema(sub));
@@ -437,7 +439,7 @@ post_subject(server::request_t rq, server::reply_t rp) {
     }
 
     auto sub_schema = co_await rq.service().schema_store().has_schema(
-      std::move(schema), inc_del, norm);
+      std::move(validated), inc_del, norm);
 
     rp.rep->write_body(
       "json",
@@ -477,9 +479,10 @@ post_subject_versions(server::request_t rq, server::reply_t rp) {
         unparsed.id = invalid_schema_id;
     }
 
+    auto canonical = co_await rq.service().schema_store().make_canonical_schema(
+      std::move(unparsed.def), norm);
     subject_schema schema{
-      co_await rq.service().schema_store().make_canonical_schema(
-        std::move(unparsed.def), norm),
+      to_unparsed_schema(std::move(canonical)),
       unparsed.version.value_or(invalid_schema_version),
       unparsed.id.value_or(invalid_schema_id),
       is_deleted::no};
@@ -674,10 +677,12 @@ compatibility_subject_version(server::request_t rq, server::reply_t rp) {
         version = parse_numerical_schema_version(ver).value();
     }
 
-    canonical_schema schema;
+    unparsed_schema validated;
     try {
-        schema = co_await rq.service().schema_store().make_canonical_schema(
-          std::move(unparsed.def));
+        auto canonical
+          = co_await rq.service().schema_store().make_canonical_schema(
+            std::move(unparsed.def));
+        validated = to_unparsed_schema(std::move(canonical));
     } catch (exception& e) {
         constexpr auto reportable = [](std::error_code ec) {
             constexpr std::array errors{
@@ -699,7 +704,7 @@ compatibility_subject_version(server::request_t rq, server::reply_t rp) {
     }
 
     auto get_res = co_await get_or_load(
-      rq, [&rq, schema{std::move(schema)}, version, is_verbose]() {
+      rq, [&rq, schema{std::move(validated)}, version, is_verbose]() {
           return rq.service().schema_store().is_compatible(
             version, schema.share(), is_verbose);
       });

src/v/pandaproxy/schema_registry/json.cc

@@ -2357,7 +2357,7 @@ result<id_to_schema_pointer> collect_bundled_schema_and_fix_refs(
 } // namespace
 
 ss::future<json_schema_definition>
-make_json_schema_definition(schema_getter&, canonical_schema schema) {
+make_json_schema_definition(schema_getter&, unparsed_schema schema) {
     auto doc
       = parse_json(schema.def().shared_raw()()).value(); // throws on error
     std::string_view name = schema.sub()();

src/v/pandaproxy/schema_registry/json.h

@@ -17,7 +17,7 @@
 namespace pandaproxy::schema_registry {
 
 ss::future<json_schema_definition>
-make_json_schema_definition(schema_getter& store, canonical_schema schema);
+make_json_schema_definition(schema_getter& store, unparsed_schema schema);
 
 ss::future<canonical_schema> make_canonical_json_schema(
   sharded_store& store, unparsed_schema def, normalize norm = normalize::no);

src/v/pandaproxy/schema_registry/protobuf.cc

@@ -233,7 +233,7 @@ class dp_error_collector final : public pb::DescriptorPool::ErrorCollector {
 ///\brief Implements ZeroCopyInputStream with a copy of the definition
 class schema_def_input_stream : public pb::io::ZeroCopyInputStream {
 public:
-    explicit schema_def_input_stream(const canonical_schema_definition& def)
+    explicit schema_def_input_stream(const unparsed_schema_definition& def)
       : _is{def.shared_raw()}
       , _impl{&_is.istream()} {}
 
@@ -255,7 +255,7 @@ class parser {
       : _parser{}
       , _fdp{} {}
 
-    const pb::FileDescriptorProto& parse(const canonical_schema& schema) {
+    const pb::FileDescriptorProto& parse(const unparsed_schema& schema) {
         schema_def_input_stream is{schema.def()};
         io_error_collector error_collector;
         pb::io::Tokenizer t{&is, &error_collector};
@@ -431,7 +431,7 @@ build_file(pb::DescriptorPool& dp, const pb::FileDescriptorProto& fdp) {
 ss::future<pb::FileDescriptorProto> build_file_with_refs(
   pb::DescriptorPool& dp,
   schema_getter& store,
-  canonical_schema schema,
+  unparsed_schema schema,
   normalize norm) {
     for (const auto& ref : schema.def().refs()) {
         if (dp.FindFileByName(ref.name)) {
@@ -442,7 +442,7 @@ ss::future<pb::FileDescriptorProto> build_file_with_refs(
         co_await build_file_with_refs(
           dp,
           store,
-          canonical_schema{subject{ref.name}, std::move(dep.schema).def()},
+          unparsed_schema{subject{ref.name}, std::move(dep.schema).def()},
           normalize::no);
     }
 
@@ -461,7 +461,7 @@ ss::future<pb::FileDescriptorProto> build_file_with_refs(
 ss::future<pb::FileDescriptorProto> import_schema(
   pb::DescriptorPool& dp,
   schema_getter& store,
-  canonical_schema schema,
+  unparsed_schema schema,
   normalize norm) {
     try {
         co_return co_await build_file_with_refs(
@@ -587,7 +587,7 @@ operator<<(std::ostream& os, const protobuf_schema_definition& def) {
 }
 
 ss::future<protobuf_schema_definition> make_protobuf_schema_definition(
-  schema_getter& store, canonical_schema schema, normalize norm) {
+  schema_getter& store, unparsed_schema schema, normalize norm) {
     auto refs = schema.def().refs();
     auto impl = ss::make_shared<protobuf_schema_definition::impl>();
     auto v2_renderer = protobuf_renderer_v2::no;
@@ -608,7 +608,7 @@ ss::future<protobuf_schema_definition> make_protobuf_schema_definition(
 }
 
 ss::future<canonical_schema_definition> validate_protobuf_schema(
-  sharded_store& store, canonical_schema schema, normalize norm) {
+  sharded_store& store, unparsed_schema schema, normalize norm) {
     auto res = co_await make_protobuf_schema_definition(
       store, std::move(schema), norm);
     co_return canonical_schema_definition{std::move(res)};
@@ -618,9 +618,9 @@ ss::future<canonical_schema> make_canonical_protobuf_schema(
   sharded_store& store, unparsed_schema schema, normalize norm) {
     auto [sub, unparsed] = std::move(schema).destructure();
     auto [def, type, refs] = std::move(unparsed).destructure();
-    canonical_schema temp{
+    unparsed_schema temp{
       sub,
-      {canonical_schema_definition::raw_string{std::move(def)()},
+      {unparsed_schema_definition::raw_string{std::move(def)()},
        type,
        std::move(refs)}};
 

src/v/pandaproxy/schema_registry/protobuf.h

@@ -21,14 +21,10 @@ class Descriptor;
 namespace pandaproxy::schema_registry {
 
 ss::future<protobuf_schema_definition> make_protobuf_schema_definition(
-  schema_getter& store,
-  canonical_schema schema,
-  normalize norm = normalize::no);
+  schema_getter& store, unparsed_schema schema, normalize norm = normalize::no);
 
 ss::future<canonical_schema_definition> validate_protobuf_schema(
-  sharded_store& store,
-  canonical_schema schema,
-  normalize norm = normalize::no);
+  sharded_store& store, unparsed_schema schema, normalize norm = normalize::no);
 
 ss::future<canonical_schema> make_canonical_protobuf_schema(
   sharded_store& store, unparsed_schema schema, normalize norm = normalize::no);

src/v/pandaproxy/schema_registry/requests/get_schemas_ids_id.h

@@ -18,7 +18,7 @@
 namespace pandaproxy::schema_registry {
 
 struct get_schemas_ids_id_response {
-    canonical_schema_definition definition;
+    unparsed_schema_definition definition;
 };
 
 template<typename Buffer>

src/v/pandaproxy/schema_registry/requests/get_subject_versions_version.h

@@ -18,7 +18,7 @@
 namespace pandaproxy::schema_registry {
 
 struct post_subject_versions_version_response {
-    canonical_schema schema;
+    unparsed_schema schema;
     schema_id id;
     schema_version version;
 };

src/v/pandaproxy/schema_registry/schema_getter.h

@@ -25,9 +25,9 @@ class schema_getter {
       std::optional<schema_version> version,
       include_deleted inc_dec)
       = 0;
-    virtual ss::future<canonical_schema_definition>
+    virtual ss::future<unparsed_schema_definition>
     get_schema_definition(schema_id id) = 0;
-    virtual ss::future<std::optional<canonical_schema_definition>>
+    virtual ss::future<std::optional<unparsed_schema_definition>>
     maybe_get_schema_definition(schema_id id) = 0;
     virtual ~schema_getter() = default;
 };

src/v/pandaproxy/schema_registry/seq_writer.cc

@@ -247,7 +247,7 @@ ss::future<std::optional<schema_id>> seq_writer::do_write_subject_version(
           .node{_node_id},
           .sub{sub},
           .version{projected.version}};
-        auto value = canonical_schema_value{
+        auto value = unparsed_schema_value{
           .schema{std::move(canonical)},
           .version{projected.version},
           .id{projected.id},
@@ -440,7 +440,7 @@ ss::future<std::optional<bool>> seq_writer::do_delete_subject_version(
     auto key = schema_key{
       .seq{write_at}, .node{_node_id}, .sub{sub}, .version{version}};
     vlog(plog.debug, "seq_writer::delete_subject_version {}", key);
-    auto value = canonical_schema_value{
+    auto value = unparsed_schema_value{
       .schema{std::move(ss.schema)},
       .version{version},
       .id{ss.id},