storage-controller: respond to table appends before txn-wal apply

misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py

@@ -2716,6 +2716,75 @@ def benchmark(self) -> MeasurementSource:
 """)
 
 
+class MultiTableTransactionCommit(Coordinator):
+    """Measure COMMIT latency of a write transaction spanning many tables.
+
+    Such a transaction goes through group commit as one txn-wal transaction
+    touching one data shard per table, so COMMIT pays the txns-shard write
+    plus whatever per-shard work sits on the group commit response path
+    (batch apply and compaction, unless deferred).
+
+    The table write worker is a single serial task: it will not start the
+    next txn's txns-shard write until the previous txn's apply and compaction
+    have finished. So to observe the response-path latency of one COMMIT in
+    isolation, the worker must be idle when that COMMIT arrives. The reads at
+    the top of each iteration drain it: they touch every table, which blocks
+    until the previous iteration's txn is fully applied across all shards.
+    Reading a single table would only wait for that one shard's apply and
+    leave the worker busy with the rest, so the measured COMMIT would queue
+    behind that leftover work and hide any response-path win.
+    """
+
+    SCALE = 3  # 1000 tables
+    # The deferred apply/compaction the response path saves is one persist
+    # write per shard, so the win grows with the table count while the fixed
+    # per-COMMIT floor (group commit, timestamp selection, pgwire) does not.
+    # A high table count is what makes that saving dominate the measured
+    # window rather than the floor.
+    FIXED_SCALE = True
+
+    def init(self) -> list[Action]:
+        creates = "\n".join(
+            f"> CREATE TABLE mttc_t{i} (x INT);" for i in range(self.n())
+        )
+        return [TdAction(f"""
+$ postgres-execute connection=postgres://mz_system:materialize@${{testdrive.materialize-internal-sql-addr}}
+ALTER SYSTEM SET max_tables = {self.n() + 200};
+
+{creates}
+""")]
+
+    def benchmark(self) -> MeasurementSource:
+        inserts = "\n".join(
+            f"> INSERT INTO mttc_t{i} VALUES (1);" for i in range(self.n())
+        )
+        # Read every table so this waits for the whole previous txn to apply,
+        # draining the serial write worker before we measure. Separate reads
+        # (rather than one wide UNION ALL) keep each peek cheap to plan and
+        # avoid the deep-expr recursion limit at high table counts.
+        drain = "\n".join(
+            f"> SELECT count(*) >= 0 FROM mttc_t{i}\ntrue" for i in range(self.n())
+        )
+
+        return Td(f"""
+{drain}
+
+> BEGIN
+
+{inserts}
+
+> SELECT 1;
+  /* A */
+1
+
+> COMMIT
+
+> SELECT 1;
+  /* B */
+1
+""")
+
+
 class ReplicaExpiration(Scenario):
     # Causes "tried to kill container, but did not receive an exit event" errors when killing container afterwards
     SCALE = 5

src/storage-controller/src/persist_handles.rs

@@ -476,10 +476,21 @@ impl TxnsTableWorker {
         // Sneak in any txns shard tidying from previous commits.
         txn.tidy(std::mem::take(&mut self.tidy));
         let txn_res = txn.commit_at(&mut self.txns, write_ts).await;
-        let response = match txn_res {
+        match txn_res {
             Ok(apply) => {
-                // TODO: Do the applying in a background task. This will be a
-                // significant INSERT latency performance win.
+                // The commit to the txns shard already made the write durable
+                // and linearized, so unblock the caller before applying. The
+                // apply is idempotent and reads of the affected data shards
+                // block until it has happened, so deferring it shifts latency
+                // from every write onto reads of the just-written shards. If
+                // we crash before applying, the next append's apply covers it,
+                // because applying is `apply_le`: it applies all unapplied
+                // txns up to its timestamp, and group commit appends at least
+                // once per timestamp interval.
+                //
+                // It is not an error for the other end to hang up.
+                let _ = tx.send(Ok(()));
+
                 debug!("applying {:?}", apply);
                 let tidy = apply.apply(&mut self.txns).await;
                 self.tidy.merge(tidy);
@@ -488,16 +499,14 @@ impl TxnsTableWorker {
                 // and compact as aggressively as we can (i.e. to the time we
                 // just wrote).
                 let () = self.txns.compact_to(write_ts).await;
-
-                Ok(())
             }
             Err(current) => {
                 self.tidy.merge(txn.take_tidy());
                 debug!(
                     "unable to commit txn at {:?} current={:?}",
                     write_ts, current
                 );
-                Err(StorageError::InvalidUppers(
+                let response = Err(StorageError::InvalidUppers(
                     self.write_handles
                         .keys()
                         .copied()
@@ -506,11 +515,11 @@ impl TxnsTableWorker {
                             current_upper: Antichain::from_elem(current),
                         })
                         .collect(),
-                ))
+                ));
+                // It is not an error for the other end to hang up.
+                let _ = tx.send(response);
             }
-        };
-        // It is not an error for the other end to hang up.
-        let _ = tx.send(response);
+        }
     }
 }