sinks: skip gratuitous key clones on the hot DiffPair path

Two changes to remove per-DiffPair Row clones that showed up as an Iceberg
perf regression in the feature benchmark:

1. Iceberg walker emits None for the key unconditionally. write_data_files
   destructures the key as _row on both the stash and drain paths and never
   uses it; the PK warner still sees the real key before we replace it with
   None, so violation detection is unchanged. Removes one Row clone per
   DiffPair.

2. PkViolationWarner stores Row::hashed() (u64) instead of a cloned
   Option<Row> to identify the current group. A hash collision could mask
   a single PK violation but this is a purely diagnostic code path, so
   the trade-off is acceptable. Removes another Row clone per group
   transition — which for snapshot workloads with unique keys is every
   DiffPair.

Author: DAlperin

src/storage/src/render/sinks.rs

@@ -158,11 +158,16 @@ fn arrange_sink_input<'scope>(
 /// Call [`PkViolationWarner::observe`] once per emitted `DiffPair`. When the
 /// current `(key, timestamp)` group changes — or when input batches finish —
 /// call [`PkViolationWarner::flush`] so the accumulated count is evaluated.
+///
+/// Keys are identified by their `Hashable::hashed()` value rather than held
+/// by value, so the hot observe path does no `Row` clones. A hash collision
+/// can mask a PK violation but this is a purely diagnostic check, so the
+/// trade-off is acceptable.
 pub(crate) struct PkViolationWarner {
     sink_id: GlobalId,
     from_id: GlobalId,
     last_warning: Instant,
-    current: Option<(Option<Row>, Timestamp)>,
+    current: Option<(u64, Timestamp)>,
     count: usize,
 }
 
@@ -181,13 +186,14 @@ impl PkViolationWarner {
     /// a new group, the previous group's count is flushed (and warned about
     /// if the count was > 1).
     pub fn observe(&mut self, key: &Option<Row>, time: Timestamp) {
-        let same = match &self.current {
-            Some((k, t)) => k == key && *t == time,
-            None => false,
-        };
+        // `None` keys hash to a distinct sentinel from any `Row::hashed()`;
+        // the exact constant doesn't matter for correctness (it just needs
+        // to be stable).
+        let hash = key.as_ref().map(|k| k.hashed()).unwrap_or(u64::MAX);
+        let same = self.current == Some((hash, time));
         if !same {
             self.flush();
-            self.current = Some((key.clone(), time));
+            self.current = Some((hash, time));
         }
         self.count += 1;
     }

src/storage/src/sink/iceberg.rs

@@ -2444,8 +2444,10 @@ fn walk_sink_arrangement<'scope>(
                         if let Some(warner) = pk_warner.as_mut() {
                             warner.observe(key, time);
                         }
-                        let emitted_key = if key_is_synthetic { None } else { key.clone() };
-                        output.give(&cap, ((emitted_key, diff_pair), time, Diff::ONE));
+                        // The arrangement key is only used downstream for grouping and PK checks;
+                        // `write_data_files` discards it on both the stash and drain paths. Emit
+                        // None unconditionally to avoid per-`DiffPair` `Row` clones on this hot path.
+                        output.give(&cap, ((None, diff_pair), time, Diff::ONE));
                     });
                 }
             }