[DNM] Run CI on some experimental peek optimizations

misc/python/materialize/checks/all_checks/statement_logging.py

@@ -16,9 +16,13 @@ class StatementLogging(Check):
     def initialize(self) -> Testdrive:
         return Testdrive(
             dedent(
+                # TODO(peek-seq): enable_frontend_peek_sequencing when it supports statement logging.
                 """
                 $ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}
                 ALTER SYSTEM SET statement_logging_max_sample_rate TO 1.0
+
+                $[version>=16200] postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}
+                ALTER SYSTEM SET enable_frontend_peek_sequencing = false;
                 """
             )
         )

misc/python/materialize/mzcompose/__init__.py

@@ -67,7 +67,9 @@ def get_minimal_system_parameters(
     version: MzVersion,
 ) -> dict[str, str]:
     """Settings we need in order to have tests run at all, but otherwise stay
-    with the defaults: not changing performance or increasing coverage."""
+    with the defaults: not changing performance or increasing coverage.
+    Note: This is not used unless we explicitly select "System Parameters: Minimal" in trigger-ci.
+    """
 
     return {
         # -----
@@ -130,6 +132,7 @@ class VariableSystemParameter:
 
 
 # TODO: The linter should check this too
+# Note: Only the default is tested unless we explicitly select "System Parameters: Random" in trigger-ci.
 def get_variable_system_parameters(
     version: MzVersion,
     force_source_table_syntax: bool,
@@ -186,6 +189,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_frontend_peek_sequencing",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "kafka_default_metadata_fetch_interval",
             "1s",

misc/python/materialize/parallel_workload/action.py

@@ -172,6 +172,7 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
                     "real-time source dropped before ingesting the upstream system's visible frontier",  # Expected, see https://buildkite.com/materialize/nightly/builds/9399#0191be17-1f4c-4321-9b51-edc4b08b71c5
                     "object state changed while transaction was in progress",  # Old error msg, can remove this ignore later
                     "another session modified the catalog while this DDL transaction was open",
+                    "was dropped while executing a statement",
                 ]
             )
         if exe.db.scenario == Scenario.Cancel:
@@ -1322,6 +1323,10 @@ def __init__(
             "314572800",  # 300 MiB, the production value
         ]
         self.flags_with_values["cluster"] = ["quickstart", "dont_exist"]
+        self.flags_with_values["enable_frontend_peek_sequencing"] = [
+            "true",
+            "false",
+        ]
 
         # If you are adding a new config flag in Materialize, consider using it
         # here instead of just marking it as uninteresting to silence the

misc/python/materialize/scalability/schema/schema.py

@@ -51,7 +51,7 @@ def init_sqls(self) -> list[str]:
                     [
                         f"CREATE TABLE t{t} (f1 INTEGER DEFAULT 1);",
                         f"INSERT INTO t{t} DEFAULT VALUES;",
-                        f"CREATE OR REPLACE MATERIALIZED VIEW mv{t} AS SELECT count(*) AS count FROM t{t};",
+                        # f"CREATE OR REPLACE MATERIALIZED VIEW mv{t} AS SELECT count(*) AS count FROM t{t};",
                     ]
                 )
 
@@ -62,9 +62,9 @@ def init_sqls(self) -> list[str]:
                 # index to be ready.
                 if self.create_index:
                     init_sqls.append(f"CREATE INDEX i{t} ON t{t} (f1);")
-                    init_sqls.append(f"CREATE INDEX mv_i{t} ON mv{t} (count);")
+                    # init_sqls.append(f"CREATE INDEX mv_i{t} ON mv{t} (count);")
                     init_sqls.append(f"SELECT f1 from t{t};")
-                    init_sqls.append(f"SELECT count from mv{t};")
+                    # init_sqls.append(f"SELECT count from mv{t};")
 
         return init_sqls
 

misc/python/materialize/scalability/workload/workloads/dml_dql_workloads.py

@@ -7,27 +7,27 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0.
 from materialize.scalability.operation.operations.operations import (
-    InsertDefaultValues,
-    SelectCount,
-    SelectCountInMv,
+    #InsertDefaultValues,
+    #SelectCount,
+    #SelectCountInMv,
     SelectLimit,
-    SelectOne,
+    #SelectOne,
     SelectStar,
-    SelectUnionAll,
-    Update,
+    #SelectUnionAll,
+    #Update,
 )
 from materialize.scalability.operation.scalability_operation import Operation
 from materialize.scalability.workload.workload_markers import DmlDqlWorkload
 
 
-class InsertWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [InsertDefaultValues()]
+# class InsertWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [InsertDefaultValues()]
 
 
-class SelectOneWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [SelectOne()]
+# class SelectOneWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [SelectOne()]
 
 
 class SelectStarWorkload(DmlDqlWorkload):
@@ -40,26 +40,26 @@ def operations(self) -> list["Operation"]:
         return [SelectLimit()]
 
 
-class SelectCountWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [SelectCount()]
+# class SelectCountWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [SelectCount()]
 
 
-class SelectUnionAllWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [SelectUnionAll()]
+# class SelectUnionAllWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [SelectUnionAll()]
 
 
-class InsertAndSelectCountInMvWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [InsertDefaultValues(), SelectCountInMv()]
+# class InsertAndSelectCountInMvWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [InsertDefaultValues(), SelectCountInMv()]
 
 
-class InsertAndSelectLimitWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [InsertDefaultValues(), SelectLimit()]
+# class InsertAndSelectLimitWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [InsertDefaultValues(), SelectLimit()]
 
 
-class UpdateWorkload(DmlDqlWorkload):
-    def operations(self) -> list["Operation"]:
-        return [Update()]
+# class UpdateWorkload(DmlDqlWorkload):
+#     def operations(self) -> list["Operation"]:
+#         return [Update()]

src/adapter/src/catalog.rs

@@ -15,7 +15,7 @@ use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::convert;
 use std::sync::Arc;
-
+use std::sync::atomic::{AtomicU64, Ordering};
 use futures::future::BoxFuture;
 use futures::{Future, FutureExt};
 use itertools::Itertools;
@@ -140,6 +140,7 @@ pub struct Catalog {
     expr_cache_handle: Option<ExpressionCacheHandle>,
     storage: Arc<tokio::sync::Mutex<Box<dyn mz_catalog::durable::DurableCatalogState>>>,
     transient_revision: u64,
+    latest_transient_revision: Arc<AtomicU64>,
 }
 
 // Implement our own Clone because derive can't unless S is Clone, which it's
@@ -152,6 +153,7 @@ impl Clone for Catalog {
             expr_cache_handle: self.expr_cache_handle.clone(),
             storage: Arc::clone(&self.storage),
             transient_revision: self.transient_revision,
+            latest_transient_revision: Arc::clone(&self.latest_transient_revision),
         }
     }
 }
@@ -490,6 +492,10 @@ impl Catalog {
         self.transient_revision
     }
 
+    pub fn latest_transient_revision(&self) -> u64 {
+        self.latest_transient_revision.load(Ordering::SeqCst) ////////// Is the ordering ok?
+    }
+
     /// Creates a debug catalog from the current
     /// `METADATA_BACKEND_URL` with parameters set appropriately for debug contexts,
     /// like in tests.

src/adapter/src/catalog/open.rs

@@ -13,6 +13,7 @@ mod builtin_item_migration;
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
+use std::sync::atomic::AtomicU64;
 use std::time::{Duration, Instant};
 
 use futures::future::{BoxFuture, FutureExt};
@@ -551,6 +552,7 @@ impl Catalog {
                 plans: CatalogPlans::default(),
                 expr_cache_handle,
                 transient_revision: 1,
+                latest_transient_revision: Arc::new(AtomicU64::new(1)),
                 storage: Arc::new(tokio::sync::Mutex::new(storage)),
             };
 

src/adapter/src/catalog/transact.rs

@@ -12,6 +12,7 @@
 use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
+use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use itertools::Itertools;
@@ -454,6 +455,7 @@ impl Catalog {
         drop(storage);
         if let Some(new_state) = new_state {
             self.transient_revision += 1;
+            self.latest_transient_revision.fetch_add(1, Ordering::SeqCst); ////////// todo: Is the ordering ok?
             self.state = new_state;
         }