From efb83c292095c0059049b11943effab2fd2c7933 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:14:58 -0500 Subject: [PATCH 1/2] bench: parquet scan with a table schema narrower than a nested column MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Registers the same wide list / struct parquet file with the file's full schema and with a narrower declared schema, plus a physically-narrow file as the floor. On main the narrow declared schema reads exactly as many bytes as the full schema (the whole column is fetched and clipped in memory by the adapter-inserted cast): list_struct_narrow_schema: bytes_scanned=25.19 MB list_struct_full_schema: bytes_scanned=25.19 MB list_struct_physically_narrow: bytes_scanned=3.32 KB select_events_narrow_schema 3.45 ms select_events_full_schema 3.36 ms select_events_physically_narrow 164 µs Baseline for schema-driven nested projection pruning (see apache/datafusion-comet#4859). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KuMaRtFSPDQesuzjN5Koyd --- datafusion/core/Cargo.toml | 5 + .../benches/parquet_nested_schema_pruning.rs | 390 ++++++++++++++++++ 2 files changed, 395 insertions(+) create mode 100644 datafusion/core/benches/parquet_nested_schema_pruning.rs diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml index 60cff658a6a97..10435220a1f0d 100644 --- a/datafusion/core/Cargo.toml +++ b/datafusion/core/Cargo.toml @@ -247,6 +247,11 @@ harness = false name = "parquet_struct_query" required-features = ["parquet"] +[[bench]] +harness = false +name = "parquet_nested_schema_pruning" +required-features = ["parquet"] + [[bench]] harness = false name = "parquet_struct_projection" diff --git a/datafusion/core/benches/parquet_nested_schema_pruning.rs b/datafusion/core/benches/parquet_nested_schema_pruning.rs new file mode 100644 index 0000000000000..989e0a8890094 --- /dev/null +++ b/datafusion/core/benches/parquet_nested_schema_pruning.rs @@ -0,0 +1,390 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Benchmarks for schema-driven nested projection pruning in Parquet. +//! +//! A table's declared (logical) schema can be *narrower* than the physical +//! parquet type of a nested column — e.g. the table declares +//! `events: LIST>` while the file contains +//! `events: LIST>`. Engines like Spark +//! communicate nested projection pruning to the scan exactly this way +//! (a clipped read schema), so the reader should fetch and decode only the +//! leaves the declared schema names. +//! +//! Each dataset shape is measured three ways: +//! +//! 1. **narrow_schema**: wide file, narrow declared table schema — the +//! interesting case; ideally close to (3) +//! 2. **full_schema**: wide file, full table schema — the cost of reading +//! everything +//! 3. **physically_narrow**: a file that only contains the narrow columns — +//! the floor +//! +//! At setup the benchmark prints the `bytes_scanned` metric for (1) and (2) +//! so the IO pattern is visible in addition to wall time. + +use arrow::array::{ + ArrayRef, Int32Array, Int64Array, ListArray, StringArray, StructArray, +}; +use arrow::buffer::OffsetBuffer; +use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; +use arrow::record_batch::RecordBatch; +use arrow::util::pretty::pretty_format_batches; +use criterion::{Criterion, criterion_group, criterion_main}; +use datafusion::datasource::listing::{ + ListingTable, ListingTableConfig, ListingTableConfigExt, +}; +use datafusion::prelude::SessionContext; +use datafusion_datasource::ListingTableUrl; +use parquet::arrow::ArrowWriter; +use parquet::file::properties::{WriterProperties, WriterVersion}; +use std::hint::black_box; +use std::sync::Arc; +use std::time::Duration; +use tempfile::NamedTempFile; +use tokio::runtime::Runtime; + +const NUM_BATCHES: usize = 2; +const ROWS_PER_BATCH: usize = 256; +const ROW_GROUP_ROW_COUNT: usize = 256; +const ELEMS_PER_ROW: usize = 3; +const NUM_PAD_FIELDS: usize = 8; +const PAD_LEN: usize = 2048; + +/// The narrow item fields: the subset of the struct the table declares. +fn narrow_item_fields() -> Fields { + Fields::from(vec![ + Field::new("x", DataType::Int64, true), + Field::new("y", DataType::Utf8, true), + ]) +} + +/// The wide item fields as written to the file: the narrow fields plus +/// `NUM_PAD_FIELDS` fat string fields the table schema does not mention. +fn wide_item_fields() -> Fields { + let mut fields = vec![ + Field::new("x", DataType::Int64, false), + Field::new("y", DataType::Utf8, true), + ]; + for i in 0..NUM_PAD_FIELDS { + fields.push(Field::new(format!("pad_{i}"), DataType::Utf8, false)); + } + Fields::from(fields) +} + +fn list_schema(item_fields: Fields) -> SchemaRef { + let item = Arc::new(Field::new("item", DataType::Struct(item_fields), true)); + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("events", DataType::List(item), true), + ])) +} + +fn struct_schema(item_fields: Fields) -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("s", DataType::Struct(item_fields), true), + ])) +} + +/// Distinct pad values so dictionary encoding cannot collapse them. +fn pad_values(count: usize, seed: usize) -> ArrayRef { + let base = "x".repeat(PAD_LEN); + let values: Vec = (0..count) + .map(|i| format!("{:08}{base}", seed + i)) + .collect(); + Arc::new(StringArray::from(values)) +} + +/// Struct children for `count` elements, restricted to `fields`. +fn item_columns(fields: &Fields, count: usize, seed: usize) -> Vec { + fields + .iter() + .enumerate() + .map(|(i, field)| match field.name().as_str() { + "x" => Arc::new(Int64Array::from_iter_values( + (0..count).map(|j| (seed + j) as i64), + )) as ArrayRef, + "y" => Arc::new(StringArray::from_iter_values( + (0..count).map(|j| format!("y-{}", seed + j)), + )) as ArrayRef, + _ => pad_values(count, seed * (i + 1)), + }) + .collect() +} + +fn list_batch(fields: &Fields, batch_id: usize) -> RecordBatch { + let num_elems = ROWS_PER_BATCH * ELEMS_PER_ROW; + let seed = batch_id * num_elems; + let struct_array = + StructArray::new(fields.clone(), item_columns(fields, num_elems, seed), None); + let item = Arc::new(Field::new("item", DataType::Struct(fields.clone()), true)); + let events = ListArray::new( + item, + OffsetBuffer::from_lengths(std::iter::repeat_n(ELEMS_PER_ROW, ROWS_PER_BATCH)), + Arc::new(struct_array), + None, + ); + let ids = Int32Array::from_iter_values( + (0..ROWS_PER_BATCH).map(|i| (batch_id * ROWS_PER_BATCH + i) as i32), + ); + RecordBatch::try_new( + list_schema(fields.clone()), + vec![Arc::new(ids), Arc::new(events)], + ) + .unwrap() +} + +fn struct_batch(fields: &Fields, batch_id: usize) -> RecordBatch { + let seed = batch_id * ROWS_PER_BATCH; + let struct_array = StructArray::new( + fields.clone(), + item_columns(fields, ROWS_PER_BATCH, seed), + None, + ); + let ids = + Int32Array::from_iter_values((0..ROWS_PER_BATCH).map(|i| (seed + i) as i32)); + RecordBatch::try_new( + struct_schema(fields.clone()), + vec![Arc::new(ids), Arc::new(struct_array)], + ) + .unwrap() +} + +fn generate_file( + schema: SchemaRef, + batch_fn: impl Fn(usize) -> RecordBatch, + prefix: &str, +) -> NamedTempFile { + let mut named_file = tempfile::Builder::new() + .prefix(prefix) + .suffix(".parquet") + .tempfile() + .unwrap(); + + let properties = WriterProperties::builder() + .set_writer_version(WriterVersion::PARQUET_2_0) + .set_dictionary_enabled(false) + .set_max_row_group_row_count(Some(ROW_GROUP_ROW_COUNT)) + .build(); + + let mut writer = + ArrowWriter::try_new(&mut named_file, schema, Some(properties)).unwrap(); + for batch_id in 0..NUM_BATCHES { + writer.write(&batch_fn(batch_id)).unwrap(); + } + let metadata = writer.close().unwrap(); + println!( + "Generated {} ({} rows, {} row groups, {} bytes)", + named_file.path().display(), + metadata.file_metadata().num_rows(), + metadata.row_groups().len(), + std::fs::metadata(named_file.path()).unwrap().len(), + ); + named_file +} + +/// Register `path` as `table`, declaring `table_schema` (which may be narrower +/// than the file's physical schema). +fn register_table( + ctx: &SessionContext, + rt: &Runtime, + table: &str, + path: &str, + table_schema: SchemaRef, +) { + let url = ListingTableUrl::parse(path).unwrap(); + let config = rt + .block_on(ListingTableConfig::new(url).infer_options(&ctx.state())) + .unwrap() + .with_schema(table_schema); + let provider = ListingTable::try_new(config).unwrap(); + ctx.register_table(table, Arc::new(provider)).unwrap(); +} + +fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) { + let df = rt.block_on(ctx.sql(sql)).unwrap(); + black_box(rt.block_on(df.collect()).unwrap()); +} + +/// Print the parquet scan's `bytes_scanned` for a query so the IO pattern is +/// visible alongside the wall-time measurements. +fn report_bytes_scanned(ctx: &SessionContext, rt: &Runtime, label: &str, sql: &str) { + let df = rt + .block_on(ctx.sql(&format!("EXPLAIN ANALYZE {sql}"))) + .unwrap(); + let batches = rt.block_on(df.collect()).unwrap(); + let text = pretty_format_batches(&batches).unwrap().to_string(); + let bytes_scanned = text + .split("bytes_scanned=") + .nth(1) + .map(|rest| { + rest.split([',', ']']) + .next() + .unwrap_or("") + .trim() + .to_string() + }) + .unwrap_or_else(|| "".to_string()); + println!("{label}: bytes_scanned={bytes_scanned} ({sql})"); +} + +struct Fixture { + ctx: SessionContext, + rt: Runtime, + _files: Vec, +} + +/// Tables: +/// `_narrow_schema`: wide file, narrow declared schema +/// `_full_schema`: wide file, full declared schema +/// `_physically_narrow`: narrow file, narrow declared schema +fn setup( + name: &str, + schema_fn: fn(Fields) -> SchemaRef, + batch_fn: fn(&Fields, usize) -> RecordBatch, +) -> Fixture { + let rt = Runtime::new().unwrap(); + let ctx = SessionContext::new(); + + let wide = wide_item_fields(); + let narrow = narrow_item_fields(); + + let wide_file = generate_file(schema_fn(wide.clone()), |i| batch_fn(&wide, i), name); + let narrow_file = generate_file( + schema_fn(narrow.clone()), + |i| batch_fn(&narrow, i), + &format!("{name}_narrow"), + ); + let wide_path = wide_file.path().display().to_string(); + let narrow_path = narrow_file.path().display().to_string(); + + register_table( + &ctx, + &rt, + &format!("{name}_narrow_schema"), + &wide_path, + schema_fn(narrow.clone()), + ); + register_table( + &ctx, + &rt, + &format!("{name}_full_schema"), + &wide_path, + schema_fn(wide.clone()), + ); + register_table( + &ctx, + &rt, + &format!("{name}_physically_narrow"), + &narrow_path, + schema_fn(narrow.clone()), + ); + + Fixture { + ctx, + rt, + _files: vec![wide_file, narrow_file], + } +} + +fn list_struct_benchmarks(c: &mut Criterion) { + let f = setup("list_struct", list_schema, list_batch); + let (ctx, rt) = (&f.ctx, &f.rt); + + for table in [ + "list_struct_narrow_schema", + "list_struct_full_schema", + "list_struct_physically_narrow", + ] { + report_bytes_scanned(ctx, rt, table, &format!("SELECT events FROM {table}")); + } + + let mut group = c.benchmark_group("list_struct"); + group.sample_size(10); + group.warm_up_time(Duration::from_secs(1)); + group.measurement_time(Duration::from_secs(3)); + + // wide file, narrow declared schema: should only read the narrow leaves + group.bench_function("select_events_narrow_schema", |b| { + b.iter(|| query(ctx, rt, "SELECT events FROM list_struct_narrow_schema")) + }); + + // wide file, full schema: the cost of reading everything + group.bench_function("select_events_full_schema", |b| { + b.iter(|| query(ctx, rt, "SELECT events FROM list_struct_full_schema")) + }); + + // narrow file: the floor + group.bench_function("select_events_physically_narrow", |b| { + b.iter(|| query(ctx, rt, "SELECT events FROM list_struct_physically_narrow")) + }); + + // aggregation over one narrow leaf through unnest + group.bench_function("sum_x_narrow_schema", |b| { + b.iter(|| { + query( + ctx, + rt, + "SELECT SUM(e['x']) FROM (SELECT UNNEST(events) AS e FROM list_struct_narrow_schema)", + ) + }) + }); + + group.finish(); +} + +fn top_level_struct_benchmarks(c: &mut Criterion) { + let f = setup("struct", struct_schema, struct_batch); + let (ctx, rt) = (&f.ctx, &f.rt); + + for table in [ + "struct_narrow_schema", + "struct_full_schema", + "struct_physically_narrow", + ] { + report_bytes_scanned(ctx, rt, table, &format!("SELECT s FROM {table}")); + } + + let mut group = c.benchmark_group("top_level_struct"); + group.sample_size(10); + group.warm_up_time(Duration::from_secs(1)); + group.measurement_time(Duration::from_secs(3)); + + group.bench_function("select_struct_narrow_schema", |b| { + b.iter(|| query(ctx, rt, "SELECT s FROM struct_narrow_schema")) + }); + + group.bench_function("select_struct_full_schema", |b| { + b.iter(|| query(ctx, rt, "SELECT s FROM struct_full_schema")) + }); + + group.bench_function("select_struct_physically_narrow", |b| { + b.iter(|| query(ctx, rt, "SELECT s FROM struct_physically_narrow")) + }); + + // get_field on a schema-narrowed struct column: the expression-level + // pruning path interacting with the schema-level narrowing + group.bench_function("sum_x_narrow_schema", |b| { + b.iter(|| query(ctx, rt, "SELECT SUM(s['x']) FROM struct_narrow_schema")) + }); + + group.finish(); +} + +criterion_group!(benches, list_struct_benchmarks, top_level_struct_benchmarks); +criterion_main!(benches); From 665c93d17512a56d6448b75a6fd0ea7a76655e51 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 19:06:20 +0000 Subject: [PATCH 2/2] bench: enforce nested-schema scan baseline via typed metrics Address review feedback on the parquet_nested_schema_pruning benchmark: - Assert the checked-in baseline instead of only printing it. narrow == full bytes_scanned is now enforced per dataset shape, with a comment to flip it to `narrow < full` once nested projection pruning lands, so the bench becomes a regression guard rather than a diagnostic. - Read bytes_scanned from the typed metrics API (MetricsSet::aggregate_by_name().sum_by_name) by executing the physical plan and walking its metrics, rather than scraping EXPLAIN ANALYZE display text that would silently break if the format changed. - Derive wide_item_fields from narrow_item_fields so the shared x/y columns match by construction, fixing the nullability drift on x. - Seed pad columns with `seed + i` instead of `seed * (i + 1)`, which collapsed to the same seed for every pad column in the first batch. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01EHGBxLhJ922jnJh4X6jgRX --- .../benches/parquet_nested_schema_pruning.rs | 143 +++++++++++++----- 1 file changed, 101 insertions(+), 42 deletions(-) diff --git a/datafusion/core/benches/parquet_nested_schema_pruning.rs b/datafusion/core/benches/parquet_nested_schema_pruning.rs index 989e0a8890094..8db67de9ffa9b 100644 --- a/datafusion/core/benches/parquet_nested_schema_pruning.rs +++ b/datafusion/core/benches/parquet_nested_schema_pruning.rs @@ -34,8 +34,12 @@ //! 3. **physically_narrow**: a file that only contains the narrow columns — //! the floor //! -//! At setup the benchmark prints the `bytes_scanned` metric for (1) and (2) -//! so the IO pattern is visible in addition to wall time. +//! At setup the benchmark reads the parquet scan's `bytes_scanned` metric for +//! (1), (2) and (3) so the IO pattern is visible in addition to wall time, and +//! asserts the current baseline: today a narrow declared schema scans the same +//! bytes as the full schema. When nested projection pruning lands, that +//! assertion is expected to fail, which is the signal to flip it to +//! `narrow < full` (see [`assert_scan_baseline`]). use arrow::array::{ ArrayRef, Int32Array, Int64Array, ListArray, StringArray, StructArray, @@ -43,11 +47,12 @@ use arrow::array::{ use arrow::buffer::OffsetBuffer; use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; -use arrow::util::pretty::pretty_format_batches; use criterion::{Criterion, criterion_group, criterion_main}; use datafusion::datasource::listing::{ ListingTable, ListingTableConfig, ListingTableConfigExt, }; +use datafusion::physical_plan::metrics::MetricsSet; +use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::SessionContext; use datafusion_datasource::ListingTableUrl; use parquet::arrow::ArrowWriter; @@ -75,11 +80,15 @@ fn narrow_item_fields() -> Fields { /// The wide item fields as written to the file: the narrow fields plus /// `NUM_PAD_FIELDS` fat string fields the table schema does not mention. +/// +/// Derived from [`narrow_item_fields`] so the shared columns (`x`, `y`) match +/// by construction — same names, types and nullability — and only the extra +/// pad fields distinguish the two. fn wide_item_fields() -> Fields { - let mut fields = vec![ - Field::new("x", DataType::Int64, false), - Field::new("y", DataType::Utf8, true), - ]; + let mut fields: Vec = narrow_item_fields() + .iter() + .map(|f| f.as_ref().clone()) + .collect(); for i in 0..NUM_PAD_FIELDS { fields.push(Field::new(format!("pad_{i}"), DataType::Utf8, false)); } @@ -122,7 +131,11 @@ fn item_columns(fields: &Fields, count: usize, seed: usize) -> Vec { "y" => Arc::new(StringArray::from_iter_values( (0..count).map(|j| format!("y-{}", seed + j)), )) as ArrayRef, - _ => pad_values(count, seed * (i + 1)), + // `seed + i` keeps each pad column's values distinct from the + // others (and matches the additive seeding used above); a + // multiplier like `seed * (i + 1)` collapses to the same seed for + // every column when `seed == 0` (the first batch). + _ => pad_values(count, seed + i), }) .collect() } @@ -221,26 +234,70 @@ fn query(ctx: &SessionContext, rt: &Runtime, sql: &str) { black_box(rt.block_on(df.collect()).unwrap()); } -/// Print the parquet scan's `bytes_scanned` for a query so the IO pattern is -/// visible alongside the wall-time measurements. -fn report_bytes_scanned(ctx: &SessionContext, rt: &Runtime, label: &str, sql: &str) { - let df = rt - .block_on(ctx.sql(&format!("EXPLAIN ANALYZE {sql}"))) - .unwrap(); - let batches = rt.block_on(df.collect()).unwrap(); - let text = pretty_format_batches(&batches).unwrap().to_string(); - let bytes_scanned = text - .split("bytes_scanned=") - .nth(1) - .map(|rest| { - rest.split([',', ']']) - .next() - .unwrap_or("") - .trim() - .to_string() - }) - .unwrap_or_else(|| "".to_string()); - println!("{label}: bytes_scanned={bytes_scanned} ({sql})"); +/// Recursively collect the metrics of every node in `plan` into `out`. +fn gather_metrics(plan: &Arc, out: &mut MetricsSet) { + if let Some(metrics) = plan.metrics() { + for metric in metrics.iter() { + out.push(Arc::clone(metric)); + } + } + for child in plan.children() { + gather_metrics(child, out); + } +} + +/// Execute `sql` and return the parquet scan's `bytes_scanned` metric, read +/// from the typed metrics API rather than scraped from display output (which +/// would silently break if the format ever changed). +fn scan_bytes(ctx: &SessionContext, rt: &Runtime, sql: &str) -> usize { + let df = rt.block_on(ctx.sql(sql)).unwrap(); + let plan = rt.block_on(df.create_physical_plan()).unwrap(); + // Fully drive the plan so the scan populates its metrics. + black_box( + rt.block_on(collect(Arc::clone(&plan), ctx.task_ctx())) + .unwrap(), + ); + + let mut metrics = MetricsSet::new(); + gather_metrics(&plan, &mut metrics); + metrics + .aggregate_by_name() + .sum_by_name("bytes_scanned") + .map(|v| v.as_usize()) + .expect("parquet scan should report a bytes_scanned metric") +} + +/// Report and assert the `bytes_scanned` baseline for one dataset shape. +/// +/// `narrow` selects from a wide file through a narrow declared schema, `full` +/// through the full schema, and `floor` from a physically-narrow file. Today +/// the extra leaves are fetched and discarded, so `narrow == full`; that +/// equality is the checked-in baseline. When nested projection pruning lands, +/// `narrow` should drop toward `floor` and this assertion is expected to fail — +/// the signal to flip it to `assert!(narrow < full)`. +fn assert_scan_baseline( + ctx: &SessionContext, + rt: &Runtime, + label: &str, + narrow_sql: &str, + full_sql: &str, + floor_sql: &str, +) { + let narrow = scan_bytes(ctx, rt, narrow_sql); + let full = scan_bytes(ctx, rt, full_sql); + let floor = scan_bytes(ctx, rt, floor_sql); + println!( + "{label}: bytes_scanned narrow_schema={narrow} full_schema={full} \ + physically_narrow={floor}" + ); + assert_eq!( + narrow, full, + "{label}: narrow declared schema scanned {narrow} bytes vs {full} for \ + the full schema. The baseline is that a narrow schema still reads \ + every leaf, so these should be equal; if narrow is now smaller, \ + nested projection pruning has likely landed — flip this to \ + `assert!(narrow < full)`." + ); } struct Fixture { @@ -306,13 +363,14 @@ fn list_struct_benchmarks(c: &mut Criterion) { let f = setup("list_struct", list_schema, list_batch); let (ctx, rt) = (&f.ctx, &f.rt); - for table in [ - "list_struct_narrow_schema", - "list_struct_full_schema", - "list_struct_physically_narrow", - ] { - report_bytes_scanned(ctx, rt, table, &format!("SELECT events FROM {table}")); - } + assert_scan_baseline( + ctx, + rt, + "list_struct", + "SELECT events FROM list_struct_narrow_schema", + "SELECT events FROM list_struct_full_schema", + "SELECT events FROM list_struct_physically_narrow", + ); let mut group = c.benchmark_group("list_struct"); group.sample_size(10); @@ -352,13 +410,14 @@ fn top_level_struct_benchmarks(c: &mut Criterion) { let f = setup("struct", struct_schema, struct_batch); let (ctx, rt) = (&f.ctx, &f.rt); - for table in [ - "struct_narrow_schema", - "struct_full_schema", - "struct_physically_narrow", - ] { - report_bytes_scanned(ctx, rt, table, &format!("SELECT s FROM {table}")); - } + assert_scan_baseline( + ctx, + rt, + "top_level_struct", + "SELECT s FROM struct_narrow_schema", + "SELECT s FROM struct_full_schema", + "SELECT s FROM struct_physically_narrow", + ); let mut group = c.benchmark_group("top_level_struct"); group.sample_size(10);