Proto decode silently truncates limits and sizes to `usize` on 32-bit targets

[adriangb]

Describe the bug

Protobuf decode paths across the plan and expression serde layers convert wire integers to usize with an unchecked as cast. On a 32-bit target (usize is 32 bits) any value above u32::MAX silently truncates instead of failing.

For row limits this is the worst possible failure mode: a fetch of 1 << 32 truncates to 0, which is a perfectly valid limit meaning "return no rows". A plan that should return rows silently returns an empty result, with no error anywhere.

This is not hypothetical — DataFusion supports wasm32 (see datafusion/wasmtest), and 32-bit targets like armv7 and i686 are in normal use.

To Reproduce

On a 32-bit target, serialize any plan carrying a large fetch/limit and deserialize it:

// GlobalLimitExec, LocalLimitExec, SortExec, SortPreservingMergeExec,
// FilterExec, CoalesceBatchesExec, CoalescePartitionsExec, AggregateExec, ...
let plan = /* plan with fetch = 1 << 32 */;
let bytes = physical_plan_to_bytes(plan)?;
let back = physical_plan_from_bytes(&bytes, &ctx)?;
// on a 32-bit target: fetch is now 0 -> empty result

Expected behavior

A value that cannot be represented as usize on the current target should produce a clear decode error, not a silently truncated value. usize::try_from(v) mapped to a plan_err! is enough.

This was just fixed for exactly one field — HashJoinExec::fetch in #24165 — using:

let fetch = node
    .fetch
    .map(|f| {
        usize::try_from(f).map_err(|_| {
            plan_datafusion_err!(
                "HashJoinExec: fetch value {f} cannot be represented as usize on this target"
            )
        })
    })
    .transpose()?;

The remaining sites should get the same treatment. A small shared helper would probably beat repeating the closure ~20 times.

Additional context

Verified present on main (92f4e8f3ee). Representative, not exhaustive:

Row limits / offsets — highest severity, truncation to 0 means silently wrong results:

Location	Expression
datafusion/physical-plan/src/limit.rs:290, :296	limit.fetch as usize, limit.skip as usize
datafusion/physical-plan/src/limit.rs:510	LocalLimitExec::new(input, limit.fetch as usize)
datafusion/physical-plan/src/filter.rs:904	filter.fetch.map(|f| f as usize)
datafusion/physical-plan/src/coalesce_batches.rs:345	coalesce_batches.fetch.map(|f| f as usize)
datafusion/physical-plan/src/coalesce_partitions.rs:396	merge.fetch.map(|f| f as usize)
datafusion/physical-plan/src/aggregates/mod.rs:2463, :2465	limit.limit as usize
datafusion/physical-plan/src/sorts/sort.rs:1656	(sort.fetch >= 0).then_some(sort.fetch as usize)
datafusion/physical-plan/src/sorts/sort_preserving_merge.rs:527	(spm.fetch >= 0).then_some(spm.fetch as usize)
datafusion/proto/src/physical_plan/mod.rs:1575	scan.fetch.map(|f| f as usize)
datafusion/proto/src/logical_plan/mod.rs:964, :969	limit.skip.max(0) as usize, limit.fetch as usize
datafusion/datasource/src/file_scan_config/proto.rs:226	sl.limit as usize

Note the sort.rs / sort_preserving_merge.rs cases are i64 guarded by >= 0, so they truncate rather than go negative — the guard does not help here.

Configuration sizes — lower severity, but the same unchecked cast:

• datafusion/physical-plan/src/filter.rs:903 — filter.batch_size as usize
• datafusion/physical-plan/src/coalesce_batches.rs:344 — target_batch_size as usize
• datafusion/proto/src/physical_plan/mod.rs:2197 — generate_series.target_batch_size as usize
• datafusion/datasource/src/file_scan_config/proto.rs:229 — batch_size.map(\|s\| s as usize)
• datafusion/proto/src/logical_plan/file_formats.rs:514, :515, :550, :551, :565, :573, :574, :588, :611, :614 — parquet page/chunk/row-group size limits

Happy to send a PR applying the checked-conversion pattern across these if that approach sounds right.

Found while auditing the ExecutionPlan serde migration (#23494).

[buraksenn]

take