Schema conflict resolution

src/handlers/http/mod.rs

@@ -139,6 +139,13 @@ pub async fn fetch_schema(
     stream_name: &str,
     tenant_id: &Option<String>,
 ) -> anyhow::Result<arrow_schema::Schema> {
+    if let Some(global) = PARSEABLE
+        .metastore
+        .get_global_schema(stream_name, tenant_id)
+        .await?
+    {
+        return Ok(global.schema);
+    }
     let res: Vec<Schema> = PARSEABLE
         .metastore
         .get_all_schemas(stream_name, tenant_id)

src/lib.rs

@@ -50,6 +50,7 @@ pub mod prism;
 pub mod query;
 pub mod rbac;
 mod response;
+pub mod schema_registry;
 pub mod sse;
 mod static_schema;
 mod stats;

src/metastore/metastore_traits.rs

@@ -23,6 +23,7 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use dashmap::DashMap;
 use erased_serde::Serialize as ErasedSerialize;
+use relative_path::RelativePathBuf;
 use tonic::async_trait;
 use ulid::Ulid;
 
@@ -36,6 +37,7 @@ use crate::{
     handlers::http::modal::NodeType,
     metastore::MetastoreError,
     option::Mode,
+    parseable::GlobalSchema,
     users::filters::Filter,
 };
 
@@ -291,6 +293,37 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<(), MetastoreError>;
+    async fn get_global_schema_locks(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Vec<String>, MetastoreError>;
+    async fn put_global_schema_lock(
+        &self,
+        lock_file_path: RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError>;
+    async fn delete_global_schema_lock(
+        &self,
+        path: RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError>;
+    /// Global schema operations used to detect and resolve schema conflicts.
+    /// A missing expected hash creates the record; an existing hash is checked
+    /// before updating. Callers coordinate updates with global-schema lock files;
+    /// the hash check and write are not atomic. `false` indicates a conflict.
+    async fn get_global_schema(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Option<GlobalSchema>, MetastoreError>;
+    async fn put_global_schema(
+        &self,
+        obj: GlobalSchema,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+        expected: Option<u64>,
+    ) -> Result<bool, MetastoreError>;
 
     /// parseable metadata
     async fn get_parseable_metadata(

src/metastore/metastores/object_store_metastore.rs

@@ -51,16 +51,17 @@ use crate::{
         metastore_traits::{Metastore, MetastoreObject},
     },
     option::Mode,
-    parseable::{DEFAULT_TENANT, PARSEABLE},
+    parseable::{DEFAULT_TENANT, GlobalSchema, PARSEABLE},
     storage::{
-        ALERTS_ROOT_DIRECTORY, ObjectStorage, ObjectStorageError, PARSEABLE_ROOT_DIRECTORY,
-        SETTINGS_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
-        TARGETS_ROOT_DIRECTORY, TOMBSTONE_ROOT_DIRECTORY,
+        ALERTS_ROOT_DIRECTORY, GLOBAL_SCHEMA_FILE_NAME, ObjectStorage, ObjectStorageError,
+        PARSEABLE_ROOT_DIRECTORY, SETTINGS_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME,
+        STREAM_ROOT_DIRECTORY, TARGETS_ROOT_DIRECTORY, TOMBSTONE_ROOT_DIRECTORY,
         object_storage::{
-            alert_json_path, alert_state_json_path, filter_path, manifest_path, mttr_json_path,
-            outbound_http_policy_json_path, parseable_json_path, schema_path, stream_json_path,
-            to_bytes,
+            alert_json_path, alert_state_json_path, filter_path, global_schema_lock_file_path,
+            global_schema_path, manifest_path, mttr_json_path, outbound_http_policy_json_path,
+            parseable_json_path, schema_path, stream_json_path, to_bytes,
         },
+        to_object_store_path,
     },
     users::filters::{Filter, migrate_v1_v2},
 };
@@ -79,6 +80,10 @@ pub fn is_missing_optional_dir(err: &ObjectStorageError) -> bool {
     }
 }
 
+fn is_legacy_schema_file(file_name: &str) -> bool {
+    file_name.contains(".schema") && !file_name.ends_with(GLOBAL_SCHEMA_FILE_NAME)
+}
+
 #[async_trait]
 impl Metastore for ObjectStoreMetastore {
     /// Since Parseable already starts with a connection to an object store, no need to implement this
@@ -1160,7 +1165,7 @@ impl Metastore for ObjectStoreMetastore {
             .storage
             .get_objects(
                 Some(&path_prefix),
-                Box::new(|file_name: String| file_name.contains(".schema")),
+                Box::new(|file_name: String| is_legacy_schema_file(&file_name)),
                 tenant_id,
             )
             .await?
@@ -1178,6 +1183,11 @@ impl Metastore for ObjectStoreMetastore {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<Bytes, MetastoreError> {
+        if let Some(global_schema) = self.get_global_schema(stream_name, tenant_id).await? {
+            let schema = to_bytes(&global_schema.schema);
+            return Ok(schema);
+        }
+
         Ok(self
             .storage
             .get_object(&schema_path(stream_name, tenant_id), tenant_id)
@@ -1196,6 +1206,87 @@ impl Metastore for ObjectStoreMetastore {
             .put_object(&path, to_bytes(&obj), tenant_id)
             .await?)
     }
+    async fn get_global_schema_locks(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Vec<String>, MetastoreError> {
+        let path = to_object_store_path(&global_schema_lock_file_path(stream_name, tenant_id));
+        let mut result_file_list = Vec::new();
+        let resp = self.storage.list_with_delimiter(Some(path)).await?;
+        for obj in resp.objects.iter() {
+            let name = obj.location.filename();
+            if let Some(name) = name
+                && name.ends_with(".lock")
+            {
+                result_file_list.push(name.to_owned());
+            }
+        }
+        Ok(result_file_list)
+    }
+    async fn put_global_schema_lock(
+        &self,
+        lock_file_path: RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError> {
+        let data = to_bytes("");
+        self.storage
+            .put_object(&lock_file_path, data, tenant_id)
+            .await?;
+        Ok(())
+    }
+    async fn delete_global_schema_lock(
+        &self,
+        path: RelativePathBuf,
+        tenant_id: &Option<String>,
+    ) -> Result<(), MetastoreError> {
+        match self.storage.delete_object(&path, tenant_id).await {
+            Ok(()) | Err(ObjectStorageError::NoSuchKey(_)) => Ok(()),
+            Err(ObjectStorageError::IoError(error))
+                if error.kind() == std::io::ErrorKind::NotFound =>
+            {
+                Ok(())
+            }
+            Err(error) => Err(error.into()),
+        }
+    }
+    /// Retrieve the persisted global schema and its stored content hash.
+    async fn get_global_schema(
+        &self,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+    ) -> Result<Option<GlobalSchema>, MetastoreError> {
+        // LocalFS does not participate in the shared registry. Only bypass
+        // explicitly unsupported backends; propagate errors from cloud stores.
+        let path = global_schema_path(stream_name, tenant_id);
+        let global_schema = match self.storage.get_object(&path, tenant_id).await {
+            Ok(obj) => serde_json::from_slice(&obj)?,
+            Err(ObjectStorageError::NoSuchKey(_) | ObjectStorageError::PathError(_)) => {
+                return Ok(None);
+            }
+            Err(e) => {
+                tracing::error!("{e}");
+                return Err(MetastoreError::ObjectStorageError(e));
+            }
+        };
+        Ok(Some(global_schema))
+    }
+
+    /// Create a missing global schema or update one whose stored hash matches.
+    /// Callers coordinate the non-atomic hash check and update with lock files.
+    async fn put_global_schema(
+        &self,
+        obj: GlobalSchema,
+        stream_name: &str,
+        tenant_id: &Option<String>,
+        expected: Option<u64>,
+    ) -> Result<bool, MetastoreError> {
+        let path = global_schema_path(stream_name, tenant_id);
+        Ok(self
+            .storage
+            .put_object_if(&path, to_bytes(&obj), expected, tenant_id)
+            .await?)
+    }
 
     async fn get_parseable_metadata(
         &self,
@@ -1469,4 +1560,40 @@ mod tests {
             Err(ObjectStorageError::NoSuchKey(_))
         ));
     }
+
+    #[test]
+    fn global_schema_paths_normalize_default_tenant_and_isolate_real_tenants() {
+        let no_tenant = None;
+        let default_tenant = Some(DEFAULT_TENANT.to_owned());
+        let tenant_a = Some("tenant-a".to_owned());
+        let tenant_b = Some("tenant-b".to_owned());
+
+        assert_eq!(
+            global_schema_path("events", &no_tenant).as_str(),
+            "events/.stream/.global.schema"
+        );
+        assert_eq!(
+            global_schema_path("events", &default_tenant),
+            global_schema_path("events", &no_tenant)
+        );
+        assert_eq!(
+            global_schema_lock_file_path("events", &default_tenant),
+            global_schema_lock_file_path("events", &no_tenant)
+        );
+        assert_eq!(
+            global_schema_path("events", &tenant_a).as_str(),
+            "tenant-a/events/.stream/.global.schema"
+        );
+        assert_ne!(
+            global_schema_path("events", &tenant_a),
+            global_schema_path("events", &tenant_b)
+        );
+        assert_ne!(
+            global_schema_lock_file_path("events", &tenant_a),
+            global_schema_lock_file_path("events", &tenant_b)
+        );
+        assert!(is_legacy_schema_file(".schema"));
+        assert!(is_legacy_schema_file(".ingestor.node-1.schema"));
+        assert!(!is_legacy_schema_file(GLOBAL_SCHEMA_FILE_NAME));
+    }
 }

src/metastore/mod.rs

@@ -49,6 +49,12 @@ pub enum MetastoreError {
     #[error("JSON schema validation error: {message}")]
     JsonSchemaError { message: String },
 
+    #[error("CAS convergence failure: {message}")]
+    CASConvergenceFailure { message: String },
+
+    #[error("LockFile failure: {message}")]
+    LockFileFailure { message: String },
+
     #[error("Invalid JSON structure: expected {expected}, found {found}")]
     InvalidJsonStructure { expected: String, found: String },
 
@@ -100,6 +106,24 @@ impl MetastoreError {
                 metadata: std::collections::HashMap::new(),
                 status_code: 400,
             },
+            MetastoreError::CASConvergenceFailure { message } => MetastoreErrorDetail {
+                operation: "CASConvergenceFailure".to_string(),
+                message: message.to_string(),
+                stream_name: None,
+                file_path: None,
+                timestamp: Some(chrono::Utc::now()),
+                metadata: std::collections::HashMap::new(),
+                status_code: 500,
+            },
+            MetastoreError::LockFileFailure { message } => MetastoreErrorDetail {
+                operation: "LockFileFailure".to_string(),
+                message: message.to_string(),
+                stream_name: None,
+                file_path: None,
+                timestamp: Some(chrono::Utc::now()),
+                metadata: std::collections::HashMap::new(),
+                status_code: 500,
+            },
             MetastoreError::ManifestCodecError(e) => MetastoreErrorDetail {
                 operation: "ManifestCodecError".to_string(),
                 message: e.to_string(),
@@ -164,9 +188,11 @@ impl MetastoreError {
             MetastoreError::JsonParseError(..) => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::ManifestCodecError(..) => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::JsonSchemaError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::InvalidJsonStructure { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::InvalidJsonStructure { .. } => StatusCode::BAD_REQUEST,
             MetastoreError::MissingJsonField { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::InvalidJsonValue { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::CASConvergenceFailure { .. }
+            | MetastoreError::LockFileFailure { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             MetastoreError::Error { status_code, .. } => *status_code,
         }
     }

src/migration/mod.rs

@@ -485,9 +485,32 @@ pub async fn setup_logstream_metadata(
         tenant_id.as_deref().unwrap_or(DEFAULT_TENANT),
     );
 
-    let schema = PARSEABLE
-        .get_or_create_stream(stream, tenant_id)
-        .updated_schema(arrow_schema.clone());
+    let stream_ref = PARSEABLE.get_or_create_stream(stream, tenant_id);
+    // Once a dataset is coordinated, its persisted schema uses canonical names
+    // while surviving IPC files still contain raw ingest names. Resolve each
+    // file independently before merging, rather than merging those raw types
+    // directly into the canonical schema during recovery.
+    // Disabling bootstrap must not revert an already canonical dataset to raw
+    // conversion: existing coordinated streams still require their mappings.
+    let schema = if PARSEABLE.options.mode == Mode::Ingest
+        && !static_schema_flag
+        && PARSEABLE
+            .metastore
+            .get_global_schema(stream, tenant_id)
+            .await?
+            .is_some()
+    {
+        stream_ref
+            .restore_global_schema(
+                arrow_schema.clone(),
+                tenant_id,
+                time_partition.as_ref(),
+                custom_partition.as_ref(),
+            )
+            .await?
+    } else {
+        stream_ref.updated_schema(arrow_schema.clone())
+    };
     let schema = HashMap::from_iter(
         schema
             .fields

src/parseable/mod.rs

@@ -34,7 +34,7 @@ use regex::Regex;
 use relative_path::RelativePathBuf;
 pub use staging::StagingError;
 use streams::StreamRef;
-pub use streams::{Stream, StreamNotFound, Streams};
+pub use streams::{GlobalSchema, Stream, StreamNotFound, Streams};
 use tokio::try_join;
 use tracing::error;
 

src/parseable/staging/mod.rs

@@ -42,4 +42,6 @@ pub enum StagingError {
     PoisonError(#[from] PoisonError<String>),
     #[error("JSON Error {0}")]
     Json(#[from] serde_json::Error),
+    #[error("Global schema resolution failed: {0}")]
+    GlobalSchema(String),
 }