feature-benchmark: Try to make runtime deterministic

misc/python/materialize/feature_benchmark/benchmark_versioning.py

@@ -23,7 +23,7 @@
 
 # Consider increasing the scenario's class #version() if changes are expected to impact results!
 SHA256_BY_SCENARIO_FILE: dict[str, str] = {
- "benchmark_main.py": "aca597f4eedb9c3e9ea08327b9997dab9ccb7202a0fe7df35181958d7413356b",
+ "benchmark_main.py": "e86cea859a64edf75896824933e08036cd92f1fb8e672dce4d3b5531a9ebdda6",
  "concurrency.py": "2e9c149c136b83b3853abc923a1adbdaf55a998ab4557712f8424c8b16f2adb1",
  "customer.py": "d1e72837a342c3ebf1f4a32ec583b1b78a78644cdba495030a6df45ebbffe703",
  "optbench.py": "e0aa427c4af0467a408ebd11fcff55b75da910bae035cf272bebd3924ebc3483",

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

@@ -1493,14 +1493,15 @@ def shared(self) -> Action:
 CREATE TABLE pk_table (pk BIGINT PRIMARY KEY, f2 BIGINT);
 INSERT INTO pk_table SELECT x, x*2 FROM generate_series(1, {self.n()}) as x;
 ALTER TABLE pk_table REPLICA IDENTITY FULL;
+
+> CREATE CLUSTER source_cluster SIZE '{self._default_size}', REPLICATION FACTOR 1;
 """
  )
 
  def before(self) -> Action:
  return TdAction(
  """
 > DROP SOURCE IF EXISTS mz_source_pgcdc CASCADE;
-> DROP CLUSTER IF EXISTS source_cluster CASCADE
  """
  )
 
@@ -1516,12 +1517,13 @@ def benchmark(self) -> MeasurementSource:
  PASSWORD SECRET pgpass
  )
 
-> CREATE CLUSTER source_cluster SIZE '{self._default_size}', REPLICATION FACTOR 1;
-
 > CREATE SOURCE mz_source_pgcdc
  IN CLUSTER source_cluster
  FROM POSTGRES CONNECTION pg_conn (PUBLICATION 'mz_source');
 
+> SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'mz_source_pgcdc';
+running
+
 > CREATE TABLE pk_table FROM SOURCE mz_source_pgcdc (REFERENCE pk_table);
  /* A */
 
@@ -1627,14 +1629,15 @@ def shared(self) -> Action:
 SET @i:=0;
 CREATE TABLE pk_table (pk BIGINT PRIMARY KEY, f2 BIGINT);
 INSERT INTO pk_table SELECT @i:=@i+1, @i*@i FROM mysql.time_zone t1, mysql.time_zone t2 LIMIT {self.n()};
+
+> CREATE CLUSTER source_cluster SIZE '{self._default_size}', REPLICATION FACTOR 1;
 """
  )
 
  def before(self) -> Action:
  return TdAction(
- """
+ f"""
 > DROP SOURCE IF EXISTS mz_source_mysqlcdc CASCADE;
-> DROP CLUSTER IF EXISTS source_cluster CASCADE
  """
  )
 
@@ -1648,11 +1651,13 @@ def benchmark(self) -> MeasurementSource:
  PASSWORD SECRET mysqlpass
  )
 
-> CREATE CLUSTER source_cluster SIZE '{self._default_size}', REPLICATION FACTOR 1;
-
 > CREATE SOURCE mz_source_mysqlcdc
  IN CLUSTER source_cluster
  FROM MYSQL CONNECTION mysql_conn;
+
+> SELECT status FROM mz_internal.mz_source_statuses WHERE name = 'mz_source_mysqlcdc';
+running
+
 > CREATE TABLE pk_table FROM SOURCE mz_source_mysqlcdc (REFERENCE public.pk_table);
  /* A */