diff --git a/crates/adapters/src/format/avro/schema.rs b/crates/adapters/src/format/avro/schema.rs index d8092395346..75ca70ca440 100644 --- a/crates/adapters/src/format/avro/schema.rs +++ b/crates/adapters/src/format/avro/schema.rs @@ -458,6 +458,12 @@ pub fn is_valid_avro_identifier(ident: &str) -> bool { && ident.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') } +/// A valid Avro fullname is a sequence of dot-separated identifiers, such as +/// `namespace.record_name`. +pub fn is_valid_avro_fullname(fullname: &str) -> bool { + fullname.split('.').all(is_valid_avro_identifier) +} + #[derive(Default)] pub struct AvroSchemaBuilder { namespace: Option, @@ -497,10 +503,26 @@ impl AvroSchemaBuilder { top_level: bool, ) -> Result { let name = name.name(); - if !is_valid_avro_identifier(&name) { - return Err(format!("'{name}' is not a valid Avro identifier")); + if !is_valid_avro_fullname(&name) { + return Err(format!( + "'{name}' is not a valid Avro name: each dot-separated segment must start with a letter or underscore and contain only letters, digits, and underscores" + )); } + // Names such as 'view.key', which the catalog assigns to the key + // schema of an indexed view, are Avro fullnames: the segments before + // the last dot form a namespace, appended to the configured one. + let (namespace, name) = match name.rsplit_once('.') { + Some((prefix, base)) => ( + Some(match &self.namespace { + Some(ns) => format!("{ns}.{prefix}"), + None => prefix.to_string(), + }), + base.to_string(), + ), + None => (self.namespace.clone(), name), + }; + let mut fields = Vec::with_capacity(struct_fields.len()); let mut lookup = BTreeMap::new(); @@ -513,10 +535,7 @@ impl AvroSchemaBuilder { } Ok(RecordSchema { - name: Name { - name: name.to_string(), - namespace: self.namespace.clone(), - }, + name: Name { name, namespace }, aliases: None, doc: None, fields, @@ -647,3 +666,90 @@ impl AvroSchemaBuilder { }) } } + +#[cfg(test)] +mod test { + use super::{AvroSchemaBuilder, is_valid_avro_fullname, schema_json}; + use apache_avro::{Schema as AvroSchema, schema::Name}; + use feldera_types::program_schema::{ColumnType, Field, Relation, SqlIdentifier}; + use std::collections::BTreeMap; + + fn relation(name: &str) -> Relation { + Relation::new( + SqlIdentifier::new(name, false), + vec![Field::new( + SqlIdentifier::new("id", false), + ColumnType::boolean(false), + )], + false, + BTreeMap::new(), + ) + } + + fn record_name(schema: &AvroSchema) -> Name { + match schema { + AvroSchema::Record(record) => record.name.clone(), + _ => panic!("expected a record schema, got {schema:?}"), + } + } + + #[test] + fn fullname_validation() { + for valid in ["key", "v.key", "_a1.b2.c3"] { + assert!(is_valid_avro_fullname(valid), "'{valid}' should be valid"); + } + for invalid in ["", ".", "v..key", ".key", "key.", "1v.key", "v-1.key"] { + assert!( + !is_valid_avro_fullname(invalid), + "'{invalid}' should be invalid" + ); + } + } + + #[test] + fn dotted_relation_name_becomes_namespace() { + let schema = AvroSchemaBuilder::new() + .relation_to_avro_schema(&relation("v.key")) + .unwrap(); + let name = record_name(&schema); + assert_eq!(name.name, "key"); + assert_eq!(name.namespace.as_deref(), Some("v")); + + // The apache_avro parser must accept the generated schema. + AvroSchema::parse_str(&schema_json(&schema)).unwrap(); + } + + #[test] + fn configured_namespace_prefixes_derived_namespace() { + let schema = AvroSchemaBuilder::new() + .with_namespace(Some("com.acme")) + .relation_to_avro_schema(&relation("v.key")) + .unwrap(); + let name = record_name(&schema); + assert_eq!(name.name, "key"); + assert_eq!(name.namespace.as_deref(), Some("com.acme.v")); + } + + #[test] + fn plain_relation_name_keeps_configured_namespace() { + let schema = AvroSchemaBuilder::new() + .with_namespace(Some("com.acme")) + .relation_to_avro_schema(&relation("v")) + .unwrap(); + let name = record_name(&schema); + assert_eq!(name.name, "v"); + assert_eq!(name.namespace.as_deref(), Some("com.acme")); + } + + #[test] + fn invalid_relation_name_is_rejected() { + for invalid in ["v..key", ".key", "key.", "v-1"] { + assert!( + AvroSchemaBuilder::new() + .relation_to_avro_schema(&relation(invalid)) + .is_err(), + "'{invalid}' should be rejected" + ); + } + } +}