perf: decode shuffle blocks against a cached schema instead of re-parsing per block

native/Cargo.toml

@@ -39,6 +39,7 @@ rust-version = "1.94.0"
 
 [workspace.dependencies]
 arrow = { version = "59.2.0", features = ["prettyprint", "ffi", "chrono-tz"] }
+arrow-data = { version = "59.2.0" }
 arrow-select = { version = "59.2.0" }
 async-trait = { version = "0.1" }
 bytes = { version = "1.11.1" }

native/shuffle/Cargo.toml

@@ -30,6 +30,7 @@ publish = false
 
 [dependencies]
 arrow = { workspace = true }
+arrow-data = { workspace = true }
 arrow-select = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }

native/shuffle/benches/shuffle_reader.rs

@@ -24,7 +24,9 @@ use arrow::ipc::reader::StreamReader;
 use arrow::ipc::writer::IpcWriteContext;
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 use datafusion::physical_plan::metrics::Time;
-use datafusion_comet_shuffle::{read_ipc_compressed, CompressionCodec, ShuffleBlockWriter};
+use datafusion_comet_shuffle::{
+    read_ipc_compressed, reset_schema_cache, CompressionCodec, ShuffleBlockWriter,
+};
 use std::hint::black_box;
 use std::io::Cursor;
 use std::sync::Arc;
@@ -100,6 +102,18 @@ fn criterion_benchmark(c: &mut Criterion) {
                 |b, block| b.iter(|| black_box(read_ipc_compressed(black_box(block)).unwrap())),
             );
 
+            // same decode with the cache cleared each iteration, so drift moves both arms together
+            group.bench_with_input(
+                BenchmarkId::new("decode_block_uncached", &id),
+                &uncompressed,
+                |b, block| {
+                    b.iter(|| {
+                        reset_schema_cache();
+                        black_box(read_ipc_compressed(black_box(block)).unwrap())
+                    })
+                },
+            );
+
             // schema parse alone: `try_new` stops before the record batch. Skips the codec tag.
             group.bench_with_input(
                 BenchmarkId::new("parse_schema_only", &id),