Revert 32598

Author: def-

misc/python/materialize/mzcompose/__init__.py

@@ -155,7 +155,7 @@ def get_default_system_parameters(
         # 16 MiB - large enough to avoid a big perf hit, small enough to get more coverage...
         "persist_blob_target_size": "16777216",
         # 5 times the default part size - 4 is the bare minimum.
-        "persist_compaction_memory_bound_bytes": "83886080",
+        #"persist_compaction_memory_bound_bytes": "83886080",
         "persist_stats_audit_percent": "100",
         "persist_use_critical_since_catalog": "true",
         "persist_use_critical_since_snapshot": "false" if zero_downtime else "true",

misc/python/materialize/parallel_workload/action.py

@@ -1064,13 +1064,6 @@ def __init__(
             "16",
             "1000",
         ]
-        self.flags_with_values["persist_compaction_memory_bound_bytes"] = [
-            # 64 MiB, 1 * 128 MiB, 4 * 128 MiB, 8 * 128 MiB
-            "67108864",
-            "134217728",
-            "536870912",
-            "1073741824",
-        ]
         self.flags_with_values["persist_part_decode_format"] = [
             "row_with_validate",
             "arrow",

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

@@ -596,13 +596,8 @@ where
 
             // Prepare memory bounds for compaction
             let in_progress_part_reserved_memory_bytes = 2 * cfg.batch.blob_target_size;
-            // - then remaining memory will go towards pulling down as many runs as we can.
-            // We'll always do at least two runs per chunk, which means we may go over this limit
-            // if parts are large or the limit is low... though we do at least increment a metric
-            // when that happens.
-            let run_reserved_memory_bytes = cfg
-                .compaction_memory_bound_bytes
-                .saturating_sub(in_progress_part_reserved_memory_bytes);
+            let run_reserved_memory_bytes =
+                cfg.compaction_memory_bound_bytes - in_progress_part_reserved_memory_bytes;
 
             // Flatten the input batches into a single list of runs
             let ordered_runs =

src/txn-wal/src/txn_cache.rs

@@ -15,7 +15,6 @@ use std::fmt::Debug;
 use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
-use differential_dataflow::consolidation::consolidate_updates;
 use differential_dataflow::hashable::Hashable;
 use differential_dataflow::lattice::Lattice;
 use itertools::Itertools;
@@ -463,7 +462,6 @@ impl<T: Timestamp + Lattice + TotalOrder + StepForward + Codec64 + Sync> TxnsCac
         // which is not what we want. If we ever expose an interface for
         // registering and committing to a data shard at the same
         // timestamp, this will also have to sort registrations first.
-        consolidate_updates(&mut entries);
         entries.sort_by(|(a, _, _), (b, _, _)| a.ts::<T>().cmp(&b.ts::<T>()));
         for (e, t, d) in entries {
             match e {