Allow cross region PrivateLink connections

src/sql/src/plan/statement/ddl/connection.rs

@@ -9,6 +9,7 @@
 
 //! Data definition language (DDL) utilities for CONNECTION objects.
 
+use regex::Regex;
 use std::collections::{BTreeMap, BTreeSet};
 
 use anyhow::Context;
@@ -262,17 +263,18 @@ impl ConnectionOptionExtracted {
                     let mut duplicate_azs: BTreeSet<String> = BTreeSet::new();
                     // Validate each AZ is supported
                     for connection_az in &connection.availability_zones {
-                        if unique_azs.contains(connection_az) {
-                            duplicate_azs.insert(connection_az.to_string());
-                        } else {
-                            unique_azs.insert(connection_az.to_string());
-                        }
-                        if !supported_azs.contains(connection_az) {
+                        if !is_valid_az_format(connection_az) {
                             return Err(PlanError::InvalidPrivatelinkAvailabilityZone {
                                 name: connection_az.to_string(),
                                 supported_azs,
                             });
                         }
+
+                        if unique_azs.contains(connection_az) {
+                            duplicate_azs.insert(connection_az.to_string());
+                        } else {
+                            unique_azs.insert(connection_az.to_string());
+                        }
                     }
                     if duplicate_azs.len() > 0 {
                         return Err(PlanError::DuplicatePrivatelinkAvailabilityZone {
@@ -737,3 +739,35 @@ fn plan_kafka_security(
 
     Ok((tls, sasl))
 }
+
+fn is_valid_az_format(az: &str) -> bool {
+    let az_pattern = Regex::new(r"^[a-z]{3,4}\d-az[1-6]$").unwrap();
+    az_pattern.is_match(az)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[mz_ore::test]
+    fn test_az_format() {
+        // Valid patterns
+        assert!(is_valid_az_format("use1-az1"));
+        assert!(is_valid_az_format("usw2-az3"));
+        assert!(is_valid_az_format("euw1-az2"));
+        assert!(is_valid_az_format("apse1-az4"));
+        assert!(is_valid_az_format("apne2-az5"));
+        assert!(is_valid_az_format("cac1-az6"));
+
+        // Invalid patterns
+        assert!(!is_valid_az_format("invalid"));
+        assert!(!is_valid_az_format("use1-az7"));
+        assert!(!is_valid_az_format("USE1-az1"));
+        assert!(!is_valid_az_format("use-az1"));
+        assert!(!is_valid_az_format("useast-az1"));
+        assert!(!is_valid_az_format("use1az1"));
+        assert!(!is_valid_az_format("use1-az0"));
+        assert!(!is_valid_az_format(""));
+        assert!(!is_valid_az_format("use1-AZ1"));
+    }
+}