diff --git a/crates/arroyo-rpc/src/api_types/connections.rs b/crates/arroyo-rpc/src/api_types/connections.rs index 6342cc83..e4e740fc 100644 --- a/crates/arroyo-rpc/src/api_types/connections.rs +++ b/crates/arroyo-rpc/src/api_types/connections.rs @@ -2,9 +2,10 @@ use crate::MetadataField; use crate::df::{ArroyoSchema, ArroyoSchemaRef}; use crate::formats::{BadData, Format, Framing}; use ahash::HashSet; -use anyhow::bail; +use anyhow::{anyhow, bail}; use arrow_schema::{DataType, Field, Fields, TimeUnit}; use arroyo_types::ArroyoExtensionType; +use datafusion::common::DFSchema; use serde::{Deserialize, Serialize, Serializer}; use std::collections::{BTreeMap, HashMap}; use std::fmt::{Display, Formatter}; @@ -420,6 +421,33 @@ impl TryFrom for SourceField { } } +fn validate_unique_field_names(fields: &[SourceField], path: &str) -> anyhow::Result<()> { + let arrow_fields = fields + .iter() + .cloned() + .map(Field::from) + .collect::>() + .into(); + DFSchema::from_unqualified_fields(arrow_fields, HashMap::new()) + .map_err(|e| anyhow!("invalid field names in {path}: {e}"))?; + + for field in fields { + validate_unique_nested_field_names(&field.field_type, &format!("{path}.{}", field.name))?; + } + + Ok(()) +} + +fn validate_unique_nested_field_names(field_type: &FieldType, path: &str) -> anyhow::Result<()> { + match field_type { + FieldType::Struct(StructField { fields }) => validate_unique_field_names(fields, path), + FieldType::List(ListField { items }) => { + validate_unique_nested_field_names(&items.field_type, &format!("{path}[]")) + } + _ => Ok(()), + } +} + #[derive(Serialize, Deserialize, Clone, Debug, ToSchema, PartialEq)] #[serde(rename_all = "snake_case", tag = "type")] pub enum SchemaDefinition { @@ -478,6 +506,8 @@ impl ConnectionSchema { } pub fn validate(self) -> anyhow::Result { + validate_unique_field_names(&self.fields, "schema")?; + let non_metadata_fields: Vec<_> = self .fields .iter() @@ -635,6 +665,50 @@ mod tests { ArrowField::new(name, dt, nullable) } + fn source_field(name: &str, field_type: FieldType) -> SourceField { + SourceField { + name: name.to_string(), + field_type, + required: false, + sql_name: None, + metadata_key: None, + } + } + + #[test] + fn connection_schema_rejects_duplicate_field_names_in_list_structs() { + let schema = ConnectionSchema { + format: None, + bad_data: None, + framing: None, + fields: vec![source_field( + "events", + FieldType::List(ListField { + items: Box::new(ListFieldItem { + name: "item".to_string(), + field_type: FieldType::Struct(StructField { + fields: vec![ + source_field("id", FieldType::Int64), + source_field("id", FieldType::String), + ], + }), + required: false, + sql_name: None, + }), + }), + )], + definition: None, + inferred: None, + primary_keys: HashSet::default(), + }; + + let err = schema.validate().unwrap_err(); + assert_eq!( + err.to_string(), + "invalid field names in schema.events[]: Schema error: Schema contains duplicate unqualified field name id" + ); + } + #[test] fn sql_type_struct_and_list() { let st = FieldType::Struct(StructField {