elt-full.sql

with source_data as (
    select * from `company-a.source.source_data`
),

reading_fields as (
    select
        JSON_VALUE(_loaded_data, "$.call_time") as call_time,
        JSON_VALUE(_loaded_data, "$.number") as caller_number,
        JSON_VALUE(_loaded_data, "$.type") as call_type,
        JSON_VALUE(_loaded_data, "$.outcome") as call_outcome,
        JSON_VALUE(_loaded_data, "$.seconds") as call_duration,
        _loaded_at,
        _loaded_data

    from
        source_data
),

cleaned_fields as (
    select
        -- upstream call_time is string representing a timestamp in CET
        parse_timestamp("%Y-%m-%dT%H:%M:%S", call_time, 'Europe/Stockholm') as ct,
        caller_number,
        case
            when call_type = "Incoming" then 'incoming'
            when call_type = "Outgoing" then 'outgoing'
        end as call_type,
        case
            when call_outcome = "Answered" then 'answered'
            when call_outcome = "Unanswered" then 'unanswered'
        end as call_outcome,
        cast(call_duration as float64) * 60 as call_duration,
    from
        reading_fields

),

clean_with_surr_key as (
    select

        {{ dbt_utils.surrogate_key(['call_time', 'caller_number', 'caller_number', 'call_type', 'call_outcome', 'call_duration']) }} as call_surr_key,
        row_number() over (
            partition by {{ dbt_utils.surrogate_key(['call_time', 'caller_number', 'caller_number', 'call_type', 'call_outcome', 'call_duration']) }}
            order by 
                _loaded_at desc 
        ) end as call_surr_key_rn,
        call_time,
        caller_number,
        call_type,
        call_outcome,
        call_duration,
        _loaded_at,
        _loaded_data

    from
        source_data
),

out as (
    select 
        * 
    except
        (call_surr_key_rn, _loaded_at, _loaded_data) 
    from 
        clean_with_surr_key 
    where
        call_surr_key_rn = 1
)

select * from out order by call_time desc