fix(proto): decode HashJoinExec fetch with a checked u64 -> usize

This branch is stacked on the `fetch` fix and, after the rebase onto
main, carries its own copy of the `HashJoinExec` decode path. Apply the
same correction here so the defect does not survive on this branch and
reappear once #24165 merges.

Review feedback on #24165: `u64 as usize` silently truncates on a 32-bit
target, where `usize` is 32 bits. A fetch of `1 << 32` decodes to `0` --
not merely a wrong limit but the worst possible one, since "fetch 0
rows" turns the query into an empty result instead of erroring.

Use `usize::try_from` and surface an out-of-range value as an error.
Truncating and saturating both misrepresent the plan; an explicit decode
failure is the honest outcome, and it is only reachable on a 32-bit
target with an absurd fetch.

`plan_datafusion_err!` rather than `internal_datafusion_err!`: this
decode path reserves the internal-error macros for genuinely malformed
nodes (an unknown `PartitionMode` discriminant, a dynamic filter that
does not downcast), which really do indicate a bug. A well-formed but
unrepresentable `fetch` is not a DataFusion bug -- it is a plan that
cannot be expressed on this target -- and `plan_err!` is already this
file's idiom for invalid plan configuration.

The encode side (`fetch.map(|f| f as u64)`) needs no change: `usize` is
at most 64 bits on every supported target, so widening to `u64` is
lossless.

`roundtrip_hash_join_fetch` now also covers `u32::MAX as usize` and
`usize::MAX`, pinning that a large fetch round-trips exactly. Both are
representable on every target, so the test stays portable. The
truncation path itself is only reachable on a 32-bit target and is
therefore not covered on a 64-bit CI host.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: adriangb

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -1896,7 +1896,7 @@ impl HashJoinExec {
         node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
         ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        use datafusion_common::internal_datafusion_err;
+        use datafusion_common::{internal_datafusion_err, plan_datafusion_err};
         use datafusion_proto_models::protobuf;
         use std::any::Any;
 
@@ -1978,17 +1978,33 @@ impl HashJoinExec {
             indices => Some(indices.iter().map(|i| *i as usize).collect()),
         };
 
+        // Restore the row limit that `limit_pushdown` may have pushed into the
+        // join. The field is presence-tracked, so a message written before it
+        // existed decodes to `None` (no limit) rather than to `Some(0)`.
+        //
+        // The conversion is checked, not `as usize`: `fetch` is a `u64` on the
+        // wire but a `usize` in the plan, and on a 32-bit target `as usize`
+        // truncates. A fetch of `1 << 32` would become `0` -- not merely a
+        // wrong limit but the worst one, silently turning the query into an
+        // empty result. Report the out-of-range value instead. Please do not
+        // "simplify" this back to `as usize`.
+        let fetch = 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()?;
+
         let mut hash_join = HashJoinExecBuilder::new(left, right, on, join_type)
             .with_filter(filter)
             .with_projection(projection)
             .with_partition_mode(partition_mode)
             .with_null_equality(null_equality)
             .with_null_aware(*null_aware)
-            // Restore the row limit that `limit_pushdown` may have pushed into
-            // the join. The field is presence-tracked, so a message written
-            // before it existed decodes to `None` (no limit) rather than to
-            // `Some(0)`.
-            .with_fetch(fetch.map(|f| f as usize))
+            .with_fetch(fetch)
             .build()?;
 
         if let Some(dynamic_filter_proto) = dynamic_filter {

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -563,7 +563,14 @@ fn roundtrip_hash_join_fetch() -> Result<()> {
         Arc::new(Column::new("col", schema_right.index_of("col")?)) as _,
     )];
 
-    for fetch in [None, Some(7)] {
+    // `usize::MAX` and `u32::MAX as usize` pin the decode-side `u64 -> usize`
+    // conversion: it is a checked `usize::try_from`, and a large fetch must
+    // survive the round trip exactly rather than being truncated or clamped.
+    // Both are representable on every target (on a 32-bit target `usize::MAX`
+    // is simply `u32::MAX`), so this stays portable. The truncating case
+    // itself -- a `u64` fetch above `usize::MAX` -- is only reachable on a
+    // 32-bit target and so is not exercised by this test on a 64-bit host.
+    for fetch in [None, Some(7), Some(u32::MAX as usize), Some(usize::MAX)] {
         let join = HashJoinExec::try_new(
             Arc::new(EmptyExec::new(Arc::clone(&schema_left))),
             Arc::new(EmptyExec::new(Arc::clone(&schema_right))),