merge into {periodcreditusage_table_name} as target
using 
(
    select 
        $1:id as id,
        $1:created_at as created_at,
        $1:updated_at as updated_at,
        $1:period_anchor as period_anchor,
        $1:credits_used as credits_used,
        $1:period_credit_limit as period_credit_limit,
        $1:extra_credits_balance as extra_credits_balance,
        $1:user_id as user_id,
        $1:period_end as period_end,
        $1:stripe_subscription_id as stripe_subscription_id,
        $1:plan_level as plan_level,
        $1:extra_credits_used as extra_credits_used,
        $1:usage_plan_id as usage_plan_id,
        $1:revcat_subscription_id as revcat_subscription_id,
        $1:is_active as is_active,
        $1:period_extra_credit_limit as period_extra_credit_limit,
        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.id = source.id and target.p_date = source.p_date and target.p_hour = source.p_hour)
when matched then
    update set 
        target.created_at = source.created_at,
        target.updated_at = source.updated_at,
        target.period_anchor = source.period_anchor,
        target.credits_used = source.credits_used,
        target.period_credit_limit = source.period_credit_limit,
        target.extra_credits_balance = source.extra_credits_balance,
        target.user_id = source.user_id,
        target.period_end = source.period_end,
        target.stripe_subscription_id = source.stripe_subscription_id,
        target.plan_level = source.plan_level,
        target.extra_credits_used = source.extra_credits_used,
        target.usage_plan_id = source.usage_plan_id,
        target.revcat_subscription_id = source.revcat_subscription_id,
        target.is_active = source.is_active,
        target.period_extra_credit_limit = source.period_extra_credit_limit,
        target.p_date = source.p_date,
        target.p_hour = source.p_hour
when not matched then
    insert (id, created_at, updated_at, period_anchor, credits_used, period_credit_limit, extra_credits_balance, user_id, period_end, stripe_subscription_id, plan_level, extra_credits_used, usage_plan_id, revcat_subscription_id, is_active, period_extra_credit_limit, p_date, p_hour)
    values (source.id, source.created_at, source.updated_at, source.period_anchor, source.credits_used, source.period_credit_limit, source.extra_credits_balance, source.user_id, source.period_end, source.stripe_subscription_id, source.plan_level, source.extra_credits_used, source.usage_plan_id, source.revcat_subscription_id, source.is_active, source.period_extra_credit_limit, source.p_date, source.p_hour);
