remove unused, update parseable server log test

Author: nikhilsinhaparseable

src/event/format/known_schema.rs

@@ -515,25 +515,19 @@ mod tests {
     }
 
     #[test]
-    fn test_rust_server_logs() {
+    fn test_parseable_server_logs() {
         let processor = EventProcessor::new(FORMATS_JSON);
         let schema = processor
             .schema_definitions
-            .get("rust_server_logs")
+            .get("parseable_server_logs")
             .unwrap();
 
         let test_logs = vec![
             // Current parseable format with ThreadId
-            "2025-09-06T10:43:01.628980875Z  WARN main ThreadId(01) parseable::handlers::http::cluster:919: node http://0.0.0.0:8010/ is not live",
-            "2025-09-06T10:44:12.62276265Z ERROR actix-rt|system:0|arbiter:17 ThreadId(163) parseable_enterprise::http::handlers::query:43: JsonParse(\"Datafusion Error: Schema error: No field named a. Valid fields are serverlogs.log\")",
-            "2025-09-06T05:16:46.092071318Z ERROR actix-rt|system:0|arbiter:21 ThreadId(167) parseable_enterprise::http::handlers::query:43: JsonParse(\"Datafusion Error: Schema error: No field named ansible.host.ip\")",
-            "2025-09-06T11:22:07.500864363Z  WARN                         main ThreadId(01) parseable_enterprise:70: Received shutdown signal, notifying server to shut down...",
-            // env_logger format
-            "[2025-09-06T10:43:01.628980875Z INFO  parseable::storage] Initializing storage backend",
-            "[2025-09-06T10:43:01.628980875Z ERROR parseable::http::ingest] Failed to parse JSON",
-            // Simple tracing format (no ThreadId)
-            "2025-09-06T10:43:01.628980875Z  INFO parseable::storage::s3: Storage configured successfully",
-            "2025-09-06T10:43:01.628980875Z DEBUG parseable::query::engine: Query executed in 45ms",
+            "01K4SHM6VQASBJ7G8V0STZN6N1 01K4SHM6VQASBJ7G8V0STZN6N1 2025-09-06T10:43:01.628980875Z  WARN main ThreadId(01) parseable::handlers::http::cluster:919: node http://0.0.0.0:8010/ is not live",
+            "01K4SHM6VQASBJ7G8V0STZN6N1 01K4SHM6VQASBJ7G8V0STZN6N1 2025-09-06T10:44:12.62276265Z ERROR actix-rt|system:0|arbiter:17 ThreadId(163) parseable_enterprise::http::handlers::query:43: JsonParse(\"Datafusion Error: Schema error: No field named a. Valid fields are serverlogs.log\")",
+            "01K4SHM6VQASBJ7G8V0STZN6N1 01K4SHM6VQASBJ7G8V0STZN6N1 2025-09-06T05:16:46.092071318Z ERROR actix-rt|system:0|arbiter:21 ThreadId(167) parseable_enterprise::http::handlers::query:43: JsonParse(\"Datafusion Error: Schema error: No field named ansible.host.ip\")",
+            "01K4SHM6VQASBJ7G8V0STZN6N1 01K4SHM6VQASBJ7G8V0STZN6N1 2025-09-06T11:22:07.500864363Z  WARN                         main ThreadId(01) parseable_enterprise:70: Received shutdown signal, notifying server to shut down...",
         ];
 
         for (i, log_text) in test_logs.iter().enumerate() {

src/query/stream_schema_provider.rs

@@ -16,7 +16,7 @@
  *
  */
 
-use std::{any::Any, collections::HashMap, ops::Bound, path::PathBuf, sync::Arc};
+use std::{any::Any, collections::HashMap, ops::Bound, sync::Arc};
 
 use arrow_array::RecordBatch;
 use arrow_schema::{Schema, SchemaRef, SortOptions};
@@ -46,13 +46,12 @@ use datafusion::{
 };
 use futures_util::TryFutureExt;
 use itertools::Itertools;
-use relative_path::RelativePathBuf;
 
 use crate::{
     catalog::{
         ManifestFile, Snapshot as CatalogSnapshot,
         column::{Column, TypedStatistics},
-        manifest::{File, Manifest},
+        manifest::File,
         snapshot::{ManifestItem, Snapshot},
     },
     event::DEFAULT_TIMESTAMP_KEY,
@@ -64,7 +63,7 @@ use crate::{
     },
     option::Mode,
     parseable::{PARSEABLE, STREAM_EXISTS},
-    storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat},
+    storage::{ObjectStorage, ObjectStoreFormat},
 };
 
 use super::listing_table_builder::ListingTableBuilder;
@@ -869,37 +868,12 @@ fn extract_timestamp_bound(
             DateTime::from_timestamp_nanos(*value).naive_utc(),
         )),
         ScalarValue::Utf8(Some(str_value)) if is_time_partition => {
-            Some((binexpr.op, str_value.parse::<NaiveDateTime>().unwrap()))
+            Some((binexpr.op, str_value.parse().unwrap()))
         }
         _ => None,
     }
 }
 
-pub async fn collect_manifest_files(
-    storage: Arc<dyn ObjectStorage>,
-    manifest_urls: Vec<String>,
-) -> Result<Vec<Manifest>, ObjectStorageError> {
-    let mut tasks = Vec::new();
-    manifest_urls.into_iter().for_each(|path| {
-        let path = RelativePathBuf::from_path(PathBuf::from(path)).expect("Invalid path");
-        let storage = Arc::clone(&storage);
-        tasks.push(tokio::task::spawn(async move {
-            storage.get_object(&path).await
-        }));
-    });
-
-    let mut op = Vec::new();
-    for task in tasks {
-        let file = task.await??;
-        op.push(file);
-    }
-
-    Ok(op
-        .into_iter()
-        .map(|res| serde_json::from_slice(&res).expect("Data is invalid for Manifest"))
-        .collect())
-}
-
 // Extract start time and end time from filter predicate
 pub fn extract_primary_filter(
     filters: &[Expr],

src/storage/azure_blob.rs

@@ -224,7 +224,7 @@ impl BlobStore {
 
         match resp {
             Ok(resp) => {
-                let body: Bytes = resp.bytes().await.unwrap();
+                let body: Bytes = resp.bytes().await?;
                 STORAGE_REQUEST_RESPONSE_TIME
                     .with_label_values(&["azure_blob", "GET", "200"])
                     .observe(elapsed);
@@ -1062,15 +1062,6 @@ impl ObjectStorage for BlobStore {
         Ok(minutes)
     }
 
-    // async fn list_manifest_files(
-    //     &self,
-    //     stream_name: &str,
-    // ) -> Result<BTreeMap<String, Vec<String>>, ObjectStorageError> {
-    //     let files = self._list_manifest_files(stream_name).await?;
-
-    //     Ok(files)
-    // }
-
     async fn upload_file(&self, key: &str, path: &Path) -> Result<(), ObjectStorageError> {
         Ok(self._upload_file(key, path).await?)
     }

src/storage/gcs.rs

@@ -185,7 +185,7 @@ impl Gcs {
 
         match resp {
             Ok(resp) => {
-                let body: Bytes = resp.bytes().await.unwrap();
+                let body: Bytes = resp.bytes().await?;
                 STORAGE_REQUEST_RESPONSE_TIME
                     .with_label_values(&["gcs", "GET", "200"])
                     .observe(elapsed);

src/storage/s3.rs

@@ -1159,15 +1159,6 @@ impl ObjectStorage for S3 {
         Ok(minutes)
     }
 
-    // async fn list_manifest_files(
-    //     &self,
-    //     stream_name: &str,
-    // ) -> Result<BTreeMap<String, Vec<String>>, ObjectStorageError> {
-    //     let files = self._list_manifest_files(stream_name).await?;
-
-    //     Ok(files)
-    // }
-
     async fn upload_file(&self, key: &str, path: &Path) -> Result<(), ObjectStorageError> {
         Ok(self._upload_file(key, path).await?)
     }