chore: use mv for course_grades and course_status

models/grading/fact_grade_status.sql

@@ -1,44 +1,12 @@
-with
-    latest_grades as (
-        select
-            org,
-            course_key,
-            actor_id,
-            splitByString('/', verb_id)[-1] as latest_state,
-            row_number() over (
-                partition by org, course_key, actor_id order by emission_time desc
-            ) as rn
-        from {{ ref("grading_events") }}
-        where
-            verb_id in (
-                'http://adlnet.gov/expapi/verbs/passed',
-                'http://adlnet.gov/expapi/verbs/failed'
-            )
-    ),
-
-    latest_course_grades as (
-        select
-            org,
-            course_key,
-            actor_id,
-            scaled_score as course_grade,
-            row_number() over (
-                partition by org, course_key, actor_id order by emission_time desc
-            ) as rn
-        from {{ ref("fact_grades") }}
-        where grade_type = 'course'
-    )
-
 select
-    lg.org as org,
-    lg.course_key as course_key,
-    lg.actor_id as actor_id,
-    lg.latest_state as latest_state,
-    lc.course_grade as course_grade
-from latest_grades lg
+    ls.org as org,
+    ls.course_key as course_key,
+    ls.actor_id as actor_id,
+    coalesce(state, 'failed') as state,
+    course_grade as course_grade
+from {{ ref("fact_learner_course_grade") }} ls
 left join
-    latest_course_grades lc
-    on lg.org = lc.org
-    and lg.course_key = lc.course_key
-    and lg.actor_id = lc.actor_id
-where lg.rn = 1 and lc.rn = 1
+    {{ ref("fact_learner_course_status") }} lg
+    on ls.org = lg.org
+    and ls.course_key = lg.course_key
+    and ls.actor_id = lg.actor_id

models/grading/fact_learner_course_grade.sql

@@ -0,0 +1,26 @@
+{{
+    config(
+        materialized="materialized_view",
+        engine=get_engine("ReplacingMergeTree()"),
+        primary_key="(org, course_key, actor_id)",
+        order_by="(org, course_key, actor_id)",
+    )
+}}
+
+with
+    ranked_grades as (
+        select
+            org,
+            course_key,
+            actor_id,
+            scaled_score as course_grade,
+            row_number() over (
+                partition by org, course_key, actor_id order by emission_time desc
+            ) as rn
+        from {{ ref("fact_grades") }}
+        where grade_type = 'course'
+    )
+
+select org, course_key, actor_id, course_grade
+from ranked_grades
+where rn = 1

models/grading/fact_learner_course_status.sql

@@ -0,0 +1,30 @@
+{{
+    config(
+        materialized="materialized_view",
+        engine=get_engine("ReplacingMergeTree()"),
+        primary_key="(org, course_key, actor_id)",
+        order_by="(org, course_key, actor_id)",
+    )
+}}
+
+with
+    ranked_status as (
+        select
+            org,
+            course_key,
+            actor_id,
+            splitByString('/', verb_id)[-1] as state,
+            row_number() over (
+                partition by org, course_key, actor_id order by emission_time desc
+            ) as rn
+        from {{ ref("grading_events") }}
+        where
+            verb_id in (
+                'http://adlnet.gov/expapi/verbs/passed',
+                'http://adlnet.gov/expapi/verbs/failed'
+            )
+    )
+
+select org, course_key, actor_id, state
+from ranked_status
+where rn = 1