Persist: plumb incremental apply APIs through

[DAlperin]

Depends on #33044

Motivation

Tips for reviewer

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

[bkirwi]

Nightly with the flag enabled seems pretty red: https://buildkite.com/materialize/nightly/builds/12728

[DAlperin]

Every day I learn a new useful itertools thing

[def-]

In the latest run: https://buildkite.com/materialize/test/builds/107434#01988262-f51c-46c1-9683-5103676fb135

#9 9.297 2025-08-07T02:42:51.793661Z  thread 'persist:003e' panicked at src/persist-client/src/internal/compact.rs:691:69:
#9 9.297 run_id must be present
#9 9.297    5: core::panicking::panic_fmt
#9 9.297    6: core::option::expect_failed
#9 9.297    7: <core::iter::adapters::map::Map<core::slice::iter::Iter<(&mz_persist_client::internal::compact::RunLocation, &differential_dataflow::trace::description::Description<mz_repr::timestamp::Timestamp>, &mz_persist_client::internal::state::RunMeta, &[mz_persist_client::internal::state::RunPart<mz_repr::timestamp::Timestamp>])>, <mz_persist_client::internal::compact::Compactor<(), (), mz_repr::timestamp::Timestamp, i64>>::compact_stream::{closure#0}::{closure#0}> as itertools::Itertools>::multiunzip::<(alloc::collections::btree::set::BTreeSet<mz_persist_client::internal::trace::SpineId>, alloc::collections::btree::set::BTreeSet<mz_persist_client::internal::state::RunId>, alloc::vec::Vec<&differential_dataflow::trace::description::Description<mz_repr::timestamp::Timestamp>>)>
#9 9.297    8: <async_stream::async_stream::AsyncStream<core::result::Result<mz_persist_client::internal::trace::FueledMergeRes<mz_repr::timestamp::Timestamp>, anyhow::Error>, <mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_stream::{closure#0}> as futures_core::stream::Stream>::poll_next
#9 9.297    9: <core::pin::Pin<&mut async_stream::async_stream::AsyncStream<core::result::Result<mz_persist_client::internal::trace::FueledMergeRes<mz_repr::timestamp::Timestamp>, anyhow::Error>, <mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_stream::{closure#0}>> as futures_util::stream::stream::StreamExt>::poll_next_unpin
#9 9.297   10: <futures_util::stream::stream::next::Next<core::pin::Pin<&mut async_stream::async_stream::AsyncStream<core::result::Result<mz_persist_client::internal::trace::FueledMergeRes<mz_repr::timestamp::Timestamp>, anyhow::Error>, <mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_stream::{closure#0}>>> as core::future::future::Future>::poll
#9 9.297   11: <tracing::instrument::Instrumented<<mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_and_apply::{closure#0}::{closure#5}> as core::future::future::Future>::poll
#9 9.297   12: <core::pin::Pin<alloc::boxed::Box<tracing::instrument::Instrumented<<mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_and_apply::{closure#0}::{closure#5}>>> as core::future::future::Future>::poll
#9 9.297   13: <tracing::instrument::Instrumented<core::pin::Pin<alloc::boxed::Box<tracing::instrument::Instrumented<<mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_and_apply::{closure#0}::{closure#5}>>>> as core::future::future::Future>::poll
#9 9.297   14: <tokio::runtime::task::core::Core<tracing::instrument::Instrumented<core::pin::Pin<alloc::boxed::Box<tracing::instrument::Instrumented<<mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_and_apply::{closure#0}::{closure#5}>>>>, alloc::sync::Arc<tokio::runtime::scheduler::multi_thread_alt::handle::Handle>>>::poll
#9 9.297   15: <tokio::runtime::task::harness::Harness<tracing::instrument::Instrumented<core::pin::Pin<alloc::boxed::Box<tracing::instrument::Instrumented<<mz_persist_client::internal::compact::Compactor<mz_storage_types::sources::SourceData, (), mz_repr::timestamp::Timestamp, i64>>::compact_and_apply::{closure#0}::{closure#5}>>>>, alloc::sync::Arc<tokio::runtime::scheduler::multi_thread::handle::Handle>>>::poll
#9 9.297   16: <tokio::runtime::scheduler::multi_thread::worker::Context>::run_task
#9 9.297   17: <tokio::runtime::scheduler::multi_thread::worker::Context>::run
#9 9.297   18: <tokio::runtime::context::scoped::Scoped<tokio::runtime::scheduler::Context>>::set::<tokio::runtime::scheduler::multi_thread::worker::run::{closure#0}::{closure#0}, ()>
#9 9.297   19: tokio::runtime::context::runtime::enter_runtime::<tokio::runtime::scheduler::multi_thread::worker::run::{closure#0}, ()>
#9 9.297   20: tokio::runtime::scheduler::multi_thread::worker::run
#9 9.297   21: <tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::multi_thread::worker::Launch>::launch::{closure#0}> as core::future::future::Future>::poll
#9 9.297   22: <tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::multi_thread::worker::Launch>::launch::{closure#0}>> as core::future::future::Future>::poll
#9 9.297   23: <tokio::runtime::task::core::Core<tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::multi_thread::worker::Launch>::launch::{closure#0}>>, tokio::runtime::blocking::schedule::BlockingSchedule>>::poll
#9 9.297   24: <tokio::runtime::task::harness::Harness<tracing::instrument::Instrumented<tokio::runtime::blocking::task::BlockingTask<<tokio::runtime::scheduler::multi_thread::worker::Launch>::launch::{closure#0}>>, tokio::runtime::blocking::schedule::BlockingSchedule>>::poll
#9 9.297   25: <tokio::runtime::task::UnownedTask<tokio::runtime::blocking::schedule::BlockingSchedule>>::run
#9 9.297   26: <tokio::runtime::blocking::pool::Inner>::run
#9 9.297 note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

[bkirwi]

Cool! Thank you for the followup!

Left a few nits. We've also discussed some possible cleanups that would make sense as a followup, but this seems overall in good shape...

[bkirwi]

✨ - thanks!

[bkirwi]

Mild preference for using chunked_runs.into_iter().enumerate() instead of the counter here... that way I don't have to check the control flow to see if it's incremented every iteration.