aggregate_purchases.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)
  )
  
  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