Add Push::size_hint and VecPush terminal operator [ci-bench]

Added `size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>))` to the
`Push` trait as the push-side analog of `Pull::size_hint`. This allows
producers to announce how many items they're about to send, enabling
downstream operators and sinks to pre-allocate.

Trait changes:
- `Push::size_hint`: default no-op implementation
- `&mut P` blanket impl: forwards to inner
- `PushVariadic::size_hint`: new required method

Propagation through combinators:
- Map, Inspect: pass through unchanged (1:1 mapping)
- Filter, FilterMap: lower bound set to 0, upper preserved
- FlatMap, Flatten: hint becomes (0, None) — output count unknown
- Fanout, Unzip: forward to both branches
- Persist: reserves on internal Vec buffer, then forwards
- DemuxVar: forwards to all downstream pushes via PushVariadic
- ForEach, SinkPush, ResolveFutures: use default no-op

New terminal operator:
- `VecPush<Buf>`: pushes items into a `Vec`, uses `size_hint` to call
  `Vec::reserve(hint.0)` for pre-allocation. Gated on `alloc` feature.
- Constructor: `push::vec_push(buf)` creates a VecPush from a `&mut Vec<T>`.


Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: MingweiSamuel

dfir_lang/src/graph/meta_graph.rs

@@ -914,9 +914,8 @@ impl DfirGraph {
                 });
                 let send_port_code = send_ports.iter().map(|ident| {
                     quote_spanned! {ident.span()=>
-                        let #ident = #root::dfir_pipes::push::for_each(|v| {
-                            #ident.give(Some(v));
-                        });
+                        let mut #ident = #ident.borrow_mut_give();
+                        let #ident = #root::dfir_pipes::push::vec_push(&mut *#ident);
                     }
                 });
 
@@ -1217,6 +1216,14 @@ impl DfirGraph {
                                                     ) -> #root::dfir_pipes::push::PushStep<Self::CanPend> {
                                                         #root::dfir_pipes::push::Push::poll_flush(self.project().inner, ctx)
                                                     }
+
+                                                    #[inline(always)]
+                                                    fn size_hint(
+                                                        self: ::std::pin::Pin<&mut Self>,
+                                                        hint: (usize, Option<usize>),
+                                                    ) {
+                                                        #root::dfir_pipes::push::Push::size_hint(self.project().inner, hint)
+                                                    }
                                                 }
 
                                                 PushGuard {

dfir_macro/src/lib.rs

@@ -288,17 +288,15 @@ pub fn derive_demux_enum(item: proc_macro::TokenStream) -> proc_macro::TokenStre
             }),
     );
 
-    let variant_pats_sink_start_send =
-        variants
-            .iter()
-            .zip(variant_localvars_sink.iter())
-            .map(|(variant, sinkvar)| {
-                let Variant { ident, fields, .. } = variant;
-                let (fields_pat, push_item) = field_pattern_item(fields);
-                quote! {
-                    Self::#ident #fields_pat => #sinkvar.as_mut().start_send(#push_item)
-                }
-            });
+    let variant_pats_sink_start_send = variants.iter().zip(variant_localvars_sink.iter()).map(
+        |(variant, sinkvar)| {
+            let Variant { ident, fields, .. } = variant;
+            let (fields_pat, push_item) = field_pattern_item(fields);
+            quote! {
+                Self::#ident #fields_pat => ::std::pin::Pin::as_mut(#sinkvar).start_send(#push_item)
+            }
+        },
+    );
 
     let (impl_generics_item, ty_generics, where_clause_item) = generics.split_for_impl();
     let (impl_generics_sink, _ty_generics_sink, where_clause_sink) =
@@ -392,11 +390,11 @@ pub fn derive_demux_enum(item: proc_macro::TokenStream) -> proc_macro::TokenStre
                 #(
                     let #headvar = {
                         let __ctx = <<#variant_generics_push as #root::dfir_pipes::push::Push<#variant_output_types, ()>>::Ctx<'_> as #root::dfir_pipes::Context<'_>>::unmerge_self(__ctx);
-                        #root::dfir_pipes::push::Push::#method_name(#headvar.as_mut(), __ctx)
+                        #root::dfir_pipes::push::Push::#method_name(::std::pin::Pin::as_mut(#headvar), __ctx)
                     };
                     let __ctx = <<#variant_generics_push as #root::dfir_pipes::push::Push<#variant_output_types, ()>>::Ctx<'_> as #root::dfir_pipes::Context<'_>>::unmerge_other(__ctx);
                 )*
-                let #lastvar = #root::dfir_pipes::push::Push::#method_name(#lastvar.as_mut(), __ctx);
+                let #lastvar = #root::dfir_pipes::push::Push::#method_name(::std::pin::Pin::as_mut(#lastvar), __ctx);
                 // If any are pending, return pending.
                 #(
                     if #variant_localvars_push.is_pending() {
@@ -531,6 +529,18 @@ pub fn derive_demux_enum(item: proc_macro::TokenStream) -> proc_macro::TokenStre
                 #push_poll_flush_body
                 #root::dfir_pipes::push::PushStep::Done
             }
+
+            fn size_hint(
+                ( #( #variant_localvars_push, )* ): &mut #variant_generics_pinned_push_all,
+                __size_hint: (usize, ::std::option::Option<usize>),
+            ) {
+                #(
+                    #root::dfir_pipes::push::Push::size_hint(
+                        ::std::pin::Pin::as_mut(#variant_localvars_push),
+                        __size_hint,
+                    );
+                )*
+            }
         }
 
         impl #impl_generics_item #root::util::demux_enum::DemuxEnumBase

dfir_pipes/src/pull/cross_singleton.rs

@@ -99,6 +99,14 @@ where
             PullStep::Ended(_) => PullStep::ended(),
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let (mut lower, upper) = self.item_pull.size_hint();
+        if self.singleton_state.borrow().is_none() {
+            lower = 0;
+        }
+        (lower, upper)
+    }
 }
 
 impl<ItemPull, SinglePull, SingleState> FusedPull

dfir_pipes/src/pull/from_fn.rs

@@ -49,4 +49,9 @@ where
         let this = self.get_mut();
         (this.func)()
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        // Depends on UDF.
+        (0, None)
+    }
 }

dfir_pipes/src/pull/mod.rs

@@ -84,8 +84,7 @@ pub use stream_ready::StreamReady;
 #[cfg(feature = "std")]
 #[cfg_attr(docsrs, doc(cfg(feature = "std")))]
 pub use symmetric_hash_join::{
-    NewTickJoinIter, NewTickJoinPull, SymmetricHashJoin, SymmetricHashJoinEither,
-    symmetric_hash_join,
+    NewTickJoinIter, SymmetricHashJoin, SymmetricHashJoinEither, symmetric_hash_join,
 };
 pub use take::Take;
 pub use take_while::TakeWhile;
@@ -220,12 +219,9 @@ pub trait Pull {
     /// That said, the implementation should provide a correct estimation,
     /// because otherwise it would be a violation of the trait's protocol.
     ///
-    /// The default implementation returns `(0, None)` which is correct for any
-    /// pull.
-    #[inline]
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        (0, None)
-    }
+    /// A default implementation should return `(0, None)` which is correct for any
+    /// pull. However this is not provided, to prevent oversight.
+    fn size_hint(&self) -> (usize, Option<usize>);
 
     /// Borrows this pull, allowing it to be used by reference.
     fn by_ref(&mut self) -> &mut Self {

dfir_pipes/src/pull/poll_fn.rs

@@ -50,4 +50,9 @@ where
         let this = self.get_mut();
         (this.func)(ctx)
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        // Depends on UDF.
+        (0, None)
+    }
 }

dfir_pipes/src/pull/symmetric_hash_join.rs

@@ -9,8 +9,8 @@ use pin_project_lite::pin_project;
 use smallvec::SmallVec;
 
 use crate::pull::half_join_state::HalfJoinState;
-use crate::pull::{FusedPull, Pull, PullStep};
-use crate::{Context, No, Toggle, Yes};
+use crate::pull::{self, FusedPull, Pull, PullStep};
+use crate::{Context, Toggle};
 
 pin_project! {
     /// Pull combinator for symmetric hash join operations.
@@ -124,6 +124,11 @@ where
             return PullStep::ended();
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        // TODO(mingwei): actual estimate
+        (0, None)
+    }
 }
 
 /// Iterator for new tick - iterates over all matches after both sides are drained.
@@ -204,6 +209,11 @@ where
             self.next_rhs_smaller()
         }
     }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        // TODO(mingwei): proper size hint estimate
+        (0, None)
+    }
 }
 
 impl<'a, Key, V1, V2, LhsState, RhsState> NewTickJoinIter<'a, Key, V1, V2, LhsState, RhsState>
@@ -291,50 +301,9 @@ where
     }
 }
 
-pin_project! {
-    /// Pull wrapper for the new tick iterator case.
-    #[must_use = "`Pull`s do nothing unless polled"]
-    pub struct NewTickJoinPull<'a, Key, V1, V2, LhsState, RhsState>
-    where
-        Key: Clone,
-        V1: Clone,
-        V2: Clone,
-    {
-        iter: NewTickJoinIter<'a, Key, V1, V2, LhsState, RhsState>,
-    }
-}
-
-impl<'a, Key, V1, V2, LhsState, RhsState> Pull
-    for NewTickJoinPull<'a, Key, V1, V2, LhsState, RhsState>
-where
-    Key: Eq + std::hash::Hash + Clone,
-    V1: Clone,
-    V2: Clone,
-    LhsState: HalfJoinState<Key, V1, V2>,
-    RhsState: HalfJoinState<Key, V2, V1>,
-{
-    type Ctx<'ctx> = ();
-
-    type Item = (Key, (V1, V2));
-    type Meta = ();
-    type CanPend = No;
-    type CanEnd = Yes;
-
-    fn pull(
-        self: Pin<&mut Self>,
-        _ctx: &mut Self::Ctx<'_>,
-    ) -> PullStep<Self::Item, Self::Meta, Self::CanPend, Self::CanEnd> {
-        let this = self.project();
-        match this.iter.next() {
-            Some(item) => PullStep::Ready(item, ()),
-            None => PullStep::Ended(Yes),
-        }
-    }
-}
-
 /// Type alias for the `Either` pull returned by [`symmetric_hash_join`].
 pub type SymmetricHashJoinEither<'a, Key, V1, V2, Lhs, Rhs, LhsState, RhsState> = Either<
-    NewTickJoinPull<'a, Key, V1, V2, LhsState, RhsState>,
+    pull::Iter<NewTickJoinIter<'a, Key, V1, V2, LhsState, RhsState>>,
     SymmetricHashJoin<Lhs, Rhs, &'a mut LhsState, &'a mut RhsState, LhsState, RhsState>,
 >;
 
@@ -370,7 +339,7 @@ where
         } else {
             NewTickJoinIter::new_rhs_smaller(lhs_state, rhs_state)
         };
-        SymmetricHashJoinEither::Left(NewTickJoinPull { iter })
+        SymmetricHashJoinEither::Left(pull::iter(iter))
     } else {
         SymmetricHashJoinEither::Right(SymmetricHashJoin::new(lhs, rhs, lhs_state, rhs_state))
     }

dfir_pipes/src/push/demux_var.rs

@@ -70,6 +70,9 @@ where
     /// unmerged context slice. All pushes are flushed even if one returns
     /// pending, so that all wakers are registered.
     fn poll_flush(self: Pin<&mut Self>, ctx: &mut Self::Ctx<'_>) -> PushStep<Self::CanPend>;
+
+    /// Inform all downstream pushes that approximately `hint` items are about to be sent.
+    fn size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>));
 }
 
 /// Recursive case: a push `P` followed by the rest of the variadic `Rest`.
@@ -108,6 +111,12 @@ where
         );
         PushStep::Done
     }
+
+    fn size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>)) {
+        let (push, rest) = pin_project_pair(self);
+        push.size_hint(hint);
+        rest.size_hint(hint);
+    }
 }
 
 /// Base case: the empty variadic. Always ready, panics on send.
@@ -130,6 +139,8 @@ where
     fn poll_flush(self: Pin<&mut Self>, _ctx: &mut Self::Ctx<'_>) -> PushStep<No> {
         PushStep::Done
     }
+
+    fn size_hint(self: Pin<&mut Self>, _hint: (usize, Option<usize>)) {}
 }
 
 /// Pin-projects a pair `(A, B)` into its two pinned components.
@@ -199,6 +210,10 @@ where
     fn poll_flush(self: Pin<&mut Self>, ctx: &mut Self::Ctx<'_>) -> PushStep<Self::CanPend> {
         self.project().pushes.poll_flush(ctx)
     }
+
+    fn size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>)) {
+        self.project().pushes.size_hint(hint);
+    }
 }
 
 /// Creates a [`DemuxVar`] push that dispatches each `(usize, Item)` pair to

dfir_pipes/src/push/fanout.rs

@@ -67,6 +67,12 @@ where
         );
         PushStep::Done
     }
+
+    fn size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>)) {
+        let this = self.project();
+        this.push_0.size_hint(hint);
+        this.push_1.size_hint(hint);
+    }
 }
 
 #[cfg(test)]

dfir_pipes/src/push/filter.rs

@@ -50,4 +50,8 @@ where
     fn poll_flush(self: Pin<&mut Self>, ctx: &mut Self::Ctx<'_>) -> PushStep<Self::CanPend> {
         self.project().next.poll_flush(ctx)
     }
+
+    fn size_hint(self: Pin<&mut Self>, hint: (usize, Option<usize>)) {
+        self.project().next.size_hint((0, hint.1));
+    }
 }