pandaproxy/sr: Refactor store internal state to use unparsed schemas

- Replace canonical schemas for unparsed schemas stored in store and
move all schema processing to sharded_store. This allows to have
lazy schema validation on startup and relax requirements for valid
schema. With this changes, a schema that had been imported before its
dependencies will be found valid when it is requested.

- Add backport to sharded_store to be able to access an unparsed schema
as-is, without processing it. This is needed as now there might be an
invalid schema loaded at startup. Operations like
delete_subject_version, which needs to reference a schema before
deleting it would be problematic.

Author: IoannisRP

src/v/pandaproxy/schema_registry/seq_writer.cc

@@ -433,17 +433,17 @@ ss::future<std::optional<bool>> seq_writer::do_delete_subject_version(
         throw as_exception(has_references(sub, version));
     }
 
-    auto s_res = co_await _store.get_subject_schema(
-      sub, version, include_deleted::yes);
-    subject_schema ss = std::move(s_res);
+    schema_id s_id = co_await _store.get_id(sub, version);
+    unparsed_schema_definition schema
+      = co_await _store.get_unparsed_schema_definition(s_id);
 
     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{
-      .schema{std::move(ss.schema)},
+    unparsed_schema_value value{
+      .schema{unparsed_schema{sub, std::move(schema)}},
       .version{version},
-      .id{ss.id},
+      .id{s_id},
       .deleted{is_deleted::yes}};
 
     batch_builder rb(write_at, sub);

src/v/pandaproxy/schema_registry/sharded_store.cc

@@ -167,14 +167,10 @@ sharded_store::get_schema_version(subject_schema schema, normalize norm) {
                            if (s_res.has_error()) {
                                return ss::make_ready_future();
                            }
-                           auto [raw, type, refs]
-                             = std::move(s_res.value()).destructure();
                            return this
                              ->make_canonical_schema(
                                unparsed_schema{
-                                 subject{},
-                                 unparsed_schema_definition{
-                                   std::move(raw), type, std::move(refs)}},
+                                 subject{}, std::move(s_res.value())},
                                norm)
                              .then([id, &ret_id, &canonical_def](
                                      canonical_schema processed) {
@@ -299,22 +295,6 @@ ss::future<bool> sharded_store::upsert(
   unparsed_schema schema,
   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) {
     auto [sub, def] = std::move(schema).destructure();
     co_await upsert_schema(id, std::move(def));
@@ -365,6 +345,16 @@ ss::future<subject_schema> sharded_store::has_schema(
             if (
               e.code() == error_code::subject_not_found
               || e.code() == error_code::subject_version_not_found) {
+            } else if (
+              // Stored schemas might be invalid if imported improperly
+              e.code() == error_code::schema_invalid) {
+                vlog(
+                  plog.warn,
+                  "Failed to parse stored schema, subject '{}', version {}. "
+                  "Error: ",
+                  schema.sub(),
+                  ver,
+                  e.what());
             } else {
                 throw;
             }
@@ -378,10 +368,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()
@@ -390,10 +380,49 @@ sharded_store::maybe_get_schema_definition(schema_id id) {
           }
           return std::move(s_res.value());
       });
+    if (!unparsed.has_value()) {
+        co_return std::nullopt;
+    }
+
+    try {
+        auto canonical = co_await make_canonical_schema(
+          {{}, std::move(unparsed.value())});
+
+        co_return std::move(canonical).def();
+    } catch (const exception& e) {
+        vlog(
+          plog.warn,
+          "Failed to parse stored schema with id {}: Error {}",
+          id,
+          e.what());
+        throw;
+    }
 }
 
 ss::future<canonical_schema_definition>
 sharded_store::get_schema_definition(schema_id id) {
+    auto unparsed = co_await _store.invoke_on(
+      shard_for(id), _smp_opts, [id](store& s) {
+          return s.get_schema_definition(id).value();
+      });
+
+    try {
+        auto canonical = co_await make_canonical_schema(
+          {{}, std::move(unparsed)});
+
+        co_return std::move(canonical).def();
+    } catch (const exception& e) {
+        vlog(
+          plog.warn,
+          "Failed to parse stored schema with id {}: Error {}",
+          id,
+          e.what());
+        throw;
+    }
+}
+
+ss::future<unparsed_schema_definition>
+sharded_store::get_unparsed_schema_definition(schema_id id) {
     co_return co_await _store.invoke_on(
       shard_for(id), _smp_opts, [id](store& s) {
           return s.get_schema_definition(id).value();
@@ -428,6 +457,17 @@ sharded_store::get_schema_subjects(schema_id id, include_deleted inc_del) {
     co_return subs;
 }
 
+ss::future<schema_id>
+sharded_store::get_id(subject sub, std::optional<schema_version> version) {
+    auto v_id = co_await _store.invoke_on(
+      shard_for(sub), _smp_opts, [sub, version](store& s) {
+          return s.get_subject_version_id(sub, version, include_deleted::yes)
+            .value();
+      });
+
+    co_return v_id.id;
+}
+
 ss::future<subject_schema> sharded_store::get_subject_schema(
   subject sub, std::optional<schema_version> version, include_deleted inc_del) {
     auto sub_shard{shard_for(sub)};
@@ -441,11 +481,23 @@ ss::future<subject_schema> sharded_store::get_subject_schema(
           return s.get_schema_definition(id).value();
       });
 
-    co_return subject_schema{
-      .schema = {sub, std::move(def)},
-      .version = v_id.version,
-      .id = v_id.id,
-      .deleted = v_id.deleted};
+    try {
+        auto canonical = co_await make_canonical_schema({sub, std::move(def)});
+
+        co_return subject_schema{
+          .schema = std::move(canonical),
+          .version = v_id.version,
+          .id = v_id.id,
+          .deleted = v_id.deleted};
+    } catch (const exception& e) {
+        vlog(
+          plog.warn,
+          "Failed to parse stored schema, subject {}, version {}: {}",
+          sub,
+          v_id.id,
+          e.what());
+        throw;
+    }
 }
 
 ss::future<chunked_vector<subject>> sharded_store::get_subjects(
@@ -703,7 +755,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

@@ -56,13 +56,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,
@@ -80,6 +73,10 @@ class sharded_store final : public schema_getter {
     ss::future<canonical_schema_definition>
     get_schema_definition(schema_id id) override;
 
+    ///\brief Return a schema definition by id, without any processing.
+    ss::future<unparsed_schema_definition>
+    get_unparsed_schema_definition(schema_id id);
+
     ss::future<std::optional<canonical_schema_definition>>
     maybe_get_schema_definition(schema_id id) override;
 
@@ -97,6 +94,10 @@ class sharded_store final : public schema_getter {
       std::optional<schema_version> version,
       include_deleted inc_dec) final;
 
+    ///\brief Return the id of a schema by subject and version (or latest).
+    ss::future<schema_id>
+    get_id(subject sub, std::optional<schema_version> version);
+
     ///\brief Return a list of subjects.
     ss::future<chunked_vector<subject>> get_subjects(
       include_deleted inc_del,
@@ -213,7 +214,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()) {
@@ -157,7 +157,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 {
@@ -166,7 +166,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,
@@ -614,7 +614,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;
@@ -630,7 +630,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;
     }
@@ -776,10 +776,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 {