Merge pull request #29590 from nvartolomei/nv/iceberg-json-schema-oneof

iceberg/json_schema: support oneOf for nullable types

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(
       []() {