Add test for relative change in column values

macros/schema_tests/column_values_basic/expect_column_value_running_abs_change_ratio_to_be

@@ -0,0 +1,63 @@
+{% test expect_column_value_running_abs_change_ratio_to_be(model, column_name,change_threshold,
+                                                   sort_column=None,
+                                                   row_condition=None,
+                                                   strictly=True,
+                                                   group_by=None) %}
+
+{%- set sort_column = column_name if not sort_column else sort_column -%}
+{%- set operator = "<" if strictly else "<=" %}
+
+with all_values as (
+
+    select
+        {{ sort_column }},
+        {%- if group_by -%}
+        {{ group_by | join(", ") }},
+        {%- endif %}
+        {{ column_name }}
+    from {{ model }}
+    {% if row_condition %}
+    where {{ row_condition }}
+    {% endif %}
+
+),
+add_lag_values as (
+
+    select
+        {{ group_by | join(", ") }}
+        , {{ sort_column }} as current_{{ sort_column | trim }}
+        , lag( {{- sort_column -}} ) over
+            {%- if not group_by -%}
+                (order by sort_column)
+            {%- else -%}
+                (partition by {{ group_by | join(", ") }} order by {{ sort_column }})
+            {%- endif  %} as previous_{{ sort_column | trim }}  
+        , {{ column_name }} as current_{{ column_name | trim }} 
+        , lag( {{- column_name -}} ) over
+            {%- if not group_by -%}
+                (order by sort_column)
+            {%- else -%}
+                (partition by {{ group_by | join(", ") }} order by {{ sort_column }})
+            {%- endif  %} as previous_{{ column_name | trim}}  
+    from
+        all_values
+
+),
+validation_errors as (
+
+    select
+        *
+        , current_{{ column_name | trim}} / nullif(previous_{{ column_name | trim}}, 0) - 1 as actual_change_ratio
+        , abs(current_{{ column_name | trim}} / nullif(previous_{{ column_name | trim}}, 0)) - 1 as abs_change_ratio 
+        , '{{ operator }} {{ change_threshold }}'  as criteria
+    from
+        add_lag_values
+    where
+        previous_{{ column_name | trim}} is not null
+        and
+        not ((abs(current_{{ column_name | trim}} / nullif(previous_{{ column_name | trim}}, 0) )-1) {{ operator }} {{ change_threshold }} )
+
+)
+select *
+from validation_errors
+{% endtest %}