Implement determine_timestamp using constraints

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

@@ -316,16 +316,16 @@ def validate(self) -> Testdrive:
                 > SHOW CREATE MATERIALIZED VIEW refresh_view_late_3
                 "materialize.public.refresh_view_late_3" "CREATE MATERIALIZED VIEW \\"materialize\\".\\"public\\".\\"refresh_view_late_3\\" IN CLUSTER \\"quickstart\\" WITH (REFRESH = AT <TIMESTAMP>::\\"mz_catalog\\".\\"mz_timestamp\\"::\\"pg_catalog\\".\\"text\\"::\\"pg_catalog\\".\\"int8\\" + 86400000) AS SELECT DISTINCT (\\"x\\") FROM \\"materialize\\".\\"public\\".\\"refresh_table\\""
 
-                $ set-regex match=(s\\d+|\\d{13}|[ ]{12}0|u\\d{1,3}|\\(\\d+-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d:\\d\\d\\.\\d\\d\\d\\)) replacement=<>
+                $ set-regex match=(s\\d+|\\d{13}|[ ]{12}0|u\\d{1,3}|\\(\\d+-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d:\\d\\d\\.\\d\\d\\d\\)|\\((\\d+)\\)) replacement=<>
 
                 > EXPLAIN TIMESTAMP FOR SELECT * FROM refresh_view_late_1
-                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_1 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n"
+                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_1 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n\\nbinding constraints:\\nlower:\\n  (StorageInput([User<>])): [<> <>]\\n  (ComputeInput([])): [<> <>]\\n"
 
                 > EXPLAIN TIMESTAMP FOR SELECT * FROM refresh_view_late_2
-                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_2 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n"
+                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_2 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n\\nbinding constraints:\\nlower:\\n  (StorageInput([User<>)])): [<> <>]\\n  (ComputeInput([])): [<> <>]\\n"
 
                 > EXPLAIN TIMESTAMP FOR SELECT * FROM refresh_view_late_3
-                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_3 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n"
+                "                query timestamp: <> <>\\n          oracle read timestamp: <> <>\\nlargest not in advance of upper: <> <>\\n                          upper:[<> <>]\\n                          since:[<> <>]\\n        can respond immediately: false\\n                       timeline: Some(EpochMilliseconds)\\n              session wall time: <> <>\\n\\nsource materialize.public.refresh_view_late_3 (<>, storage):\\n                  read frontier:[<> <>]\\n                 write frontier:[<> <>]\\n\\nbinding constraints:\\nlower:\\n  (StorageInput([User<>])): [<> <>]\\n  (ComputeInput([])): [<> <>]\\n"
            """
             )
         )

src/adapter-types/Cargo.toml

@@ -17,6 +17,7 @@ mz-storage-types = { path = "../storage-types" }
 serde = "1.0.219"
 timely = "0.19.0"
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
+tracing = "0.1.37"
 
 [package.metadata.cargo-udeps.ignore]
 normal = ["workspace-hack"]

src/adapter-types/src/dyncfgs.rs

@@ -13,6 +13,8 @@ use std::time::Duration;
 
 use mz_dyncfg::{Config, ConfigSet};
 
+use crate::timestamp_oracle::ConstraintBasedTimestampSelection;
+
 pub const ALLOW_USER_SESSIONS: Config<bool> = Config::new(
     "allow_user_sessions",
     true,
@@ -116,6 +118,12 @@ pub const ENABLE_MULTI_REPLICA_SOURCES: Config<bool> = Config::new(
     "Enable multi-replica sources.",
 );
 
+pub const CONSTRAINT_BASED_TIMESTAMP_SELECTION: Config<&'static str> = Config::new(
+    "constraint_based_timestamp_selection",
+    ConstraintBasedTimestampSelection::const_default().as_str(),
+    "Whether to use the constraint-based timestamp selection, one of: enabled, disabled, verify",
+);
+
 /// Adds the full set of all compute `Config`s.
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
@@ -134,4 +142,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_CONTINUAL_TASK_BUILTINS)
         .add(&ENABLE_EXPRESSION_CACHE)
         .add(&ENABLE_MULTI_REPLICA_SOURCES)
+        .add(&CONSTRAINT_BASED_TIMESTAMP_SELECTION)
 }

src/adapter-types/src/timestamp_oracle.rs

@@ -9,6 +9,9 @@
 
 use std::time::Duration;
 
+use serde::{Deserialize, Serialize};
+use tracing;
+
 /// Default value for `DynamicConfig::pg_connection_pool_max_size`.
 pub const DEFAULT_PG_TIMESTAMP_ORACLE_CONNPOOL_MAX_SIZE: usize = 50;
 
@@ -26,3 +29,43 @@ pub const DEFAULT_PG_TIMESTAMP_ORACLE_CONNECT_TIMEOUT: Duration = Duration::from
 
 /// Default value for `DynamicConfig::pg_connection_pool_tcp_user_timeout`.
 pub const DEFAULT_PG_TIMESTAMP_ORACLE_TCP_USER_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Whether to use the constraint-based timestamp selection.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum ConstraintBasedTimestampSelection {
+    Enabled,
+    Disabled,
+    Verify,
+}
+
+impl std::default::Default for ConstraintBasedTimestampSelection {
+    fn default() -> Self {
+        Self::Verify
+    }
+}
+
+impl ConstraintBasedTimestampSelection {
+    pub const fn const_default() -> Self {
+        Self::Verify
+    }
+
+    pub fn from_str(s: &str) -> Self {
+        match s {
+            "enabled" => Self::Enabled,
+            "disabled" => Self::Disabled,
+            "verify" => Self::Verify,
+            _ => {
+                tracing::error!("invalid value for ConstraintBasedTimestampSelection: {}", s);
+                ConstraintBasedTimestampSelection::default()
+            }
+        }
+    }
+
+    pub const fn as_str(&self) -> &'static str {
+        match self {
+            Self::Enabled => "enabled",
+            Self::Disabled => "disabled",
+            Self::Verify => "verify",
+        }
+    }
+}