bench: parquet scan with a table schema narrower than a nested column

[adriangb]

Which issue does this PR close?

• Related to Comet native Parquet scan reads full nested columns despite pruned Spark ReadSchema datafusion-comet#4859. Second of a 3-PR stack: refactor: extract parquet projection read plan into its own module #23396 (refactor), this benchmark, and the feature PR (nested schema pruning for the parquet reader).

Rationale for this change

When a table's declared schema is narrower than a parquet file's nested column (logical events: LIST<STRUCT<x, y>> over a physical LIST<STRUCT<x, y, +8 pads>>), the reader currently fetches and decodes every leaf of the column and discards the extra subfields in memory via the adapter-inserted cast. This is how engines like Spark (via Comet) communicate nested projection pruning to the scan — as a clipped read schema — and it is where Comet measured reading 1.35 TB where Spark read 30.9 GB for the same pruned ReadSchema.

This PR adds a benchmark that documents the current behavior as a checked-in baseline, independent of any fix:

list_struct_narrow_schema:      bytes_scanned=25.19 MB   3.45 ms
list_struct_full_schema:        bytes_scanned=25.19 MB   3.36 ms   <- narrow == full today
list_struct_physically_narrow:  bytes_scanned= 3.32 KB    164 µs   <- the floor

What changes are included in this PR?

A criterion benchmark, datafusion/core/benches/parquet_nested_schema_pruning.rs, that registers the same wide list<struct> (and top-level struct) parquet file with both its full schema and a narrower declared schema, plus a physically-narrow file as the floor, and prints the scans' bytes_scanned at setup so the IO pattern is visible alongside wall time.

Are these changes tested?

It is a benchmark; it compiles under cargo bench --no-run and runs green.

Are there any user-facing changes?

No.

🤖 Generated with Claude Code

https://claude.ai/code/session_01KuMaRtFSPDQesuzjN5Koyd

[mbutrovich]

Thanks for this. The module doc (lines 40-60) lays out the narrow/full/floor comparison clearly, and disabling dictionary encoding with distinct pad values is the right call for making the wide case actually pay the IO. I also like that query borrows &SessionContext/&str instead of cloning every iteration, which is cleaner than what the sibling bench does.

The core idea is solid and I want to see it land. My one real ask before it does is around report_bytes_scanned: the value of this PR is a checked-in baseline, but right now that baseline is printed rather than enforced, and it's read by scraping display text. If we assert the narrow == full relationship through the metrics API, this stops being a diagnostic and becomes a real regression guard, and PR-3's fix will flip the bench where you can see it. I left the specifics inline. The rest of my comments are quality suggestions I'd recommend but leave to you.

[mbutrovich]

The baseline claim, that narrow scans the same bytes as full today, is printed but never enforced, so a later change could shift it and nobody would notice. The sibling bench asserts its physical-read invariants (parquet_struct_projection.rs:167-178). Can we do the same on the relationship here?

assert_eq!(narrow_bytes, full_bytes, "narrow declared schema still reads the full column today");
assert!(physically_narrow_bytes < narrow_bytes);

The nice part: when PR-3 lands, the bench fails right here and prompts you to flip it to assert!(narrow_bytes < full_bytes), so the fix documents itself. Asserting the relationship instead of an absolute byte count keeps it from breaking on formatting changes.

(See the related note on line 233 about reading the value from the metrics API, which gives you a typed number to assert on.)

[mbutrovich]

bytes_scanned is a real Count metric, so we can read it in process instead of scraping pretty_format_batches output. The current parse returns <not found> if the display format ever shifts, which is easy to miss.

let plan = df.create_physical_plan().await?;
let _ = collect(plan.clone(), task_ctx).await?;
let bytes = plan.metrics().unwrap()
    .aggregate_by_name()
    .sum_by_name("bytes_scanned")
    .map(|v| v.as_usize());

plan.metrics().unwrap().aggregate_by_name() is already used this way at tests/sql/explain_analyze.rs:162. It hands back a typed usize, which also makes it easy to assert the narrow == full baseline instead of just printing it.

[mbutrovich]

This is close to parquet_struct_projection.rs:137-188, and there's a third copy in parquet_struct_query.rs. Benches can share code across criterion binaries here: benches/data_utils/mod.rs is reused by around nine benches via mod data_utils; use data_utils::.... A small shared module for generate_file, query, and a metrics-based bytes reader would keep the three parquet_struct* benches in sync. Only worth it if the duplication bugs you; just noting the option.

[mbutrovich]

This helper is duplicated in both sibling benches (parquet_struct_projection.rs, parquet_struct_query.rs), so it's a candidate for the shared bench module too. If it does get hoisted, keep this borrowing version rather than the cloning one.

[mbutrovich]

NUM_BATCHES and ROW_GROUP_ROW_COUNT (line 63) match parquet_struct_projection.rs:45,47, and ROWS_PER_BATCH is that file's WRITE_RECORD_BATCH_SIZE. Natural to fold into a shared bench module if you decide to dedupe the helpers.

[mbutrovich]

x is Int64, false here but Int64, true in narrow_item_fields (line 71), while y is true in both. Reads like unintended drift for the same logical column. It's harmless today, but deriving wide from narrow keeps the shared prefix aligned by construction:

fn wide_item_fields() -> Fields {
    let mut fields: Vec<Field> =
        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));
    }
    Fields::from(fields)
}

[mbutrovich]

pad_values already makes columns distinct through its per-element index, so the seed * (i + 1) multiplier isn't doing anything load-bearing. seed + i matches the additive convention used elsewhere and reads more plainly. An explicit name if name.starts_with("pad_") arm would also state the intent instead of leaning on the _ catch-all. Minor.

[mbutrovich]

More of a question than a request. This measures the same thing as parquet_struct_projection.rs at a complementary altitude, schema-level narrowing versus expression-level pruning, so a schema_narrowed group inside that file would put the whole story in one place and land PR-3's fix next to the assert it flips. The trade-off is a longer file and generalizing the helpers (register_parquet versus ListingTableConfig::with_schema). Fine either way; just worth a look before committing to a second binary.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
⚠️ Please upload report for BASE (main@c2e3473). Learn more about missing BASE report.

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #23397   +/-   ##
=======================================
  Coverage        ?   80.67%           
=======================================
  Files           ?     1088           
  Lines           ?   367591           
  Branches        ?   367591           
=======================================
  Hits            ?   296549           
  Misses          ?    53382           
  Partials        ?    17660           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[mbutrovich]

Thanks @adriangb!

[adriangb]

Sorry I never replied, I was mulling over merging the benchmarks, pulling out helpers. TLDR: I think lets keep it split for now. If there's more duplication in the future we can merge / extract.