Persist: plumb incremental apply APIs through

src/persist-client/src/cli/admin.rs

@@ -40,7 +40,7 @@ use crate::internal::compact::{CompactConfig, CompactReq, Compactor};
 use crate::internal::encoding::Schemas;
 use crate::internal::gc::{GarbageCollector, GcReq};
 use crate::internal::machine::Machine;
-use crate::internal::trace::{CompactionInput, FueledMergeRes};
+use crate::internal::trace::FueledMergeRes;
 use crate::rpc::{NoopPubSubSender, PubSubSender};
 use crate::write::{WriteHandle, WriterId};
 use crate::{
@@ -455,17 +455,17 @@ where
             let req = CompactReq {
                 shard_id,
                 desc: req.desc,
-                inputs: req
-                    .inputs
-                    .into_iter()
-                    .map(|b| Arc::unwrap_or_clone(b.batch))
-                    .collect(),
+                inputs: req.inputs,
             };
-            let parts = req.inputs.iter().map(|x| x.part_count()).sum::<usize>();
+            let parts = req
+                .inputs
+                .iter()
+                .map(|x| x.batch.part_count())
+                .sum::<usize>();
             let bytes = req
                 .inputs
                 .iter()
-                .map(|x| x.encoded_size_bytes())
+                .map(|x| x.batch.encoded_size_bytes())
                 .sum::<usize>();
             let start = Instant::now();
             info!(
@@ -514,7 +514,7 @@ where
             let (apply_res, maintenance) = machine
                 .merge_res(&FueledMergeRes {
                     output: res.output,
-                    input: CompactionInput::Legacy,
+                    input: res.input,
                     new_active_compaction: None,
                 })
                 .await;
@@ -749,14 +749,18 @@ pub async fn dangerous_force_compaction_and_break_pushdown<K, V, T, D>(
         let (reqs, mut maintenance) = machine.spine_exert(fuel).await;
         for req in reqs {
             info!(
-                "force_compaction {} {} compacting {} batches in {} parts totaling {} bytes: lower={:?} upper={:?} since={:?}",
+                "force_compaction {} {} compacting {} batches in {} parts with {} runs totaling {} bytes: lower={:?} upper={:?} since={:?}",
                 machine.applier.shard_metrics.name,
                 machine.applier.shard_metrics.shard_id,
                 req.inputs.len(),
-                req.inputs.iter().flat_map(|x| &x.parts).count(),
+                req.inputs.iter().flat_map(|x| &x.batch.parts).count(),
+                req.inputs
+                    .iter()
+                    .map(|x| x.batch.runs().count())
+                    .sum::<usize>(),
                 req.inputs
                     .iter()
-                    .flat_map(|x| &x.parts)
+                    .flat_map(|x| &x.batch.parts)
                     .map(|x| x.encoded_size_bytes())
                     .sum::<usize>(),
                 req.desc.lower().elements(),
@@ -801,13 +805,18 @@ pub async fn dangerous_force_compaction_and_break_pushdown<K, V, T, D>(
 
         // NB: This check is intentionally at the end so that it's safe to call
         // this method in a loop.
-        let num_batches = machine.applier.all_batches().len();
-        if num_batches < 2 {
+        let num_runs: usize = machine
+            .applier
+            .all_batches()
+            .iter()
+            .map(|x| x.runs().count())
+            .sum();
+        if num_runs <= 1 {
             info!(
-                "force_compaction {} {} exiting with {} batches",
+                "force_compaction {} {} exiting with {} runs",
                 machine.applier.shard_metrics.name,
                 machine.applier.shard_metrics.shard_id,
-                num_batches
+                num_runs
             );
             return;
         }

src/persist-client/src/fetch.rs

@@ -1255,8 +1255,11 @@ where
         //   truncated bounds must be ignored. Not every user batch is
         //   truncated.
         // - Batches written by compaction. These always have an inline desc
-        //   that exactly matches the one they are registered with. The since
-        //   can be anything.
+        //   lower and upper that matches the registered desc lower and upper,
+        //   and a since that is less than or equal to the registered desc.
+        //   The inline since may be less than the registered desc since,
+        //   this is because of incremental compaction, where we might rewrite
+        //   certain runs in a batch but not others.
         let inline_desc = &parsed.desc;
         let needs_truncation = inline_desc.lower() != registered_desc.lower()
             || inline_desc.upper() != registered_desc.upper();
@@ -1297,10 +1300,28 @@ where
                 registered_desc
             );
         } else {
+            assert!(
+                PartialOrder::less_equal(inline_desc.since(), registered_desc.since()),
+                "key={} inline={:?} registered={:?}",
+                printable_name,
+                inline_desc,
+                registered_desc
+            );
+            assert_eq!(
+                inline_desc.lower(),
+                registered_desc.lower(),
+                "key={} inline={:?} registered={:?}",
+                printable_name,
+                inline_desc,
+                registered_desc
+            );
             assert_eq!(
-                inline_desc, &registered_desc,
+                inline_desc.upper(),
+                registered_desc.upper(),
                 "key={} inline={:?} registered={:?}",
-                printable_name, inline_desc, registered_desc
+                printable_name,
+                inline_desc,
+                registered_desc
             );
         }