fix: flatten before detect (#1361)

This PR ensures the detect schema API used 
for detecting the schema of a given JSON file
returns the exact, flattened schema as server
would store the event.

Author: nikhilsinhaparseable

build.rs

@@ -151,7 +151,7 @@ mod ui {
                 .expect("has segemnts")
                 .find(|v| v.starts_with('v'))
                 .expect("version segement");
-            println!("cargo:rustc-env=UI_VERSION={}", ui_version);
+            println!("cargo:rustc-env=UI_VERSION={ui_version}");
         }
 
         Ok(())

src/alerts/alerts_utils.rs

@@ -88,7 +88,7 @@ async fn execute_base_query(
     original_query: &str,
 ) -> Result<DataFrame, AlertError> {
     let stream_name = query.first_table_name().ok_or_else(|| {
-        AlertError::CustomError(format!("Table name not found in query- {}", original_query))
+        AlertError::CustomError(format!("Table name not found in query- {original_query}"))
     })?;
 
     let time_partition = PARSEABLE.get_stream(&stream_name)?.get_time_partition();
@@ -412,7 +412,7 @@ pub fn get_filter_string(where_clause: &Conditions) -> Result<String, String> {
                                 let value = match NumberOrString::from_string(value.to_owned()) {
                                     NumberOrString::Number(val) => format!("{val}"),
                                     NumberOrString::String(val) => {
-                                        format!("'{}'", val)
+                                        format!("'{val}'")
                                     }
                                 };
                                 format!("{} {}", condition.operator, value)
@@ -456,23 +456,23 @@ fn match_alert_operator(expr: &ConditionConfig) -> Expr {
             WhereConfigOperator::BeginsWith => Expr::BinaryExpr(BinaryExpr::new(
                 Box::new(col(column)),
                 Operator::RegexIMatch,
-                Box::new(lit(format!("^{}", value))),
+                Box::new(lit(format!("^{value}"))),
             )),
             WhereConfigOperator::EndsWith => Expr::BinaryExpr(BinaryExpr::new(
                 Box::new(col(column)),
                 Operator::RegexIMatch,
-                Box::new(lit(format!("{}$", value))),
+                Box::new(lit(format!("{value}$"))),
             )),
             WhereConfigOperator::DoesNotContain => col(column).not_ilike(lit(value)),
             WhereConfigOperator::DoesNotBeginWith => Expr::BinaryExpr(BinaryExpr::new(
                 Box::new(col(column)),
                 Operator::RegexNotIMatch,
-                Box::new(lit(format!("^{}", value))),
+                Box::new(lit(format!("^{value}"))),
             )),
             WhereConfigOperator::DoesNotEndWith => Expr::BinaryExpr(BinaryExpr::new(
                 Box::new(col(column)),
                 Operator::RegexNotIMatch,
-                Box::new(lit(format!("{}$", value))),
+                Box::new(lit(format!("{value}$"))),
             )),
             _ => unreachable!("value must not be null for operators other than `is null` and `is not null`. Should've been caught in validation")
         }

src/alerts/mod.rs

@@ -365,7 +365,7 @@ impl Conditions {
                         format!("{} {}", expr2.column, expr2.operator)
                     };
 
-                    format!("[{} {op} {}]", expr1_msg, expr2_msg)
+                    format!("[{expr1_msg} {op} {expr2_msg}]")
                 }
             },
             None => {

src/catalog/mod.rs

@@ -459,8 +459,8 @@ pub fn partition_path(
     let lower = lower_bound.date_naive().format("%Y-%m-%d").to_string();
     let upper = upper_bound.date_naive().format("%Y-%m-%d").to_string();
     if lower == upper {
-        RelativePathBuf::from_iter([stream, &format!("date={}", lower)])
+        RelativePathBuf::from_iter([stream, &format!("date={lower}")])
     } else {
-        RelativePathBuf::from_iter([stream, &format!("date={}:{}", lower, upper)])
+        RelativePathBuf::from_iter([stream, &format!("date={lower}:{upper}")])
     }
 }

src/cli.rs

@@ -563,8 +563,7 @@ impl Options {
         } else {
             if endpoint.starts_with("http") {
                 panic!(
-                    "Invalid value `{}`, please set the environment variable `{}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.",
-                    endpoint, env_var
+                    "Invalid value `{endpoint}`, please set the environment variable `{env_var}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.",
                 );
             }
             endpoint.to_string()
@@ -579,15 +578,14 @@ impl Options {
 
         if addr_parts.len() != 2 {
             panic!(
-                "Invalid value `{}`, please set the environment variable to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.",
-                endpoint
+                "Invalid value `{endpoint}`, please set the environment variable to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details."
             );
         }
 
         let hostname = self.resolve_env_var(addr_parts[0]);
         let port = self.resolve_env_var(addr_parts[1]);
 
-        self.build_url(&format!("{}:{}", hostname, port))
+        self.build_url(&format!("{hostname}:{port}"))
     }
 
     /// resolve the env var
@@ -597,15 +595,13 @@ impl Options {
         if let Some(env_var) = value.strip_prefix('$') {
             let resolved_value = env::var(env_var).unwrap_or_else(|_| {
                 panic!(
-                    "The environment variable `{}` is not set. Please set it to a valid value. Refer to the documentation: https://logg.ing/env for more details.",
-                    env_var
+                    "The environment variable `{env_var}` is not set. Please set it to a valid value. Refer to the documentation: https://logg.ing/env for more details."
                 );
             });
 
             if resolved_value.starts_with("http") {
                 panic!(
-                    "Invalid value `{}`, please set the environment variable `{}` to `<ip address / DNS>` without the scheme (e.g., 192.168.1.1 or example.com). Please refer to the documentation: https://logg.ing/env for more details.",
-                    resolved_value, env_var
+                    "Invalid value `{resolved_value}`, please set the environment variable `{env_var}` to `<ip address / DNS>` without the scheme (e.g., 192.168.1.1 or example.com). Please refer to the documentation: https://logg.ing/env for more details.",
                 );
             }
 
@@ -621,8 +617,7 @@ impl Options {
             .parse::<Url>()
             .unwrap_or_else(|err| {
                 panic!(
-                    "{err}, failed to parse `{}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details.",
-                    address
+                    "{err}, failed to parse `{address}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details."
                 );
             })
     }

src/connectors/common/mod.rs

@@ -95,7 +95,7 @@ impl FromStr for BadData {
             "drop" => Ok(BadData::Drop),
             "fail" => Ok(BadData::Fail),
             "dlt" => Ok(BadData::Dlt),
-            _ => Err(format!("Invalid bad data policy: {}", s)),
+            _ => Err(format!("Invalid bad data policy: {s}")),
         }
     }
 }

src/connectors/kafka/config.rs

@@ -827,7 +827,7 @@ impl std::str::FromStr for SecurityProtocol {
             "SSL" => Ok(SecurityProtocol::Ssl),
             "SASL_SSL" => Ok(SecurityProtocol::SaslSsl),
             "SASL_PLAINTEXT" => Ok(SecurityProtocol::SaslPlaintext),
-            _ => Err(format!("Invalid security protocol: {}", s)),
+            _ => Err(format!("Invalid security protocol: {s}")),
         }
     }
 }
@@ -965,7 +965,7 @@ impl std::str::FromStr for SaslMechanism {
             "SCRAM-SHA-512" => Ok(SaslMechanism::ScramSha512),
             "GSSAPI" => Ok(SaslMechanism::Gssapi),
             "OAUTHBEARER" => Ok(SaslMechanism::OAuthBearer),
-            _ => Err(format!("Invalid SASL mechanism: {}", s)),
+            _ => Err(format!("Invalid SASL mechanism: {s}")),
         }
     }
 }
@@ -985,7 +985,7 @@ impl std::str::FromStr for Acks {
             "0" => Ok(Acks::None),
             "1" => Ok(Acks::Leader),
             "all" => Ok(Acks::All),
-            _ => Err(format!("Invalid acks value: {}", s)),
+            _ => Err(format!("Invalid acks value: {s}")),
         }
     }
 }

src/handlers/airplane.rs

@@ -253,7 +253,7 @@ impl FlightService for AirServiceImpl {
 
         let time = time.elapsed().as_secs_f64();
         QUERY_EXECUTE_TIME
-            .with_label_values(&[&format!("flight-query-{}", stream_name)])
+            .with_label_values(&[&format!("flight-query-{stream_name}")])
             .observe(time);
 
         // Airplane takes off 🛫

src/handlers/http/cluster/mod.rs

@@ -783,13 +783,12 @@ pub async fn remove_node(node_url: Path<String>) -> Result<impl Responder, PostE
 
     if removed_ingestor || removed_indexer || removed_querier || removed_prism {
         return Ok((
-            format!("node {} removed successfully", domain_name),
+            format!("node {domain_name} removed successfully"),
             StatusCode::OK,
         ));
     }
     Err(PostError::Invalid(anyhow::anyhow!(
-        "node {} not found",
-        domain_name
+        "node {domain_name} not found"
     )))
 }
 

src/handlers/http/cluster/utils.rs

@@ -203,5 +203,5 @@ pub fn to_url_string(str: String) -> String {
         return str;
     }
 
-    format!("http://{}/", str)
+    format!("http://{str}/")
 }