catch within-batch type mismatches on new fields

remove time partition headers from create/update logstream

Author: nikhilsinhaparseable

src/event/format/mod.rs

@@ -584,21 +584,22 @@ pub fn rename_per_record_type_mismatches(
     existing_schema: &HashMap<String, Arc<Field>>,
     schema_version: SchemaVersion,
 ) -> Vec<Value> {
-    if values.len() <= 1 || existing_schema.is_empty() {
-        return values;
-    }
-    // Bail out unless at least one inferred field collides with storage at
-    // the same type. Without that, arrow's inference can't have hidden a
-    // mixed-type batch behind a matching aggregate type.
-    let needs_check = inferred_schema.fields().iter().any(|f| {
-        existing_schema
-            .get(f.name())
-            .is_some_and(|s| s.data_type() == f.data_type())
-    });
-    if !needs_check {
+    if values.len() <= 1 {
         return values;
     }
 
+    // Lookup of inferred field types — used as the reference type when a field
+    // isn't yet in storage (e.g. first batch for the stream, or a new column).
+    // Without this, mixed-type batches for new fields slip through: arrow picks
+    // a single aggregate type (Utf8 wins over Bool, etc.), the batch-level
+    // conflict check sees nothing to compare against in empty storage, and
+    // records carrying the loser type later fail `fields_mismatch`.
+    let inferred_types: HashMap<&str, &DataType> = inferred_schema
+        .fields()
+        .iter()
+        .map(|f| (f.name().as_str(), f.data_type()))
+        .collect();
+
     values
         .into_iter()
         .map(|value| {
@@ -611,11 +612,16 @@ pub fn rename_per_record_type_mismatches(
                     if val.is_null() {
                         return (key, val);
                     }
-                    let Some(existing_field) = existing_schema.get(&key) else {
+                    // Prefer storage's declared type; fall back to the inferred
+                    // type so within-batch mismatches on new fields are caught.
+                    let target_type = existing_schema
+                        .get(&key)
+                        .map(|f| f.data_type())
+                        .or_else(|| inferred_types.get(key.as_str()).copied());
+                    let Some(target_type) = target_type else {
                         return (key, val);
                     };
-                    if value_compatible_with_type(&val, existing_field.data_type(), schema_version)
-                    {
+                    if value_compatible_with_type(&val, target_type, schema_version) {
                         return (key, val);
                     }
                     let suffix = get_datatype_suffix(&datatype_for_value(&val));

src/handlers/http/modal/utils/logstream_utils.rs

@@ -23,15 +23,14 @@ use crate::{
     handlers::{
         CUSTOM_PARTITION_KEY, DATASET_LABELS_KEY, DATASET_TAG_KEY, DATASET_TAGS_KEY, DatasetTag,
         INFER_TIMESTAMP_KEY, LOG_SOURCE_KEY, STATIC_SCHEMA_FLAG, STREAM_TYPE_KEY,
-        TELEMETRY_TYPE_KEY, TIME_PARTITION_KEY, TIME_PARTITION_LIMIT_KEY, TelemetryType,
-        UPDATE_STREAM_KEY, parse_dataset_labels, parse_dataset_tags,
+        TELEMETRY_TYPE_KEY, TIME_PARTITION_LIMIT_KEY, TelemetryType, UPDATE_STREAM_KEY,
+        parse_dataset_labels, parse_dataset_tags,
     },
     storage::StreamType,
 };
 
 #[derive(Debug)]
 pub struct PutStreamHeaders {
-    pub time_partition: String,
     pub time_partition_limit: String,
     pub custom_partition: Option<String>,
     pub static_schema_flag: bool,
@@ -49,7 +48,6 @@ pub struct PutStreamHeaders {
 impl Default for PutStreamHeaders {
     fn default() -> Self {
         Self {
-            time_partition: String::default(),
             time_partition_limit: String::default(),
             custom_partition: None,
             static_schema_flag: false,
@@ -80,10 +78,6 @@ impl From<&HeaderMap> for PutStreamHeaders {
             .get(TELEMETRY_TYPE_KEY)
             .and_then(|v| v.to_str().ok());
         PutStreamHeaders {
-            time_partition: headers
-                .get(TIME_PARTITION_KEY)
-                .map_or("", |v| v.to_str().unwrap())
-                .to_string(),
             time_partition_limit: headers
                 .get(TIME_PARTITION_LIMIT_KEY)
                 .map_or("", |v| v.to_str().unwrap())

src/parseable/mod.rs

@@ -695,7 +695,6 @@ impl Parseable {
         tenant_id: &Option<String>,
     ) -> Result<HeaderMap, StreamError> {
         let PutStreamHeaders {
-            time_partition,
             time_partition_limit,
             custom_partition,
             static_schema_flag,
@@ -763,7 +762,6 @@ impl Parseable {
                 .update_stream(
                     headers,
                     stream_name,
-                    &time_partition,
                     static_schema_flag,
                     &time_partition_limit,
                     custom_partition.as_ref(),
@@ -782,25 +780,18 @@ impl Parseable {
             validate_custom_partition(custom_partition)?;
         }
 
-        if !time_partition.is_empty() && custom_partition.is_some() {
-            return Err(StreamError::Custom {
-                msg: "Cannot set both time partition and custom partition".to_string(),
-                status: StatusCode::BAD_REQUEST,
-            });
-        }
-
         let schema = validate_static_schema(
             body,
             stream_name,
-            &time_partition,
+            "",
             custom_partition.as_ref(),
             static_schema_flag,
         )?;
 
         let log_source_entry = LogSourceEntry::new(log_source, HashSet::new());
         self.create_stream(
             stream_name.to_string(),
-            &time_partition,
+            "",
             time_partition_in_days,
             custom_partition.as_ref(),
             static_schema_flag,
@@ -818,12 +809,10 @@ impl Parseable {
         Ok(headers.clone())
     }
 
-    #[allow(clippy::too_many_arguments)]
     async fn update_stream(
         &self,
         headers: &HeaderMap,
         stream_name: &str,
-        time_partition: &str,
         static_schema_flag: bool,
         time_partition_limit: &str,
         custom_partition: Option<&String>,
@@ -832,12 +821,6 @@ impl Parseable {
         if !self.streams.contains(stream_name, tenant_id) {
             return Err(StreamNotFound(stream_name.to_string()).into());
         }
-        if !time_partition.is_empty() {
-            return Err(StreamError::Custom {
-                msg: "Altering the time partition of an existing stream is restricted.".to_string(),
-                status: StatusCode::BAD_REQUEST,
-            });
-        }
         if static_schema_flag {
             return Err(StreamError::Custom {
                 msg: "Altering the schema of an existing stream is restricted.".to_string(),