price_with_promotion_history.sql

WITH price_history_per_shop_per_article AS (
  SELECT
    shop_id, 
    article_id,
    ARRAY_AGG((SELECT AS STRUCT T.* EXCEPT(shop_id, article_id)) ORDER BY start_datetime) as price_history
  FROM `supermarket.sell_price_history` T
  GROUP BY shop_id, article_id
)
, promotion_history_per_shop_per_article AS (
  SELECT
    shop_id, 
    article_id,
    ARRAY_AGG((SELECT AS STRUCT T.* EXCEPT(shop_id, article_id)) ORDER BY start_datetime) as promotion_history
  FROM `supermarket.promotion_history` T
  GROUP BY shop_id, article_id
)
, price_and_promotion_history AS (
  SELECT 
    shop_id,
    article_id,
    price_history,
    promotion_history,
  FROM price_history_per_shop_per_article AS price
  LEFT JOIN promotion_history_per_shop_per_article AS promotion
  USING (shop_id, article_id)
)
, price_and_promotion_history_with_next_event_datetime AS (
  SELECT 
    shop_id,
    article_id,
    ARRAY(
      SELECT AS STRUCT
        *,
        LAG(start_datetime) OVER (ORDER BY start_datetime DESC) as next_event_datetime
      FROM UNNEST(price_history)
      ORDER BY start_datetime
    ) as price_history,
    ARRAY(
      SELECT AS STRUCT
        *,
        LAG(start_datetime) OVER (ORDER BY start_datetime DESC) as next_event_datetime
      FROM UNNEST(promotion_history)
      ORDER BY start_datetime
    ) as promotion_history,
  FROM price_and_promotion_history
)
, price_and_promotion_events AS (
  SELECT 
    * EXCEPT(price_history, promotion_history),
    ARRAY (
      SELECT AS STRUCT
        sell_price,
        CAST(NULL as STRING) as promotion_id,
        NULL as promotion_value,
        start_datetime,
      FROM UNNEST(price_history)
      UNION ALL
      SELECT AS STRUCT
        -1 as sell_price,
        CAST(NULL as STRING) as promotion_id,
        NULL as promotion_value,
        end_datetime,
      FROM UNNEST(price_history)
      WHERE end_datetime < next_event_datetime OR next_event_datetime IS NULL
      UNION ALL
      SELECT AS STRUCT
        NULL as sell_price,
        promotion_id,
        promotion_value,
        start_datetime,
      FROM UNNEST(promotion_history)
      UNION ALL
      SELECT AS STRUCT
        NULL as sell_price,
        "" as promotion_id,
        -1 as promotion_value,
        end_datetime,
      FROM UNNEST(promotion_history)
      WHERE end_datetime < next_event_datetime OR next_event_datetime IS NULL
      ORDER BY start_datetime
    ) as events,
  FROM price_and_promotion_history_with_next_event_datetime
)
, price_with_promotion_history_before_cleaning AS (
  SELECT
    * EXCEPT(events),
    ARRAY(
      SELECT AS STRUCT
        LAST_VALUE(sell_price IGNORE NULLS) OVER (ORDER BY start_datetime) as sell_price,
        LAST_VALUE(promotion_id IGNORE NULLS) OVER (ORDER BY start_datetime) as promotion_id,
        LAST_VALUE(promotion_value IGNORE NULLS) OVER (ORDER BY start_datetime) as promotion_value,
        start_datetime,
        LEAD(start_datetime) OVER (ORDER BY start_datetime) as end_datetime
      FROM UNNEST(events)
      ORDER BY start_datetime
    ) as price_and_promotion
  FROM price_and_promotion_events
)
, price_with_promotion_history AS (
  SELECT 
    * EXCEPT(price_and_promotion),
    ARRAY(
      SELECT AS STRUCT
        NULLIF(sell_price, -1) as sell_price,
        NULLIF(promotion_id, "") as promotion_id,
        NULLIF(promotion_value, -1) as promotion_value,
        start_datetime,
        end_datetime
      FROM UNNEST(price_and_promotion)
      WHERE start_datetime < end_datetime
    ) as price_and_promotion
  FROM price_with_promotion_history_before_cleaning
)
SELECT 
    shop_id,
    article_id,
    sell_price,
    promotion_id,
    promotion_value,
    COALESCE(promotion_value, sell_price) as price_with_promotion_included,
    start_datetime,
    end_datetime
FROM price_with_promotion_history
LEFT JOIN UNNEST(price_and_promotion)

promotion_history.csv

sell_price_history.csv