[v25.1.x] CORE-15421: add protovalidate as well-known type

MODULE.bazel

@@ -52,6 +52,8 @@ bazel_dep(name = "bzip2", version = "1.0.8.bcr.2")
 bazel_dep(name = "buildifier_prebuilt", version = "7.3.1", dev_dependency = True)
 bazel_dep(name = "rules_python", version = "1.0.0", dev_dependency = True)
 
+bazel_dep(name = "protovalidate", version = "1.0.0")
+
 python = use_extension("@rules_python//python/extensions:python.bzl", "python", dev_dependency = True)
 python.toolchain(
     ignore_root_user_error = True,

src/v/pandaproxy/BUILD

@@ -186,6 +186,7 @@ redpanda_cc_library(
         "@protobuf//src/google/protobuf/compiler:importer",
         "@protobuf//src/google/protobuf/io",
         "@protobuf//src/google/protobuf/io:tokenizer",
+        "@protovalidate//proto/protovalidate/buf/validate:validate_proto_cc",
         "@rapidjson",
         "@re2",
         "@seastar",

src/v/pandaproxy/schema_registry/protobuf.cc

@@ -31,6 +31,7 @@
 #include <absl/strings/escaping.h>
 #include <boost/algorithm/string/trim.hpp>
 #include <boost/range/combine.hpp>
+#include <buf/validate/validate.pb.h>
 #include <confluent/meta.pb.h>
 #include <confluent/types/decimal.pb.h>
 #include <fmt/core.h>
@@ -146,7 +147,9 @@ static const known_types_set known_types{
   google::protobuf::FieldMask::GetDescriptor()->file(),
   google::protobuf::Struct::GetDescriptor()->file(),
   google::protobuf::Timestamp::GetDescriptor()->file(),
-  google::protobuf::FieldDescriptorProto::GetDescriptor()->file()};
+  google::protobuf::FieldDescriptorProto::GetDescriptor()->file(),
+  buf::validate::Rule::GetDescriptor()->file(),
+};
 
 class io_error_collector final : public pb::io::ErrorCollector {
     enum class level {

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

@@ -284,6 +284,7 @@ import "google/type/quaternion.proto";
 import "google/type/timeofday.proto";
 import "confluent/meta.proto";
 import "confluent/types/decimal.proto";
+import "buf/validate/validate.proto";
 
 message well_known_types {
   google.protobuf.Any any = 1;
@@ -333,6 +334,15 @@ message well_known_types {
   google.type.TimeOfDay time_of_day = 46;
   confluent.Meta c_meta = 47;
   confluent.type.Decimal c_decimal = 48;
+  buf.validate.FieldRules field_rules = 49;
+  buf.validate.StringRules string_rules = 50;
+  buf.validate.Int32Rules int32_rules = 51;
+  buf.validate.MessageRules message_rules = 52;
+  buf.validate.RepeatedRules repeated_rules = 53;
+  buf.validate.MapRules map_rules = 54;
+  buf.validate.AnyRules any_rules = 55;
+  buf.validate.DurationRules duration_rules = 56;
+  buf.validate.TimestampRules timestamp_rules = 57;
 })",
         pps::schema_type::protobuf}};
     store.insert(schema.share(), pps::schema_version{1});

tests/rptest/tests/schema_registry_test.py

@@ -280,6 +280,19 @@ def get_subject_name(sns: str, topic: str, field: MessageField, record_name: str
   google.protobuf.Timestamp timestamp = 1;
 }"""
 
+validate_proto_def = """
+syntax = "proto3";
+
+import "buf/validate/validate.proto";
+
+message TestValidate {
+  buf.validate.FieldRules field_rules = 1;
+  buf.validate.StringRules string_rules = 2;
+  buf.validate.Int32Rules int32_rules = 3;
+  buf.validate.MessageRules message_rules = 4;
+  buf.validate.TimestampRules timestamp_rules = 5;
+}"""
+
 json_number_schema_def = '{"type":"number"}'
 
 validation_schemas = dict(
@@ -5277,6 +5290,12 @@ def test_references(self):
         result = self.sr_client.register_schema(well_known_subject, well_known_schema)
         assert result == 3, f"Result: {result}"
 
+        validate_subject = "topic_4-key"
+        validate_schema = Schema(validate_proto_def, "PROTOBUF")
+
+        result = self.sr_client.register_schema(validate_subject, validate_schema)
+        assert result == 4, f"Result: {result}"
+
         result = self.sr_client.get_schema(1)
         assert result == simple_schema, f"Result: {result}"
 
@@ -5286,6 +5305,9 @@ def test_references(self):
         result = self.sr_client.get_schema(3)
         assert result == well_known_schema, f"Result: {result}"
 
+        result = self.sr_client.get_schema(4)
+        assert result == validate_schema, f"Result: {result}"
+
 
 # dataset for SchemaRegistryCompatibilityModes: schemas is a list of 3 schemas compatible for `mode`, `antimode` is a suitable mode that will make the compat check for schemas fail
 CompatDataset = NamedTuple(