fix: data type for static schema

if string parsable to int, consider it valid
if string parsable to float, consider it valid

Author: nikhilsinhaparseable

src/event/format/json.rs

@@ -62,6 +62,7 @@ impl EventFormat for Event {
         schema: &HashMap<String, Arc<Field>>,
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
+        static_schema_flag: bool,
     ) -> Result<(Self::Data, Vec<Arc<Field>>, bool), anyhow::Error> {
         let stream_schema = schema;
 
@@ -111,7 +112,7 @@ impl EventFormat for Event {
 
         if value_arr
             .iter()
-            .any(|value| fields_mismatch(&schema, value, schema_version))
+            .any(|value| fields_mismatch(&schema, value, schema_version, static_schema_flag))
         {
             return Err(anyhow!(
                 "Could not process this event due to mismatch in datatype"
@@ -253,39 +254,64 @@ fn collect_keys<'a>(values: impl Iterator<Item = &'a Value>) -> Result<Vec<&'a s
     Ok(keys)
 }
 
-fn fields_mismatch(schema: &[Arc<Field>], body: &Value, schema_version: SchemaVersion) -> bool {
+fn fields_mismatch(
+    schema: &[Arc<Field>],
+    body: &Value,
+    schema_version: SchemaVersion,
+    static_schema_flag: bool,
+) -> bool {
     for (name, val) in body.as_object().expect("body is of object variant") {
         if val.is_null() {
             continue;
         }
         let Some(field) = get_field(schema, name) else {
             return true;
         };
-        if !valid_type(field.data_type(), val, schema_version) {
+        if !valid_type(field, val, schema_version, static_schema_flag) {
             return true;
         }
     }
     false
 }
 
-fn valid_type(data_type: &DataType, value: &Value, schema_version: SchemaVersion) -> bool {
-    match data_type {
+fn valid_type(
+    field: &Field,
+    value: &Value,
+    schema_version: SchemaVersion,
+    static_schema_flag: bool,
+) -> bool {
+    match field.data_type() {
         DataType::Boolean => value.is_boolean(),
-        DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 => value.is_i64(),
+        DataType::Int8 | DataType::Int16 | DataType::Int32 | DataType::Int64 => {
+            if static_schema_flag {
+                if let Value::String(s) = value {
+                    return s.parse::<i64>().is_ok();
+                }
+            }
+            value.is_i64()
+        }
         DataType::UInt8 | DataType::UInt16 | DataType::UInt32 | DataType::UInt64 => value.is_u64(),
         DataType::Float16 | DataType::Float32 => value.is_f64(),
-        // All numbers can be cast as Float64 from schema version v1
-        DataType::Float64 if schema_version == SchemaVersion::V1 => value.is_number(),
-        DataType::Float64 if schema_version != SchemaVersion::V1 => value.is_f64(),
+        DataType::Float64 => {
+            if static_schema_flag {
+                if let Value::String(s) = value.clone() {
+                    return s.parse::<f64>().is_ok() || s.parse::<i64>().is_ok();
+                }
+                return value.is_number();
+            }
+            match schema_version {
+                SchemaVersion::V1 => value.is_number(),
+                _ => value.is_f64(),
+            }
+        }
         DataType::Utf8 => value.is_string(),
         DataType::List(field) => {
-            let data_type = field.data_type();
             if let Value::Array(arr) = value {
                 for elem in arr {
                     if elem.is_null() {
                         continue;
                     }
-                    if !valid_type(data_type, elem, schema_version) {
+                    if !valid_type(field, elem, schema_version, static_schema_flag) {
                         return false;
                     }
                 }
@@ -303,7 +329,7 @@ fn valid_type(data_type: &DataType, value: &Value, schema_version: SchemaVersion
                         if value.is_null() {
                             continue;
                         }
-                        if !valid_type(field.data_type(), value, schema_version) {
+                        if !valid_type(field, value, schema_version, static_schema_flag) {
                             return false;
                         }
                     } else {
@@ -317,7 +343,11 @@ fn valid_type(data_type: &DataType, value: &Value, schema_version: SchemaVersion
         }
         DataType::Timestamp(_, _) => value.is_string() || value.is_number(),
         _ => {
-            error!("Unsupported datatype {:?}, value {:?}", data_type, value);
+            error!(
+                "Unsupported datatype {:?}, value {:?}",
+                field.data_type(),
+                value
+            );
             unreachable!()
         }
     }

src/event/format/mod.rs

@@ -102,6 +102,7 @@ pub trait EventFormat: Sized {
         schema: &HashMap<String, Arc<Field>>,
         time_partition: Option<&String>,
         schema_version: SchemaVersion,
+        static_schema_flag: bool,
     ) -> Result<(Self::Data, EventSchema, bool), AnyError>;
 
     fn decode(data: Self::Data, schema: Arc<Schema>) -> Result<RecordBatch, AnyError>;
@@ -117,8 +118,12 @@ pub trait EventFormat: Sized {
         schema_version: SchemaVersion,
     ) -> Result<(RecordBatch, bool), AnyError> {
         let p_timestamp = self.get_p_timestamp();
-        let (data, mut schema, is_first) =
-            self.to_data(storage_schema, time_partition, schema_version)?;
+        let (data, mut schema, is_first) = self.to_data(
+            storage_schema,
+            time_partition,
+            schema_version,
+            static_schema_flag,
+        )?;
 
         if get_field(&schema, DEFAULT_TIMESTAMP_KEY).is_some() {
             return Err(anyhow!(