Add nested file-scan pipeline (stream of per-file streams)

[robertbuessow]

Summary

• Nested pipeline API: scan_nested! / iceberg_file_scan_stream returns an outer FileScanStream that yields one IcebergFileScan per file. Each scan carries the filename, record count, and a prefetched inner IcebergArrowStream of Arrow IPC batches. The existing flat scan! is re-implemented as a thin flattening wrapper over the same nested pipeline so all ordering, backpressure, and stats logic lives in one place.
• Backwards compatible: all existing scan! call sites and tests are unchanged.
• IcebergScan.file_io made non-optional: was always set at construction; removing Option<> enforces this statically and eliminates dead .take().ok_or_else(...) guards in both stream ops.
• unexpected() crate-wide helper: replaces all inline iceberg::Error::new(iceberg::ErrorKind::Unexpected, ...) in catalog.rs, lib.rs, and ordered_file_pipeline.rs.
• timed! macro: replaces the let start / .await / add_elapsed pattern for the three async phases (fetch+decode, serialize, semaphore). Block form timed!(field, { expr }) keeps the important computation visually prominent.
• PipelineStats improvements: add_elapsed / store_elapsed helpers; file_dispatch_wait_ns metric tracks outer-channel backpressure (time run_nested is blocked waiting for the consumer to call next_file_scan); pipeline_wall_ns is now also set for the nested path (previously always 0, breaking the throughput display).
• 12 Rust unit tests for all PipelineStats methods.
• Julia tests: 6 new @testset blocks covering basic iteration, nested_arrow_stream called directly, row-count consistency vs flat scan, correct data, builder-method compatibility, and safe early drop.

New Julia API

# Returns a FileScanStream — one FileScan per file
stream = scan_nested!(scan)                  # or nested_arrow_stream(scan) after build!

fs_ptr = next_file_scan(stream)              # Ptr{Cvoid}, C_NULL when done
while fs_ptr != C_NULL
    fname  = file_scan_filename(fs_ptr)      # String
    nrec   = file_scan_record_count(fs_ptr)  # Int64 (before deletes)
    inner  = file_scan_arrow_stream(fs_ptr)  # borrowed ArrowStream — do NOT free_stream
    batch  = next_batch(inner)
    while batch != C_NULL
        # ... process Arrow IPC batch ...
        free_batch(batch)
        batch = next_batch(inner)
    end
    free_file_scan!(fs_ptr)                  # frees filename + inner stream
    fs_ptr = next_file_scan(stream)
end
free_file_scan_stream!(stream)

Ownership rules

• file_scan_arrow_stream returns a borrowed pointer into the IcebergFileScan struct. Callers must not call free_stream on it — free_file_scan! handles cleanup.
• Dropping a FileScan without reading all inner batches is safe: the dropped receiver causes the background file task to exit cleanly.
• Dropping the FileScanStream without consuming all files is equally safe.

Test plan

• make test passes (all existing tests unchanged)
• New nested-API tests pass (6 testsets)
• cargo test passes (12 PipelineStats unit tests)

Based on @hall-alex sorted iterator change.
Overarching task tracking Iceberg-load optimizations: RAI-49519.

🤖 Generated with Claude Code

[hall-alex]

Cool 💪🚀

[hall-alex]

Optional: would it make sense to extract a helper for this to use here and above (lines 151-174)?

[robertbuessow]

Done — extracted into resolve_pipeline_params in the eaa30d4 commit.

[hall-alex]

So format!("Serialization task panicked: {e}") is the same as format!("Serialization task panicked: {}", e)?

[robertbuessow]

Yes, they're identical at runtime. {e} is Rust 1.58 "captured identifier" syntax — it implicitly captures the local variable e as the format argument, so you don't need to pass it explicitly after the format string. Pure style preference; I tend to use it for single-variable cases to keep the string self-contained.

[hall-alex]

Perhaps change this comment. Would be nice to keep this for now at least.

[robertbuessow]

Updated in eaa30d4 — the comment now describes the module as permanent pipeline statistics rather than temporary profiling.

[gbrgr]

Wondering here whether, as done in the export work, it would make sense to have a thread pool of N workers that do the serialization work idle in the background?

[robertbuessow]

Done. Added a global rayon::ThreadPool

[gbrgr]

Shall we expose that as a Julia function as well, or do you prefer users to use the ccall macro?

[robertbuessow]

Added print_pipeline_stats() and reset_pipeline_stats() Julia wrappers in full.jl, both exported from the module.

[codecov-commenter]

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

❌ Patch coverage is 75.67568% with 9 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.89%. Comparing base (aaca529) to head (b41d539).
⚠️ Report is 10 commits behind head on main.

Files with missing lines	Patch %	Lines
src/full.jl	75.67%	9 Missing ⚠️
❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff             @@
##             main      #91      +/-   ##
==========================================
+ Coverage   83.86%   83.89%   +0.03%     
==========================================
  Files           9        9              
  Lines         849      913      +64     
==========================================
+ Hits          712      766      +54     
- Misses        137      147      +10     

☔ View full report in Codecov by Sentry.
📢 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.