CORE-6860 SR: Wire up support for ?verbose compatibility checks

src/go/rpk/pkg/cli/registry/schema/check_compatibility.go

@@ -23,7 +23,8 @@ import (
 )
 
 type compatCheckResponse struct {
-	Compatible bool `json:"compatible" yaml:"compatible"`
+	Compatible bool     `json:"compatible" yaml:"compatible"`
+	Messages   []string `json:"messages,omitempty" yaml:"messages,omitempty"`
 }
 
 func newCheckCompatibilityCommand(fs afero.Fs, p *config.Params) *cobra.Command {
@@ -66,16 +67,19 @@ func newCheckCompatibilityCommand(fs afero.Fs, p *config.Params) *cobra.Command
 				Type:       t,
 				References: references,
 			}
-			compatible, err := cl.CheckCompatibility(cmd.Context(), subject, version, schema)
+			ctx := sr.WithParams(cmd.Context(), sr.Verbose)
+			compatible, err := cl.CheckCompatibility(ctx, subject, version, schema)
 			out.MaybeDie(err, "unable to check compatibility: %v", err)
-			if isText, _, s, err := f.Format(compatCheckResponse{compatible.Is}); !isText {
+			if isText, _, s, err := f.Format(compatCheckResponse{compatible.Is, compatible.Messages}); !isText {
 				out.MaybeDie(err, "unable to print in the required format %q: %v", f.Kind, err)
 				out.Exit(s)
 			}
 			if compatible.Is {
 				fmt.Println("Schema is compatible.")
 			} else {
 				fmt.Println("Schema is not compatible.")
+				messages := strings.Join(compatible.Messages, "\n")
+				fmt.Println(messages)
 			}
 		},
 	}

src/v/pandaproxy/BUILD

@@ -141,6 +141,7 @@ redpanda_cc_library(
         "rest/proxy.cc",
         "schema_registry/api.cc",
         "schema_registry/avro.cc",
+        "schema_registry/compatibility.cc",
         "schema_registry/configuration.cc",
         "schema_registry/error.cc",
         "schema_registry/handlers.cc",
@@ -189,6 +190,7 @@ redpanda_cc_library(
         "rest/proxy.h",
         "schema_registry/api.h",
         "schema_registry/avro.h",
+        "schema_registry/compatibility.h",
         "schema_registry/configuration.h",
         "schema_registry/error.h",
         "schema_registry/errors.h",

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

@@ -1148,6 +1148,12 @@
             "schema": {
               "$ref": "#/definitions/schema_def"
             }
+          },
+          {
+            "name": "verbose",
+            "in": "query",
+            "required": false,
+            "type": "boolean"
           }
         ],
         "produces": [

src/v/pandaproxy/api/api-doc/schema_registry_definitions.def.json

@@ -108,6 +108,12 @@
       "properties": {
         "is_compatible": {
           "type": "boolean"
-        }
+        },
+        "messages": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        },
       }
     }

src/v/pandaproxy/schema_registry/CMakeLists.txt

@@ -14,6 +14,7 @@ v_cc_library(
   SRCS
     api.cc
     configuration.cc
+    compatibility.cc
     handlers.cc
     error.cc
     service.cc

src/v/pandaproxy/schema_registry/avro.cc

@@ -17,6 +17,7 @@
 #include "json/document.h"
 #include "json/json.h"
 #include "json/types.h"
+#include "pandaproxy/schema_registry/compatibility.h"
 #include "pandaproxy/schema_registry/error.h"
 #include "pandaproxy/schema_registry/errors.h"
 #include "pandaproxy/schema_registry/sharded_store.h"
@@ -550,9 +551,13 @@ sanitize_avro_schema_definition(unparsed_schema_definition def) {
       def.refs()};
 }
 
-bool check_compatible(
-  const avro_schema_definition& reader, const avro_schema_definition& writer) {
-    return check_compatible(*reader().root(), *writer().root());
+compatibility_result check_compatible(
+  const avro_schema_definition& reader,
+  const avro_schema_definition& writer,
+  verbose is_verbose [[maybe_unused]]) {
+    // TODO(gellert.nagy): start using the is_verbose flag in a follow up PR
+    return compatibility_result{
+      .is_compat = check_compatible(*reader().root(), *writer().root())};
 }
 
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/avro.h

@@ -23,7 +23,9 @@ make_avro_schema_definition(sharded_store& store, canonical_schema schema);
 result<canonical_schema_definition>
 sanitize_avro_schema_definition(unparsed_schema_definition def);
 
-bool check_compatible(
-  const avro_schema_definition& reader, const avro_schema_definition& writer);
+compatibility_result check_compatible(
+  const avro_schema_definition& reader,
+  const avro_schema_definition& writer,
+  verbose is_verbose = verbose::no);
 
 } // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/compatibility.cc

@@ -0,0 +1,174 @@
+/*
+ * Copyright 2024 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "pandaproxy/schema_registry/compatibility.h"
+
+namespace pandaproxy::schema_registry {
+
+std::ostream& operator<<(std::ostream& os, const avro_incompatibility_type& t) {
+    switch (t) {
+    case avro_incompatibility_type::name_mismatch:
+        return os << "NAME_MISMATCH";
+    case avro_incompatibility_type::fixed_size_mismatch:
+        return os << "FIXED_SIZE_MISMATCH";
+    case avro_incompatibility_type::missing_enum_symbols:
+        return os << "MISSING_ENUM_SYMBOLS";
+    case avro_incompatibility_type::reader_field_missing_default_value:
+        return os << "READER_FIELD_MISSING_DEFAULT_VALUE";
+    case avro_incompatibility_type::type_mismatch:
+        return os << "TYPE_MISMATCH";
+    case avro_incompatibility_type::missing_union_branch:
+        return os << "MISSING_UNION_BRANCH";
+    case avro_incompatibility_type::unknown:
+        return os << "UNKNOWN";
+    };
+    __builtin_unreachable();
+}
+
+std::string_view description_for_type(avro_incompatibility_type t) {
+    switch (t) {
+    case avro_incompatibility_type::name_mismatch:
+        return "The name of the schema has changed (path '{path}')";
+    case avro_incompatibility_type::fixed_size_mismatch:
+        return "The size of FIXED type field at path '{path}' in the "
+               "{{reader}} schema does not match with the {{writer}} schema";
+    case avro_incompatibility_type::missing_enum_symbols:
+        return "The {{reader}} schema is missing enum symbols '{additional}' "
+               "at path '{path}' in the {{writer}} schema";
+    case avro_incompatibility_type::reader_field_missing_default_value:
+        return "The field '{additional}' at path '{path}' in the {{reader}} "
+               "schema has "
+               "no default value and is missing in the {{writer}}";
+    case avro_incompatibility_type::type_mismatch:
+        return "The type (path '{path}') of a field in the {{reader}} schema "
+               "does not match with the {{writer}} schema";
+    case avro_incompatibility_type::missing_union_branch:
+        return "The {{reader}} schema is missing a type inside a union field "
+               "at path '{path}' in the {{writer}} schema";
+    case avro_incompatibility_type::unknown:
+        return "{{reader}} schema is not compatible with {{writer}} schema: "
+               "check '{path}'";
+    };
+    __builtin_unreachable();
+}
+
+std::ostream& operator<<(std::ostream& os, const avro_incompatibility& v) {
+    fmt::print(
+      os,
+      "{{errorType:'{}', description:'{}', additionalInfo:'{}'}}",
+      v._type,
+      v.describe(),
+      v._additional_info);
+    return os;
+}
+
+ss::sstring avro_incompatibility::describe() const {
+    return fmt::format(
+      fmt::runtime(description_for_type(_type)),
+      fmt::arg("path", _path.string()),
+      fmt::arg("additional", _additional_info));
+}
+
+std::ostream&
+operator<<(std::ostream& os, const proto_incompatibility_type& t) {
+    switch (t) {
+    case proto_incompatibility_type::message_removed:
+        return os << "MESSAGE_REMOVED";
+    case proto_incompatibility_type::field_kind_changed:
+        return os << "FIELD_KIND_CHANGED";
+    case proto_incompatibility_type::field_scalar_kind_changed:
+        return os << "FIELD_SCALAR_KIND_CHANGED";
+    case proto_incompatibility_type::field_named_type_changed:
+        return os << "FIELD_NAMED_TYPE_CHANGED";
+    case proto_incompatibility_type::required_field_added:
+        return os << "REQUIRED_FIELD_ADDED";
+    case proto_incompatibility_type::required_field_removed:
+        return os << "REQUIRED_FIELD_REMOVED";
+    case proto_incompatibility_type::oneof_field_removed:
+        return os << "ONEOF_FIELD_REMOVED";
+    case proto_incompatibility_type::multiple_fields_moved_to_oneof:
+        return os << "MULTIPLE_FIELDS_MOVED_TO_ONEOF";
+    case proto_incompatibility_type::unknown:
+        return os << "UNKNOWN";
+    }
+    __builtin_unreachable();
+}
+
+std::string_view description_for_type(proto_incompatibility_type t) {
+    switch (t) {
+    case proto_incompatibility_type::message_removed:
+        return "The {{reader}} schema is missing a field of type MESSAGE at "
+               "path '{path}' in the {{writer}} schema";
+    case proto_incompatibility_type::field_kind_changed:
+        return "The type of a field at path '{path}' in the {{reader}} "
+               "schema does  not match the {{writer}} schema";
+    case proto_incompatibility_type::field_scalar_kind_changed:
+        return "The kind of a SCALAR field at path '{path}' in the {{reader}} "
+               "schema does not match its kind in the {{writer}} schema";
+    case proto_incompatibility_type::field_named_type_changed:
+        return "The type of a MESSAGE field at path '{path}' in the {{reader}} "
+               "schema does not match its type in the {{writer}} schema ";
+    case proto_incompatibility_type::required_field_added:
+        return "A required field  at path '{path}' in the {{reader}} schema "
+               "is missing in the {{writer}} schema";
+    case proto_incompatibility_type::required_field_removed:
+        return "The {{reader}} schema is missing a required field at path: "
+               "'{path}' in the {{writer}} schema";
+    case proto_incompatibility_type::oneof_field_removed:
+        return "The {{reader}} schema is missing a oneof field at path "
+               "'{path}' in the {{writer}} schema";
+    case proto_incompatibility_type::multiple_fields_moved_to_oneof:
+        return "Multiple fields in the oneof at path '{path}' in the "
+               "{{reader}} schema are outside a oneof in the {{writer}} "
+               "schema ";
+    case proto_incompatibility_type::unknown:
+        return "{{reader}} schema is not compatible with {{writer}} schema: "
+               "check '{path}'";
+    }
+    __builtin_unreachable();
+}
+
+std::ostream& operator<<(std::ostream& os, const proto_incompatibility& v) {
+    fmt::print(
+      os, "{{errorType:'{}', description:'{}'}}", v._type, v.describe());
+    return os;
+}
+
+ss::sstring proto_incompatibility::describe() const {
+    return fmt::format(
+      fmt::runtime(description_for_type(_type)),
+      fmt::arg("path", _path.string()));
+}
+
+compatibility_result
+raw_compatibility_result::operator()(verbose is_verbose) && {
+    compatibility_result result = {.is_compat = !has_error()};
+    if (is_verbose) {
+        result.messages.reserve(_errors.size());
+        std::transform(
+          std::make_move_iterator(_errors.begin()),
+          std::make_move_iterator(_errors.end()),
+          std::back_inserter(result.messages),
+          [](auto e) {
+              return std::visit(
+                [](auto&& e) { return fmt::format("{{{}}}", e); }, e);
+          });
+    }
+    return result;
+}
+
+void raw_compatibility_result::merge(raw_compatibility_result&& other) {
+    _errors.reserve(_errors.size() + other._errors.size());
+    std::move(
+      other._errors.begin(), other._errors.end(), std::back_inserter(_errors));
+}
+
+} // namespace pandaproxy::schema_registry