Enable persist_enable_incremental_compaction in CI

Author: DAlperin

misc/python/materialize/mzcompose/__init__.py

@@ -315,6 +315,15 @@ def get_variable_system_parameters(
             "83886080",
             ["67108864", "134217728", "536870912", "1073741824"],
         ),
+        VariableSystemParameter(
+            "persist_enable_incremental_compaction",
+            ("true" if version > MzVersion.parse_mz("v0.154.0-dev") else "false"),
+            (
+                ["true", "false"]
+                if version > MzVersion.parse_mz("v0.154.0-dev")
+                else ["false"]
+            ),
+        ),
         VariableSystemParameter(
             "persist_use_critical_since_catalog", "true", ["true", "false"]
         ),
@@ -566,7 +575,6 @@ def get_default_system_parameters(
     "compute_peek_response_stash_read_memory_budget_bytes",
     "compute_peek_stash_num_batches",
     "compute_peek_stash_batch_size",
-    "persist_enable_incremental_compaction",
     "storage_statistics_retention_duration",
     "enable_ctp_cluster_protocols",
     "enable_paused_cluster_readhold_downgrade",

src/persist-client/src/internal/trace.rs

@@ -66,7 +66,7 @@ use serde::{Serialize, Serializer};
 use timely::PartialOrder;
 use timely::progress::frontier::AntichainRef;
 use timely::progress::{Antichain, Timestamp};
-use tracing::error;
+use tracing::{error, info};
 
 use crate::internal::paths::WriterKey;
 use crate::internal::state::{HollowBatch, RunId};
@@ -1685,6 +1685,7 @@ impl<T: Timestamp + Lattice> Spine<T> {
             // If any merges exist, we can directly call `apply_fuel`.
             self.apply_fuel(&fuel, log);
         } else {
+            info!("no merges in progress, introducing empty batch");
             // Otherwise, we'll need to introduce fake updates to move merges
             // along.
 
@@ -1754,15 +1755,18 @@ impl<T: Timestamp + Lattice> Spine<T> {
     /// words, there are either zero runs (fully empty) or exactly one logical
     /// run of data remaining.
     fn reduced(&self) -> bool {
-        self.spine_batches()
+        let total_runs = self
+            .spine_batches()
             .map(|b| {
                 b.parts
                     .iter()
                     .map(|p| p.batch.run_meta.len())
                     .sum::<usize>()
             })
-            .sum::<usize>()
-            < 2
+            .sum::<usize>();
+        let total_spine_batches = self.spine_batches().count();
+        info!(?total_runs, ?total_spine_batches, "checking if reduced");
+        total_spine_batches < 2
     }
 
     /// Describes the merge progress of layers in the trace.