Add trino__equality_fct_source_fanout test

integration_tests/dbt_project_evaluator/dbt_project.yml

@@ -55,6 +55,9 @@ models:
         fct_model_naming_conventions:
           +enabled: false
       dag:
+        # Overridden by trino__fct_source_fanout
+        fct_source_fanout:
+          +enabled: false
         # Overridden by trino__fct_model_fanout
         fct_model_fanout:
           +enabled: false
@@ -67,6 +70,9 @@ tests:
       # Overridden by trino__equality_fct_duplicate_sources
       equality_fct_duplicate_sources:
         +enabled: false
+      # Overridden by trino__equality_fct_source_fanout
+      equality_fct_source_fanout:
+        +enabled: false
       # Overridden by trino__equality_fct_model_fanout
       equality_fct_model_fanout:
         +enabled: false
@@ -100,6 +106,9 @@ seeds:
       test_fct_model_naming_conventions:
         +enabled: false
     dag:
+      # Overridden by trino__test_fct_source_fanout
+      test_fct_source_fanout:
+        +enabled: false
       # Overridden by trino__test_fct_model_fanout
       test_fct_model_fanout:
         +enabled: false

integration_tests/dbt_project_evaluator/models/marts/dag/trino__fct_source_fanout.sql

@@ -0,0 +1,30 @@
+-- this model finds cases where a source is used in multiple direct downstream models
+with direct_source_relationships as (
+    select  
+        *
+    from {{ ref('int_all_dag_relationships') }}
+    where distance = 1
+    and parent_resource_type = 'source'
+    and child_resource_type = 'model'
+    and not parent_is_excluded
+    and not child_is_excluded
+    -- we order the CTE so that listagg returns values correctly sorted for some warehouses
+    order by child
+),
+
+source_fanout as (
+    select
+        parent,
+        {{ dbt.listagg(
+            measure='child', 
+            delimiter_text="', '", 
+            order_by_clause='order by child')
+        }} as model_children
+    from direct_source_relationships
+    group by 1
+    having count(*) > 1
+)
+
+select * from source_fanout
+
+{{ trino__filter_exceptions(model.name) }}

integration_tests/dbt_project_evaluator/seeds/dag/dag_seeds.yml

@@ -9,6 +9,15 @@ seeds:
           compare_columns:
             - source_names
 
+  - name: trino__test_fct_source_fanout
+    tests:
+      - dbt_utils.equality:
+          name: trino__equality_fct_source_fanout
+          compare_model: ref('trino__fct_source_fanout')
+          compare_columns:
+            - parent
+            - model_children
+
   - name: trino__test_fct_model_fanout
     tests:
       - dbt_utils.equality:

integration_tests/dbt_project_evaluator/seeds/dag/trino__test_fct_source_fanout.csv

@@ -0,0 +1,3 @@
+parent,model_children
+source_1.table_2,"int_model_4, stg_model_2"
+source_1.table_1,"stg_model_1, stg_model_2"