add project_id and body for logs. checkpoint before attempt at deriving schema

Author: tonyalaribe

src/database.rs

@@ -11,6 +11,8 @@ use datafusion::{
 };
 use deltalake::{storage::StorageOptions, DeltaOps, DeltaTable, DeltaTableBuilder};
 use tokio::sync::RwLock;
+use datafusion::prelude::DataFrame;
+
 type ProjectConfig = (String, StorageOptions, Arc<RwLock<DeltaTable>>);
 
 pub type ProjectConfigs = Arc<RwLock<HashMap<String, ProjectConfig>>>;

src/main.rs

@@ -154,17 +154,6 @@ async fn main() -> anyhow::Result<()> {
                 warn!("'datafusion' catalog not found; proceeding with empty context");
             }
 
-            let test_schema = Arc::new(Schema::new(vec![
-                Field::new("id", DataType::Int32, false),
-                Field::new("name", DataType::Utf8, false),
-            ]));
-            let test_batch = RecordBatch::try_new(
-                test_schema.clone(),
-                vec![Arc::new(Int32Array::from(vec![1, 2, 3])), Arc::new(StringArray::from(vec!["a", "b", "c"]))],
-            )?;
-            ctx.register_batch("test_table", test_batch)?;
-            info!("Registered dummy table: test_table");
-
             register_pg_settings_table(&ctx)?;
             register_set_config_udf(&ctx);
 
@@ -271,7 +260,7 @@ async fn main() -> anyhow::Result<()> {
                         if !records.is_empty() {
                             info!("Flushing {} enqueued records", records.len());
                             for record in records {
-                                process_record(&db_clone, &queue_clone, &status_store_clone, "default", record).await;
+                                process_record(&db_clone, &queue_clone, "default", record).await;
                             }
                         }
                     }

src/persistent_queue.rs

@@ -12,6 +12,7 @@ use tokio::{
 #[derive(Serialize, Deserialize, Clone)]
 pub struct IngestRecord {
     // Top-level fields
+    pub project_id:         String,
     pub timestamp:          i64,
     pub observed_timestamp: i64,
 
@@ -27,6 +28,7 @@ pub struct IngestRecord {
     pub level:                      Option<String>, // same as severity text
     pub severity___severity_text:   Option<String>,
     pub severity___severity_number: Option<String>,
+    pub body:                       Option<String>, // body as json json
 
     pub duration:   u64, // nanoseconds
     pub start_time: i64,
@@ -192,4 +194,3 @@ impl PersistentQueue {
         Ok(Some(record))
     }
 }
-