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/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/metrics/mod.rs

@@ -25,7 +25,9 @@ use actix_web::Responder;
 use actix_web_prometheus::{PrometheusMetrics, PrometheusMetricsBuilder};
 use error::MetricsError;
 use once_cell::sync::Lazy;
-use prometheus::{HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry};
+use prometheus::{
+    HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry,
+};
 
 pub const METRICS_NAMESPACE: &str = env!("CARGO_PKG_NAME");
 

src/parseable/streams.rs

@@ -742,7 +742,8 @@ impl Stream {
             let column_path = ColumnPath::new(vec!["metric_name".to_string()]);
             props = props
                 .set_column_bloom_filter_enabled(column_path.clone(), true)
-                .set_column_bloom_filter_ndv(column_path, METRIC_NAME_BLOOM_FILTER_NDV);
+                .set_column_bloom_filter_ndv(column_path, METRIC_NAME_BLOOM_FILTER_NDV)
+                .set_bloom_filter_position(parquet::file::properties::BloomFilterPosition::End);
         }
         sorting_column_vec.push(SortingColumn {
             column_idx: time_partition_idx as i32,