Replace canonical schemas for unparsed schemas stored in store

Author: IoannisRP

src/v/pandaproxy/schema_registry/sharded_store.cc

@@ -137,9 +137,10 @@ sharded_store::get_schema_version(subject_schema schema) {
     // Validate the schema (may throw)
     co_await validate_schema(schema.schema.share());
 
+    auto unparsed_def = to_unparsed(schema.schema.def().share());
     // Determine if the definition already exists
-    auto map = [&schema](store& s) {
-        return s.get_schema_id(schema.schema.def());
+    auto map = [&unparsed_def](store& s) {
+        return s.get_schema_id(unparsed_def);
     };
     auto reduce = [](
                     std::optional<schema_id> acc,
@@ -246,22 +247,7 @@ ss::future<bool> sharded_store::upsert(
   schema_id id,
   schema_version version,
   is_deleted deleted) {
-    auto norm = normalize{
-      config::shard_local_cfg().schema_registry_normalize_on_startup()};
-    co_return co_await upsert(
-      marker,
-      co_await make_canonical_schema(std::move(schema), norm),
-      id,
-      version,
-      deleted);
-}
-
-ss::future<bool> sharded_store::upsert(
-  seq_marker marker,
-  canonical_schema schema,
-  schema_id id,
-  schema_version version,
-  is_deleted deleted) {
+    // TODO(ik):Normalization on startup??
     auto [sub, def] = std::move(schema).destructure();
     co_await upsert_schema(id, std::move(def));
     co_return co_await upsert_subject(
@@ -324,10 +310,10 @@ ss::future<subject_schema> sharded_store::has_schema(
 
 ss::future<std::optional<canonical_schema_definition>>
 sharded_store::maybe_get_schema_definition(schema_id id) {
-    co_return co_await _store.invoke_on(
+    auto unparsed = co_await _store.invoke_on(
       shard_for(id),
       _smp_opts,
-      [id](store& s) -> std::optional<canonical_schema_definition> {
+      [id](store& s) -> std::optional<unparsed_schema_definition> {
           auto s_res = s.get_schema_definition(id);
           if (
             s_res.has_error()
@@ -336,14 +322,28 @@ sharded_store::maybe_get_schema_definition(schema_id id) {
           }
           return std::move(s_res.value());
       });
+    if (!unparsed.has_value()) {
+        co_return std::nullopt;
+    }
+
+    // TODO(ik): This may throw. Intercept exception
+    auto canonical = co_await make_canonical_schema(
+      {{}, std::move(unparsed.value())});
+
+    co_return std::move(canonical).def();
 }
 
 ss::future<canonical_schema_definition>
 sharded_store::get_schema_definition(schema_id id) {
-    co_return co_await _store.invoke_on(
+    auto unparsed = co_await _store.invoke_on(
       shard_for(id), _smp_opts, [id](store& s) {
           return s.get_schema_definition(id).value();
       });
+
+    // TODO(ik): This may throw. Intercept exception
+    auto canonical = co_await make_canonical_schema({{}, std::move(unparsed)});
+
+    co_return std::move(canonical).def();
 }
 
 ss::future<chunked_vector<subject_version>>
@@ -387,8 +387,11 @@ ss::future<subject_schema> sharded_store::get_subject_schema(
           return s.get_schema_definition(id).value();
       });
 
+    // TODO(ik): this may throw. Intercept exception
+    auto canonical = co_await make_canonical_schema({sub, std::move(def)});
+
     co_return subject_schema{
-      .schema = {sub, std::move(def)},
+      .schema = std::move(canonical),
       .version = v_id.version,
       .id = v_id.id,
       .deleted = v_id.deleted};
@@ -649,7 +652,7 @@ sharded_store::clear_compatibility(seq_marker marker, subject sub) {
 }
 
 ss::future<bool>
-sharded_store::upsert_schema(schema_id id, canonical_schema_definition def) {
+sharded_store::upsert_schema(schema_id id, unparsed_schema_definition def) {
     co_await maybe_update_max_schema_id(id);
     co_return co_await _store.invoke_on(
       shard_for(id), _smp_opts, [id, def{std::move(def)}](store& s) mutable {

src/v/pandaproxy/schema_registry/sharded_store.h

@@ -55,13 +55,6 @@ class sharded_store final : public schema_getter {
     };
     ss::future<insert_result> project_ids(subject_schema schema);
 
-    ss::future<bool> upsert(
-      seq_marker marker,
-      canonical_schema schema,
-      schema_id id,
-      schema_version version,
-      is_deleted deleted);
-
     ss::future<bool> upsert(
       seq_marker marker,
       unparsed_schema schema,
@@ -212,7 +205,7 @@ class sharded_store final : public schema_getter {
       schema_version version, canonical_schema new_schema, verbose is_verbose);
 
     ss::future<bool>
-    upsert_schema(schema_id id, canonical_schema_definition def);
+    upsert_schema(schema_id id, unparsed_schema_definition def);
     ss::future<> delete_schema(schema_id id);
 
     struct insert_subject_result {

src/v/pandaproxy/schema_registry/store.h

@@ -87,15 +87,15 @@ class store {
     /// version.
     ///
     /// return the schema_version and schema_id, and whether it's new.
-    insert_result insert(canonical_schema schema) {
+    insert_result insert(unparsed_schema schema) {
         auto [sub, def] = std::move(schema).destructure();
         auto id = insert_schema(std::move(def)).id;
         auto [version, inserted] = insert_subject(std::move(sub), id);
         return {version, id, inserted};
     }
 
     ///\brief Return a schema definition by id.
-    result<canonical_schema_definition>
+    result<unparsed_schema_definition>
     get_schema_definition(const schema_id& id) const {
         auto it = _schemas.find(id);
         if (it == _schemas.end()) {
@@ -106,7 +106,7 @@ class store {
 
     ///\brief Return the id of the schema, if it already exists.
     std::optional<schema_id>
-    get_schema_id(const canonical_schema_definition& def) const {
+    get_schema_id(const unparsed_schema_definition& def) const {
         const auto s_it = std::find_if(
           _schemas.begin(), _schemas.end(), [&](const auto& s) {
               const auto& entry = s.second;
@@ -169,7 +169,7 @@ class store {
     }
 
     ///\brief Return a schema by subject and version.
-    result<subject_schema> get_subject_schema(
+    result<unparsed_subject_schema> get_subject_schema(
       const subject& sub,
       std::optional<schema_version> version,
       include_deleted inc_del) const {
@@ -178,7 +178,7 @@ class store {
 
         auto def = BOOST_OUTCOME_TRYX(get_schema_definition(v_id.id));
 
-        return subject_schema{
+        return unparsed_subject_schema{
           .schema = {sub, std::move(def)},
           .version = v_id.version,
           .id = v_id.id,
@@ -626,7 +626,7 @@ class store {
         schema_id id;
         bool inserted;
     };
-    insert_schema_result insert_schema(canonical_schema_definition def) {
+    insert_schema_result insert_schema(unparsed_schema_definition def) {
         const auto s_it = std::find_if(
           _schemas.begin(), _schemas.end(), [&](const auto& s) {
               const auto& entry = s.second;
@@ -642,7 +642,7 @@ class store {
         return {id, inserted};
     }
 
-    bool upsert_schema(schema_id id, canonical_schema_definition def) {
+    bool upsert_schema(schema_id id, unparsed_schema_definition def) {
         return _schemas.insert_or_assign(id, schema_entry(std::move(def)))
           .second;
     }
@@ -785,10 +785,10 @@ class store {
 
 private:
     struct schema_entry {
-        explicit schema_entry(canonical_schema_definition definition)
+        explicit schema_entry(unparsed_schema_definition definition)
           : definition{std::move(definition)} {}
 
-        canonical_schema_definition definition;
+        unparsed_schema_definition definition;
     };
 
     class subject_entry {

src/v/pandaproxy/schema_registry/test/store.cc

@@ -11,6 +11,7 @@
 
 #include "pandaproxy/schema_registry/error.h"
 #include "pandaproxy/schema_registry/test/compatibility_avro.h"
+#include "pandaproxy/schema_registry/types.h"
 #include "pandaproxy/schema_registry/util.h"
 
 #include <absl/algorithm/container.h>
@@ -21,13 +22,13 @@ namespace pps = pandaproxy::schema_registry;
 constexpr std::string_view sv_string_def0{R"({"type":"string"})"};
 constexpr std::string_view sv_string_def1{R"({"type": "string"})"};
 constexpr std::string_view sv_int_def0{R"({"type": "int"})"};
-const pps::canonical_schema_definition string_def0{
+const pps::unparsed_schema_definition string_def0{
   pps::make_schema_definition<json::UTF8<>>(sv_string_def0).value(),
   pps::schema_type::avro};
-const pps::canonical_schema_definition string_def1{
+const pps::unparsed_schema_definition string_def1{
   pps::make_schema_definition<json::UTF8<>>(sv_string_def1).value(),
   pps::schema_type::avro};
-const pps::canonical_schema_definition int_def0{
+const pps::unparsed_schema_definition int_def0{
   pps::make_schema_definition<json::UTF8<>>(sv_int_def0).value(),
   pps::schema_type::avro};
 const pps::subject subject0{"subject0"};
@@ -72,7 +73,7 @@ BOOST_AUTO_TEST_CASE(test_store_insert) {
 bool upsert(
   pps::store& store,
   pps::subject sub,
-  pps::canonical_schema_definition def,
+  pps::unparsed_schema_definition def,
   pps::schema_type,
   pps::schema_id id,
   pps::schema_version version,
@@ -207,8 +208,7 @@ BOOST_AUTO_TEST_CASE(test_store_get_schema_subject_versions) {
     pps::seq_marker dummy_marker;
 
     // First insert, expect id{1}
-    auto ins_res = s.insert(
-      {subject0, pps::canonical_schema_definition(schema1.share())});
+    auto ins_res = s.insert(pps::to_unparsed({subject0, schema1.share()}));
     BOOST_REQUIRE(ins_res.inserted);
     BOOST_REQUIRE_EQUAL(ins_res.id, pps::schema_id{1});
     BOOST_REQUIRE_EQUAL(ins_res.version, pps::schema_version{1});
@@ -222,8 +222,7 @@ BOOST_AUTO_TEST_CASE(test_store_get_schema_subject_versions) {
     BOOST_REQUIRE(versions.empty());
 
     // Second insert, expect id{2}
-    ins_res = s.insert(
-      {subject0, pps::canonical_schema_definition(schema2.share())});
+    ins_res = s.insert(pps::to_unparsed({subject0, schema2.share()}));
     BOOST_REQUIRE(ins_res.inserted);
     BOOST_REQUIRE_EQUAL(ins_res.id, pps::schema_id{2});
     BOOST_REQUIRE_EQUAL(ins_res.version, pps::schema_version{2});
@@ -259,8 +258,7 @@ BOOST_AUTO_TEST_CASE(test_store_get_schema_subjects) {
     pps::seq_marker dummy_marker;
 
     // First insert, expect id{1}
-    auto ins_res = s.insert(
-      {subject0, pps::canonical_schema_definition(schema1.share())});
+    auto ins_res = s.insert(pps::to_unparsed({subject0, schema1.share()}));
     BOOST_REQUIRE(ins_res.inserted);
     BOOST_REQUIRE_EQUAL(ins_res.id, pps::schema_id{1});
     BOOST_REQUIRE_EQUAL(ins_res.version, pps::schema_version{1});
@@ -271,15 +269,13 @@ BOOST_AUTO_TEST_CASE(test_store_get_schema_subjects) {
     BOOST_REQUIRE_EQUAL(absl::c_count_if(subjects, is_equal(subject0)), 1);
 
     // Second insert, same schema, expect id{1}
-    ins_res = s.insert(
-      {subject1, pps::canonical_schema_definition(schema1.share())});
+    ins_res = s.insert(pps::to_unparsed({subject1, schema1.share()}));
     BOOST_REQUIRE(ins_res.inserted);
     BOOST_REQUIRE_EQUAL(ins_res.id, pps::schema_id{1});
     BOOST_REQUIRE_EQUAL(ins_res.version, pps::schema_version{1});
 
     // Insert yet another schema associated with a different subject
-    ins_res = s.insert(
-      {subject2, pps::canonical_schema_definition(schema2.share())});
+    ins_res = s.insert(pps::to_unparsed({subject2, schema2.share()}));
     BOOST_REQUIRE(ins_res.inserted);
     BOOST_REQUIRE_EQUAL(ins_res.id, pps::schema_id{2});
     BOOST_REQUIRE_EQUAL(ins_res.version, pps::schema_version{1});

src/v/pandaproxy/schema_registry/types.cc

@@ -93,4 +93,15 @@ std::ostream& operator<<(std::ostream& os, const compatibility_result& res) {
     return os;
 }
 
+unparsed_schema_definition
+to_unparsed(canonical_schema_definition&& canonical) {
+    auto [raw, type, refs] = std::move(canonical).destructure();
+    return {std::move(raw), type, std::move(refs)};
+}
+
+unparsed_schema to_unparsed(canonical_schema&& canonical) {
+    auto [sub, def] = std::move(canonical).destructure();
+    return {std::move(sub), to_unparsed(std::move(def))};
+}
+
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/types.h

@@ -204,6 +204,9 @@ using unparsed_schema_definition
 using canonical_schema_definition
   = typed_schema_definition<struct canonical_schema_definition_tag>;
 
+///\brief Util function for when a canonical schema need to be re-ingested
+unparsed_schema_definition to_unparsed(canonical_schema_definition&&);
+
 static const unparsed_schema_definition invalid_schema_definition{
   "", schema_type::avro};
 
@@ -468,16 +471,25 @@ class typed_schema {
 using unparsed_schema = typed_schema<unparsed_schema_definition::tag>;
 using canonical_schema = typed_schema<canonical_schema_definition::tag>;
 
-///\brief Complete description of a subject and schema for a version.
-struct subject_schema {
-    canonical_schema schema;
+///\brief Util function for when a canonical schema need to be re-ingested
+unparsed_schema to_unparsed(canonical_schema&&);
+
+template<typename tag>
+struct typed_subject_schema {
+    typed_schema<tag> schema;
     schema_version version{invalid_schema_version};
     schema_id id{invalid_schema_id};
     is_deleted deleted{false};
-    subject_schema share() const {
+    typed_subject_schema share() const {
         return {schema.share(), version, id, deleted};
     }
 };
+///\brief Complete description of a subject and schema for a version, as stored
+/// in store
+using unparsed_subject_schema
+  = typed_subject_schema<unparsed_schema_definition::tag>;
+///\brief Complete description of a subject and schema for a version.
+using subject_schema = typed_subject_schema<canonical_schema_definition::tag>;
 
 enum class compatibility_level {
     none = 0,