use crate::{
debug, debug_verbose,
schema::core::{make_promoted_scalar_key, SchemaInferenceConfig},
};
use serde_json::{json, Map, Value};
fn normalise_nullable(v: &Value) -> &Value {
let mut current = v;
loop {
if let Some(arr) = current.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".to_string())) {
current = arr
.iter()
.find(|x| *x != &Value::String("null".to_string()))
.unwrap();
continue;
}
}
return current;
}
}
fn try_make_nullable_union(a: &Value, b: &Value) -> Option<Value> {
if a.get("type") == Some(&Value::String("null".into())) {
if let Some(other_type) = b.get("type") {
if other_type != &Value::String("null".into()) {
let mut result = b.clone();
result
.as_object_mut()?
.insert("type".to_string(), json!(["null", other_type]));
return Some(result);
}
}
}
None
}
fn schemas_compatible(existing: &Value, new: &Value) -> Option<Value> {
if existing == new {
return Some(existing.clone());
}
if let Some(result) =
try_make_nullable_union(existing, new).or_else(|| try_make_nullable_union(new, existing))
{
return Some(result);
}
let extract_nullable_info = |schema: &Value| -> (bool, Value) {
if let Some(Value::Array(type_arr)) = schema.get("type") {
if type_arr.len() == 2 && type_arr.contains(&Value::String("null".into())) {
if let Some(non_null_type) = type_arr
.iter()
.find(|t| *t != &Value::String("null".into()))
{
let mut non_null_schema = schema.clone();
if let Some(obj) = non_null_schema.as_object_mut() {
obj.insert("type".to_string(), non_null_type.clone());
(true, non_null_schema)
} else {
(false, schema.clone())
}
} else {
(false, schema.clone())
}
} else {
(false, schema.clone())
}
} else {
(false, schema.clone())
}
};
let (existing_nullable, existing_inner) = extract_nullable_info(existing);
let (new_nullable, new_inner) = extract_nullable_info(new);
if existing_inner == new_inner {
if existing_nullable || new_nullable {
let mut nullable_schema = existing_inner.clone();
if let Some(inner_type) = existing_inner.get("type") {
if let Some(obj) = nullable_schema.as_object_mut() {
obj.insert("type".to_string(), json!(["null", inner_type]));
}
}
return Some(nullable_schema);
} else {
return Some(existing_inner);
}
}
None
}
fn is_scalar_schema(schema: &Value) -> bool {
if let Value::Array(arr) = schema {
if arr.len() == 2 && arr.contains(&Value::String("null".to_string())) {
let inner_schema = arr
.iter()
.find(|v| *v != &Value::String("null".to_string()))
.unwrap();
return is_scalar_schema(inner_schema); }
}
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
return matches!(type_str, "string" | "number" | "integer" | "boolean");
}
if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
let non_null_type = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str());
return matches!(
non_null_type,
Some("string" | "number" | "integer" | "boolean")
);
}
}
}
false
}
fn is_object_schema(schema: &Value) -> bool {
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
return type_str == "object" && schema.get("properties").is_some();
}
if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
let non_null_type = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str());
return non_null_type == Some("object") && schema.get("properties").is_some();
}
}
}
false
}
fn is_map_schema(schema: &Value) -> bool {
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
return type_str == "object" && schema.get("additionalProperties").is_some();
}
if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
let non_null_type = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str());
return non_null_type == Some("object")
&& schema.get("additionalProperties").is_some();
}
}
}
false
}
fn is_array_schema(schema: &Value) -> bool {
if let Value::Array(arr) = schema {
if arr.len() == 2 && arr.contains(&Value::String("null".to_string())) {
let inner_schema = arr
.iter()
.find(|v| *v != &Value::String("null".to_string()))
.unwrap();
return is_array_schema(inner_schema); }
}
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
return type_str == "array";
}
if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
let non_null_type = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str());
return non_null_type == Some("array");
}
}
}
false
}
fn get_scalar_type_name(schema: &Value) -> Option<String> {
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
if matches!(type_str, "string" | "number" | "integer" | "boolean") {
return Some(type_str.to_string());
}
}
if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
let non_null_type = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str());
if matches!(
non_null_type,
Some("string" | "number" | "integer" | "boolean")
) {
return non_null_type.map(|s| s.to_string());
}
}
}
}
None
}
fn try_scalar_promotion(
object_schema: &Value,
scalar_schema: &Value,
field_name: &str,
scalar_side: &str,
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
let Some(scalar_type) = get_scalar_type_name(scalar_schema) else {
debug!(config, "Cannot determine scalar type for promotion");
return None;
};
let wrapped_key = make_promoted_scalar_key(field_name, &scalar_type);
debug!(
config,
"Promoting scalar on {} side: wrapping {} into object under key `{}`",
scalar_side,
scalar_type,
wrapped_key
);
let mut wrapped_props = Map::new();
wrapped_props.insert(wrapped_key, scalar_schema.clone());
let promoted = json!({
"type": "object",
"properties": wrapped_props
});
check_unifiable_schemas(
&[object_schema.clone(), promoted],
&format!("{path}.{}", field_name),
config,
)
}
fn extract_field_from_nullable_schema<'a>(
schema: &'a Value,
field_name: &str,
) -> Option<&'a Value> {
if let Value::Array(arr) = schema {
if arr.len() == 2 && arr.contains(&Value::String("null".to_string())) {
let inner_schema = arr
.iter()
.find(|v| *v != &Value::String("null".to_string()))?;
return extract_field_from_nullable_schema(inner_schema, field_name);
}
}
if let Some(Value::Array(type_arr)) = schema.get("type") {
if type_arr.len() == 2 && type_arr.contains(&Value::String("null".into())) {
return schema.get(field_name);
}
}
schema.get(field_name)
}
fn unify_array_schemas(
schemas: &[Value],
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
debug!(
config,
"{}: Attempting to unify {} array schemas",
path,
schemas.len()
);
if schemas.is_empty() {
return None;
}
let mut items_schemas = Vec::new();
for (i, schema) in schemas.iter().enumerate() {
if let Some(items) = extract_field_from_nullable_schema(schema, "items") {
debug_verbose!(
config,
"{}: Array schema[{}] items: {}",
path,
i,
serde_json::to_string(items).unwrap_or_default()
);
items_schemas.push(items.clone());
} else {
debug!(config, "{}: Array schema[{}] missing items", path, i);
return None;
}
}
if let Some(unified_items) =
check_unifiable_schemas(&items_schemas, &format!("{}.items", path), config)
{
debug!(config, "{}: Successfully unified array items", path);
Some(json!({
"type": "array",
"items": unified_items
}))
} else {
debug!(config, "{}: Failed to unify array items", path);
None
}
}
fn unify_scalar_schemas(
schemas: &[Value],
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
if schemas.is_empty() {
debug!(config, "Empty schema at {}", path);
return None;
}
let mut base_types = std::collections::HashSet::new();
for schema in schemas {
if let Some(type_val) = schema.get("type") {
if let Some(type_str) = type_val.as_str() {
base_types.insert(type_str.to_string());
} else if let Some(arr) = type_val.as_array() {
if arr.len() == 2 && arr.contains(&Value::String("null".into())) {
if let Some(non_null_type) = arr
.iter()
.find(|t| *t != &Value::String("null".into()))
.and_then(|t| t.as_str())
{
base_types.insert(non_null_type.to_string());
}
}
}
}
}
if base_types.len() == 1 {
let base_type = base_types.iter().next().unwrap();
debug!(
config,
"{}: Unified scalar schemas to nullable {}", path, base_type
);
return Some(json!({"type": ["null", base_type]}));
}
if config.debug {
let mut sorted_types: Vec<_> = base_types.into_iter().collect();
sorted_types.sort();
debug!(
config,
"{}: Cannot unify incompatible scalar types: {:?}", path, sorted_types
);
}
None
}
fn unify_map_schemas(
schemas: &[Value],
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
debug!(
config,
"{}: Attempting to unify {} map schemas",
path,
schemas.len()
);
if schemas.is_empty() {
return None;
}
let mut additional_props_schemas = Vec::new();
for (i, schema) in schemas.iter().enumerate() {
if let Some(additional_props) =
extract_field_from_nullable_schema(schema, "additionalProperties")
{
debug_verbose!(
config,
"{}: Map schema[{}] additionalProperties: {}",
path,
i,
serde_json::to_string(additional_props).unwrap_or_default()
);
additional_props_schemas.push(additional_props.clone());
} else {
debug!(
config,
"{}: Map schema[{}] missing additionalProperties", path, i
);
return None;
}
}
if let Some(unified_additional_props) = check_unifiable_schemas(
&additional_props_schemas,
&format!("{}.additionalProperties", path),
config,
) {
debug!(
config,
"{}: Successfully unified map additionalProperties", path
);
Some(json!({
"type": "object",
"additionalProperties": unified_additional_props
}))
} else {
debug!(config, "{}: Failed to unify map additionalProperties", path);
None
}
}
fn unify_record_schemas(
schemas: &[Value],
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
debug!(
config,
"{}: Attempting to unify {} record schemas",
path,
schemas.len()
);
let mut all_fields = ordermap::OrderMap::new();
let mut field_counts = std::collections::HashMap::new();
for (i, schema) in schemas.iter().enumerate() {
if let Some(Value::Object(props)) = extract_field_from_nullable_schema(schema, "properties")
{
for (field_name, field_schema) in props {
*field_counts.entry(field_name.clone()).or_insert(0) += 1;
match all_fields.entry(field_name.clone()) {
ordermap::map::Entry::Vacant(e) => {
debug_verbose!(config, "Schema[{i}] introduces new field `{field_name}`");
e.insert(normalise_nullable(field_schema).clone());
}
ordermap::map::Entry::Occupied(mut e) => {
let existing = normalise_nullable(e.get()).clone();
let new = normalise_nullable(field_schema).clone();
let existing = if existing.get("anyOf").is_some() {
if let Some(Value::Array(any_of_schemas)) = existing.get("anyOf") {
if let Some(unified) =
unify_anyof_schemas(any_of_schemas, field_name, config)
{
unified
} else {
existing
}
} else {
existing
}
} else {
existing
};
let new = if new.get("anyOf").is_some() {
if let Some(Value::Array(any_of_schemas)) = new.get("anyOf") {
if let Some(unified) =
unify_anyof_schemas(any_of_schemas, field_name, config)
{
unified
} else {
new
}
} else {
new
}
} else {
new
};
if let Some(compatible_schema) = schemas_compatible(&existing, &new) {
debug_verbose!(config, "Field `{field_name}` compatible (nullable/non-nullable unification)");
e.insert(compatible_schema);
} else if is_array_schema(&existing) && is_array_schema(&new) {
debug!(config,
"Field `{field_name}` has conflicting array schemas, attempting recursive unify"
);
if let Some(unified) = check_unifiable_schemas(
&[existing.clone(), new.clone()],
&format!("{path}.{}", field_name),
config,
) {
debug!(
config,
"Field `{field_name}` unified successfully after array recursion"
);
e.insert(unified);
} else {
debug!(config, "{path}.{}: failed to unify arrays", field_name);
return None;
}
} else if is_object_schema(&existing) && is_object_schema(&new) {
debug!(config,
"Field `{field_name}` has conflicting object schemas, attempting recursive unify"
);
if let Some(unified) = check_unifiable_schemas(
&[existing.clone(), new.clone()],
&format!("{path}.{}", field_name),
config,
) {
debug!(
config,
"Field `{field_name}` unified successfully after recursion"
);
e.insert(unified);
} else {
debug!(config, "{path}.{}: failed to unify", field_name);
return None;
}
} else if config.wrap_scalars {
let existing_is_obj = is_object_schema(&existing);
let existing_is_scalar = is_scalar_schema(&existing);
let new_is_obj = is_object_schema(&new);
let new_is_scalar = is_scalar_schema(&new);
if existing_is_obj && new_is_scalar {
if let Some(unified) = try_scalar_promotion(
&existing, &new, field_name, "new", path, config,
) {
debug!(config, "Field `{field_name}` unified successfully after scalar promotion");
e.insert(unified);
continue;
}
} else if new_is_obj && existing_is_scalar {
if let Some(unified) = try_scalar_promotion(
&new, &existing, field_name, "existing", path, config,
) {
debug!(config, "Field `{field_name}` unified successfully after scalar promotion");
e.insert(unified);
continue;
}
} else if existing_is_scalar && new_is_scalar {
debug!(config, "Field `{field_name}` has scalar vs scalar conflict, attempting promotion");
if let Some(unified) = try_mixed_scalar_promotion(
&existing, &new, field_name, path, config,
) {
debug!(config, "Field `{field_name}` unified successfully after mixed scalar promotion");
e.insert(unified);
continue;
}
}
debug!(config,
"{path}.{field_name}: incompatible types (promotion failed):\n existing={:#?}\n new={:#?}",
existing, new
);
return None;
} else {
debug!(config,
"{path}.{field_name}: incompatible types:\n existing={:#?}\n new={:#?}",
existing, new
);
return None; }
}
}
}
} else {
debug!(config, "Schema[{i}] has no properties object");
return None;
}
}
let total_schemas = schemas.len();
let mut unified_properties = Map::new();
let mut required_fields = Vec::new();
for (field_name, field_type) in &all_fields {
let count = field_counts.get(field_name).unwrap_or(&0);
if *count == total_schemas {
debug_verbose!(
config,
"Field `{field_name}` present in all schemas → keeping non-nullable"
);
unified_properties.insert(field_name.clone(), field_type.clone());
required_fields.push(field_name.clone()); }
}
for (field_name, field_type) in &all_fields {
let count = field_counts.get(field_name).unwrap_or(&0);
if *count < total_schemas {
debug_verbose!(
config,
"Field `{field_name}` missing in {}/{} schemas → making nullable",
total_schemas - count,
total_schemas
);
if let Some(type_str) = field_type.get("type").and_then(|t| t.as_str()) {
if type_str == "null" {
unified_properties.insert(field_name.clone(), field_type.clone());
} else {
let mut nullable_field = field_type.clone();
nullable_field["type"] = json!(["null", type_str]);
unified_properties.insert(field_name.clone(), nullable_field);
}
} else if let Some(_type_arr) = field_type.get("type").and_then(|t| t.as_array()) {
unified_properties.insert(field_name.clone(), field_type.clone());
} else {
let nullable_schema = json!({
"anyOf": [
{"type": "null"},
field_type
]
});
unified_properties.insert(field_name.clone(), nullable_schema);
}
}
}
debug!(config, "{}: Record schemas unified successfully", path);
let mut result = json!({
"type": "object",
"properties": unified_properties
});
if !required_fields.is_empty() {
result["required"] = json!(required_fields);
}
Some(result)
}
fn try_mixed_scalar_promotion(
existing: &Value,
new: &Value,
field_name: &str,
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
let existing_type = get_scalar_type_name(existing)?;
let new_type = get_scalar_type_name(new)?;
if existing_type == new_type {
return None;
}
debug!(
config,
"{}: Promoting mixed scalars {} and {} for field '{}'",
path,
existing_type,
new_type,
field_name
);
let existing_key = make_promoted_scalar_key(field_name, &existing_type);
let new_key = make_promoted_scalar_key(field_name, &new_type);
let mut properties = Map::new();
properties.insert(existing_key.clone(), existing.clone());
properties.insert(new_key.clone(), new.clone());
let promoted = json!({
"type": "object",
"properties": properties
});
Some(promoted)
}
pub(crate) fn unify_anyof_schemas(
schemas: &[Value],
field_name: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
if !config.wrap_scalars {
return None;
}
let has_scalars = schemas.iter().any(is_scalar_schema);
let has_objects = schemas.iter().any(is_object_schema);
if !has_scalars || !has_objects {
return None; }
debug!(
config,
"anyOf unification: promoting scalars for field '{}'", field_name
);
let mut promoted_schemas = Vec::new();
for schema in schemas {
if is_scalar_schema(schema) {
if let Some(scalar_type) = get_scalar_type_name(schema) {
let wrapped_key = make_promoted_scalar_key(field_name, &scalar_type);
let promoted = json!({
"type": "object",
"properties": {
wrapped_key: schema.clone()
}
});
promoted_schemas.push(promoted);
} else {
return None;
}
} else {
promoted_schemas.push(schema.clone());
}
}
check_unifiable_schemas(&promoted_schemas, field_name, config)
}
pub(crate) fn check_unifiable_schemas(
schemas: &[Value],
path: &str,
config: &SchemaInferenceConfig,
) -> Option<Value> {
debug_verbose!(
config,
"=== check_unifiable_schemas called with path='{}' and {} schemas:",
path,
schemas.len()
);
for (i, schema) in schemas.iter().enumerate() {
debug_verbose!(
config,
" Schema[{}]: {}",
i,
serde_json::to_string(schema).unwrap_or_default()
);
}
if schemas.is_empty() {
debug!(config, "{path}: failed (empty schema list)");
return None;
}
if schemas.iter().all(is_array_schema) {
debug!(
config,
"{}: All schemas are arrays, attempting array unification", path
);
return unify_array_schemas(schemas, path, config);
}
if schemas.iter().all(is_map_schema) {
debug!(
config,
"{}: All schemas are maps, attempting map unification", path
);
return unify_map_schemas(schemas, path, config);
}
if schemas.iter().all(is_object_schema) {
debug!(
config,
"{}: All schemas are records, attempting record unification", path
);
return unify_record_schemas(schemas, path, config);
}
if schemas.iter().all(is_scalar_schema) {
debug!(
config,
"{}: All schemas are scalars, attempting scalar unification", path
);
return unify_scalar_schemas(schemas, path, config);
}
debug!(
config,
"{}: Mixed schema types not supported for unification", path
);
for (i, schema) in schemas.iter().enumerate() {
let schema_type = if is_array_schema(schema) {
"array"
} else if is_map_schema(schema) {
"map"
} else if is_object_schema(schema) {
"record"
} else if is_scalar_schema(schema) {
"scalar"
} else {
"unknown"
};
debug!(
config,
" Schema[{}] type: {} - {}",
i,
schema_type,
serde_json::to_string(schema).unwrap_or_default()
);
}
None
}
#[cfg(test)]
mod tests {
include!("../../tests/unification.rs");
}