updated comments and styling changes

Author: nikhilsinhaparseable

src/catalog/mod.rs

@@ -450,19 +450,14 @@ pub async fn get_first_event(
 ///   first event timestamp if found, `Ok(None)` if no timestamps are found, or an `ObjectStorageError`
 ///   if an error occurs.
 ///
-/// # Errors
-///
-/// This function will return an error if:
-/// * The stream metadata cannot be retrieved from the storage.
-///
 /// # Examples
 /// ```ignore
 /// ```rust
 /// let result = get_first_event_from_storage("my_stream").await;
 /// match result {
 ///     Ok(Some(first_event)) => println!("first-event-at: {}", first_event),
 ///     Ok(None) => println!("first-event-at not found"),
-///     Err(e) => eprintln!("Error retrieving first-event-at from storage: {}", e),
+///     Err(err) => println!("Error: {:?}", err),
 /// }
 /// ```
 pub async fn get_first_event_from_storage(

src/event/format/json.rs

@@ -94,8 +94,8 @@ impl EventFormat for Event {
         };
 
         if value_arr
-                .iter()
-                .any(|value| fields_mismatch(&schema, value, schema_version))
+            .iter()
+            .any(|value| fields_mismatch(&schema, value, schema_version))
         {
             return Err(anyhow!(
                 "Could not process this event due to mismatch in datatype"

src/event/format/mod.rs

@@ -112,11 +112,8 @@ pub trait EventFormat: Sized {
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
     ) -> Result<(RecordBatch, bool), AnyError> {
-        let (data, mut schema, is_first) = self.to_data(
-            storage_schema,
-            time_partition,
-            schema_version,
-        )?;
+        let (data, mut schema, is_first) =
+            self.to_data(storage_schema, time_partition, schema_version)?;
 
         if get_field(&schema, DEFAULT_TIMESTAMP_KEY).is_some() {
             return Err(anyhow!(

src/handlers/http/modal/ingest_server.rs

@@ -249,7 +249,7 @@ impl IngestServer {
                             web::put()
                                 .to(ingestor_logstream::put_stream)
                                 .authorize_for_stream(Action::CreateStream),
-                        )
+                        ),
                 )
                 .service(
                     // GET "/logstream/{logstream}/info" ==> Get info for given log stream

src/metadata.rs

@@ -224,8 +224,9 @@ impl StreamInfo {
 
     /// Removes the `first_event_at` timestamp for the specified stream from the LogStreamMetadata.
     ///
-    /// This function acquires a write lock, retrieves the LogStreamMetadata for the given stream,
-    /// and removes the `first_event_at` timestamp if it exists.
+    /// This function is called during the retention task, when the parquet files along with the manifest files are deleted from the storage.
+    /// The manifest path is removed from the snapshot in the stream.json
+    /// and the first_event_at value in the stream.json is removed.
     ///
     /// # Arguments
     ///
@@ -236,11 +237,6 @@ impl StreamInfo {
     /// * `Result<(), MetadataError>` - Returns `Ok(())` if the `first_event_at` timestamp is successfully removed,
     ///   or a `MetadataError` if the stream metadata is not found.
     ///
-    /// # Errors
-    ///
-    /// This function will return an error if:
-    /// * The stream metadata cannot be found.
-    ///
     /// # Examples
     /// ```ignore
     /// ```rust