enhancement: stream name validation (#921)

removed existing stream name validation
added validation to allow below special characters -
Hyphen (-)
Underscore (_)
fixes: #917

Author: nikhilsinhaparseable

Cargo.lock

@@ -34,7 +34,7 @@ mime = "0.3.17"
 ### other dependencies
 anyhow = { version = "1.0", features = ["backtrace"] }
 argon2 = "0.5.0"
-async-trait = "0.1"
+async-trait = "0.1.82"
 base64 = "0.22.0"
 lazy_static = "1.4"
 bytes = "1.4"
@@ -71,7 +71,7 @@ prometheus = { version = "0.13", features = ["process"] }
 rand = "0.8"
 regex = "1.7.3"
 relative-path = { version = "1.7", features = ["serde"] }
-reqwest = { version = "0.11.27", default_features = false, features = [
+reqwest = { version = "0.11.27", default-features = false, features = [
   "rustls-tls",
   "json",
 ] }         # cannot update cause rustls is not latest `see rustls`
@@ -113,7 +113,7 @@ sha1_smol = { version = "1.0", features = ["std"] }
 static-files = "0.2"
 ureq = "2.6"
 vergen = { version = "8.1", features = ["build", "git", "cargo", "gitcl"] }
-zip = { version = "1.1.1", default_features = false, features = ["deflate"] }
+zip = { version = "1.1.1", default-features = false, features = ["deflate"] }
 url = "2.4.0"
 prost-build = "0.12.3"
 

server/Cargo.toml

@@ -322,7 +322,7 @@ impl S3 {
             stream_json_check.push(task);
         }
 
-        stream_json_check.try_collect().await?;
+        stream_json_check.try_collect::<()>().await?;
 
         Ok(dirs.into_iter().map(|name| LogStream { name }).collect())
     }
@@ -633,7 +633,7 @@ impl ObjectStorage for S3 {
             stream_json_check.push(task);
         }
 
-        stream_json_check.try_collect().await?;
+        stream_json_check.try_collect::<()>().await?;
 
         Ok(dirs.into_iter().map(|name| LogStream { name }).collect())
     }

server/src/storage/s3.rs

@@ -98,10 +98,8 @@ pub fn record_batches_to_json(records: &[&RecordBatch]) -> Result<Vec<Map<String
 
     let buf = writer.into_inner();
 
-    let json_rows: Vec<Map<String, Value>> = match serde_json::from_reader(buf.as_slice()) {
-        Ok(json) => json,
-        Err(_) => vec![],
-    };
+    let json_rows: Vec<Map<String, Value>> =
+        serde_json::from_reader(buf.as_slice()).unwrap_or_default();
 
     Ok(json_rows)
 }

server/src/utils/arrow.rs

@@ -31,6 +31,8 @@ const DENIED_NAMES: &[&str] = &[
     "select", "from", "where", "group", "by", "order", "limit", "offset", "join", "and",
 ];
 
+const ALLOWED_SPECIAL_CHARS: &[char] = &['-', '_'];
+
 pub fn alert(alerts: &Alerts) -> Result<(), AlertValidationError> {
     let alert_name: Vec<&str> = alerts.alerts.iter().map(|a| a.name.as_str()).collect();
     let mut alert_name_dedup = alert_name.clone();
@@ -80,34 +82,15 @@ pub fn stream_name(stream_name: &str, stream_type: &str) -> Result<(), StreamNam
         return Err(StreamNameValidationError::EmptyName);
     }
 
-    if stream_name.chars().all(char::is_numeric) {
-        return Err(StreamNameValidationError::NameNumericOnly(
-            stream_name.to_owned(),
-        ));
-    }
-
-    if stream_name.chars().next().unwrap().is_numeric() {
-        return Err(StreamNameValidationError::NameCantStartWithNumber(
-            stream_name.to_owned(),
-        ));
-    }
-
     for c in stream_name.chars() {
         match c {
             ' ' => {
                 return Err(StreamNameValidationError::NameWhiteSpace(
                     stream_name.to_owned(),
                 ))
             }
-            c if !c.is_alphanumeric() => {
-                return Err(StreamNameValidationError::NameSpecialChar(
-                    stream_name.to_owned(),
-                ))
-            }
-            c if c.is_ascii_uppercase() => {
-                return Err(StreamNameValidationError::NameUpperCase(
-                    stream_name.to_owned(),
-                ))
+            c if !c.is_alphanumeric() && !ALLOWED_SPECIAL_CHARS.contains(&c) => {
+                return Err(StreamNameValidationError::NameSpecialChar { c })
             }
             _ => {}
         }
@@ -182,16 +165,10 @@ pub mod error {
     pub enum StreamNameValidationError {
         #[error("Stream name cannot be empty")]
         EmptyName,
-        #[error("Invalid stream name with numeric values only")]
-        NameNumericOnly(String),
-        #[error("Stream name cannot start with a number")]
-        NameCantStartWithNumber(String),
         #[error("Stream name cannot contain whitespace")]
         NameWhiteSpace(String),
-        #[error("Stream name cannot contain special characters")]
-        NameSpecialChar(String),
-        #[error("Uppercase character in stream name")]
-        NameUpperCase(String),
+        #[error("Stream name cannot contain special character: {c}")]
+        NameSpecialChar { c: char },
         #[error("SQL keyword cannot be used as stream name")]
         SQLKeyword(String),
         #[error("The stream {0} is reserved for internal use and cannot be used for user defined streams")]