WITH installs AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_installs_app_version
), ratings AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_ratings_app_version
), crashes AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_crashes_app_version
), install_metrics AS (
  SELECT
    *,
    SUM(device_installs) OVER (PARTITION BY app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    SUM(device_uninstalls) OVER (PARTITION BY app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls
  FROM installs
), app_version_join AS (
  SELECT
    COALESCE(install_metrics.date_day, ratings.date_day, crashes.date_day) AS date_day, /* these 3 columns are the grain of this model */
    COALESCE(install_metrics.app_version_code, ratings.app_version_code, crashes.app_version_code) AS app_version_code,
    COALESCE(install_metrics.package_name, ratings.package_name, crashes.package_name) AS package_name,
    COALESCE(install_metrics.active_devices_last_30_days, 0) AS active_devices_last_30_days, /* metrics based on unique devices + users */
    COALESCE(install_metrics.device_installs, 0) AS device_installs,
    COALESCE(install_metrics.device_uninstalls, 0) AS device_uninstalls,
    COALESCE(install_metrics.device_upgrades, 0) AS device_upgrades,
    COALESCE(install_metrics.user_installs, 0) AS user_installs,
    COALESCE(install_metrics.user_uninstalls, 0) AS user_uninstalls,
    COALESCE(crashes.crashes, 0) AS crashes, /* metrics based on events. a user or device can have multiple events in one day */
    COALESCE(crashes.anrs, 0) AS anrs,
    COALESCE(install_metrics.install_events, 0) AS install_events,
    COALESCE(install_metrics.uninstall_events, 0) AS uninstall_events,
    COALESCE(install_metrics.update_events, 0) AS update_events,
    install_metrics.total_device_installs,
    install_metrics.total_device_uninstalls,
    ratings.average_rating, /* this one actually isn't rolling but we won't coalesce days with no reviews to 0 rating */
    ratings.rolling_total_average_rating
  FROM install_metrics
  FULL OUTER JOIN ratings
    ON install_metrics.date_day = ratings.date_day
    AND install_metrics.package_name = ratings.package_name
    AND /* choosing an arbitrary negative integer as we can't coalesce with a string like 'null_version_code'. null app version codes will cause fanout */ COALESCE(install_metrics.app_version_code, -5) = COALESCE(ratings.app_version_code, -5) /* this really doesn't happen IRL but let's be safe */
  FULL OUTER JOIN crashes
    ON COALESCE(install_metrics.date_day, ratings.date_day) = crashes.date_day
    AND COALESCE(install_metrics.package_name, ratings.package_name) = crashes.package_name
    AND COALESCE(install_metrics.app_version_code, ratings.app_version_code, -5) = COALESCE(crashes.app_version_code, -5)
), create_partitions /* to backfill in days with NULL values for rolling metrics, we'll create partitions to batch them together with records that have non-null values */ /* we can't just use last_value(ignore nulls) because of postgres :/ */ AS (
  SELECT
    *,
    SUM(CASE WHEN rolling_total_average_rating IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating_partition,
    SUM(CASE WHEN total_device_installs IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs_partition,
    SUM(CASE WHEN total_device_uninstalls IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls_partition
  FROM app_version_join
), fill_values /* now we'll take the non-null value for each partitioned batch and propagate it across the rows included in the batch */ AS (
  SELECT
    date_day,
    app_version_code,
    package_name,
    active_devices_last_30_days,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    crashes,
    anrs,
    install_events,
    uninstall_events,
    update_events,
    average_rating,
    FIRST_VALUE(rolling_total_average_rating) OVER (PARTITION BY rolling_total_average_rating_partition, app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating,
    FIRST_VALUE(total_device_installs) OVER (PARTITION BY total_device_installs_partition, app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    FIRST_VALUE(total_device_uninstalls) OVER (PARTITION BY total_device_uninstalls_partition, app_version_code, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls
  FROM create_partitions
), final AS (
  SELECT
    date_day,
    app_version_code,
    package_name,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    crashes,
    anrs,
    install_events,
    uninstall_events,
    update_events,
    active_devices_last_30_days,
    average_rating,
    rolling_total_average_rating, /* leave null if there are no ratings yet */
    COALESCE(total_device_installs, 0) AS total_device_installs, /* the first day will have NULL values, let's make it 0 */
    COALESCE(total_device_uninstalls, 0) AS total_device_uninstalls,
    COALESCE(total_device_installs, 0) /* calculate difference rolling metric */ - COALESCE(total_device_uninstalls, 0) AS net_device_installs
  FROM fill_values
)
SELECT
  *
FROM final

WITH installs AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_installs_country
), ratings AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_ratings_country
), store_performance AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__store_performance_country
), country_codes AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.google_play__country_codes
), install_metrics AS (
  SELECT
    *,
    SUM(device_installs) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    SUM(device_uninstalls) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls
  FROM installs
), store_performance_metrics AS (
  SELECT
    *,
    SUM(store_listing_acquisitions) OVER (PARTITION BY country_region, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_acquisitions,
    SUM(store_listing_visitors) OVER (PARTITION BY country_region, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_visitors
  FROM store_performance
), country_join AS (
  SELECT
    COALESCE(install_metrics.date_day, ratings.date_day, store_performance_metrics.date_day) AS date_day, /* these 3 columns are the grain of this model */
    COALESCE(install_metrics.country, ratings.country, store_performance_metrics.country_region) AS country,
    COALESCE(
      install_metrics.package_name,
      ratings.package_name,
      store_performance_metrics.package_name
    ) AS package_name,
    COALESCE(install_metrics.active_devices_last_30_days, 0) AS active_devices_last_30_days, /* metrics based on unique devices + users */
    COALESCE(install_metrics.device_installs, 0) AS device_installs,
    COALESCE(install_metrics.device_uninstalls, 0) AS device_uninstalls,
    COALESCE(install_metrics.device_upgrades, 0) AS device_upgrades,
    COALESCE(install_metrics.user_installs, 0) AS user_installs,
    COALESCE(install_metrics.user_uninstalls, 0) AS user_uninstalls,
    COALESCE(store_performance_metrics.store_listing_acquisitions, 0) AS store_listing_acquisitions,
    COALESCE(store_performance_metrics.store_listing_visitors, 0) AS store_listing_visitors,
    store_performance_metrics.store_listing_conversion_rate, /* not coalescing if there aren't any visitors */
    COALESCE(install_metrics.install_events, 0) AS install_events, /* metrics based on events. a user or device can have multiple installs in one day */
    COALESCE(install_metrics.uninstall_events, 0) AS uninstall_events,
    COALESCE(install_metrics.update_events, 0) AS update_events,
    install_metrics.total_device_installs,
    install_metrics.total_device_uninstalls,
    ratings.average_rating, /* this one actually isn't rolling but we won't coalesce days with no reviews to 0 rating */
    ratings.rolling_total_average_rating,
    store_performance_metrics.total_store_acquisitions,
    store_performance_metrics.total_store_visitors
  FROM install_metrics
  FULL OUTER JOIN ratings
    ON install_metrics.date_day = ratings.date_day
    AND install_metrics.package_name = ratings.package_name
    AND /* coalesce null countries otherwise they'll cause fanout with the full outer join */ COALESCE(install_metrics.country, 'null_country') = COALESCE(ratings.country, 'null_country') /* in the source package we aggregate all null country records together into one batch per day */
  FULL OUTER JOIN store_performance_metrics
    ON store_performance_metrics.date_day = COALESCE(install_metrics.date_day, ratings.date_day)
    AND store_performance_metrics.package_name = COALESCE(install_metrics.package_name, ratings.package_name)
    AND COALESCE(store_performance_metrics.country_region, 'null_country') = COALESCE(install_metrics.country, ratings.country, 'null_country')
), create_partitions /* to backfill in days with NULL values for rolling metrics, we'll create partitions to batch them together with records that have non-null values */ /* we can't just use last_value(ignore nulls) because of postgres :/ */ AS (
  SELECT
    *,
    SUM(CASE WHEN rolling_total_average_rating IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating_partition,
    SUM(CASE WHEN total_device_installs IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs_partition,
    SUM(CASE WHEN total_device_uninstalls IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls_partition,
    SUM(CASE WHEN total_store_acquisitions IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_acquisitions_partition,
    SUM(CASE WHEN total_store_visitors IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_visitors_partition
  FROM country_join
), fill_values /* now we'll take the non-null value for each partitioned batch and propagate it across the rows included in the batch */ AS (
  SELECT
    date_day,
    country,
    package_name,
    active_devices_last_30_days,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    install_events,
    uninstall_events,
    update_events,
    store_listing_acquisitions,
    store_listing_visitors,
    store_listing_conversion_rate,
    average_rating,
    FIRST_VALUE(rolling_total_average_rating) OVER (PARTITION BY rolling_total_average_rating_partition, country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating,
    FIRST_VALUE(total_device_installs) OVER (PARTITION BY total_device_installs_partition, country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    FIRST_VALUE(total_device_uninstalls) OVER (PARTITION BY total_device_uninstalls_partition, country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls,
    FIRST_VALUE(total_store_acquisitions) OVER (PARTITION BY total_store_acquisitions_partition, country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_acquisitions,
    FIRST_VALUE(total_store_visitors) OVER (PARTITION BY total_store_visitors_partition, country, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_store_visitors
  FROM create_partitions
), final AS (
  SELECT
    date_day,
    country AS country_short,
    COALESCE(country_codes.alternative_country_name, country_codes.country_name) AS country_long,
    country_codes.region,
    country_codes.sub_region,
    package_name,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    install_events,
    uninstall_events,
    update_events,
    store_listing_acquisitions,
    store_listing_visitors,
    store_listing_conversion_rate,
    active_devices_last_30_days,
    average_rating,
    rolling_total_average_rating, /* leave null if there are no ratings yet */
    COALESCE(total_device_installs, 0) AS total_device_installs, /* the first day will have NULL values, let's make it 0 */
    COALESCE(total_device_uninstalls, 0) AS total_device_uninstalls,
    COALESCE(total_store_acquisitions, 0) AS total_store_acquisitions,
    COALESCE(total_store_visitors, 0) AS total_store_visitors,
    ROUND(
      CAST(total_store_acquisitions AS DECIMAL(28, 6)) / NULLIF(total_store_visitors, 0),
      4
    ) AS rolling_store_conversion_rate, /* calculate percentage and difference rolling metrics */
    COALESCE(total_device_installs, 0) - COALESCE(total_device_uninstalls, 0) AS net_device_installs
  FROM fill_values
  LEFT JOIN country_codes
    ON country_codes.country_code_alpha_2 = fill_values.country
)
SELECT
  *
FROM final

WITH installs AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_installs_device
), ratings AS (
  SELECT
    *
  FROM TEST.PUBLIC_google_play_source.stg_google_play__stats_ratings_device
), install_metrics AS (
  SELECT
    *,
    SUM(device_installs) OVER (PARTITION BY device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    SUM(device_uninstalls) OVER (PARTITION BY device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls
  FROM installs
), device_join AS (
  SELECT
    COALESCE(install_metrics.date_day, ratings.date_day) AS date_day, /* these 3 columns are the grain of this model */
    COALESCE(install_metrics.device, ratings.device) AS device, /* device type */
    COALESCE(install_metrics.package_name, ratings.package_name) AS package_name,
    COALESCE(install_metrics.active_devices_last_30_days, 0) AS active_devices_last_30_days, /* metrics based on unique devices + users */
    COALESCE(install_metrics.device_installs, 0) AS device_installs,
    COALESCE(install_metrics.device_uninstalls, 0) AS device_uninstalls,
    COALESCE(install_metrics.device_upgrades, 0) AS device_upgrades,
    COALESCE(install_metrics.user_installs, 0) AS user_installs,
    COALESCE(install_metrics.user_uninstalls, 0) AS user_uninstalls,
    COALESCE(install_metrics.install_events, 0) AS install_events, /* metrics based on events. a user or device can have multiple installs in one day */
    COALESCE(install_metrics.uninstall_events, 0) AS uninstall_events,
    COALESCE(install_metrics.update_events, 0) AS update_events,
    install_metrics.total_device_installs,
    install_metrics.total_device_uninstalls,
    ratings.average_rating, /* this one actually isn't rolling but we won't coalesce days with no reviews to 0 rating */
    ratings.rolling_total_average_rating
  FROM install_metrics
  FULL OUTER JOIN ratings
    ON install_metrics.date_day = ratings.date_day
    AND install_metrics.package_name = ratings.package_name
    AND /* coalesce null device types otherwise they'll cause fanout with the full outer join */ COALESCE(install_metrics.device, 'null_device') = COALESCE(ratings.device, 'null_device') /* in the source package we aggregate all null device-type records together into one batch per day */
), create_partitions /* to backfill in days with NULL values for rolling metrics, we'll create partitions to batch them together with records that have non-null values */ /* we can't just use last_value(ignore nulls) because of postgres :/ */ AS (
  SELECT
    *,
    SUM(CASE WHEN rolling_total_average_rating IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating_partition,
    SUM(CASE WHEN total_device_installs IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs_partition,
    SUM(CASE WHEN total_device_uninstalls IS NULL THEN 0 ELSE 1 END) OVER (PARTITION BY device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls_partition
  FROM device_join
), fill_values /* now we'll take the non-null value for each partitioned batch and propagate it across the rows included in the batch */ AS (
  SELECT
    date_day,
    device,
    package_name,
    active_devices_last_30_days,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    install_events,
    uninstall_events,
    update_events,
    average_rating,
    FIRST_VALUE(rolling_total_average_rating) OVER (PARTITION BY rolling_total_average_rating_partition, device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS rolling_total_average_rating,
    FIRST_VALUE(total_device_installs) OVER (PARTITION BY total_device_installs_partition, device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_installs,
    FIRST_VALUE(total_device_uninstalls) OVER (PARTITION BY total_device_uninstalls_partition, device, package_name ORDER BY date_day ASC rows BETWEEN UNBOUNDED preceding AND CURRENT ROW) AS total_device_uninstalls
  FROM create_partitions
), final AS (
  SELECT
    date_day,
    device,
    package_name,
    device_installs,
    device_uninstalls,
    device_upgrades,
    user_installs,
    user_uninstalls,
    install_events,
    uninstall_events,
    update_events,
    active_devices_last_30_days,
    average_rating,
    rolling_total_average_rating, /* leave null if there are no ratings yet */
    COALESCE(total_device_installs, 0) AS total_device_installs, /* the first day will have NULL values, let's make it 0 */
    COALESCE(total_device_uninstalls, 0) AS total_device_uninstalls,
    COALESCE(total_device_installs, 0) /* calculate difference rolling metric */ - COALESCE(total_device_uninstalls, 0) AS net_device_installs
  FROM fill_values
)
SELECT
  *
FROM final