feat: add feature to provide static schema in logstream creation

set header x-p-static-schema-flag = true and schema in body in below format -
{
   "fields":[
    {
        "name": name of the field",
        "data_type": "data type of the field, out of these (int, double, float, boolean, string, datetime, string_list, int_list, double_list, float_list, boolean_list)",
    }
   ]
}
once provided, schema is persisted in the storage and in metadata
if static schema provided, ingest api will verify if event log schema matches the static schema provided in stream creation
if schema does not match, ingestion is rejected

Author: nikhilsinhaparseable

server/src/event/format.rs

@@ -46,10 +46,12 @@ pub trait EventFormat: Sized {
     fn decode(data: Self::Data, schema: Arc<Schema>) -> Result<RecordBatch, AnyError>;
     fn into_recordbatch(
         self,
-        schema: HashMap<String, Arc<Field>>,
+        storage_schema: HashMap<String, Arc<Field>>,
         time_partition: Option<String>,
+        static_schema_flag: Option<String>,
     ) -> Result<(RecordBatch, bool), AnyError> {
-        let (data, mut schema, is_first, tags, metadata) = self.to_data(schema, time_partition)?;
+        let (data, mut schema, is_first, tags, metadata) =
+            self.to_data(storage_schema.clone(), time_partition)?;
 
         if get_field(&schema, DEFAULT_TAGS_KEY).is_some() {
             return Err(anyhow!("field {} is a reserved field", DEFAULT_TAGS_KEY));
@@ -90,19 +92,50 @@ pub trait EventFormat: Sized {
         )));
 
         // prepare the record batch and new fields to be added
-        let schema = Arc::new(Schema::new(schema));
-        let rb = Self::decode(data, schema.clone())?;
+        let new_schema = Arc::new(Schema::new(schema));
+        if !Self::is_schema_matching(new_schema.clone(), storage_schema, static_schema_flag) {
+            return Err(anyhow!("Schema mismatch"));
+        }
+        let rb = Self::decode(data, new_schema.clone())?;
         let tags_arr = StringArray::from_iter_values(std::iter::repeat(&tags).take(rb.num_rows()));
         let metadata_arr =
             StringArray::from_iter_values(std::iter::repeat(&metadata).take(rb.num_rows()));
         // modify the record batch to add fields to respective indexes
         let rb = utils::arrow::replace_columns(
-            Arc::clone(&schema),
+            Arc::clone(&new_schema),
             &rb,
             &[tags_index, metadata_index],
             &[Arc::new(tags_arr), Arc::new(metadata_arr)],
         );
 
         Ok((rb, is_first))
     }
+
+    fn is_schema_matching(
+        new_schema: Arc<Schema>,
+        storage_schema: HashMap<String, Arc<Field>>,
+        static_schema_flag: Option<String>,
+    ) -> bool {
+        if static_schema_flag.is_none() {
+            return true;
+        }
+        for (field_name, field) in new_schema
+            .fields()
+            .iter()
+            .map(|field| (field.name().to_owned(), field.clone()))
+            .collect::<HashMap<String, Arc<Field>>>()
+        {
+            if let Some(storage_field) = storage_schema.get(&field_name) {
+                if field_name != *storage_field.name() {
+                    return false;
+                }
+                if field.data_type() != storage_field.data_type() {
+                    return false;
+                }
+            } else {
+                return false;
+            }
+        }
+        true
+    }
 }

server/src/handlers.rs

@@ -24,6 +24,7 @@ const PREFIX_META: &str = "x-p-meta-";
 const STREAM_NAME_HEADER_KEY: &str = "x-p-stream";
 const LOG_SOURCE_KEY: &str = "x-p-log-source";
 const TIME_PARTITION_KEY: &str = "x-p-time-partition";
+const STATIC_SCHEMA_FLAG: &str = "x-p-static-schema-flag";
 const AUTHORIZATION_KEY: &str = "authorization";
 const SEPARATOR: char = '^';
 

server/src/handlers/http/ingest.rs

@@ -16,25 +16,26 @@
  *
  */
 
-use crate::event::error::EventError;
-use crate::event::format::EventFormat;
-use crate::event::{self, format};
+use super::logstream::error::CreateStreamError;
+use super::{kinesis, otel};
+use crate::event::{
+    self,
+    error::EventError,
+    format::{self, EventFormat},
+};
 use crate::handlers::{
     LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL, PREFIX_META, PREFIX_TAGS, SEPARATOR,
     STREAM_NAME_HEADER_KEY,
 };
 use crate::metadata::STREAM_INFO;
 use crate::utils::header_parsing::{collect_labelled_headers, ParseHeaderError};
 use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
-use arrow_schema::Field;
+use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use http::StatusCode;
 use serde_json::Value;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
-
-use super::logstream::error::CreateStreamError;
-use super::{kinesis, otel};
 // Handler for POST /api/v1/ingest
 // ingests events by extracting stream name from header
 // creates if stream does not exist
@@ -99,14 +100,18 @@ async fn push_logs(stream_name: String, req: HttpRequest, body: Bytes) -> Result
             .ok_or(PostError::StreamNotFound(stream_name.clone()))?
             .schema
             .clone();
-
         let time_partition = hash_map
             .get(&stream_name)
             .ok_or(PostError::StreamNotFound(stream_name.clone()))?
             .time_partition
             .clone();
+        let static_schema_flag = hash_map
+            .get(&stream_name)
+            .ok_or(PostError::StreamNotFound(stream_name.clone()))?
+            .static_schema_flag
+            .clone();
 
-        into_event_batch(req, body, schema, time_partition)?
+        into_event_batch(req, body, schema, time_partition, static_schema_flag)?
     };
 
     event::Event {
@@ -127,6 +132,7 @@ fn into_event_batch(
     body: Bytes,
     schema: HashMap<String, Arc<Field>>,
     time_partition: Option<String>,
+    static_schema_flag: Option<String>,
 ) -> Result<(usize, arrow_array::RecordBatch, bool), PostError> {
     let tags = collect_labelled_headers(&req, PREFIX_TAGS, SEPARATOR)?;
     let metadata = collect_labelled_headers(&req, PREFIX_META, SEPARATOR)?;
@@ -137,7 +143,7 @@ fn into_event_batch(
         tags,
         metadata,
     };
-    let (rb, is_first) = event.into_recordbatch(schema, time_partition)?;
+    let (rb, is_first) = event.into_recordbatch(schema, time_partition, static_schema_flag)?;
     Ok((size, rb, is_first))
 }
 
@@ -146,7 +152,8 @@ pub async fn create_stream_if_not_exists(stream_name: &str) -> Result<(), PostEr
     if STREAM_INFO.stream_exists(stream_name) {
         return Ok(());
     }
-    super::logstream::create_stream(stream_name.to_string(), "").await?;
+    super::logstream::create_stream(stream_name.to_string(), "", "", Arc::new(Schema::empty()))
+        .await?;
     Ok(())
 }
 
@@ -250,6 +257,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();
 
@@ -297,6 +305,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();
 
@@ -335,6 +344,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             schema,
             None,
+            None,
         )
         .unwrap();
 
@@ -372,6 +382,7 @@ mod tests {
             req,
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             schema,
+            None,
             None
         )
         .is_err());
@@ -397,6 +408,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             schema,
             None,
+            None,
         )
         .unwrap();
 
@@ -414,6 +426,7 @@ mod tests {
             req,
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
+            None,
             None
         )
         .is_err())
@@ -444,6 +457,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();
 
@@ -498,6 +512,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();
 
@@ -552,6 +567,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             schema,
             None,
+            None,
         )
         .unwrap();
 
@@ -598,6 +614,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();
 
@@ -648,6 +665,7 @@ mod tests {
             req,
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             schema,
+            None,
             None
         )
         .is_err());
@@ -683,6 +701,7 @@ mod tests {
             Bytes::from(serde_json::to_vec(&json).unwrap()),
             HashMap::default(),
             None,
+            None,
         )
         .unwrap();