@@ -50,6 +50,92 @@ pub enum SortOrder {
5050pub type SortInfo = ( String , SortOrder ) ;
5151pub const CURRENT_MANIFEST_VERSION : & str = "v1" ;
5252
53+ /// Leading bytes of a zstd frame, used to tell a compressed manifest from a
54+ /// plain JSON one. Manifests written before compression was introduced are
55+ /// still read as-is, and the file name is identical either way, so this is the
56+ /// only thing distinguishing the two encodings.
57+ const ZSTD_MAGIC : [ u8 ; 4 ] = [ 0x28 , 0xB5 , 0x2F , 0xFD ] ;
58+
59+ /// Compression level for manifests. Manifest JSON repeats the same column names
60+ /// across thousands of file entries, so it compresses roughly 15-25x even at
61+ /// this cheap level; higher levels cost write time for very little extra.
62+ const ZSTD_LEVEL : i32 = 3 ;
63+
64+ /// Cap on manifests being decoded at once by [`decode_manifest_blocking`].
65+ ///
66+ /// Decoding is CPU bound, so there is nothing to gain past the core count, and
67+ /// each in-flight decode holds a fully decompressed manifest in memory — which
68+ /// is two orders of magnitude larger than the compressed form.
69+ static DECODE_PERMITS : std:: sync:: LazyLock < std:: sync:: Arc < tokio:: sync:: Semaphore > > =
70+ std:: sync:: LazyLock :: new ( || std:: sync:: Arc :: new ( tokio:: sync:: Semaphore :: new ( num_cpus:: get ( ) ) ) ) ;
71+
72+ #[ derive( Debug , thiserror:: Error ) ]
73+ pub enum ManifestCodecError {
74+ #[ error( "failed to compress manifest: {0}" ) ]
75+ Compress ( std:: io:: Error ) ,
76+
77+ #[ error( "manifest decode task failed: {0}" ) ]
78+ Join ( #[ from] tokio:: task:: JoinError ) ,
79+
80+ #[ error( "failed to decompress manifest: {0}" ) ]
81+ Decompress ( std:: io:: Error ) ,
82+
83+ #[ error( "failed to serialize manifest: {0}" ) ]
84+ Serialize ( serde_json:: Error ) ,
85+
86+ #[ error( "failed to parse manifest: {0}" ) ]
87+ Parse ( serde_json:: Error ) ,
88+ }
89+
90+ /// Encode a manifest for storage. This and [`decode_manifest`] are the only
91+ /// places manifest bytes are produced or interpreted; going around them would
92+ /// write a manifest nothing else can read.
93+ pub fn encode_manifest ( manifest : & Manifest ) -> Result < bytes:: Bytes , ManifestCodecError > {
94+ let json = serde_json:: to_vec ( manifest) . map_err ( ManifestCodecError :: Serialize ) ?;
95+ let compressed =
96+ zstd:: encode_all ( json. as_slice ( ) , ZSTD_LEVEL ) . map_err ( ManifestCodecError :: Compress ) ?;
97+ Ok ( bytes:: Bytes :: from ( compressed) )
98+ }
99+
100+ /// Decode a manifest, accepting both the compressed and the plain JSON form.
101+ ///
102+ /// Existing manifests are never rewritten, so uncompressed ones stay readable
103+ /// indefinitely rather than being migrated.
104+ pub fn decode_manifest ( bytes : & [ u8 ] ) -> Result < Manifest , ManifestCodecError > {
105+ if bytes. starts_with ( & ZSTD_MAGIC ) {
106+ let json = zstd:: decode_all ( bytes) . map_err ( ManifestCodecError :: Decompress ) ?;
107+ serde_json:: from_slice ( & json) . map_err ( ManifestCodecError :: Parse )
108+ } else {
109+ serde_json:: from_slice ( bytes) . map_err ( ManifestCodecError :: Parse )
110+ }
111+ }
112+
113+ /// Decode a manifest on the blocking pool.
114+ ///
115+ /// Decompressing and parsing a manifest is hundreds of milliseconds of pure
116+ /// CPU. Run inline on an async task it serialises against every other decode
117+ /// that task drives — callers fetch manifests with `buffered(..)`, which gives
118+ /// concurrent I/O but polls all those futures from one task, so the decodes
119+ /// queue up behind each other and each pending GET appears to take longer and
120+ /// longer. Handing the work to the blocking pool lets them actually run in
121+ /// parallel.
122+ pub async fn decode_manifest_blocking ( bytes : bytes:: Bytes ) -> Result < Manifest , ManifestCodecError > {
123+ // Owned permit moved into the task: dropping this future cancels the await
124+ // but not the blocking work, so the permit has to outlive it to keep
125+ // detached decodes counted.
126+ let permit = DECODE_PERMITS
127+ . clone ( )
128+ . acquire_owned ( )
129+ . await
130+ . expect ( "decode semaphore is never closed" ) ;
131+
132+ tokio:: task:: spawn_blocking ( move || {
133+ let _permit = permit;
134+ decode_manifest ( & bytes)
135+ } )
136+ . await ?
137+ }
138+
53139/// An entry in a manifest which points to a single file.
54140/// Additionally, it is meant to store the statistics for the file it
55141/// points to. Used for pruning file at planning level.
@@ -199,3 +285,101 @@ fn column_statistics(row_groups: &[RowGroupMetaData]) -> HashMap<String, Column>
199285 }
200286 columns
201287}
288+
289+ #[ cfg( test) ]
290+ mod codec_tests {
291+ use super :: * ;
292+
293+ fn sample_manifest ( file_count : usize ) -> Manifest {
294+ let files = ( 0 ..file_count)
295+ . map ( |i| File {
296+ file_path : format ! ( "stream/date=2026-07-29/hour=00/file-{i}.parquet" ) ,
297+ num_rows : 262_144 ,
298+ file_size : 1_048_576 ,
299+ ingestion_size : 2_097_152 ,
300+ columns : ( 0 ..64 )
301+ . map ( |c| Column {
302+ name : format ! ( "some_reasonably_long_column_name_{c}" ) ,
303+ stats : None ,
304+ uncompressed_size : 1024 ,
305+ compressed_size : 512 ,
306+ } )
307+ . collect ( ) ,
308+ sort_order_id : Vec :: new ( ) ,
309+ } )
310+ . collect ( ) ;
311+ Manifest {
312+ version : CURRENT_MANIFEST_VERSION . to_string ( ) ,
313+ files,
314+ }
315+ }
316+
317+ #[ test]
318+ fn round_trips ( ) {
319+ let manifest = sample_manifest ( 4 ) ;
320+ let decoded = decode_manifest ( & encode_manifest ( & manifest) . unwrap ( ) ) . unwrap ( ) ;
321+
322+ assert_eq ! ( decoded. version, manifest. version) ;
323+ assert_eq ! ( decoded. files. len( ) , manifest. files. len( ) ) ;
324+ assert_eq ! ( decoded. files[ 0 ] . file_path, manifest. files[ 0 ] . file_path) ;
325+ assert_eq ! (
326+ decoded. files[ 0 ] . columns. len( ) ,
327+ manifest. files[ 0 ] . columns. len( )
328+ ) ;
329+ }
330+
331+ #[ test]
332+ fn encodes_as_zstd ( ) {
333+ let encoded = encode_manifest ( & sample_manifest ( 1 ) ) . unwrap ( ) ;
334+ assert ! (
335+ encoded. starts_with( & ZSTD_MAGIC ) ,
336+ "manifest was not compressed"
337+ ) ;
338+ }
339+
340+ /// Manifests written before compression are never rewritten, so plain JSON
341+ /// has to stay readable forever.
342+ #[ test]
343+ fn reads_uncompressed_manifests ( ) {
344+ let manifest = sample_manifest ( 3 ) ;
345+ let plain = serde_json:: to_vec ( & manifest) . unwrap ( ) ;
346+ assert ! ( !plain. starts_with( & ZSTD_MAGIC ) ) ;
347+
348+ let decoded = decode_manifest ( & plain) . unwrap ( ) ;
349+ assert_eq ! ( decoded. files. len( ) , 3 ) ;
350+ assert_eq ! ( decoded. files[ 2 ] . file_path, manifest. files[ 2 ] . file_path) ;
351+ }
352+
353+ /// Guards against the codec silently degrading to a no-op: this shape is
354+ /// the whole reason compression is worth doing.
355+ #[ test]
356+ fn compresses_repetitive_manifests_substantially ( ) {
357+ let manifest = sample_manifest ( 64 ) ;
358+ let plain_len = serde_json:: to_vec ( & manifest) . unwrap ( ) . len ( ) ;
359+ let encoded_len = encode_manifest ( & manifest) . unwrap ( ) . len ( ) ;
360+
361+ assert ! (
362+ encoded_len * 10 < plain_len,
363+ "expected >10x compression, got {plain_len} -> {encoded_len}"
364+ ) ;
365+ }
366+
367+ #[ test]
368+ fn rejects_truncated_compressed_input ( ) {
369+ let encoded = encode_manifest ( & sample_manifest ( 8 ) ) . unwrap ( ) ;
370+ let truncated = & encoded[ ..encoded. len ( ) / 2 ] ;
371+
372+ assert ! ( matches!(
373+ decode_manifest( truncated) ,
374+ Err ( ManifestCodecError :: Decompress ( _) )
375+ ) ) ;
376+ }
377+
378+ #[ test]
379+ fn rejects_garbage_input ( ) {
380+ assert ! ( matches!(
381+ decode_manifest( b"not json, not zstd" ) ,
382+ Err ( ManifestCodecError :: Parse ( _) )
383+ ) ) ;
384+ }
385+ }
0 commit comments