chore: clean up

Author: Eshanatnight

server/src/catalog.rs

@@ -230,8 +230,7 @@ async fn create_manifest(
                 .ok_or(IOError::new(
                     ErrorKind::Other,
                     "Failed to create upper bound for manifest",
-                ))
-                .map_err(ObjectStorageError::IoError)?,
+                ))?,
         )
         .and_utc();
 

server/src/event/writer/file_writer.rs

@@ -49,10 +49,7 @@ impl FileWriter {
     ) -> Result<(), StreamWriterError> {
         match self.get_mut(schema_key) {
             Some(writer) => {
-                writer
-                    .writer
-                    .write(record)
-                    .map_err(StreamWriterError::Writer)?;
+                writer.writer.write(record)?;
             }
             // entry is not present thus we create it
             None => {
@@ -100,8 +97,6 @@ fn init_new_stream_writer_file(
     let mut stream_writer = StreamWriter::try_new(file, &record.schema())
         .expect("File and RecordBatch both are checked");
 
-    stream_writer
-        .write(record)
-        .map_err(StreamWriterError::Writer)?;
+    stream_writer.write(record)?;
     Ok((path, stream_writer))
 }

server/src/handlers/http/cluster/mod.rs

@@ -456,12 +456,12 @@ pub async fn get_cluster_metrics() -> Result<impl Responder, PostError> {
             .await;
 
         if let Ok(res) = res {
-            let text = res.text().await.map_err(PostError::NetworkError)?;
+            let text = res.text().await?;
             let lines: Vec<Result<String, std::io::Error>> =
                 text.lines().map(|line| Ok(line.to_owned())).collect_vec();
 
             let sample = prometheus_parse::Scrape::parse(lines.into_iter())
-                .map_err(|err| PostError::CustomError(err.to_string()))?
+                .map_err(PostError::Error)?
                 .samples;
 
             dresses.push(Metrics::from_prometheus_samples(

server/src/handlers/http/ingest.rs

@@ -124,7 +124,7 @@ async fn push_logs(stream_name: String, req: HttpRequest, body: Bytes) -> Result
     let object_store_format = glob_storage
         .get_object_store_format(&stream_name)
         .await
-        .map_err(|_err| PostError::StreamNotFound(stream_name.clone()))?;
+        .map_err(|_| PostError::StreamNotFound(stream_name.clone()))?;
 
     let time_partition = object_store_format.time_partition;
     let time_partition_limit = object_store_format.time_partition_limit;
@@ -387,6 +387,9 @@ pub enum PostError {
     #[error("{0}")]
     CreateStream(#[from] CreateStreamError),
     #[error("Error: {0}")]
+    Error(std::io::Error),
+    #[allow(unused)]
+    #[error("Error: {0}")]
     CustomError(String),
     #[error("Error: {0}")]
     NetworkError(#[from] reqwest::Error),
@@ -415,6 +418,7 @@ impl actix_web::ResponseError for PostError {
             PostError::ObjectStorageError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::DashboardError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::FiltersError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            PostError::Error(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 

server/src/handlers/http/query.rs

@@ -53,7 +53,7 @@ use crate::utils::actix::extract_session_key_from_req;
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct Query {
-    pub pub query: String,
+    pub query: String,
     pub start_time: String,
     pub end_time: String,
     #[serde(default)]

server/src/handlers/http/users/dashboards.rs

@@ -98,7 +98,7 @@ pub async fn delete(req: HttpRequest) -> Result<HttpResponse, PostError> {
 //     version: String,
 //     name: String,
 //     id: String,
-//     time-filter: `type_not_defined`
+//     time_filter: TimeFilter
 //     refresh_interval: u64,
 //     pannels: Vec<Pannel>,
 // }
@@ -112,6 +112,12 @@ pub async fn delete(req: HttpRequest) -> Result<HttpResponse, PostError> {
 //     headers: Vec<String>,
 //     dimensions: (u64, u64),
 // }
+//
+// #[derive(Debug, Serialize, Deserialize)]
+// pub struct TimeFilter {
+//      to: String,
+//      from: String
+// }
 
 #[derive(Debug, thiserror::Error)]
 pub enum DashboardError {

server/src/handlers/http/users/filters.rs

@@ -161,5 +161,5 @@ impl actix_web::ResponseError for FiltersError {
 //     stream_name: String,
 //     filter_name: String,
 //     query: String,
-//     time-filter: `type_not_defined`
+//     time-filter: TimeFilter
 // }

server/src/query/listing_table_builder.rs

@@ -167,7 +167,6 @@ impl ListingTableBuilder {
             })
             .try_collect()
             .await
-            // TODO: make the err map better
             .map_err(|err| DataFusionError::External(Box::new(err)))?;
 
         let mut res = res.into_iter().flatten().collect_vec();

server/src/query/stream_schema_provider.rs

@@ -180,8 +180,8 @@ async fn collect_from_snapshot(
             .map(|item| item.manifest_path)
             .collect(),
     )
-    .await
-    .map_err(DataFusionError::ObjectStore)?;
+    .await?;
+
     let mut manifest_files: Vec<_> = manifest_files
         .into_iter()
         .flat_map(|file| file.files)

server/src/response.rs

@@ -33,8 +33,8 @@ impl QueryResponse {
     pub fn to_http(&self) -> Result<impl Responder, QueryError> {
         log::info!("{}", "Returning query results");
         let records: Vec<&RecordBatch> = self.records.iter().collect();
-        let mut json_records = record_batches_to_json(&records)
-            .map_err(|err| QueryError::JsonParse(err.to_string()))?;
+        let mut json_records = record_batches_to_json(&records)?;
+
         if self.fill_null {
             for map in &mut json_records {
                 for field in &self.fields {