Merge branch 'main' into alert-promql

Author: nikhilsinhaparseable

src/enterprise/utils.rs

@@ -99,7 +99,7 @@ pub async fn fetch_parquet_file_paths(
 
     let obs = PARSEABLE
         .metastore
-        .get_all_stream_jsons(stream, None, tenant_id)
+        .get_all_stream_jsons(stream, None, tenant_id, false)
         .await;
     if let Ok(obs) = obs {
         for ob in obs {

src/event/format/json.rs

@@ -109,7 +109,12 @@ impl EventFormat for Event {
 
             // If there are conflicts, rename the fields in JSON values
             let value_arr = if !conflicts.is_empty() {
-                rename_conflicting_fields_in_json(value_arr, &conflicts)
+                rename_conflicting_fields_in_json(
+                    value_arr,
+                    &conflicts,
+                    stream_schema,
+                    schema_version,
+                )
             } else {
                 value_arr
             };

src/event/format/mod.rs

@@ -528,9 +528,21 @@ pub fn detect_schema_conflicts(
 
 /// Renames fields in JSON values according to the provided field mapping.
 /// Used to resolve schema conflicts by renaming fields with conflicting types.
+/// Renames conflicting JSON fields to typed sibling columns, but only for
+/// records whose value is incompatible with the existing field type.
+///
+/// `field_mapping` is produced by `detect_schema_conflicts` and maps an
+/// original field name to its typed sibling name, e.g. `level -> level_utf8`.
+/// The mapping applies per value, not blindly to every record in the batch:
+/// if existing schema has `level: Float64`, then `{ "level": 30 }` stays as
+/// `level`, while `{ "level": "info" }` becomes `{ "level_utf8": "info" }`.
+/// This avoids routing compatible values into the sibling column and then
+/// failing validation, e.g. `level_utf8` expecting `Utf8` but receiving `30`.
 pub fn rename_conflicting_fields_in_json(
     values: Vec<Value>,
     field_mapping: &HashMap<String, String>,
+    existing_schema: &HashMap<String, Arc<Field>>,
+    schema_version: SchemaVersion,
 ) -> Vec<Value> {
     if field_mapping.is_empty() {
         return values;
@@ -544,7 +556,13 @@ pub fn rename_conflicting_fields_in_json(
                     .into_iter()
                     .map(|(key, val)| {
                         if let Some(new_key) = field_mapping.get(&key) {
-                            (new_key.clone(), val)
+                            if existing_schema.get(&key).is_some_and(|field| {
+                                value_compatible_with_type(&val, field.data_type(), schema_version)
+                            }) {
+                                (key, val)
+                            } else {
+                                (new_key.clone(), val)
+                            }
                         } else {
                             (key, val)
                         }
@@ -776,8 +794,8 @@ mod tests {
     #[test]
     fn test_rename_conflicting_fields_in_json() {
         let values = vec![
-            json!({"body_timestamp": "2025-01-01T00:00:00Z", "message": "hello"}),
-            json!({"body_timestamp": "2025-01-02T00:00:00Z", "message": "world"}),
+            json!({"body_timestamp": "not a timestamp", "message": "hello"}),
+            json!({"body_timestamp": "also not a timestamp", "message": "world"}),
         ];
 
         let mut field_mapping = HashMap::new();
@@ -786,7 +804,22 @@ mod tests {
             "body_timestamp_timestamp_ms".to_string(),
         );
 
-        let renamed = rename_conflicting_fields_in_json(values, &field_mapping);
+        let mut existing_schema: HashMap<String, Arc<Field>> = HashMap::new();
+        existing_schema.insert(
+            "body_timestamp".to_string(),
+            Arc::new(Field::new(
+                "body_timestamp",
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                true,
+            )),
+        );
+
+        let renamed = rename_conflicting_fields_in_json(
+            values,
+            &field_mapping,
+            &existing_schema,
+            SchemaVersion::V1,
+        );
 
         assert_eq!(renamed.len(), 2);
         assert!(renamed[0].get("body_timestamp_timestamp_ms").is_some());
@@ -799,12 +832,44 @@ mod tests {
         let values = vec![json!({"body_timestamp": "2025-01-01T00:00:00Z"})];
 
         let field_mapping = HashMap::new();
-        let renamed = rename_conflicting_fields_in_json(values.clone(), &field_mapping);
+        let existing_schema = HashMap::new();
+        let renamed = rename_conflicting_fields_in_json(
+            values.clone(),
+            &field_mapping,
+            &existing_schema,
+            SchemaVersion::V1,
+        );
 
         // Should return values unchanged
         assert_eq!(renamed, values);
     }
 
+    #[test]
+    fn test_rename_conflicting_fields_in_json_only_incompatible_records() {
+        let values = vec![json!({"level": 30}), json!({"level": "info"})];
+
+        let mut field_mapping = HashMap::new();
+        field_mapping.insert("level".to_string(), "level_utf8".to_string());
+
+        let mut existing_schema: HashMap<String, Arc<Field>> = HashMap::new();
+        existing_schema.insert(
+            "level".to_string(),
+            Arc::new(Field::new("level", DataType::Float64, true)),
+        );
+
+        let renamed = rename_conflicting_fields_in_json(
+            values,
+            &field_mapping,
+            &existing_schema,
+            SchemaVersion::V1,
+        );
+
+        assert_eq!(renamed[0].get("level"), Some(&json!(30)));
+        assert!(renamed[0].get("level_utf8").is_none());
+        assert_eq!(renamed[1].get("level_utf8"), Some(&json!("info")));
+        assert!(renamed[1].get("level").is_none());
+    }
+
     #[test]
     fn test_detect_schema_conflicts_timestamp_vs_utf8() {
         // Existing schema has body_timestamp as Timestamp

src/handlers/http/cluster/mod.rs

@@ -872,7 +872,7 @@ pub async fn fetch_stats_from_ingestors(
 ) -> Result<Vec<utils::QueriedStats>, StreamError> {
     let obs = PARSEABLE
         .metastore
-        .get_all_stream_jsons(stream_name, Some(Mode::Ingest), tenant_id)
+        .get_all_stream_jsons(stream_name, Some(Mode::Ingest), tenant_id, false)
         .await?;
 
     let mut ingestion_size = 0u64;

src/handlers/http/middleware.rs

@@ -186,6 +186,10 @@ where
                         permissions,
                         &user.tenant,
                     );
+                    req.headers_mut().insert(
+                        HeaderName::from_static(TENANT_ID),
+                        HeaderValue::from_str(tenant).unwrap(),
+                    );
                     req.extensions_mut().insert(session_key);
                     Some(session_id)
                 }

src/handlers/http/modal/query/querier_logstream.rs

@@ -197,7 +197,7 @@ pub async fn get_stats(
         if !date_value.is_empty() {
             let obs = PARSEABLE
                 .metastore
-                .get_all_stream_jsons(&stream_name, None, &tenant_id)
+                .get_all_stream_jsons(&stream_name, None, &tenant_id, false)
                 .await?;
 
             let mut stream_jsons = Vec::new();

src/handlers/http/oidc.rs

@@ -44,7 +44,7 @@ use url::Url;
 use crate::{
     handlers::{
         COOKIE_AGE_DAYS, SESSION_COOKIE_NAME, USER_COOKIE_NAME, USER_ID_COOKIE_NAME,
-        http::modal::OIDC_CLIENT,
+        http::{cluster::sync_user_creation, modal::OIDC_CLIENT},
     },
     oauth::OAuthSession,
     parseable::{DEFAULT_TENANT, PARSEABLE},
@@ -238,6 +238,21 @@ pub async fn reply_login(
         }
     };
 
+    if !PARSEABLE.options.is_multi_tenant() {
+        let roles = Some(user.roles.clone());
+        if let Err(e) = sync_user_creation(
+            &req,
+            user.clone(),
+            &roles,
+            &tenant_id,
+            &PARSEABLE.options.username,
+        )
+        .await
+        {
+            tracing::error!("Failed to sync OAuth user with roles to cluster nodes: {e}");
+        }
+    }
+
     let id = Ulid::new();
     Users.new_session(&user, SessionKey::SessionId(id), expires_in);
 

src/lib.rs

@@ -32,7 +32,7 @@ pub mod handlers;
 pub mod hottier;
 pub mod interactive;
 mod livetail;
-mod metadata;
+pub mod metadata;
 pub mod metastore;
 pub mod metrics;
 pub mod migration;

src/metastore/metastore_traits.rs

@@ -247,6 +247,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         stream_name: &str,
         mode: Option<Mode>,
         tenant_id: &Option<String>,
+        is_migration: bool,
     ) -> Result<Vec<Bytes>, MetastoreError>;
 
     /// Fetch manifests only for the explicitly requested date keys

src/metastore/metastores/object_store_metastore.rs

@@ -892,6 +892,7 @@ impl Metastore for ObjectStoreMetastore {
         stream_name: &str,
         mode: Option<Mode>,
         tenant_id: &Option<String>,
+        _is_migration: bool,
     ) -> Result<Vec<Bytes>, MetastoreError> {
         let root = tenant_id.as_deref().unwrap_or("");
         let path = RelativePathBuf::from_iter([root, stream_name, STREAM_ROOT_DIRECTORY]);