tests: add mzcompose test for retain history

ci/nightly/pipeline.template.yml

@@ -1002,6 +1002,16 @@ steps:
       - ./ci/plugins/mzcompose:
           composition: pubsub-disruption
 
+  - id: retain-history
+    label: "Check retain history"
+    timeout_in_minutes: 15
+    artifact_paths: junit_*.xml
+    agents:
+      queue: linux-x86_64
+    plugins:
+      - ./ci/plugins/mzcompose:
+          composition: retain-history
+
   - id: data-ingest
     label: "Data Ingest"
     artifact_paths: junit_*.xml

test/retain-history/mzcompose

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+#
+# mzcompose — runs Docker Compose with Materialize customizations.
+
+exec "$(dirname "$0")"/../../bin/pyactivate -m materialize.cli.mzcompose "$@"

test/retain-history/mzcompose.py

@@ -0,0 +1,208 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+import time
+from datetime import datetime
+from textwrap import dedent
+
+from materialize.mzcompose.composition import Composition
+from materialize.mzcompose.services.cockroach import Cockroach
+from materialize.mzcompose.services.materialized import Materialized
+from materialize.mzcompose.services.testdrive import Testdrive
+
+SERVICES = [
+    Cockroach(setup_materialize=True),
+    Materialized(propagate_crashes=True, external_cockroach=True),
+    Testdrive(),
+]
+
+
+def workflow_default(c: Composition) -> None:
+    """Test the retain history feature."""
+    run_test(c)
+
+
+def run_test(c: Composition) -> None:
+    c.up("materialized")
+    c.up("testdrive", persistent=True)
+
+    mv_on_mv1_retention_in_sec = 1
+    mv_on_mv_on_mv1_retention_in_sec = 60
+
+    mz_time0 = fetch_now_from_mz(c)
+
+    testdrive(
+        c,
+        f"""
+            $ postgres-execute connection=postgres://mz_system:materialize@${{testdrive.materialize-internal-sql-addr}}
+            ALTER SYSTEM SET enable_logical_compaction_window = true
+
+            > CREATE TABLE retain_history_table (key INT, value INT);
+
+            > INSERT INTO retain_history_table VALUES (1, 100), (2, 200);
+
+            > CREATE MATERIALIZED VIEW retain_history_mv1 WITH (RETAIN HISTORY FOR '10s') AS
+                    SELECT * FROM retain_history_table;
+
+            > CREATE MATERIALIZED VIEW retain_history_mv_on_mv1 WITH (RETAIN HISTORY FOR '{mv_on_mv1_retention_in_sec}s') AS
+                    SELECT * FROM retain_history_mv1;
+
+            > CREATE MATERIALIZED VIEW retain_history_mv_on_mv_on_mv1 WITH (RETAIN HISTORY FOR '{mv_on_mv_on_mv1_retention_in_sec}s') AS
+                    SELECT * FROM retain_history_mv_on_mv1;
+
+            > SELECT count(*) FROM retain_history_mv1;
+            2
+            """,
+    )
+
+    mz_time1 = fetch_now_from_mz(c)
+    test_time1 = datetime.now()
+
+    testdrive(
+        c,
+        f"""
+            > UPDATE retain_history_table SET value = value + 1;
+            > INSERT INTO retain_history_table VALUES (3, 300);
+
+            > SELECT * FROM retain_history_mv1;
+            1 101
+            2 201
+            3 300
+
+            ! SELECT count(*) FROM retain_history_mv1 AS OF '{mz_time0}'::TIMESTAMP;
+            contains: is not valid for all inputs
+
+            > SELECT count(*) FROM retain_history_mv1 AS OF AT LEAST'{mz_time0}'::TIMESTAMP;
+            0
+
+            > SELECT * FROM retain_history_mv1 AS OF '{mz_time1}'::TIMESTAMP;
+            1 100
+            2 200
+
+            > INSERT INTO retain_history_table VALUES (4, 400);
+            """,
+    )
+
+    mz_time2 = fetch_now_from_mz(c)
+
+    testdrive(
+        c,
+        """
+            > DELETE FROM retain_history_table WHERE key IN (3, 4);
+            """,
+    )
+
+    mz_time3 = fetch_now_from_mz(c)
+
+    testdrive(
+        c,
+        f"""
+            > SELECT count(*) FROM retain_history_mv1;
+            2
+
+            > SELECT * FROM retain_history_mv1 AS OF '{mz_time1}'::TIMESTAMP;
+            1 100
+            2 200
+
+            > SELECT * FROM retain_history_mv_on_mv1 AS OF '{mz_time1}'::TIMESTAMP;
+            1 100
+            2 200
+
+            > SELECT * FROM retain_history_mv1 AS OF '{mz_time2}'::TIMESTAMP;
+            1 101
+            2 201
+            3 300
+            4 400
+
+            > SELECT * FROM retain_history_mv1 AS OF AT LEAST '{mz_time2}'::TIMESTAMP;
+            1 101
+            2 201
+            3 300
+            4 400
+
+            > SELECT sum(value), max(value) FROM retain_history_mv1 AS OF '{mz_time2}'::TIMESTAMP;
+            1002 400
+
+            > SELECT count(*) FROM retain_history_mv1 AS OF '{mz_time3}'::TIMESTAMP;
+            2
+
+            ? EXPLAIN SELECT * FROM retain_history_mv1 AS OF '{mz_time2}'::TIMESTAMP;
+            Explained Query:
+              ReadStorage materialize.public.retain_history_mv1
+
+            > SELECT mv1a.key, mv1b.key
+              FROM retain_history_mv1 mv1a
+              LEFT OUTER JOIN retain_history_mv1 mv1b
+              ON mv1a.key = mv1b.key
+              AS OF '{mz_time2}'::TIMESTAMP;
+            1 1
+            2 2
+            3 3
+            4 4
+
+            ! SELECT t.key, mv.key
+              FROM retain_history_table t
+              LEFT OUTER JOIN retain_history_mv1 mv
+              ON t.key = mv.key
+              AS OF '{mz_time2}'::TIMESTAMP;
+            contains: is not valid for all inputs
+
+            > UPDATE retain_history_table SET key = 9 WHERE key = 1;
+            """,
+    )
+
+    mz_time4 = fetch_now_from_mz(c)
+
+    testdrive(
+        c,
+        f"""
+            > SELECT count(*) FROM retain_history_mv1 WHERE key = 1 AS OF '{mz_time3}'::TIMESTAMP;
+            1
+
+            > SELECT count(*) FROM retain_history_mv1 WHERE key = 1 AS OF '{mz_time4}'::TIMESTAMP;
+            0
+
+            > SELECT 1 WHERE 1 = (SELECT count(*) FROM retain_history_mv1 WHERE key = 1) AS OF '{mz_time3}'::TIMESTAMP;
+            1
+            """,
+    )
+
+    test_time5 = datetime.now()
+
+    if (test_time5 - test_time1).total_seconds() <= mv_on_mv1_retention_in_sec:
+        time.sleep(1)
+
+    assert (
+        test_time5 - test_time1
+    ).total_seconds() < mv_on_mv_on_mv1_retention_in_sec, "test precondition not satisfied, consider increasing 'mv_on_mv_on_mv1_retention_in_sec'"
+
+    testdrive(
+        c,
+        f"""
+            # retain period exceeded
+            ! SELECT * FROM retain_history_mv_on_mv1 AS OF '{mz_time1}'::TIMESTAMP;
+            contains: is not valid for all inputs
+
+            # retain period on wrapping mv still valid
+            > SELECT * FROM retain_history_mv_on_mv_on_mv1 AS OF '{mz_time1}'::TIMESTAMP;
+            1 100
+            2 200
+            """,
+    )
+
+
+def testdrive(c: Composition, sql: str) -> None:
+    c.testdrive(
+        dedent(sql),
+        default_timeout="5s",
+        no_reset=True,
+    )
+
+
+def fetch_now_from_mz(c: Composition) -> str:
+    return c.sql_query("SELECT now()")[0][0]