Methods to achieve null safety for deduplicate

macros/sql/deduplicate.sql

@@ -1,8 +1,40 @@
 {%- macro deduplicate(relation, partition_by, order_by) -%}
-    {{ return(adapter.dispatch('deduplicate', 'dbt_utils')(relation, partition_by, order_by)) }}
+    {{ return(adapter.dispatch('deduplicate', 'dbt_utils')(relation, partition_by, order_by, **kwargs)) }}
 {% endmacro %}
 
-{%- macro default__deduplicate(relation, partition_by, order_by) -%}
+{#
+-- ⚠️ This macro drops rows that contain NULL values ⚠️
+
+-- The implementation below uses a natural join which avoids returning an
+-- extra column at the cost of not being null safe.
+
+-- dbt_utils._safe_deduplicate is an alternative that avoids dropping rows
+-- that contain NULL values at the cost of adding an extra column.
+#}
+{%- macro _unsafe_deduplicate(relation, partition_by, order_by) -%}
+
+{%- set error_message = "
+Warning: the implementation of the `deduplicate` macro for the `{}` adapter is not null safe. \
+
+Set `row_alias` within calls to `deduplicate` to achieve null safety (which will also add it \
+as an extra column to the output).
+
+e.g.,
+    {{
+        dbt_utils.deduplicate(
+            'my_cte',
+            partition_by='user_id',
+            order_by='version desc',
+            row_alias='rn'
+        ) | indent
+    }}
+
+Warning triggered by model: {}.{}
+dbt project / package: {}
+path: {}
+".format(target.type, model.package_name, model.name, model.package_name, model.original_file_path) -%}
+
+{%- do exceptions.warn(error_message) -%}
 
     with row_numbered as (
         select
@@ -29,6 +61,63 @@
 
 {%- endmacro -%}
 
+{#
+-- For data platforms that don't support QUALIFY or an equivalent, the
+-- best we can do to ensure null safety is to use a window function +
+-- filter (which returns an extra column):
+-- https://modern-sql.com/caniuse/qualify
+#}
+{%- macro _safe_deduplicate(relation, partition_by, order_by, row_alias="rn", columns=none) -%}
+
+    {% if not row_alias %}
+        {% set row_alias = "rn" %}
+    {% endif %}
+
+    with row_numbered as (
+        select
+
+        {% if columns != None %}
+            {% for column in columns %}
+            {{ column }},
+            {% endfor %}
+        {% else %}
+            _inner.*,
+        {% endif %}
+
+            row_number() over (
+                partition by {{ partition_by }}
+                order by {{ order_by }}
+            ) as {{ row_alias }}
+        from {{ relation }} as _inner
+    )
+
+    select *
+    from row_numbered
+    where {{ row_alias }} = 1
+
+{%- endmacro -%}
+
+{#
+-- ⚠️ This macro drops rows that contain NULL values unless one of the following is true:
+--   - `relation` parameter is a non-CTE dbt Relation
+--   - `row_alias` parameter is included
+--   - `columns` parameter is included
+#}
+{%- macro default__deduplicate(relation, partition_by, order_by) -%}
+    {% set row_alias = kwargs.get('row_alias') %}
+    {% set columns = kwargs.get('columns') %}
+
+    {% if relation.is_cte is defined and not relation.is_cte %}
+        {% set columns = dbt_utils.get_filtered_columns_in_relation(relation) %}
+        {{ dbt_utils._safe_deduplicate(relation, partition_by, order_by, columns=columns) }}
+    {% elif row_alias != None or columns != None %}
+        {{ dbt_utils._safe_deduplicate(relation, partition_by, order_by, row_alias=row_alias, columns=columns) }}
+    {% else %}
+        {{ dbt_utils._unsafe_deduplicate(relation, partition_by, order_by) }}
+    {% endif %}
+
+{%- endmacro -%}
+
 -- Redshift has the `QUALIFY` syntax:
 -- https://docs.aws.amazon.com/redshift/latest/dg/r_QUALIFY_clause.html
 {% macro redshift__deduplicate(relation, partition_by, order_by) -%}