sinks: walk the input arrangement via cursor per batch

[DAlperin]

Summary

Refactor the sink rendering path so sinks walk their input arrangement's cursors directly, eliminating the Vec<DiffPair<Row>>-per-group materialization and the trace-reader overhead that zip_into_diff_pairs → combine_at_timestamp → flat_map previously incurred. Motivated by profiles of very large snapshot sinks where that pipeline dominated allocation and caused major page faults.

What this buys

• No per-(key, time) Vec<DiffPair<Row>> allocation. Biggest direct win — this was the dominant allocator in zip_into_diff_pairs on the profile.
• Key owned once per key, not per (key, time) group.
• Fewer operator boundaries. Kafka: arrangement → encode (was: arrangement → combine_at_timestamp → flat_map → encode). Iceberg: one fewer hop too.
• Spine compacts aggressively. Dropping the TraceAgent lets spine compaction advance its frontier, releasing historical batch state rather than accumulating it.

What this does not fix

• The arrangement's pre-spine batcher still buffers un-sealed updates (unavoidable at this layer).
• For a pure-insertion snapshot with no retractions, spine compaction has limited work to do — savings come primarily from the Vec/operator overhead, not from the spine itself shrinking.
• Persist snapshot forwarding semantics and sink commit-on-frontier semantics are unchanged.

Follow-ups (not in this PR) worth considering:

• A batcher-only operator that skips the spine entirely for sinks that don't need a trace.
• An append-mode Iceberg fast-path that skips arrangement altogether when no key is configured.

Test plan

• cargo check --workspace --all-targets
• cargo clippy -p mz-interchange -p mz-storage --tests
• bin/lint (check-no-diff fails locally due to jj colocation; all substantive checks pass)
• cargo test -p mz-interchange --lib envelopes:: (7 tests, all pass)
• Full CI (pending)
• Kafka sink testdrive (snapshot + steady-state, Upsert and Debezium envelopes)
• Iceberg sink testdrive (snapshot + steady-state, Upsert and Append envelopes)
• Re-profile a very large snapshot sink to confirm the `zip_into_diff_pairs` hotspot is gone

🤖 Generated with Claude Code

[def-]

When a Kafka or Iceberg sink is configured with a non-unique key (KEY (...) NOT ENFORCED), Materialize is supposed to log a warning ("primary key error") whenever multiple rows share the same key at the same timestamp. With this PR the warning doesn't appear when no new data arrives afterwards. Not sure how bad that is.

Reproducer:

diff --git a/test/source-sink-errors/mzcompose.py b/test/source-sink-errors/mzcompose.py
index ac2b7c3e92..5a9bbeb7ee 100644
--- a/test/source-sink-errors/mzcompose.py
+++ b/test/source-sink-errors/mzcompose.py
@@ -13,6 +13,7 @@ Disruption and then checking the mz_internal.mz_*_statuses tables
 """

 import random
+import time
 from collections.abc import Callable
 from dataclasses import dataclass
 from textwrap import dedent
@@ -561,3 +562,52 @@ def unsupported_pg_table(c: Composition) -> None:
                  $ postgres-execute connection=postgres://postgres:postgres@postgres
                  INSERT INTO source1 VALUES (3, '[2:3]={2,2}')
                  """))
+
+
+def workflow_test_pk_violation_warning(c: Composition) -> None:
+    seed = random.randint(0, 256**4)
+    c.up("redpanda", "materialized", Service("testdrive", idle=True))
+
+    with c.override(
+        Testdrive(
+            no_reset=True,
+            seed=seed,
+            entrypoint_extra=["--initial-backoff=1s", "--backoff-factor=0"],
+        )
+    ):
+        c.testdrive(dedent("""
+                > CREATE CONNECTION kafka_conn
+                  TO KAFKA (BROKER '${testdrive.kafka-addr}', SECURITY PROTOCOL PLAINTEXT);
+
+                > CREATE CONNECTION IF NOT EXISTS csr_conn TO CONFLUENT SCHEMA REGISTRY (
+                  URL '${testdrive.schema-registry-url}'
+                  );
+
+                > CREATE TABLE t (a INT, b INT);
+
+                > CREATE SINK pk_test_sink FROM t
+                  INTO KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-pk-test-${testdrive.seed}')
+                  KEY (a) NOT ENFORCED
+                  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
+                  ENVELOPE UPSERT;
+
+                > SELECT status FROM mz_internal.mz_sink_statuses WHERE name = 'pk_test_sink'
+                running
+        """))
+
+        # PkViolationWarner rate-limits to once per 10s from construction.
+        time.sleep(12)
+
+        c.testdrive(dedent("""
+                > INSERT INTO t VALUES (1, 10), (1, 20);
+
+                $ set-sql-timeout duration=60s
+                $ kafka-verify-topic sink=materialize.public.pk_test_sink await-value-schema=true
+        """))
+
+        time.sleep(5)
+
+        logs = c.invoke("logs", "materialized", capture=True)
+        assert "primary key error" in logs.stdout, (
+            "PkViolationWarner::flush() not called after batch"
+        )