perf: parallelize compute_metadata() with rayon

Materialize all secondary slivers upfront and use par_iter over all
n_shards columns for primary encoding + hashing, matching the pattern
in encode_with_metadata(). Each rayon task creates its own encoder.
Replaces two sequential encoding loops with a single parallel pass.

Author: wbbradley

crates/walrus-core/src/encoding/blob_encoding.rs

@@ -441,16 +441,19 @@ impl<'a> BlobEncoder<'a> {
             u64::try_from(self.blob.len()).expect("any valid blob size fits into a `u64`");
         let n_shards = self.inner.n_shards_usize();
 
-        let n_non_systematic_secondary_slivers = n_shards - self.inner.n_columns_usize();
-        // Use a dummy sliver index for the non-systematic secondary slivers.
-        let mut non_systematic_secondary_slivers =
-            vec![
-                self.inner.empty_sliver::<Secondary>(SliverIndex(0));
-                n_non_systematic_secondary_slivers
-            ];
-
-        // Compute the non-systematic secondary slivers by encoding the rows (i.e., primary
-        // slivers).
+        // Materialize all secondary slivers (systematic + non-systematic).
+        let mut secondary_slivers = self.inner.empty_slivers::<Secondary>();
+
+        // Copy systematic columns from blob data.
+        for (column, sliver) in self.column_symbols().zip(secondary_slivers.iter_mut()) {
+            sliver
+                .symbols
+                .to_symbols_mut()
+                .zip(column)
+                .for_each(|(dest, src)| dest[..src.len()].copy_from_slice(src))
+        }
+
+        // Compute non-systematic secondary slivers by encoding rows (sequential).
         let mut secondary_encoder = self.inner.get_encoder::<Secondary>();
         let mut buffer = Symbols::zeros(self.inner.n_columns_usize(), self.inner.symbol_size);
         let row_length_bytes =
@@ -467,48 +470,39 @@ impl<'a> BlobEncoder<'a> {
             let encode_result = secondary_encoder
                 .encode(data)
                 .expect("size has already been checked");
-            for (symbol, sliver) in encode_result
-                .recovery_iter()
-                .zip(non_systematic_secondary_slivers.iter_mut())
-            {
+            for (symbol, sliver) in encode_result.recovery_iter().zip(
+                secondary_slivers
+                    .iter_mut()
+                    .skip(self.inner.n_columns_usize()),
+            ) {
                 sliver.copy_symbol_to(r, symbol);
             }
         }
         drop(secondary_encoder);
 
-        // Now we can encode all secondary slivers, computing all symbol hashes.
+        // Parallel primary encoding + hashing over all secondary slivers.
         let mut symbol_hashes = vec![Node::Empty; n_shards * n_shards];
 
-        // First encode the systematic secondary slivers.
-        let mut primary_encoder = self.inner.get_encoder::<Primary>();
-        let mut buffer = Symbols::zeros(self.inner.n_rows_usize(), self.inner.symbol_size);
-        for (col_index, column_symbols) in self.column_symbols().enumerate() {
-            buffer.data_mut().fill(0);
-            buffer
-                .to_symbols_mut()
-                .zip(column_symbols)
-                .for_each(|(dest, src)| dest[..src.len()].copy_from_slice(src));
-            let symbols = primary_encoder
-                .encode_all_ref(buffer.data())
-                .expect("size has already been checked");
-            for (row_index, symbol) in symbols.to_symbols().enumerate() {
-                symbol_hashes[n_shards * row_index + col_index] = leaf_hash::<Blake2b256>(symbol);
-            }
-        }
+        let column_results: Vec<(usize, Vec<Node>)> = secondary_slivers
+            .par_iter()
+            .enumerate()
+            .map(|(col_index, column)| {
+                let mut encoder = self.inner.get_encoder::<Primary>();
+                let symbols = encoder
+                    .encode_all(column.symbols.data())
+                    .expect("size has already been checked");
+                let hashes: Vec<Node> = symbols.to_symbols().map(leaf_hash::<Blake2b256>).collect();
+                (col_index, hashes)
+            })
+            .collect();
 
-        // Then encode the non-systematic secondary slivers.
-        for (col_index, column) in non_systematic_secondary_slivers.iter().enumerate() {
-            let col_index = col_index + self.inner.n_columns_usize();
-            let symbols = primary_encoder
-                .encode_all_ref(column.symbols.data())
-                .expect("size has already been checked");
-            for (row_index, symbol) in symbols.to_symbols().enumerate() {
-                symbol_hashes[n_shards * row_index + col_index] = leaf_hash::<Blake2b256>(symbol);
+        // Sequential scatter.
+        for (col_index, hashes) in column_results {
+            for (row_index, hash) in hashes.into_iter().enumerate() {
+                symbol_hashes[n_shards * row_index + col_index] = hash;
             }
         }
 
-        drop(primary_encoder);
-
         BlobEncoderData::compute_metadata_from_symbol_hashes(
             self.inner.config,
             &symbol_hashes,