refactor: simplify logsource write to store

Author: Devdutt Shenoi

src/event/format/mod.rs

@@ -28,7 +28,7 @@ use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use serde_json::{json, Value};
+use serde_json::Value;
 
 use crate::{
     metadata::SchemaVersion,
@@ -107,15 +107,6 @@ impl LogSourceEntry {
             fields,
         }
     }
-
-    pub fn add_log_source(&mut self, log_source_format: LogSource, fields: HashSet<String>) {
-        self.log_source_format = log_source_format;
-        self.fields = fields;
-    }
-
-    pub fn to_value(&self) -> Value {
-        json!(self)
-    }
 }
 
 // Global Trait for event format

src/migration/stream_metadata_migration.rs

@@ -191,14 +191,13 @@ pub fn v5_v6(mut stream_metadata: Value) -> Value {
         "version".to_owned(),
         Value::String(storage::CURRENT_SCHEMA_VERSION.into()),
     );
-    let log_source = stream_metadata_map.get("log_source");
     let mut log_source_entry = LogSourceEntry::default();
-    if log_source.is_some() {
-        if let Ok(log_source) = serde_json::from_value::<LogSource>(log_source.unwrap().clone()) {
-            log_source_entry.add_log_source(log_source, HashSet::new());
+    if let Some(log_source) = stream_metadata_map.get("log_source") {
+        if let Ok(log_source) = serde_json::from_value::<LogSource>(log_source.clone()) {
+            log_source_entry = LogSourceEntry::new(log_source, HashSet::new());
         }
     }
-    stream_metadata_map.insert("log_source".to_owned(), log_source_entry.to_value());
+    stream_metadata_map.insert("log_source".to_owned(), json!(log_source_entry));
     stream_metadata
 }