fix: commit schema bug

Author: parmesant

src/event/mod.rs

@@ -86,19 +86,22 @@ impl Event {
             }
         }
 
+        let stream = PARSEABLE.get_or_create_stream(&self.stream_name, &self.tenant_id);
+
         if self.is_first_event {
             commit_schema(&self.stream_name, self.rb.schema(), &self.tenant_id)?;
+            if !stream.get_static_schema_flag() {
+                stream.stage_schema_file(stream.get_schema().as_ref().clone())?;
+            }
         }
 
-        PARSEABLE
-            .get_or_create_stream(&self.stream_name, &self.tenant_id)
-            .push(
-                &key,
-                &self.rb,
-                self.parsed_timestamp,
-                &self.custom_partition_values,
-                self.stream_type,
-            )?;
+        stream.push(
+            &key,
+            &self.rb,
+            self.parsed_timestamp,
+            &self.custom_partition_values,
+            self.stream_type,
+        )?;
 
         let tenant = self.tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         update_stats(

src/parseable/streams.rs

@@ -497,28 +497,35 @@ impl Stream {
         // check if there is already a schema file in staging pertaining to this stream
         // if yes, then merge them and save
 
-        if let Some(mut schema) = schema {
+        if let Some(schema) = schema {
             let static_schema_flag = self.get_static_schema_flag();
             if !static_schema_flag {
-                // schema is dynamic, read from staging and merge if present
+                self.stage_schema_file(schema)?;
+            }
+        }
+
+        Ok(())
+    }
 
-                // need to add something before .schema to make the file have an extension of type `schema`
-                let path = RelativePathBuf::from_iter([format!("{}.schema", self.stream_name)])
-                    .to_path(&self.data_path);
+    pub fn stage_schema_file(&self, mut schema: Schema) -> Result<(), StagingError> {
+        // schema is dynamic, read from staging and merge if present
+        fs::create_dir_all(&self.data_path)?;
 
-                let staging_schemas = self.get_schemas_if_present();
-                if let Some(mut staging_schemas) = staging_schemas {
-                    staging_schemas.push(schema);
-                    schema = Schema::try_merge(staging_schemas)?;
-                }
+        // need to add something before .schema to make the file have an extension of type `schema`
+        let path = RelativePathBuf::from_iter([format!("{}.schema", self.stream_name)])
+            .to_path(&self.data_path);
 
-                // save the merged schema on staging disk
-                // the path should be stream/.ingestor.{id}.schema
-                info!("writing schema to path - {path:?}");
-                write(path, to_bytes(&schema))?;
-            }
+        let staging_schemas = self.get_schemas_if_present();
+        if let Some(mut staging_schemas) = staging_schemas {
+            staging_schemas.push(schema);
+            schema = Schema::try_merge(staging_schemas)?;
         }
 
+        // save the merged schema on staging disk
+        // the path should be stream/.ingestor.{id}.schema
+        info!("writing schema to path - {path:?}");
+        write(path, to_bytes(&schema))?;
+
         Ok(())
     }