Refactor/base reader/writer

Cargo.toml

@@ -35,7 +35,7 @@ rand = "0.8" # Added rand dependency
 # {{ Add missing lapin trait dependencies }}
 tokio-executor-trait = "2.1"
 tokio-reactor-trait = "1.1"
-chrono = "0.4.41"
+chrono = { version = "0.4.41", features = ["serde"] }
 serde_yaml = "0.9.34"
 whatlang = "0.16.4"
 anyhow = "1.0.98"

src/config/parquet.rs

@@ -3,9 +3,9 @@ use serde::Deserialize;
 // Keep existing config
 #[derive(Deserialize, Debug, Clone)]
 pub struct ParquetInputConfig {
-    pub path: String,              // Path to the Parquet file or directory
-    pub text_column: String,       // Name of the column containing the main text
-    pub id_column: Option<String>, // Optional: Name of a column to use as document ID
+    pub path: String,        // Path to the Parquet file or directory
+    pub text_column: String, // Name of the column containing the main text
+    pub id_column: String,   // Name of a column to use as document ID
     // Add other column mappings as needed (e.g., for metadata)
     pub batch_size: Option<usize>, // Optional: Arrow batch size for reading
 }

src/data_model.rs

@@ -1,14 +1,15 @@
+use chrono::{NaiveDate, NaiveDateTime};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
-#[derive(Debug, Clone, Serialize, Deserialize)] // Clone might be needed depending on ownership strategy
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct TextDocument {
-    pub id: String,     // Unique identifier (e.g., filename, database ID)
-    pub source: String, // Original source location/identifier
+    pub id: String,
     pub content: String,
-    pub metadata: HashMap<String, String>, // For intermediate results or context
-                                           // pub score: Option<f64>, // Example field added by a scoring step
-                                           // Add other fields as needed by your steps
+    pub source: String,
+    pub added: Option<NaiveDate>, // or NaiveDateTime, depending
+    pub created: Option<(NaiveDateTime, NaiveDateTime)>, // updated
+    pub metadata: HashMap<String, String>,
 }
 
 // TODO: Add more fields as they are discovered

src/pipeline/filters/c4_filters.rs

@@ -549,6 +549,7 @@ mod tests {
             source: "test_source".to_string(),
             content: content.to_string(),
             metadata: HashMap::new(),
+            ..Default::default()
         }
     }
 

src/pipeline/filters/fineweb_quality.rs

@@ -206,6 +206,7 @@ mod tests {
             source: "test_source".to_string(),
             content: content.to_string(),
             metadata: StdHashMap::new(),
+            ..Default::default()
         }
     }
 

src/pipeline/filters/gopher_quality.rs

@@ -332,6 +332,7 @@ mod tests {
             source: "gopher_test_source".to_string(),
             content: content.to_string(),
             metadata: StdHashMap::new(),
+            ..Default::default()
         }
     }
 

src/pipeline/filters/gopher_rep.rs

@@ -265,6 +265,7 @@ mod tests {
             source: "rep_test_source".to_string(),
             content: content.to_string(),
             metadata: StdHashMap::new(),
+            ..Default::default()
         }
     }
 

src/pipeline/filters/language_filter.rs

@@ -76,6 +76,7 @@ mod tests {
             source: "rep_test_source".to_string(),
             content: content.to_string(),
             metadata: Default::default(),
+            ..Default::default()
         }
     }
 

src/pipeline/readers/base_reader.rs

@@ -0,0 +1,6 @@
+use crate::data_model::TextDocument;
+use crate::error::Result;
+
+pub trait BaseReader {
+    fn read_documents(&self) -> Result<Box<dyn Iterator<Item = Result<TextDocument>>>>;
+}

src/pipeline/readers/mod.rs

@@ -1,6 +1,7 @@
 // src/pipeline/readers/mod.rs
 
+pub mod base_reader;
 pub mod parquet_reader; // Looks for src/pipeline/readers/parquet_reader.rs
-
-// Often good practice to re-export the main types
+                        // Often good practice to re-export the main types
+pub use base_reader::BaseReader;
 pub use parquet_reader::ParquetReader; // Assuming ParquetReader struct/enum exists