Docs: Define the WAP materialization observation target

Signed-off-by: mday-io <mdaytn@gmail.com>

Author: mday-io

docs/reference/execution_observations.md

@@ -125,8 +125,13 @@ SQLMesh calls `finish` for an audit before it runs later audits or completes
 write-audit-publish (WAP). A later audit or WAP publication can still fail. An audit
 query failure can produce a `finish` callback without an `AuditResult`. The
 [audits guide](../concepts/audits.md) describes audit behavior in plans and
-runs. The physical materialization callback ends after its surrounding
-transaction and session exit.
+runs.
+
+For `physical/materialize`, `target` is the logical destination table from the
+snapshot. During WAP, SQLMesh writes to a staging destination, but the `target`
+fact remains the logical destination. The `finish` callback runs after the
+materialization's transaction and session exit. It does not report the result
+of later audits or WAP publication.
 
 An observed operation can do no work, issue multiple statements, or trigger
 internal driver retries. The observation does not predict how many statements

tests/core/test_snapshot_evaluator.py

@@ -5071,9 +5071,15 @@ def test_multiple_engine_virtual_layer(snapshot: Snapshot, adapters, make_snapsh
     )
 
 
+@pytest.mark.parametrize("observer_enabled", [False, True])
 def test_wap_basic(
-    adapter_mock: Mock, make_snapshot: t.Callable[..., Snapshot], mocker: MockerFixture
+    adapter_mock: Mock,
+    make_snapshot: t.Callable[..., Snapshot],
+    mocker: MockerFixture,
+    observer_enabled: bool,
 ) -> None:
+    from sqlmesh.core.execution_observation import observer_scope
+
     evaluator = SnapshotEvaluator(adapter_mock)
 
     model = SqlModel(
@@ -5089,19 +5095,25 @@ def test_wap_basic(
     expected_wap_table = "test_schema.test_table.branch_wap_12345678"
     adapter_mock.wap_prepare.return_value = expected_wap_table
 
-    wap_id = evaluator.evaluate(
-        snapshot,
-        start="2020-01-01",
-        end="2020-01-01",
-        execution_time="2020-01-01",
-        snapshots={},
-        target_table_exists=True,  # Use parameter to control table existence
-    )
+    observations = []
+
+    def observer(phase, facts, error):
+        observations.append((phase, dict(facts), error))
+
+    expected_table_name = snapshot.table_name()
+    with observer_scope(observer if observer_enabled else None):
+        wap_id = evaluator.evaluate(
+            snapshot,
+            start="2020-01-01",
+            end="2020-01-01",
+            execution_time="2020-01-01",
+            snapshots={},
+            target_table_exists=True,  # Use parameter to control table existence
+        )
 
     assert wap_id is not None
     assert len(wap_id) == 8
 
-    expected_table_name = snapshot.table_name()
     adapter_mock.wap_prepare.assert_called_once_with(expected_table_name, wap_id)
     adapter_mock.replace_query.assert_called_once_with(
         expected_wap_table,
@@ -5118,6 +5130,21 @@ def test_wap_basic(
         source_columns=mocker.ANY,
     )
 
+    materialization_observations = [
+        (phase, facts, error)
+        for phase, facts, error in observations
+        if facts.get("kind") == "physical" and facts.get("action") == "materialize"
+    ]
+    if observer_enabled:
+        assert [
+            (phase, facts["target"], error) for phase, facts, error in materialization_observations
+        ] == [
+            ("start", expected_table_name, None),
+            ("finish", expected_table_name, None),
+        ]
+    else:
+        assert observations == []
+
 
 def test_wap_model_wap_supported(
     adapter_mock: Mock, make_snapshot: t.Callable[..., Snapshot], mocker: MockerFixture