replace etag, version with hash

Author: parmesant

src/handlers/http/mod.rs

@@ -144,7 +144,7 @@ pub async fn fetch_schema(
         .get_global_schema(stream_name, tenant_id)
         .await?
     {
-        return Ok(serde_json::from_slice(&global.schema)?);
+        return Ok(global.schema);
     }
     let res: Vec<Schema> = PARSEABLE
         .metastore

src/metastore/metastore_traits.rs

@@ -23,7 +23,6 @@ use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use dashmap::DashMap;
 use erased_serde::Serialize as ErasedSerialize;
-use object_store::UpdateVersion;
 use tonic::async_trait;
 use ulid::Ulid;
 
@@ -304,10 +303,10 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
     ) -> Result<Option<GlobalSchema>, MetastoreError>;
     async fn put_global_schema(
         &self,
-        obj: Schema,
+        obj: GlobalSchema,
         stream_name: &str,
         tenant_id: &Option<String>,
-        expected: Option<UpdateVersion>,
+        expected: Option<u64>,
     ) -> Result<bool, MetastoreError>;
 
     /// parseable metadata

src/metastore/metastores/object_store_metastore.rs

@@ -1184,7 +1184,8 @@ impl Metastore for ObjectStoreMetastore {
         tenant_id: &Option<String>,
     ) -> Result<Bytes, MetastoreError> {
         if let Some(global_schema) = self.get_global_schema(stream_name, tenant_id).await? {
-            return Ok(global_schema.schema);
+            let schema = to_bytes(&global_schema.schema);
+            return Ok(schema);
         }
 
         Ok(self
@@ -1214,24 +1215,23 @@ impl Metastore for ObjectStoreMetastore {
     ) -> Result<Option<GlobalSchema>, MetastoreError> {
         // LocalFS does not participate in the shared registry. Only bypass
         // explicitly unsupported backends; propagate errors from cloud stores.
-        if !self.storage.supports_conditional_writes() {
-            return Ok(None);
-        }
         let path = global_schema_path(stream_name, tenant_id);
+        let global_schema =
+            serde_json::from_slice(&self.storage.get_object(&path, tenant_id).await?)?;
         let Some((schema, version)) = self.storage.get_object_versioned(&path, tenant_id).await?
         else {
             return Ok(None);
         };
-        Ok(Some(GlobalSchema::new(schema, version)))
+        Ok(Some(global_schema))
     }
 
     /// Atomically create a missing global schema or compare-and-swap an existing one.
     async fn put_global_schema(
         &self,
-        obj: Schema,
+        obj: GlobalSchema,
         stream_name: &str,
         tenant_id: &Option<String>,
-        expected: Option<UpdateVersion>,
+        expected: Option<u64>,
     ) -> Result<bool, MetastoreError> {
         let path = global_schema_path(stream_name, tenant_id);
         Ok(self

src/parseable/streams.rs

@@ -23,7 +23,6 @@ use arrow::{
 };
 use arrow_array::{ArrayRef, RecordBatch, UInt32Array};
 use arrow_schema::{ArrowError, Field, Fields, Schema, SortOptions};
-use bytes::Bytes;
 use chrono::{NaiveDate, NaiveDateTime, Timelike, Utc};
 use derive_more::derive::{Deref, DerefMut};
 use itertools::Itertools;
@@ -52,7 +51,8 @@ use rayon::{
     slice::ParallelSliceMut,
 };
 use relative_path::RelativePathBuf;
-use std::sync::PoisonError;
+use rustc_hash::FxHasher;
+use serde::{Deserialize, Serialize};
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     fs::{self, File, OpenOptions, remove_file, write},
@@ -62,6 +62,10 @@ use std::{
     sync::{Arc, Mutex, RwLock},
     time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
+use std::{
+    hash::{Hash, Hasher},
+    sync::PoisonError,
+};
 use tokio::task::JoinSet;
 use tracing::{error, info, info_span, instrument, trace, warn};
 use ulid::Ulid;
@@ -80,7 +84,7 @@ use crate::{
     option::Mode,
     parseable::{DEFAULT_TENANT, PARSEABLE},
     schema_registry::Resolution,
-    storage::{StreamType, object_storage::to_bytes, retention::Retention},
+    storage::{ComparableBytes, StreamType, object_storage::to_bytes, retention::Retention},
     sync::FLUSH_AND_CONVERT_RUNTIME,
     utils::{
         arrow::adapt_batch,
@@ -374,15 +378,24 @@ fn minute_from_system_time(time: SystemTime) -> u128 {
         / 60000
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct GlobalSchema {
-    pub schema: Bytes,
-    pub version: object_store::UpdateVersion,
+    pub schema: Schema,
+    pub hash: u64,
 }
 
 impl GlobalSchema {
-    pub fn new(schema: Bytes, version: object_store::UpdateVersion) -> Self {
-        Self { schema, version }
+    pub fn from_schema(schema: Schema) -> Self {
+        let mut hasher = FxHasher::with_seed(0);
+        schema.hash(&mut hasher);
+        let hash = hasher.finish();
+        Self { schema, hash }
+    }
+}
+
+impl ComparableBytes for GlobalSchema {
+    fn get_hash(&self) -> u64 {
+        self.hash
     }
 }
 
@@ -1179,7 +1192,7 @@ impl Stream {
         {
             // The registry is authoritative. Raw schemas staged before the
             // first coordinated conversion may contain the losing field type.
-            schema = global.as_ref().clone();
+            schema = global.as_ref().to_owned();
         } else {
             let staging_schemas = self.get_schemas_if_present()?;
             if !staging_schemas.is_empty() {
@@ -1876,12 +1889,11 @@ impl Stream {
     }
 
     fn can_use_global_schema(&self) -> bool {
-        self.options.mode == Mode::Ingest
-            && !self.get_static_schema_flag()
-            && PARSEABLE
-                .storage
-                .get_object_store()
-                .supports_conditional_writes()
+        self.options.mode == Mode::Ingest && !self.get_static_schema_flag()
+        // && PARSEABLE
+        //     .storage
+        //     .get_object_store()
+        //     .supports_conditional_writes()
     }
 
     fn uses_global_schema(&self) -> bool {
@@ -1973,7 +1985,7 @@ impl Stream {
                         "global schema disappeared after resolution".to_owned(),
                     )
                 })?;
-            let canonical = serde_json::from_slice::<Schema>(&global.schema)?;
+            let canonical = global.schema;
             Ok::<_, StagingError>((resolution, Arc::new(canonical)))
         })?;
         resolution.include_identity_schema(&canonical);
@@ -2146,6 +2158,7 @@ impl Stream {
 
     /// Converts one Parquet output group and removes its Arrow sources only
     /// after the final Parquet rename succeeds.
+    #[allow(clippy::too_many_arguments)]
     fn convert_arrow_group(
         &self,
         parquet_path: PathBuf,
@@ -2225,6 +2238,7 @@ impl Stream {
 
     #[cfg_attr(feature = "hotpath", hotpath::measure)]
     /// Writes one conversion group to a temporary Parquet file.
+    #[allow(clippy::too_many_arguments)]
     fn write_parquet_part_file(
         &self,
         part_path: &Path,
@@ -2287,6 +2301,7 @@ impl Stream {
 
     /// Writes a metric conversion group with pipelined row-group preparation
     /// and encoding while preserving row-group output order.
+    #[allow(clippy::too_many_arguments)]
     fn write_metric_parquet_part_file(
         &self,
         part_path: &Path,
@@ -2632,7 +2647,7 @@ impl Stream {
             .ok_or_else(|| {
                 StagingError::GlobalSchema("global schema disappeared during startup".to_owned())
             })?;
-        let canonical = serde_json::from_slice::<Schema>(&global.schema)?;
+        let canonical = global.schema;
         resolution.include_identity_schema(&canonical);
         *self.global_schema.write().expect(LOCK_EXPECT) =
             Some((Arc::new(canonical.clone()), Arc::new(resolution)));

src/schema_registry.rs

@@ -30,7 +30,6 @@ use std::{
 
 use arrow_array::{RecordBatch, RecordBatchOptions};
 use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema};
-use bytes::Bytes;
 use tonic::async_trait;
 
 use crate::{
@@ -49,8 +48,8 @@ trait RegistryBackend: Send + Sync {
     async fn read_legacy_schemas(&self) -> Result<Vec<Schema>, MetastoreError>;
     async fn compare_and_swap(
         &self,
-        schema: Schema,
-        expected: Option<object_store::UpdateVersion>,
+        global_schema: GlobalSchema,
+        expected: Option<u64>,
     ) -> Result<bool, MetastoreError>;
 }
 
@@ -76,11 +75,11 @@ impl RegistryBackend for MetastoreRegistryBackend<'_> {
 
     async fn compare_and_swap(
         &self,
-        schema: Schema,
-        expected: Option<object_store::UpdateVersion>,
+        global_schema: GlobalSchema,
+        expected: Option<u64>,
     ) -> Result<bool, MetastoreError> {
         self.metastore
-            .put_global_schema(schema, self.stream_name, self.tenant_id, expected)
+            .put_global_schema(global_schema, self.stream_name, self.tenant_id, expected)
             .await
     }
 }
@@ -264,7 +263,7 @@ async fn resolve_backend(
         let global = backend.read_global().await?;
         let (canonical, original_schema) = match global.as_ref() {
             Some(global) => {
-                let original = parse_global_schema(&global.schema)?;
+                let original = global.schema.to_owned();
                 let normalized = normalize_registry(&original).map_err(schema_error)?;
                 (normalized, Some(original))
             }
@@ -290,8 +289,11 @@ async fn resolve_backend(
 
         // `None` is explicitly create-only; a snapshot version is an
         // update-only CAS. A failed CAS invalidates this whole resolution.
-        let expected = global.as_ref().map(|snapshot| snapshot.version.clone());
-        if backend.compare_and_swap(proposed, expected).await? {
+        let expected = global.as_ref().map(|snapshot| snapshot.hash.clone());
+        if backend
+            .compare_and_swap(GlobalSchema::from_schema(proposed), expected)
+            .await?
+        {
             return Ok(resolution);
         }
 
@@ -306,10 +308,6 @@ async fn resolve_backend(
     )))
 }
 
-fn parse_global_schema(bytes: &Bytes) -> Result<Schema, MetastoreError> {
-    serde_json::from_slice(bytes).map_err(MetastoreError::from)
-}
-
 fn schema_error(message: impl Into<String>) -> MetastoreError {
     MetastoreError::JsonSchemaError {
         message: message.into(),

src/storage/azure_blob.rs

@@ -623,10 +623,6 @@ impl BlobStore {
 
 #[async_trait]
 impl ObjectStorage for BlobStore {
-    fn supports_conditional_writes(&self) -> bool {
-        true
-    }
-
     async fn parallel_chunked_download(
         &self,
         path: &RelativePath,
@@ -850,12 +846,12 @@ impl ObjectStorage for BlobStore {
         &self,
         path: &RelativePath,
         data: Bytes,
-        expected: Option<object_store::UpdateVersion>,
+        expected: Option<u64>,
         tenant_id: &Option<String>,
     ) -> Result<bool, ObjectStorageError> {
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let date = Utc::now().date_naive().to_string();
-        let result = crate::storage::put_object_if(
+        let result = crate::storage::put_schema_if(
             &*self.client,
             &to_object_store_path(path),
             data,

src/storage/gcs.rs

@@ -590,10 +590,6 @@ impl Gcs {
 
 #[async_trait]
 impl ObjectStorage for Gcs {
-    fn supports_conditional_writes(&self) -> bool {
-        true
-    }
-
     async fn parallel_chunked_download(
         &self,
         path: &RelativePath,
@@ -832,12 +828,12 @@ impl ObjectStorage for Gcs {
         &self,
         path: &RelativePath,
         data: Bytes,
-        expected: Option<object_store::UpdateVersion>,
+        expected: Option<u64>,
         tenant_id: &Option<String>,
     ) -> Result<bool, ObjectStorageError> {
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let date = Utc::now().date_naive().to_string();
-        let result = crate::storage::put_object_if(
+        let result = crate::storage::put_schema_if(
             &*self.client,
             &to_object_store_path(path),
             data,