Merge remote-tracking branch 'origin/main'

Author: Devdutt Shenoi

Cargo.lock

@@ -9,16 +9,16 @@ build = "build.rs"
 
 [dependencies]
 # Arrow and DataFusion ecosystem
-arrow = "53.0.0"
-arrow-array = "53.0.0" 
-arrow-flight = { version = "53.0.0", features = ["tls"] }
-arrow-ipc = { version = "53.0.0", features = ["zstd"] }
-arrow-json = "53.0.0"
-arrow-schema = { version = "53.0.0", features = ["serde"] }
-arrow-select = "53.0.0"
-datafusion = "44.0.0"
+arrow = "54.0.0"
+arrow-array = "54.0.0" 
+arrow-flight = { version = "54.0.0", features = ["tls"] }
+arrow-ipc = { version = "54.0.0", features = ["zstd"] }
+arrow-json = "54.0.0"
+arrow-schema = { version = "54.0.0", features = ["serde"] }
+arrow-select = "54.0.0"
+datafusion = "45.0.0"
 object_store = { version = "0.11.2", features = ["cloud", "aws", "azure"] }
-parquet = "53.0.0"
+parquet = "54.0.0"
 
 # Web server and HTTP-related
 actix-cors = "0.7.0"
@@ -135,7 +135,7 @@ anyhow = "1.0"
 
 [dev-dependencies]
 rstest = "0.23.0"
-arrow = "53.0.0"
+arrow = "54.0.0"
 temp-dir = "0.1.14"
 
 [package.metadata.parseable_ui]

Cargo.toml

@@ -27,7 +27,7 @@ use tracing::{debug, error};
 
 use crate::{
     connectors::common::processor::Processor,
-    event::format::{json, EventFormat, LogSource},
+    event::format::{json, EventFormat, LogSource, LogSourceEntry},
     parseable::PARSEABLE,
     storage::StreamType,
 };
@@ -43,9 +43,14 @@ impl ParseableSinkProcessor {
             .first()
             .map(|r| r.topic.as_str())
             .unwrap_or_default();
+        let log_source_entry = LogSourceEntry::default();
 
         PARSEABLE
-            .create_stream_if_not_exists(stream_name, StreamType::UserDefined, LogSource::Json)
+            .create_stream_if_not_exists(
+                stream_name,
+                StreamType::UserDefined,
+                vec![log_source_entry],
+            )
             .await?;
 
         let mut json_vec = Vec::with_capacity(records.len());

src/connectors/kafka/processor.rs

@@ -55,7 +55,7 @@ use crate::{
 struct JsonPartition {
     batch: Vec<Json>,
     schema: Vec<Arc<Field>>,
-    date: NaiveDate,
+    parsed_timestamp: NaiveDateTime,
 }
 
 pub struct Event {
@@ -209,7 +209,7 @@ impl EventFormat for Event {
                 .collect()
         };
 
-        if fields_mismatch(&schema, data, schema_version) {
+        if fields_mismatch(&schema, data, schema_version, static_schema_flag) {
             return Err(anyhow!(
                 "Could not process this event due to mismatch in datatype"
             ));
@@ -289,7 +289,7 @@ impl EventFormat for Event {
                     JsonPartition {
                         batch: vec![json],
                         schema,
-                        date: parsed_timestamp.date(),
+                        parsed_timestamp,
                     },
                 );
             }
@@ -301,7 +301,7 @@ impl EventFormat for Event {
             JsonPartition {
                 batch,
                 schema,
-                date,
+                parsed_timestamp,
             },
         ) in json_partitions
         {
@@ -313,14 +313,19 @@ impl EventFormat for Event {
                 schema_version,
             )?;
 
-            partitions.insert(prefix, PartitionEvent { rb: batch, date });
+            partitions.insert(
+                prefix,
+                PartitionEvent {
+                    rb: batch,
+                    parsed_timestamp,
+                },
+            );
         }
 
         Ok(super::Event {
             origin_format: "json",
             origin_size,
             is_first_event,
-            time_partition: None,
             partitions,
             stream_type,
         })
@@ -401,66 +406,124 @@ fn collect_keys(object: &Json) -> HashSet<&str> {
 }
 
 // Returns true when the field doesn't exist in schema or has an invalid type
-fn fields_mismatch(schema: &[Arc<Field>], body: &Json, schema_version: SchemaVersion) -> bool {
+fn fields_mismatch(
+    schema: &[Arc<Field>],
+    body: &Json,
+    schema_version: SchemaVersion,
+    static_schema_flag: bool,
+) -> bool {
     body.iter().any(|(key, value)| {
         !value.is_null()
             && get_field(schema, key)
-                .is_none_or(|field| !valid_type(field.data_type(), value, schema_version))
+                .is_none_or(|field| !valid_type(field, value, schema_version, static_schema_flag))
     })
 }
 
-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 => {
+            validate_int(value, static_schema_flag)
+        }
         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 => validate_float(value, schema_version, static_schema_flag),
         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) {
-                        return false;
-                    }
-                }
-            }
-            true
-        }
+        DataType::List(field) => validate_list(field, value, schema_version, static_schema_flag),
         DataType::Struct(fields) => {
-            if let Value::Object(val) = value {
-                for (key, value) in val {
-                    let field = (0..fields.len())
-                        .find(|idx| fields[*idx].name() == key)
-                        .map(|idx| &fields[idx]);
-
-                    if let Some(field) = field {
-                        if value.is_null() {
-                            continue;
-                        }
-                        if !valid_type(field.data_type(), value, schema_version) {
-                            return false;
-                        }
-                    } else {
-                        return false;
-                    }
-                }
-                true
-            } else {
-                false
+            validate_struct(fields, value, schema_version, static_schema_flag)
+        }
+        DataType::Date32 => {
+            if let Value::String(s) = value {
+                return NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok();
             }
+            false
         }
         DataType::Timestamp(_, _) => value.is_string() || value.is_number(),
         _ => {
-            error!("Unsupported datatype {:?}, value {:?}", data_type, value);
-            unreachable!()
+            error!(
+                "Unsupported datatype {:?}, value {:?}",
+                field.data_type(),
+                value
+            );
+            false
+        }
+    }
+}
+
+fn validate_int(value: &Value, static_schema_flag: bool) -> bool {
+    // allow casting string to int for static schema
+    if static_schema_flag {
+        if let Value::String(s) = value {
+            return s.trim().parse::<i64>().is_ok();
+        }
+    }
+    value.is_i64()
+}
+
+fn validate_float(value: &Value, schema_version: SchemaVersion, static_schema_flag: bool) -> bool {
+    // allow casting string to int for static schema
+    if static_schema_flag {
+        if let Value::String(s) = value.clone() {
+            let trimmed = s.trim();
+            return trimmed.parse::<f64>().is_ok() || trimmed.parse::<i64>().is_ok();
+        }
+        return value.is_number();
+    }
+    match schema_version {
+        SchemaVersion::V1 => value.is_number(),
+        _ => value.is_f64(),
+    }
+}
+
+fn validate_list(
+    field: &Field,
+    value: &Value,
+    schema_version: SchemaVersion,
+    static_schema_flag: bool,
+) -> bool {
+    if let Value::Array(arr) = value {
+        for elem in arr {
+            if elem.is_null() {
+                continue;
+            }
+            if !valid_type(field, elem, schema_version, static_schema_flag) {
+                return false;
+            }
+        }
+    }
+    true
+}
+
+fn validate_struct(
+    fields: &Fields,
+    value: &Value,
+    schema_version: SchemaVersion,
+    static_schema_flag: bool,
+) -> bool {
+    if let Value::Object(val) = value {
+        for (key, value) in val {
+            let field = fields.iter().find(|f| f.name() == key);
+
+            if let Some(field) = field {
+                if value.is_null() {
+                    continue;
+                }
+                if !valid_type(field, value, schema_version, static_schema_flag) {
+                    return false;
+                }
+            } else {
+                return false;
+            }
         }
+        true
+    } else {
+        false
     }
 }
 

src/event/format/json.rs

@@ -48,7 +48,7 @@ static TIME_FIELD_NAME_PARTS: [&str; 2] = ["time", "date"];
 type EventSchema = Vec<Arc<Field>>;
 
 /// Source of the logs, used to perform special processing for certain sources
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash)]
 pub enum LogSource {
     // AWS Kinesis sends logs in the format of a json array
     Kinesis,
@@ -98,6 +98,23 @@ impl Display for LogSource {
 
 pub type IsFirstEvent = bool;
 
+/// Contains the format name and a list of known field names that are associated with the said format.
+/// Stored on disk as part of `ObjectStoreFormat` in stream.json
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LogSourceEntry {
+    pub log_source_format: LogSource,
+    pub fields: HashSet<String>,
+}
+
+impl LogSourceEntry {
+    pub fn new(log_source_format: LogSource, fields: HashSet<String>) -> Self {
+        LogSourceEntry {
+            log_source_format,
+            fields,
+        }
+    }
+}
+
 // Global Trait for event format
 // This trait is implemented by all the event formats
 pub trait EventFormat: Sized {

src/event/format/mod.rs

@@ -26,42 +26,48 @@ use std::sync::Arc;
 
 use self::error::EventError;
 use crate::{metadata::update_stats, parseable::Stream, storage::StreamType};
-use chrono::NaiveDate;
+use chrono::NaiveDateTime;
 use std::collections::HashMap;
 
 pub const DEFAULT_TIMESTAMP_KEY: &str = "p_timestamp";
 
 #[derive(Debug)]
 pub struct PartitionEvent {
     pub rb: RecordBatch,
-    pub date: NaiveDate,
+    pub parsed_timestamp: NaiveDateTime,
 }
 
 #[derive(Debug)]
 pub struct Event {
     pub origin_format: &'static str,
     pub origin_size: usize,
     pub is_first_event: bool,
-    pub time_partition: Option<String>,
     pub partitions: HashMap<String, PartitionEvent>,
     pub stream_type: StreamType,
 }
 
 // Events holds the schema related to a each event for a single log stream
 impl Event {
     pub fn process(self, stream: &Stream) -> Result<(), EventError> {
-        for (prefix, PartitionEvent { rb, date }) in self.partitions {
+        for (
+            prefix,
+            PartitionEvent {
+                rb,
+                parsed_timestamp,
+            },
+        ) in self.partitions
+        {
             if self.is_first_event {
                 stream.commit_schema(rb.schema().as_ref().clone())?;
             }
-            stream.push(&prefix, &rb, self.stream_type)?;
+            stream.push(&prefix, parsed_timestamp, &rb, self.stream_type)?;
 
             update_stats(
                 &stream.stream_name,
                 self.origin_format,
                 self.origin_size,
                 rb.num_rows(),
-                date,
+                parsed_timestamp.date(),
             );
 
             crate::livetail::LIVETAIL.process(&stream.stream_name, &rb);
@@ -71,7 +77,12 @@ impl Event {
 
     pub fn process_unchecked(&self, stream: &Stream) -> Result<(), EventError> {
         for (prefix, partition) in &self.partitions {
-            stream.push(prefix, &partition.rb, self.stream_type)?;
+            stream.push(
+                prefix,
+                partition.parsed_timestamp,
+                &partition.rb,
+                self.stream_type,
+            )?;
         }
 
         Ok(())