Merge pull request #29563 from michael-redpanda/sr/contexts/core-15196

CORE-15196 - Add support for `referenceFormat`

Author: michael-redpanda

src/v/pandaproxy/api/api-doc/schema_registry.json

@@ -461,6 +461,13 @@
             "required": false,
             "type": "string",
             "description": "Optional qualified subject to search for the schema under. Use <:.context:> for context-only lookup, or <:.context:subject> to also verify the schema is associated with that subject. Defaults to searching the default context if unspecified."
+          },
+          {
+            "name": "referenceFormat",
+            "in": "query",
+            "required": false,
+            "type": "string",
+            "description": "If set to 'qualified', schema references are returned in context-qualified form. Otherwise, unqualified references are returned."
           }
         ],
         "produces": [

src/v/pandaproxy/schema_registry/handlers.cc

@@ -96,6 +96,13 @@ output_format parse_output_format(const ss::http::request& req) {
       .value_or(output_format::none);
 }
 
+reference_format parse_reference_format(const ss::http::request& req) {
+    return parse::query_param<std::optional<ss::sstring>>(
+             req, "referenceFormat")
+      .and_then(&from_string_view<reference_format>)
+      .value_or(reference_format::none);
+}
+
 template<ppj::impl::RjsonParseHandler Handler>
 typename ss::future<typename Handler::rjson_parse_result>
 rjson_parse(ss::http::request& req, Handler handler) {
@@ -1046,6 +1053,7 @@ ss::future<ctx_server<service>::reply_t> get_subject_versions_version(
       parse::query_param<std::optional<include_deleted>>(*rq.req, "deleted")
         .value_or(include_deleted::no)};
     const auto format = parse_output_format(*rq.req);
+    const auto reference_format = parse_reference_format(*rq.req);
 
     co_await rq.service().writer().read_sync();
 
@@ -1061,12 +1069,14 @@ ss::future<ctx_server<service>::reply_t> get_subject_versions_version(
       std::move(def), format);
 
     auto resp = ppj::rjson_serialize_iobuf(
-      get_subject_versions_version_response{.stored_schema{
-        .schema = {std::move(subject), std::move(formatted_schema)},
-        .version = get_res.version,
-        .id = get_res.id,
-        .deleted = get_res.deleted,
-      }});
+      get_subject_versions_version_response{
+        .stored_schema{
+          .schema = {std::move(subject), std::move(formatted_schema)},
+          .version = get_res.version,
+          .id = get_res.id,
+          .deleted = get_res.deleted,
+        },
+        .format = reference_format});
     log_response(*rq.req, resp);
     rp.rep->write_body("json", ppj::as_body_writer(std::move(resp)));
     co_return rp;

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

@@ -13,18 +13,21 @@
 
 #include "json/iobuf_writer.h"
 #include "pandaproxy/schema_registry/rjson.h"
+#include "pandaproxy/schema_registry/types.h"
 
 namespace pandaproxy::schema_registry {
 
 struct get_subject_versions_version_response {
     stored_schema stored_schema;
+    reference_format format{reference_format::none};
 };
 
 template<typename Buffer>
 void rjson_serialize(
   ::json::iobuf_writer<Buffer>& w,
   const get_subject_versions_version_response& res) {
     w.StartObject();
+    bool is_qualified = res.stored_schema.schema.sub().ctx != default_context;
     w.Key("subject");
     w.String(res.stored_schema.schema.sub().to_string());
     w.Key("version");
@@ -36,7 +39,13 @@ void rjson_serialize(
     ::json::rjson_serialize(w, to_string_view(type));
     if (!res.stored_schema.schema.def().refs().empty()) {
         w.Key("references");
-        ::json::rjson_serialize(w, res.stored_schema.schema.def().refs());
+        ::json::rjson_serialize(
+          w,
+          res.stored_schema.schema.def().refs(),
+          is_qualified && res.format == reference_format::qualified
+            ? std::make_optional<std::reference_wrapper<const context>>(
+                res.stored_schema.schema.sub().ctx)
+            : std::nullopt);
     }
     ::json::rjson_serialize(w, res.stored_schema.schema.def().meta());
     w.Key("schema");

src/v/pandaproxy/schema_registry/rjson.h

@@ -26,17 +26,40 @@ void rjson_serialize(
 
 template<typename Writer>
 void rjson_serialize(
-  Writer& w, const pandaproxy::schema_registry::schema_reference& ref) {
+  Writer& w,
+  const pandaproxy::schema_registry::schema_reference& ref,
+  std::optional<
+    std::reference_wrapper<const pandaproxy::schema_registry::context>> ctx
+  = std::nullopt) {
     w.StartObject();
     w.Key("name");
     ::json::rjson_serialize(w, ref.name);
     w.Key("subject");
-    w.String(ref.sub.to_string());
+    if (ctx.has_value()) {
+        w.String(ref.sub.resolve(ctx->get()).to_string());
+    } else {
+        w.String(ref.sub.to_string());
+    }
+
     w.Key("version");
     ::json::rjson_serialize(w, ref.version);
     w.EndObject();
 }
 
+template<typename Writer>
+void rjson_serialize(
+  Writer& w,
+  const pandaproxy::schema_registry::schema_definition::references& refs,
+  std::optional<
+    std::reference_wrapper<const pandaproxy::schema_registry::context>> ctx
+  = std::nullopt) {
+    w.StartArray();
+    for (const auto& ref : refs) {
+        rjson_serialize(w, ref, ctx);
+    }
+    w.EndArray();
+}
+
 template<typename Writer>
 void rjson_serialize(
   Writer& w,

src/v/pandaproxy/schema_registry/types.cc

@@ -28,6 +28,10 @@ std::ostream& operator<<(std::ostream& os, const output_format& v) {
     return os << to_string_view(v);
 }
 
+std::ostream& operator<<(std::ostream& os, const reference_format& v) {
+    return os << to_string_view(v);
+}
+
 std::ostream& operator<<(std::ostream& os, const seq_marker& v) {
     if (v.seq.has_value() && v.node.has_value()) {
         fmt::print(

src/v/pandaproxy/schema_registry/types.h

@@ -124,6 +124,31 @@ from_string_view<output_format>(std::string_view sv) {
 
 std::ostream& operator<<(std::ostream& os, const output_format& of);
 
+enum class reference_format { none = 0, qualified };
+
+constexpr std::string_view to_string_view(reference_format rf) {
+    switch (rf) {
+    case reference_format::qualified:
+        return "qualified";
+    case reference_format::none:
+        break;
+    }
+    return "";
+}
+
+template<>
+inline std::optional<reference_format>
+from_string_view<reference_format>(std::string_view sv) {
+    return string_switch<std::optional<reference_format>>(sv)
+      .match(to_string_view(reference_format::none), reference_format::none)
+      .match(
+        to_string_view(reference_format::qualified),
+        reference_format::qualified)
+      .default_match(std::nullopt);
+}
+
+std::ostream& operator<<(std::ostream& os, const reference_format& rf);
+
 ///\brief Type representing a global resource for ACLs.
 using registry_resource = named_type<ss::sstring, struct registry_resource_tag>;