get_product_name.sql

WITH lastest_purchase_time as (
    SELECT
        u.user_id,
        _event_time,
        u.purchased
   FROM
        kafkaWorkshop.CustomerPurchaseActivities u
    WHERE
        u.user_id = :userid
    ORDER BY
        _event_time desc
    LIMIT
        1
), activity as(

   SELECT
        u.product_id,
        u._event_type,
        u.user_id
    FROM
        lastest_purchase_time l,
        kafkaWorkshop.CustomerClickActivities u
    WHERE
        l._event_time < u._event_time
        AND  TRY_CAST( u.user_id as string) = :userid
        AND (
            u._event_type = 'viewed'
            or u._event_type = 'added_to_cart'
        )
    ORDER BY
        u._event_time desc
), cohorts as(

 SELECT
        u.user_id as other_users,
        u.purchased as otherItemsPurchased,
        i.user_id as user_id,
        i.product_id as productInterest
    FROM

         kafkaWorkshop.CustomerPurchaseActivities u,
         activity i
    WHERE
        ARRAY_CONTAINS(u.purchased, i.product_id)
  ), aggregate as (
  
  SELECT
        all_purch.purchase,
        COUNT(*) as purch_count
    from
        cohorts u,
        unnest(u.otherItemsPurchased as purchase) as all_purch
    GROUP BY
        all_purch.purchase
    ORDER BY
        purch_count DESC
  ), filterResult as (
  
  select
    *
from
    aggregate a
where
    a.purchase not in (
        select
            product_id
        from
            activity
    )
  )
  Select
	 TRY_CAST(fr.purchase as string) as purchase,
     p.product_name,
     p.price
from
    filterResult fr
    join kafkaWorkshop.ProductInformation p on TRY_CAST(fr.purchase as string) = p._id
order by
    fr.purch_count desc
limit 3
;