{% macro string_mapping(table_name, primary_key, run_mode=none) -%}
    {{ return(adapter.dispatch('string_mapping', 'dynamics_365_crm')(table_name, primary_key, run_mode)) }}
{% endmacro %}

{% macro default__string_mapping(table_name, primary_key, run_mode=none) %}
    {{ config(enabled=var('dynamics_365_crm_using_' ~ table_name, True)) }}
    {% if run_mode is not none %}
        {% do exceptions.warn("The `run_mode` argument in `string_mapping` is deprecated and no longer has any effect. It will be removed in a future release.") if execute %}
    {% endif %}
    {%- set columns = adapter.get_columns_in_relation(source('dynamics_365_crm', table_name)) -%}
    {# Retrieves the attribute names available for the subject table #}
    {%- set stringmap_columns = adapter.get_columns_in_relation(source('dynamics_365_crm', 'stringmap')) | map(attribute='name') | map('lower') | list -%}
    {%- set attribute_column = 'renamed_attributename' if 'renamed_attributename' in stringmap_columns else 'attributename' -%}
    {%- set attributes = dbt_utils.get_column_values(
        table=source('dynamics_365_crm', 'stringmap'),
        where="lower(objecttypecode) = '" ~ table_name ~ "'",
        column=attribute_column) -%}

    {# Create two lists: 1. fields for mapping 2. all the remaining fields #}
    {%- set fields = [] -%}
    {%- set non_pivot_fields = [] -%}
    {%- for col in columns -%}
        {%- set data_type = col.data_type|lower %}
        {# For Databricks, we use partial string matches since col.is_number() doesn't work as expected.
            We want partial matches to catch types like int2, int4, bigint, float4, double precision, etc. #}
        {%- set is_number = ('int' in data_type
            or 'float' in data_type
            or 'numeric' in data_type
            or 'decimal' in data_type
            or 'serial' in data_type
            or 'real' in data_type
            or 'double' in data_type
        ) if target.type == 'databricks' else col.is_number() -%}

        {%- if col.name | lower in attributes | map('lower') and is_number -%}
            {%- do fields.append(col.name) -%}
        {%- else -%}
            {%- do non_pivot_fields.append(col.name) -%}
        {%- endif -%}
    {%- endfor -%}

    {# Only pivot/unpivot if there's at least one column to map. Otherwise pass the source through unchanged #}
    {% if fields | length > 0 %}
    with base as(
        select *
        from {{ source('dynamics_365_crm', table_name) }}
    
    -- Select only the primary key to shorten the compiled query--rejoin remaining fields later
    ), unpivoted as (
        {%- for field in fields -%}
        select
            {{ primary_key }},
            cast('{{ field }}' as {{ dbt.type_string() }}) as fieldname,
            cast({{ field }} as {{ dbt.type_int() }}) as fieldvalue
        from base

        {{ 'union all' if not loop.last }}
        {% endfor %}

    ), stringmaps as (
        select
            stringmapid,
            cast(attributevalue as {{ dbt.type_int() }}) as attributevalue,
            cast({{ attribute_column }} as {{ dbt.type_string() }}) as attributename,
            cast(objecttypecode as {{ dbt.type_string() }}) as objecttypecode,
            cast(value as {{ dbt.type_string()}}) as stringmap_value
        from {{ source('dynamics_365_crm', 'stringmap')}}
        where lower(objecttypecode) = {{ "'" ~ table_name ~ "'" }}
        and not coalesce(_fivetran_deleted, false)

    ), joined as (
        -- the long format table can easily be joined with the stringmap table
        select
            unpivoted.*,
            stringmaps.stringmap_value as fieldvalue_name
        from unpivoted
        left join stringmaps
        on lower(unpivoted.fieldname) = lower(stringmaps.attributename)
        and unpivoted.fieldvalue = stringmaps.attributevalue

    ), repivoted as (
        -- convert back to wide format, now with the human readable columns
        select
            {% for non_pivot_field in non_pivot_fields -%}
                base.{{ non_pivot_field }},
            {% endfor %}
            {% for field in fields -%}
                max(case when lower(joined.fieldname) = lower('{{ field }}') then joined.fieldvalue else null end) as {{ field }},
                max(case when lower(joined.fieldname) = lower('{{ field }}') then joined.fieldvalue_name else null end) as {{ field }}_label{{ ',' if not loop.last }}
            {% endfor %}
        from joined
        left join base
            on joined.{{ primary_key }} = base.{{ primary_key }}
        {{ dbt_utils.group_by(non_pivot_fields | length) }}
    )

    select *
    from repivoted

    {%- else %}

        {# When no columns to string map, pass the source through unchanged #}
        select *
        from {{ source('dynamics_365_crm', table_name) }}

    {% endif %}
{% endmacro %}