-- Upsert data from RDS to Snowflake
-- This query inserts new records and updates existing ones based on SESSION_ID and partition

merge into {session_history_table_name} as target
using 
(
    select 
        $1:session_id as session_id,
        $1:session_type as session_type,
        $1:user_id as user_id,
        $1:ip as ip,
        $1:created_at as created_at,
        $1:updated_at as updated_at,
        $1:expire_time as expire_time,
        $1:length_in_sec as length_in_sec,
        $1:properties as properties,
        $1:platform as platform,
        date(to_timestamp(created_at)) as p_date,
        hour(to_timestamp(created_at)) as p_hour
    from {stage_path} (file_format => 'BASIC_PARQUET_LOADER', pattern => ('.*.parquet'))
) as source
on (target.session_id = source.session_id and target.p_date = source.p_date and target.p_hour = source.p_hour)
when matched then
    update set 
        target.session_type = source.session_type,
        target.user_id = source.user_id,
        target.ip = source.ip,
        target.created_at = source.created_at,
        target.updated_at = source.updated_at,
        target.expire_time = source.expire_time,
        target.length_in_sec = source.length_in_sec,
        target.properties = source.properties,
        target.platform = source.platform,
        target.p_date = source.p_date,
        target.p_hour = source.p_hour
when not matched then
    insert (session_id, session_type, user_id, ip, created_at, updated_at, expire_time, length_in_sec, properties, platform, p_date, p_hour)
    values (source.session_id, source.session_type, source.user_id, source.ip, source.created_at, source.updated_at, source.expire_time, source.length_in_sec, source.properties, source.platform, source.p_date, source.p_hour);
