iceberg/json_schema: support oneOf for nullable types

Add support for the oneOf keyword, limited to the T|null
pattern used to express nullable fields. Constraints from
oneOf branches are intersected with the parent schema for
types, format, properties, and items. Properties and items
must appear in only one branch; specifying them on both
sides is rejected. Format must either match across branches
or appear in only one.

Author: nvartolomei

src/v/iceberg/conversion/ir_json.cc

@@ -49,8 +49,13 @@ class type_set {
     static type_set none() { return type_set{}; }
 
     void set(json_type t) { bits_.set(index(t)); }
+    void intersect(const type_set& other) { bits_ &= other.bits_; }
     bool test(json_type t) const { return bits_.test(index(t)); }
 
+    bool is_null_only() const {
+        return bits_.count() == 1 && bits_.test(index(json_type::null));
+    }
+
     /// Returns the single non-null type if exactly one is set.
     std::optional<json_type> single_non_null_type() const {
         auto copy = bits_;
@@ -113,12 +118,56 @@ struct constraint {
 
     // Object properties, keyed by name.
     std::map<std::string, constraint> properties = {};
+    bool additional_properties_allowed = true;
 
     // Array item constraints.
     // - nullopt: no "items" keyword
     // - empty vector: "items": []
     // - non-empty: item schema(s) to validate
     std::optional<std::vector<constraint>> items = std::nullopt;
+
+    conversion_outcome<void> intersect(const constraint& other) {
+        types.intersect(other.types);
+
+        if (format.has_value() && other.format.has_value()) {
+            if (format != other.format) {
+                return conversion_exception(
+                  "Conflicting format annotations across branches");
+            }
+        } else if (other.format.has_value()) {
+            format = other.format;
+        }
+
+        if (!properties.empty() && !other.properties.empty()) {
+            return conversion_exception(
+              "Intersecting constraints with properties on both sides "
+              "is not supported");
+        } else if (!other.properties.empty()) {
+            if (!additional_properties_allowed) {
+                return conversion_exception(
+                  "additionalProperties: false conflicts with properties "
+                  "defined in another branch");
+            }
+            properties = other.properties;
+        } else if (
+          !properties.empty() && !other.additional_properties_allowed) {
+            return conversion_exception(
+              "additionalProperties: false conflicts with properties "
+              "defined in another branch");
+        }
+        additional_properties_allowed = additional_properties_allowed
+                                        && other.additional_properties_allowed;
+
+        if (items.has_value() && other.items.has_value()) {
+            return conversion_exception(
+              "Intersecting constraints with items on both sides is not "
+              "supported");
+        } else if (other.items.has_value()) {
+            items = other.items;
+        }
+
+        return outcome::success();
+    }
 };
 
 /// Context for the resolution phase.
@@ -149,6 +198,48 @@ collect(collect_context& ctx, const conversion::json_schema::subschema&);
 
 conversion_outcome<field_type> resolve(resolution_context&, const constraint&);
 
+// This is not full fidelity oneOf support - only T|null is supported.
+// In the general case, oneOf is a XOR over multiple schemas. For T|null
+// it is reduced to OR.
+conversion_outcome<constraint> collect_one_of_t_xor_null(
+  collect_context& ctx,
+  const conversion::json_schema::const_list_view& one_of) {
+    constexpr std::string_view unsupported_one_of_msg
+      = "oneOf keyword is supported only for exclusive T|null structures";
+    if (one_of.size() != 2) {
+        return conversion_exception(std::string{unsupported_one_of_msg});
+    }
+
+    auto c1 = collect(ctx, one_of.at(0));
+    if (c1.has_error()) {
+        return c1.error();
+    }
+    auto c2 = collect(ctx, one_of.at(1));
+    if (c2.has_error()) {
+        return c2.error();
+    }
+
+    // Accept only exclusive oneOf(T, null):
+    // - exactly one branch is null-only
+    // - the non-null branch must not include null
+    const bool c1_is_null_only = c1.value().types.is_null_only();
+    const bool c2_is_null_only = c2.value().types.is_null_only();
+
+    if (c1_is_null_only == c2_is_null_only) {
+        return conversion_exception(std::string{unsupported_one_of_msg});
+    }
+
+    const constraint& non_null_branch = c1_is_null_only ? c2.value()
+                                                        : c1.value();
+    if (non_null_branch.types.test(json_type::null)) {
+        return conversion_exception(std::string{unsupported_one_of_msg});
+    }
+
+    auto c = non_null_branch;
+    c.types.set(json_type::null);
+    return c;
+}
+
 /// Collect item constraints from a JSON Schema array.
 conversion_outcome<std::optional<std::vector<constraint>>> collect_items(
   collect_context& ctx, const conversion::json_schema::subschema& s) {
@@ -232,12 +323,14 @@ collect(collect_context& ctx, const conversion::json_schema::subschema& s) {
             c.properties[name] = std::move(prop_constraint.value());
         }
 
-        if (
-          s.additional_properties()
-          && s.additional_properties()->get().boolean_subschema() != false) {
-            return conversion_exception(
-              "Only 'false' subschema is supported "
-              "for additionalProperties keyword");
+        if (s.additional_properties()) {
+            if (s.additional_properties()->get().boolean_subschema() == false) {
+                c.additional_properties_allowed = false;
+            } else {
+                return conversion_exception(
+                  "Only 'false' subschema is supported "
+                  "for additionalProperties keyword");
+            }
         }
     }
 
@@ -250,6 +343,17 @@ collect(collect_context& ctx, const conversion::json_schema::subschema& s) {
         c.items = std::move(items_result.value());
     }
 
+    if (!s.one_of().empty()) {
+        auto one_of_constraint = collect_one_of_t_xor_null(ctx, s.one_of());
+        if (one_of_constraint.has_error()) {
+            return one_of_constraint.error();
+        }
+        auto intersect_result = c.intersect(one_of_constraint.value());
+        if (intersect_result.has_error()) {
+            return intersect_result.error();
+        }
+    }
+
     return c;
 }
 

src/v/iceberg/conversion/json_schema/frontend.cc

@@ -300,7 +300,6 @@ constexpr auto banned_keywords = std::to_array({
   "else",
   "allOf",
   "anyOf",
-  "oneOf",
 });
 
 }; // namespace
@@ -525,6 +524,12 @@ class frontend::frontend_impl {
             }
         }
 
+        if (
+          const auto one_of_node = find_keyword(
+            node, "oneOf", {json_value_type::array})) {
+            compile_one_of(ctx, *sub, *one_of_node);
+        }
+
         if (
           const auto format_node = find_keyword(
             node, "format", {json_value_type::string})) {
@@ -657,6 +662,33 @@ class frontend::frontend_impl {
               "The items keyword must be an object or an array of objects");
         }
     }
+
+    void compile_one_of(
+      compile_context& ctx, subschema& sub, const json::Value& node) const {
+        vassert(
+          sub.one_of_.empty(),
+          "oneOf subschema vector should be empty at this point");
+
+        if (!node.IsArray() || node.GetArray().Empty()) {
+            throw std::runtime_error(
+              "The oneOf keyword must be a non-empty array");
+        }
+
+        std::vector<ss::shared_ptr<subschema>> one_of_subschemas;
+        one_of_subschemas.reserve(node.GetArray().Size());
+        for (const auto& item : node.GetArray()) {
+            one_of_subschemas.push_back(compile_subschema(ctx, item));
+        }
+
+        for (size_t i = 0; i < one_of_subschemas.size(); ++i) {
+            auto& one_of_subschema = one_of_subschemas[i];
+            auto [it, inserted] = sub.subschemas_.emplace(
+              fmt::format("oneOf/{}", i), one_of_subschema);
+            vassert(inserted, "unique insertion should have succeeded");
+        }
+
+        sub.one_of_ = std::move(one_of_subschemas);
+    }
 };
 
 frontend::frontend()

src/v/iceberg/conversion/json_schema/ir.h

@@ -303,6 +303,9 @@ class subschema {
                                  : std::nullopt;
     }
 
+    /// \return Subschemas in the "oneOf" keyword if specified.
+    const_list_view one_of() const { return const_list_view(one_of_); }
+
     /// \return Format of this schema object if specified.
     /// See https://www.learnjsonschema.com/2020-12/validation/format
     const std::optional<format>& format() const { return format_; }
@@ -333,6 +336,8 @@ class subschema {
     items_t items_;
     ss::shared_ptr<subschema> additional_items_;
 
+    std::vector<ss::shared_ptr<subschema>> one_of_;
+
     std::optional<enum format> format_;
 
     std::optional<std::string> ref_value_;

src/v/iceberg/conversion/json_schema/tests/frontend_test.cc

@@ -923,6 +923,87 @@ TEST(frontend_test, non_object_or_boolean_subschema) {
         StrEq("Subschema must be an object or a boolean")));
 }
 
+TEST(frontend_test, one_of) {
+    auto schema = frontend{}.compile(
+      parse_json(R"({
+        "$id": "https://example.com/root.json",
+        "oneOf": [
+            { "type": "null" },
+            { "type": "string" },
+            { "type": "integer" }
+        ]
+      })"),
+      "https://example.com/irrelevant-base.json",
+      dialect::draft7);
+
+    auto expected = R"(# (document root)
+  base uri: https://example.com/root.json
+  dialect: http://json-schema.org/draft-07/schema#
+#/oneOf/0
+  base uri: https://example.com/root.json
+  dialect: http://json-schema.org/draft-07/schema#
+  types: [null]
+#/oneOf/1
+  base uri: https://example.com/root.json
+  dialect: http://json-schema.org/draft-07/schema#
+  types: [string]
+#/oneOf/2
+  base uri: https://example.com/root.json
+  dialect: http://json-schema.org/draft-07/schema#
+  types: [integer]
+)";
+
+    ASSERT_EQ(expected, ir_tree_printer::to_string(schema));
+    ASSERT_EQ(schema.root().one_of().size(), 3);
+}
+
+TEST(frontend_test, duplicate_one_of) {
+    // Not required by json schema but we disallow duplicate keywords to reduce
+    // ambiguity.
+    EXPECT_THAT(
+      []() {
+          frontend{}.compile(
+            parse_json(R"({
+              "$id": "https://example.com/root.json",
+              "oneOf": [{ "type": "string" }],
+              "oneOf": [{ "type": "integer" }]
+            })"),
+            "https://example.com/irrelevant-base.json",
+            dialect::draft7);
+      },
+      ThrowsMessage<std::runtime_error>(StrEq("Duplicate keyword: oneOf")));
+}
+
+TEST(frontend_test, one_of_empty_array) {
+    EXPECT_THAT(
+      []() {
+          frontend{}.compile(
+            parse_json(R"({
+              "$id": "https://example.com/root.json",
+              "oneOf": []
+            })"),
+            "https://example.com/irrelevant-base.json",
+            dialect::draft7);
+      },
+      ThrowsMessage<std::runtime_error>(
+        StrEq("The oneOf keyword must be a non-empty array")));
+}
+
+TEST(frontend_test, one_of_non_array) {
+    EXPECT_THAT(
+      []() {
+          frontend{}.compile(
+            parse_json(R"({
+              "$id": "https://example.com/root.json",
+              "oneOf": "not an array"
+            })"),
+            "https://example.com/irrelevant-base.json",
+            dialect::draft7);
+      },
+      ThrowsMessage<std::runtime_error>(
+        StrEq("Invalid type for keyword oneOf. Expected one of: [array].")));
+}
+
 TEST(frontend_test, non_string_dialect) {
     EXPECT_THAT(
       []() {