Zstd manifest

Cargo.toml

@@ -186,6 +186,7 @@ thiserror = "2.0"
 ulid = { version = "1.2.1", features = ["serde"] }
 uuid = { version = "1.23.3", features = ["v4"] }
 xxhash-rust = { version = "0.8.15", features = ["xxh3"] }
+zstd = "0.13.3"
 rustc-hash = "2.1.2"
 futures-core = "0.3.31"
 tempfile = "3.20.0"

src/catalog/manifest.rs

@@ -50,6 +50,84 @@ pub enum SortOrder {
 pub type SortInfo = (String, SortOrder);
 pub const CURRENT_MANIFEST_VERSION: &str = "v1";
 
+/// Leading bytes of a zstd frame, used to tell a compressed manifest from a
+/// plain JSON one. Manifests written before compression was introduced are
+/// still read as-is, and the file name is identical either way, so this is the
+/// only thing distinguishing the two encodings.
+const ZSTD_MAGIC: [u8; 4] = [0x28, 0xB5, 0x2F, 0xFD];
+
+/// Compression level for manifests. Manifest JSON repeats the same column names
+/// across thousands of file entries, so it compresses roughly 15-25x even at
+/// this cheap level; higher levels cost write time for very little extra.
+const ZSTD_LEVEL: i32 = 3;
+
+/// Cap on manifests being decoded at once by [`decode_manifest_blocking`].
+///
+/// Decoding is CPU bound, so there is nothing to gain past the core count, and
+/// each in-flight decode holds a fully decompressed manifest in memory — which
+/// is two orders of magnitude larger than the compressed form.
+static DECODE_PERMITS: std::sync::LazyLock<tokio::sync::Semaphore> =
+    std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(num_cpus::get()));
+
+#[derive(Debug, thiserror::Error)]
+pub enum ManifestCodecError {
+    #[error("failed to compress manifest: {0}")]
+    Compress(std::io::Error),
+
+    #[error("manifest decode task failed: {0}")]
+    Join(#[from] tokio::task::JoinError),
+
+    #[error("failed to decompress manifest: {0}")]
+    Decompress(std::io::Error),
+
+    #[error("failed to serialize manifest: {0}")]
+    Serialize(serde_json::Error),
+
+    #[error("failed to parse manifest: {0}")]
+    Parse(serde_json::Error),
+}
+
+/// Encode a manifest for storage. This and [`decode_manifest`] are the only
+/// places manifest bytes are produced or interpreted; going around them would
+/// write a manifest nothing else can read.
+pub fn encode_manifest(manifest: &Manifest) -> Result<bytes::Bytes, ManifestCodecError> {
+    let json = serde_json::to_vec(manifest).map_err(ManifestCodecError::Serialize)?;
+    let compressed =
+        zstd::encode_all(json.as_slice(), ZSTD_LEVEL).map_err(ManifestCodecError::Compress)?;
+    Ok(bytes::Bytes::from(compressed))
+}
+
+/// Decode a manifest, accepting both the compressed and the plain JSON form.
+///
+/// Existing manifests are never rewritten, so uncompressed ones stay readable
+/// indefinitely rather than being migrated.
+pub fn decode_manifest(bytes: &[u8]) -> Result<Manifest, ManifestCodecError> {
+    if bytes.starts_with(&ZSTD_MAGIC) {
+        let json = zstd::decode_all(bytes).map_err(ManifestCodecError::Decompress)?;
+        serde_json::from_slice(&json).map_err(ManifestCodecError::Parse)
+    } else {
+        serde_json::from_slice(bytes).map_err(ManifestCodecError::Parse)
+    }
+}
+
+/// Decode a manifest on the blocking pool.
+///
+/// Decompressing and parsing a manifest is hundreds of milliseconds of pure
+/// CPU. Run inline on an async task it serialises against every other decode
+/// that task drives — callers fetch manifests with `buffered(..)`, which gives
+/// concurrent I/O but polls all those futures from one task, so the decodes
+/// queue up behind each other and each pending GET appears to take longer and
+/// longer. Handing the work to the blocking pool lets them actually run in
+/// parallel.
+pub async fn decode_manifest_blocking(bytes: bytes::Bytes) -> Result<Manifest, ManifestCodecError> {
+    let _permit = DECODE_PERMITS
+        .acquire()
+        .await
+        .expect("decode semaphore is never closed");
+
+    tokio::task::spawn_blocking(move || decode_manifest(&bytes)).await?
+}
+
 /// An entry in a manifest which points to a single file.
 /// Additionally, it is meant to store the statistics for the file it
 /// points to. Used for pruning file at planning level.
@@ -199,3 +277,101 @@ fn column_statistics(row_groups: &[RowGroupMetaData]) -> HashMap<String, Column>
     }
     columns
 }
+
+#[cfg(test)]
+mod codec_tests {
+    use super::*;
+
+    fn sample_manifest(file_count: usize) -> Manifest {
+        let files = (0..file_count)
+            .map(|i| File {
+                file_path: format!("stream/date=2026-07-29/hour=00/file-{i}.parquet"),
+                num_rows: 262_144,
+                file_size: 1_048_576,
+                ingestion_size: 2_097_152,
+                columns: (0..64)
+                    .map(|c| Column {
+                        name: format!("some_reasonably_long_column_name_{c}"),
+                        stats: None,
+                        uncompressed_size: 1024,
+                        compressed_size: 512,
+                    })
+                    .collect(),
+                sort_order_id: Vec::new(),
+            })
+            .collect();
+        Manifest {
+            version: CURRENT_MANIFEST_VERSION.to_string(),
+            files,
+        }
+    }
+
+    #[test]
+    fn round_trips() {
+        let manifest = sample_manifest(4);
+        let decoded = decode_manifest(&encode_manifest(&manifest).unwrap()).unwrap();
+
+        assert_eq!(decoded.version, manifest.version);
+        assert_eq!(decoded.files.len(), manifest.files.len());
+        assert_eq!(decoded.files[0].file_path, manifest.files[0].file_path);
+        assert_eq!(
+            decoded.files[0].columns.len(),
+            manifest.files[0].columns.len()
+        );
+    }
+
+    #[test]
+    fn encodes_as_zstd() {
+        let encoded = encode_manifest(&sample_manifest(1)).unwrap();
+        assert!(
+            encoded.starts_with(&ZSTD_MAGIC),
+            "manifest was not compressed"
+        );
+    }
+
+    /// Manifests written before compression are never rewritten, so plain JSON
+    /// has to stay readable forever.
+    #[test]
+    fn reads_uncompressed_manifests() {
+        let manifest = sample_manifest(3);
+        let plain = serde_json::to_vec(&manifest).unwrap();
+        assert!(!plain.starts_with(&ZSTD_MAGIC));
+
+        let decoded = decode_manifest(&plain).unwrap();
+        assert_eq!(decoded.files.len(), 3);
+        assert_eq!(decoded.files[2].file_path, manifest.files[2].file_path);
+    }
+
+    /// Guards against the codec silently degrading to a no-op: this shape is
+    /// the whole reason compression is worth doing.
+    #[test]
+    fn compresses_repetitive_manifests_substantially() {
+        let manifest = sample_manifest(64);
+        let plain_len = serde_json::to_vec(&manifest).unwrap().len();
+        let encoded_len = encode_manifest(&manifest).unwrap().len();
+
+        assert!(
+            encoded_len * 10 < plain_len,
+            "expected >10x compression, got {plain_len} -> {encoded_len}"
+        );
+    }
+
+    #[test]
+    fn rejects_truncated_compressed_input() {
+        let encoded = encode_manifest(&sample_manifest(8)).unwrap();
+        let truncated = &encoded[..encoded.len() / 2];
+
+        assert!(matches!(
+            decode_manifest(truncated),
+            Err(ManifestCodecError::Decompress(_))
+        ));
+    }
+
+    #[test]
+    fn rejects_garbage_input() {
+        assert!(matches!(
+            decode_manifest(b"not json, not zstd"),
+            Err(ManifestCodecError::Parse(_))
+        ));
+    }
+}

src/event/format/json.rs

@@ -526,9 +526,38 @@ fn validate_struct(
 mod tests {
     use std::str::FromStr;
 
+    use arrow_array::UInt64Array;
     use serde_json::json;
 
     use super::*;
+    use crate::otel::metrics::SERIES_HASH_COLUMN;
+
+    #[test]
+    fn series_hash_is_decoded_as_uint64_without_precision_loss() {
+        let hash = u64::MAX;
+        let event = Event::new(json!({ "__series_hash_u64": hash }), Utc::now());
+
+        let (data, fields, _) = event
+            .to_data(&HashMap::new(), None, SchemaVersion::V1, false, true)
+            .unwrap();
+        let schema = Arc::new(Schema::new(fields));
+        assert_eq!(
+            schema
+                .field_with_name(SERIES_HASH_COLUMN)
+                .unwrap()
+                .data_type(),
+            &DataType::UInt64
+        );
+
+        let batch = Event::decode(data, schema).unwrap();
+        let hashes = batch
+            .column_by_name(SERIES_HASH_COLUMN)
+            .unwrap()
+            .as_any()
+            .downcast_ref::<UInt64Array>()
+            .unwrap();
+        assert_eq!(hashes.value(0), hash);
+    }
 
     #[test]
     fn parse_time_parition_from_value() {

src/event/format/mod.rs

@@ -34,6 +34,7 @@ use tracing::info_span;
 use crate::{
     handlers::TelemetryType,
     metadata::SchemaVersion,
+    otel::metrics::SERIES_HASH_COLUMN,
     storage::StreamType,
     utils::arrow::{add_parseable_fields, get_field},
 };
@@ -369,6 +370,12 @@ pub fn override_data_type(
             let mut field_name = field.name().to_string();
             normalize_field_name(&mut field_name);
             match (schema_version, map.get(field.name())) {
+                // Series identity must retain all 64 bits. Schema V1 normally
+                // coerces JSON numbers to Float64, which cannot represent the
+                // full u64 range exactly.
+                (_, Some(Value::Number(_))) if field_name == SERIES_HASH_COLUMN => {
+                    Field::new(field_name, DataType::UInt64, true)
+                }
                 // in V1 for new fields in json named "time"/"date" or such and having
                 // inferred type string, that can be parsed as timestamp, use the
                 // timestamp type. Gated on `infer_timestamp` (default true) — settable

src/metastore/metastore_traits.rs

@@ -277,9 +277,12 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         manifest_url: Option<String>,
         tenant_id: &Option<String>,
     ) -> Result<Option<Manifest>, MetastoreError>;
+    /// Takes a concrete `&Manifest` rather than `&dyn MetastoreObject` because
+    /// manifests have their own encoding (see `catalog::manifest::encode_manifest`)
+    /// and must not go through the generic JSON path used by other metadata.
     async fn put_manifest(
         &self,
-        obj: &dyn MetastoreObject,
+        manifest: &Manifest,
         stream_name: &str,
         lower_bound: DateTime<Utc>,
         upper_bound: DateTime<Utc>,

src/metastore/metastores/object_store_metastore.rs

@@ -38,7 +38,10 @@ use crate::{
         outbound_http_policy::AlertTargetPolicyConfig,
         target::Target,
     },
-    catalog::{manifest::Manifest, partition_path},
+    catalog::{
+        manifest::{Manifest, decode_manifest_blocking, encode_manifest},
+        partition_path,
+    },
     handlers::http::{
         modal::{Metadata, NodeMetadata, NodeType},
         users::USERS_ROOT_DIR,
@@ -1020,7 +1023,7 @@ impl Metastore for ObjectStoreMetastore {
                                 .storage
                                 .get_object(&RelativePathBuf::from(path), &tenant)
                                 .await?;
-                            Ok::<Manifest, MetastoreError>(serde_json::from_slice(&bytes)?)
+                            Ok::<Manifest, MetastoreError>(decode_manifest_blocking(bytes).await?)
                         }
                     })
                     .buffer_unordered(16)
@@ -1083,7 +1086,7 @@ impl Metastore for ObjectStoreMetastore {
         };
         match self.storage.get_object(&path, tenant_id).await {
             Ok(bytes) => {
-                let manifest = serde_json::from_slice(&bytes)?;
+                let manifest = decode_manifest_blocking(bytes).await?;
                 Ok(Some(manifest))
             }
             Err(ObjectStorageError::NoSuchKey(_)) => Ok(None),
@@ -1120,7 +1123,7 @@ impl Metastore for ObjectStoreMetastore {
 
     async fn put_manifest(
         &self,
-        obj: &dyn MetastoreObject,
+        manifest: &Manifest,
         stream_name: &str,
         lower_bound: DateTime<Utc>,
         upper_bound: DateTime<Utc>,
@@ -1132,7 +1135,7 @@ impl Metastore for ObjectStoreMetastore {
 
         Ok(self
             .storage
-            .put_object(&path, to_bytes(obj), tenant_id)
+            .put_object(&path, encode_manifest(manifest)?, tenant_id)
             .await?)
     }
 

src/metastore/mod.rs

@@ -43,6 +43,9 @@ pub enum MetastoreError {
     #[error("JSON parsing error: {0}")]
     JsonParseError(#[from] serde_json::Error),
 
+    #[error("Manifest encoding error: {0}")]
+    ManifestCodecError(#[from] crate::catalog::manifest::ManifestCodecError),
+
     #[error("JSON schema validation error: {message}")]
     JsonSchemaError { message: String },
 
@@ -97,6 +100,15 @@ impl MetastoreError {
                 metadata: std::collections::HashMap::new(),
                 status_code: 400,
             },
+            MetastoreError::ManifestCodecError(e) => MetastoreErrorDetail {
+                operation: "ManifestCodecError".to_string(),
+                message: e.to_string(),
+                stream_name: None,
+                file_path: None,
+                timestamp: Some(chrono::Utc::now()),
+                metadata: std::collections::HashMap::new(),
+                status_code: 500,
+            },
             MetastoreError::JsonSchemaError { message } => MetastoreErrorDetail {
                 operation: "JsonSchemaError".to_string(),
                 message: message.clone(),
@@ -150,6 +162,7 @@ impl MetastoreError {
         match self {
             MetastoreError::ObjectStorageError(..) => StatusCode::INTERNAL_SERVER_ERROR,
             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::MissingJsonField { .. } => StatusCode::INTERNAL_SERVER_ERROR,

src/otel/metrics.rs

@@ -37,6 +37,8 @@ use super::otel_utils::{
     convert_epoch_nano_to_timestamp, insert_attributes, insert_number_if_some,
 };
 
+pub const SERIES_HASH_COLUMN: &str = "__series_hash_u64";
+
 pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 38] = [
     "metric_name",
     "metric_description",
@@ -80,14 +82,11 @@ pub const OTEL_METRICS_KNOWN_FIELD_LIST: [&str; 38] = [
     "resource_dropped_attributes_count",
     "resource_schema_url",
     // Precomputed per-sample identity of the physical series. Stable
-    // u64 hash of `metric_name` + sorted attribute key/value pairs,
-    // stored as a decimal-encoded string so arrow-json infers Utf8 and
-    // we get byte-exact roundtrip. Int64/Float64 inference dropped bits
-    // for hashes near the high range; string sidesteps that entirely.
+    // u64 hash of `metric_name` + sorted attribute key/value pairs.
     // Lets the query layer group samples into physical series via a
     // single column read instead of decoding every label column and
     // hashing per row.
-    "__series_hash",
+    SERIES_HASH_COLUMN,
 ];
 
 static OTEL_METRICS_KNOWN_FIELDS: Lazy<HashSet<&'static str>> =
@@ -673,12 +672,9 @@ fn process_resource_metrics<T, S, M>(
                     // perspective). Computed once per data point — O(label
                     // count) per sample, ~200 ns at typical attribute counts.
                     let series_hash = compute_series_hash(&dp);
-                    // Stored as decimal-encoded string. Arrow-json
-                    // infers Utf8, preserving all 64 bits — Int64/Float64
-                    // inference truncated values near the high range.
                     dp.insert(
-                        "__series_hash".to_string(),
-                        Value::String(series_hash.to_string()),
+                        SERIES_HASH_COLUMN.to_string(),
+                        Value::Number(series_hash.into()),
                     );
                     vec_otel_json.push(Value::Object(dp));
                 }