page size to 20mb, add data type for Date

Author: nikhilsinhaparseable

src/event/format/json.rs

@@ -217,6 +217,7 @@ fn valid_type(data_type: &DataType, value: &Value, schema_version: SchemaVersion
                 false
             }
         }
+        DataType::Date32 => value.is_string() && is_parsable_as_date(value),
         DataType::Timestamp(_, _) => value.is_string() || value.is_number(),
         _ => {
             error!("Unsupported datatype {:?}, value {:?}", data_type, value);
@@ -231,3 +232,10 @@ pub fn is_parsable_as_number(value: &Value) -> bool {
     };
     s.parse::<i64>().is_ok()
 }
+
+pub fn is_parsable_as_date(value: &Value) -> bool {
+    let Value::String(s) = value else {
+        return false;
+    };
+    chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok()
+}

src/event/format/mod.rs

@@ -26,7 +26,7 @@ use std::{
 use anyhow::{anyhow, Error as AnyError};
 use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
-use chrono::DateTime;
+use chrono::{DateTime, NaiveDate};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
@@ -325,6 +325,16 @@ pub fn override_data_type(
                     // Update the field's data type to Float64
                     Field::new(field_name, DataType::Int64, true)
                 }
+                (SchemaVersion::V1, Some(Value::String(s)))
+                    if TIME_FIELD_NAME_PARTS
+                        .iter()
+                        .any(|part| field_name.to_lowercase().contains(part))
+                        && field.data_type() == &DataType::Utf8
+                        && NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok() =>
+                {
+                    // Update the field's data type to Timestamp
+                    Field::new(field_name, DataType::Date32, true)
+                }
 
                 // in V1 for new fields in json with inferred type number, cast as float64.
                 (SchemaVersion::V1, Some(Value::Number(_))) if field.data_type().is_integer() => {

src/static_schema.rs

@@ -102,6 +102,7 @@ pub fn convert_static_schema_to_arrow_schema(
                     "double" | "float" => DataType::Float64,
                     "boolean" => DataType::Boolean,
                     "string" => DataType::Utf8,
+                    "date" => DataType::Date32,
                     "datetime" => DataType::Timestamp(TimeUnit::Millisecond, None),
                     "string_list" => {
                         DataType::List(Arc::new(Field::new("item", DataType::Utf8, true)))

src/storage/staging.rs

@@ -325,7 +325,8 @@ pub fn parquet_writer_props(
         .set_column_encoding(
             ColumnPath::new(vec![time_partition_field]),
             Encoding::DELTA_BINARY_PACKED,
-        );
+        )
+        .set_data_page_size_limit(20 * 1024 * 1024);
 
     for (field, index) in custom_partition_fields {
         let field = ColumnPath::new(vec![field]);