feldera · mihaibudiu · Jul 15, 2026 · Jul 14, 2026
diff --git 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<String>,
@@ -497,10 +503,26 @@ impl AvroSchemaBuilder {
         top_level: bool,
     ) -> Result<RecordSchema, String> {
         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"
+            );
+        }
+    }
+}