removed p_timestamp from primary filter, used Utf8View instead of Utf8

Author: nikhilsinhaparseable

src/event/format/json.rs

@@ -20,7 +20,7 @@
 #![allow(deprecated)]
 
 use anyhow::anyhow;
-use arrow_array::RecordBatch;
+use arrow_array::{RecordBatch, StringArray, StringViewArray};
 use arrow_json::reader::{infer_json_schema_from_iterator, ReaderBuilder};
 use arrow_schema::{DataType, Field, Fields, Schema};
 use datafusion::arrow::util::bit_util::round_upto_multiple_of_64;
@@ -105,22 +105,61 @@ impl EventFormat for Event {
         Ok((value_arr, schema, is_first))
     }
 
-    // Convert the Data type (defined above) to arrow record batch
-    fn decode(data: Self::Data, schema: Arc<Schema>) -> Result<RecordBatch, anyhow::Error> {
-        let array_capacity = round_upto_multiple_of_64(data.len());
-        let mut reader = ReaderBuilder::new(schema)
-            .with_batch_size(array_capacity)
-            .with_coerce_primitive(false)
-            .build_decoder()?;
-
-        reader.serialize(&data)?;
-        match reader.flush() {
-            Ok(Some(recordbatch)) => Ok(recordbatch),
-            Err(err) => Err(anyhow!("Failed to create recordbatch due to {:?}", err)),
-            Ok(None) => unreachable!("all records are added to one rb"),
+
+fn decode(data: Self::Data, schema: Arc<Schema>) -> Result<RecordBatch, anyhow::Error> {
+    // First create a schema with Utf8 instead of Utf8View
+    let temp_schema = Schema::new(
+        schema
+            .fields()
+            .iter()
+            .map(|field| {
+                if matches!(field.data_type(), DataType::Utf8View) {
+                    Arc::new(Field::new(field.name(), DataType::Utf8, field.is_nullable()))
+                } else {
+                    field.clone()
+                }
+            })
+            .collect::<Vec<_>>(),
+    );
+
+    let array_capacity = round_upto_multiple_of_64(data.len());
+    let mut reader = ReaderBuilder::new(Arc::new(temp_schema))
+        .with_batch_size(array_capacity)
+        .with_coerce_primitive(false)
+        .with_strict_mode(false)
+        .build_decoder()?;
+
+    reader.serialize(&data)?;
+    
+    match reader.flush() {
+        Ok(Some(temp_batch)) => {
+            // Convert Utf8 arrays to Utf8View arrays where needed
+            let new_columns: Vec<Arc<dyn arrow_array::Array>> = temp_batch
+                .columns()
+                .iter()
+                .zip(schema.fields())
+                .map(|(col, field)| {
+                    if matches!(field.data_type(), DataType::Utf8View) {
+                        let string_array = col
+                            .as_any()
+                            .downcast_ref::<StringArray>()
+                            .expect("Expected StringArray");
+                        Arc::new(StringViewArray::from(
+                            string_array.iter().map(|s| s.map(|s| s.to_string())).collect::<Vec<_>>()
+                        ))
+                    } else {
+                        col.clone()
+                    }
+                })
+                .collect();
+
+            Ok(RecordBatch::try_new(schema, new_columns)?)
         }
+        Err(err) => Err(anyhow!("Failed to create recordbatch due to {:?}", err)),
+        Ok(None) => unreachable!("all records are added to one rb"),
     }
 }
+}
 
 // Returns arrow schema with the fields that are present in the request body
 // This schema is an input to convert the request body to arrow record batch
@@ -179,7 +218,7 @@ fn valid_type(data_type: &DataType, value: &Value, schema_version: SchemaVersion
         DataType::Float16 | DataType::Float32 | DataType::Float64 => value.is_f64(),
         // All numbers can be cast as Float64 from schema version v1
         DataType::Int64 => value.is_i64() || is_parsable_as_number(value),
-        DataType::Utf8 => value.is_string(),
+        DataType::Utf8View => value.is_string(),
         DataType::List(field) => {
             let data_type = field.data_type();
             if let Value::Array(arr) = value {

src/event/format/mod.rs

@@ -298,7 +298,7 @@ pub fn override_data_type(
                     if TIME_FIELD_NAME_PARTS
                         .iter()
                         .any(|part| field_name.to_lowercase().contains(part))
-                        && field.data_type() == &DataType::Utf8
+                        && field.data_type() == &DataType::Utf8View
                         && (DateTime::parse_from_rfc3339(s).is_ok()
                             || DateTime::parse_from_rfc2822(s).is_ok()) =>
                 {
@@ -319,7 +319,7 @@ pub fn override_data_type(
                                 .unwrap()
                                 .data_type()
                                 == &DataType::Int64
-                            && field.data_type() == &DataType::Utf8
+                            && field.data_type() == &DataType::Utf8View
                             && s.parse::<i64>().is_ok()) =>
                 {
                     // Update the field's data type to Float64
@@ -329,7 +329,7 @@ pub fn override_data_type(
                     if TIME_FIELD_NAME_PARTS
                         .iter()
                         .any(|part| field_name.to_lowercase().contains(part))
-                        && field.data_type() == &DataType::Utf8
+                        && field.data_type() == &DataType::Utf8View
                         && NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok() =>
                 {
                     // Update the field's data type to Timestamp

src/query/mod.rs

@@ -494,7 +494,7 @@ fn transform(
 ) -> Transformed<LogicalPlan> {
     plan.transform(&|plan| match plan {
         LogicalPlan::TableScan(table) => {
-            let mut new_filters = vec![];
+            let new_filters = vec![];
             if !table_contains_any_time_filters(&table, time_partition) {
                 let mut _start_time_filter: Expr;
                 let mut _end_time_filter: Expr;
@@ -529,8 +529,8 @@ fn transform(
                     }
                 }
 
-                new_filters.push(_start_time_filter);
-                new_filters.push(_end_time_filter);
+                // new_filters.push(_start_time_filter);
+                // new_filters.push(_end_time_filter);
             }
             let new_filter = new_filters.into_iter().reduce(and);
             if let Some(new_filter) = new_filter {

src/query/stream_schema_provider.rs

@@ -435,9 +435,9 @@ impl TableProvider for StandardTableProvider {
             .map_err(|err| DataFusionError::Plan(err.to_string()))?;
         let time_partition = object_store_format.time_partition;
         let mut time_filters = extract_primary_filter(filters, &time_partition);
-        if time_filters.is_empty() {
-            return Err(DataFusionError::Plan("potentially unbounded query on time range. Table scanning requires atleast one time bound".to_string()));
-        }
+        // if time_filters.is_empty() {
+        //     return Err(DataFusionError::Plan("potentially unbounded query on time range. Table scanning requires atleast one time bound".to_string()));
+        // }
 
         if include_now(filters, &time_partition) {
             if let Some(records) =

src/static_schema.rs

@@ -101,7 +101,7 @@ pub fn convert_static_schema_to_arrow_schema(
                     "int" => DataType::Int64,
                     "double" | "float" => DataType::Float64,
                     "boolean" => DataType::Boolean,
-                    "string" => DataType::Utf8,
+                    "string" => DataType::Utf8View,
                     "date" => DataType::Date32,
                     "datetime" => DataType::Timestamp(TimeUnit::Millisecond, None),
                     "string_list" => {

src/storage/staging.rs

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