fix(proto): preserve HashJoinExec fetch across serialization

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

@@ -1841,6 +1841,7 @@ impl ExecutionPlan for HashJoinExec {
                         },
                         null_aware: self.null_aware,
                         dynamic_filter,
+                        fetch: self.fetch.map(|f| f as u64),
                     },
                 )),
             ),
@@ -1855,7 +1856,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;
 
@@ -1932,17 +1933,35 @@ impl HashJoinExec {
             indices => Some(indices.iter().map(|i| *i as usize).collect()),
         };
 
-        let mut hash_join = HashJoinExec::try_new(
-            left,
-            right,
-            on,
-            filter,
-            &join_type,
-            projection,
-            partition_mode,
-            null_equality,
-            hashjoin.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)`.
+        //
+        // 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 = hashjoin
+            .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(hashjoin.null_aware)
+            .with_fetch(fetch)
+            .build()?;
 
         if let Some(dynamic_filter_proto) = &hashjoin.dynamic_filter {
             // The dynamic filter is a `DynamicFilterPhysicalExpr` over the probe

datafusion/proto-models/proto/datafusion.proto

@@ -1324,6 +1324,14 @@ message HashJoinExecNode {
   bool null_aware = 10;
   // Optional dynamic filter expression for pushing down to the probe side.
   PhysicalExprNode dynamic_filter = 11;
+  // Optional row limit pushed into the join by the `limit_pushdown` rule.
+  //
+  // This is presence-tracked (`optional`) on purpose: messages produced by
+  // versions predating this field carry no `fetch` at all, and a plain proto3
+  // scalar would decode that absence as `0`, i.e. "fetch 0 rows", silently
+  // turning old plans into empty results. With `optional`, absent decodes to
+  // `None`, which is the correct reading of an older message.
+  optional uint64 fetch = 12;
 }
 
 enum StreamPartitionMode {

datafusion/proto-models/src/generated/pbjson.rs

@@ -9026,6 +9026,9 @@ impl serde::Serialize for HashJoinExecNode {
         if self.dynamic_filter.is_some() {
             len += 1;
         }
+        if self.fetch.is_some() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("datafusion.HashJoinExecNode", len)?;
         if let Some(v) = self.left.as_ref() {
             struct_ser.serialize_field("left", v)?;
@@ -9063,6 +9066,11 @@ impl serde::Serialize for HashJoinExecNode {
         if let Some(v) = self.dynamic_filter.as_ref() {
             struct_ser.serialize_field("dynamicFilter", v)?;
         }
+        if let Some(v) = self.fetch.as_ref() {
+            #[allow(clippy::needless_borrow)]
+            #[allow(clippy::needless_borrows_for_generic_args)]
+            struct_ser.serialize_field("fetch", ToString::to_string(&v).as_str())?;
+        }
         struct_ser.end()
     }
 }
@@ -9088,6 +9096,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
             "nullAware",
             "dynamic_filter",
             "dynamicFilter",
+            "fetch",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -9102,6 +9111,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
             Projection,
             NullAware,
             DynamicFilter,
+            Fetch,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -9133,6 +9143,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
                             "projection" => Ok(GeneratedField::Projection),
                             "nullAware" | "null_aware" => Ok(GeneratedField::NullAware),
                             "dynamicFilter" | "dynamic_filter" => Ok(GeneratedField::DynamicFilter),
+                            "fetch" => Ok(GeneratedField::Fetch),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -9162,6 +9173,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
                 let mut projection__ = None;
                 let mut null_aware__ = None;
                 let mut dynamic_filter__ = None;
+                let mut fetch__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Left => {
@@ -9227,6 +9239,14 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
                             }
                             dynamic_filter__ = map_.next_value()?;
                         }
+                        GeneratedField::Fetch => {
+                            if fetch__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("fetch"));
+                            }
+                            fetch__ = 
+                                map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0)
+                            ;
+                        }
                     }
                 }
                 Ok(HashJoinExecNode {
@@ -9240,6 +9260,7 @@ impl<'de> serde::Deserialize<'de> for HashJoinExecNode {
                     projection: projection__.unwrap_or_default(),
                     null_aware: null_aware__.unwrap_or_default(),
                     dynamic_filter: dynamic_filter__,
+                    fetch: fetch__,
                 })
             }
         }

datafusion/proto-models/src/generated/prost.rs

@@ -2029,6 +2029,15 @@ pub struct HashJoinExecNode {
     /// Optional dynamic filter expression for pushing down to the probe side.
     #[prost(message, optional, tag = "11")]
     pub dynamic_filter: ::core::option::Option<PhysicalExprNode>,
+    /// Optional row limit pushed into the join by the `limit_pushdown` rule.
+    ///
+    /// This is presence-tracked (`optional`) on purpose: messages produced by
+    /// versions predating this field carry no `fetch` at all, and a plain proto3
+    /// scalar would decode that absence as `0`, i.e. "fetch 0 rows", silently
+    /// turning old plans into empty results. With `optional`, absent decodes to
+    /// `None`, which is the correct reading of an older message.
+    #[prost(uint64, optional, tag = "12")]
+    pub fetch: ::core::option::Option<u64>,
 }
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SymmetricHashJoinExecNode {

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -542,6 +542,70 @@ fn roundtrip_hash_join_projection_states() -> Result<()> {
     Ok(())
 }
 
+/// Regression: `HashJoinExecNode` had no `fetch` field, so the row limit that
+/// the `limit_pushdown` physical optimizer rule pushes into the join via
+/// `ExecutionPlan::with_fetch` was silently dropped by serde. Because that rule
+/// also removes the enclosing `GlobalLimitExec` once the join absorbs the limit,
+/// a round-tripped plan had no limit left at all and a distributed executor
+/// returned more rows than the query asked for.
+///
+/// Note this cannot be covered by `roundtrip_test`: that helper compares
+/// `format!("{plan:?}")`, and `HashJoinExec`'s `Debug` output does not include
+/// `fetch`, so the before/after strings match even when the value is lost. The
+/// assertions below therefore inspect `fetch()` directly.
+#[test]
+fn roundtrip_hash_join_fetch() -> Result<()> {
+    let field_a = Field::new("col", DataType::Int64, false);
+    let schema_left = Arc::new(Schema::new(vec![field_a.clone()]));
+    let schema_right = Arc::new(Schema::new(vec![field_a]));
+    let on = vec![(
+        Arc::new(Column::new("col", schema_left.index_of("col")?)) as _,
+        Arc::new(Column::new("col", schema_right.index_of("col")?)) as _,
+    )];
+
+    // `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))),
+            on.clone(),
+            None,
+            &JoinType::Inner,
+            None,
+            PartitionMode::Partitioned,
+            NullEquality::NullEqualsNothing,
+            false,
+        )?;
+
+        let plan: Arc<dyn ExecutionPlan> = match fetch {
+            // This is how `limit_pushdown` installs the limit.
+            Some(fetch) => join
+                .with_fetch(Some(fetch))
+                .expect("HashJoinExec supports fetch"),
+            None => Arc::new(join),
+        };
+        assert_eq!(plan.fetch(), fetch);
+
+        let ctx = SessionContext::new();
+        let codec = DefaultPhysicalExtensionCodec {};
+        let proto_converter = DefaultPhysicalProtoConverter {};
+        let deserialized =
+            roundtrip_test_and_return(plan, &ctx, &codec, &proto_converter)?;
+
+        let deserialized_join = deserialized
+            .downcast_ref::<HashJoinExec>()
+            .expect("should be a HashJoinExec");
+        assert_eq!(deserialized_join.fetch(), fetch);
+    }
+    Ok(())
+}
+
 /// Same regression coverage for `NestedLoopJoinExec`, which shares the
 /// `repeated uint32 projection` proto field shape with `HashJoinExec`.
 #[test]