ADAP-761: Add retry logic to flaky MV tests to avoid "cannot open relation with OID" error

.changes/unreleased/Under the Hood-20230808-141645.yaml

@@ -0,0 +1,7 @@
+kind: Under the Hood
+body: Update flaky MV tests to use retry logic to avoid "cannot open relation with
+  OID" error
+time: 2023-08-08T14:16:45.227308-04:00
+custom:
+  Author: mikealfare
+  Issue: "569"

tests/functional/adapter/materialized_view_tests/test_materialized_views.py

@@ -12,13 +12,14 @@
     MaterializedViewChangesFailMixin,
 )
 from dbt.tests.adapter.materialized_view.files import MY_TABLE, MY_VIEW
-from dbt.tests.util import get_model_file, set_model_file
+from dbt.tests.util import assert_message_in_logs, get_model_file, set_model_file
 
 from tests.functional.adapter.materialized_view_tests.utils import (
     query_autorefresh,
     query_dist,
     query_relation_type,
     query_sort,
+    run_dbt_and_capture_with_retries,
 )
 
 
@@ -61,6 +62,12 @@ def query_row_count(project, relation: BaseRelation) -> int:
     def query_relation_type(project, relation: BaseRelation) -> Optional[str]:
         return query_relation_type(project, relation)
 
+    def test_materialized_view_create_idempotent(self, project, my_materialized_view):
+        # setup creates it once; verify it's there and run once
+        assert self.query_relation_type(project, my_materialized_view) == "materialized_view"
+        run_dbt_and_capture_with_retries(["run", "--models", my_materialized_view.identifier])
+        assert self.query_relation_type(project, my_materialized_view) == "materialized_view"
+
     @pytest.mark.skip(
         "The current implementation does not support overwriting materialized views with tables."
     )
@@ -120,16 +127,77 @@ def check_state_replace_change_is_applied(project, materialized_view):
 class TestRedshiftMaterializedViewChangesApply(
     RedshiftMaterializedViewChanges, MaterializedViewChangesApplyMixin
 ):
-    pass
+    def test_change_is_applied_via_alter(self, project, my_materialized_view):
+        self.check_start_state(project, my_materialized_view)
+
+        self.change_config_via_alter(project, my_materialized_view)
+        _, logs = run_dbt_and_capture_with_retries(
+            ["--debug", "run", "--models", my_materialized_view.name]
+        )
+
+        self.check_state_alter_change_is_applied(project, my_materialized_view)
+
+        assert_message_in_logs(f"Applying ALTER to: {my_materialized_view}", logs)
+        assert_message_in_logs(f"Applying REPLACE to: {my_materialized_view}", logs, False)
+
+    def test_change_is_applied_via_replace(self, project, my_materialized_view):
+        self.check_start_state(project, my_materialized_view)
+
+        self.change_config_via_alter(project, my_materialized_view)
+        self.change_config_via_replace(project, my_materialized_view)
+        _, logs = run_dbt_and_capture_with_retries(
+            ["--debug", "run", "--models", my_materialized_view.name]
+        )
+
+        self.check_state_alter_change_is_applied(project, my_materialized_view)
+        self.check_state_replace_change_is_applied(project, my_materialized_view)
+
+        assert_message_in_logs(f"Applying REPLACE to: {my_materialized_view}", logs)
 
 
 class TestRedshiftMaterializedViewChangesContinue(
     RedshiftMaterializedViewChanges, MaterializedViewChangesContinueMixin
 ):
-    pass
+    def test_change_is_not_applied_via_alter(self, project, my_materialized_view):
+        self.check_start_state(project, my_materialized_view)
+
+        self.change_config_via_alter(project, my_materialized_view)
+        _, logs = run_dbt_and_capture_with_retries(
+            ["--debug", "run", "--models", my_materialized_view.name]
+        )
+
+        self.check_start_state(project, my_materialized_view)
+
+        assert_message_in_logs(
+            f"Configuration changes were identified and `on_configuration_change` was set"
+            f" to `continue` for `{my_materialized_view}`",
+            logs,
+        )
+        assert_message_in_logs(f"Applying ALTER to: {my_materialized_view}", logs, False)
+        assert_message_in_logs(f"Applying REPLACE to: {my_materialized_view}", logs, False)
+
+    def test_change_is_not_applied_via_replace(self, project, my_materialized_view):
+        self.check_start_state(project, my_materialized_view)
+
+        self.change_config_via_alter(project, my_materialized_view)
+        self.change_config_via_replace(project, my_materialized_view)
+        _, logs = run_dbt_and_capture_with_retries(
+            ["--debug", "run", "--models", my_materialized_view.name]
+        )
+
+        self.check_start_state(project, my_materialized_view)
+
+        assert_message_in_logs(
+            f"Configuration changes were identified and `on_configuration_change` was set"
+            f" to `continue` for `{my_materialized_view}`",
+            logs,
+        )
+        assert_message_in_logs(f"Applying ALTER to: {my_materialized_view}", logs, False)
+        assert_message_in_logs(f"Applying REPLACE to: {my_materialized_view}", logs, False)
 
 
 class TestRedshiftMaterializedViewChangesFail(
     RedshiftMaterializedViewChanges, MaterializedViewChangesFailMixin
 ):
+    # Note: using retries doesn't work when we expect `dbt_run` to fail
     pass

tests/functional/adapter/materialized_view_tests/utils.py

@@ -1,6 +1,7 @@
-from typing import Optional
+from typing import List, Optional
 
 from dbt.adapters.base.relation import BaseRelation
+from dbt.tests.util import run_dbt_and_capture
 
 from dbt.adapters.redshift.relation import RedshiftRelation
 
@@ -67,3 +68,20 @@ def query_autorefresh(project, relation: RedshiftRelation) -> bool:
         and trim(mv.db_name) ilike '{ relation.database }'
     """
     return project.run_sql(sql, fetch="one")[0]
+
+
+def run_dbt_and_capture_with_retries(args: List[str], max_retries: int = 10):
+    """
+    We need to retry `run_dbt` calls on Redshift because we get sporadic failures of the form:
+
+        Database Error in model my_materialized_view (models/my_materialized_view.sql)
+        could not open relation with OID 14957412
+    """
+    retries = 0
+    while retries < max_retries:
+        try:
+            # there's no point to using this with expect_pass=False
+            return run_dbt_and_capture(args, expect_pass=True)
+        except AssertionError:
+            retries += 1
+    return None