scpw.sql

CREATE TEMP FUNCTION
  get_priority(txn_type string)
  RETURNS int64 AS (
  IF
    (txn_type LIKE '%DR%',
      1,
      0));
CREATE TEMP FUNCTION
  capitalize(str STRING) AS ((
    SELECT
      STRING_AGG(CONCAT(UPPER(SUBSTR(word, 1, 1)), LOWER(SUBSTR(word, 2))), ' '
      ORDER BY
        pos)
    FROM
      UNNEST(SPLIT(REPLACE(REPLACE(str, '-', ' '), '_', ' '), ' ')) word
    WITH
    OFFSET
      pos ));
CREATE TEMP FUNCTION
  get_plan_wise_mrr_split(
    group_ids ARRAY<int64>,
    txn_ids ARRAY<int64>,
    txn_descriptions ARRAY<string>,
    products ARRAY<string>,
    mrr_changes ARRAY<float64>,
    channel_types ARRAY<string>,
    ids_list ARRAY<int64>,
    entry_plan_ids_list ARRAY<int64>,
    exit_plan_ids_list ARRAY<int64>,
    new_monthly_price_list ARRAY<float64>,
    txn_types ARRAY<string>)
  RETURNS ARRAY<STRUCT< transaction_id int64,
  id int64,
  entry_plan_id int64,
  exit_plan_id int64,
  new_monthly_price float64,
  mrr_change float64,
  transaction_type string,
  ss_mrr_change float64,
  st_mrr_change float64,
  ss_cumulative_mrr float64,
  st_cumulative_mrr float64,
  total_cumulative_mrr float64,
  total_mrr_change float64,
  channel_type string,
  product_ss_cumulative_mrr float64,
  product_st_cumulative_mrr float64 >>
  LANGUAGE js AS """
function add(a, b) {
  sum = a + b
  return parseFloat(sum.toFixed(5))  // retain only 5 decimal places
}
function initialize_product_mrr(previous_product_mrrs, product, txn_description) {
  previous_product_cumulative_mrr = 0
  if (!previous_product_mrrs.hasOwnProperty(product)) {
    if (!['new', 'repurchase', 'addition'].includes(txn_description)) {
      existing_products = Object.keys(previous_product_mrrs)
      for (let index = 0; index < existing_products.length; index++) {
        previous_product_cumulative_mrr = add(previous_product_cumulative_mrr, previous_product_mrrs[existing_products[index]])
      }
    }
    previous_product_mrrs[product] = previous_product_cumulative_mrr
  }
}
const SS_CHANNEL = 'SS'
const ST_CHANNEL = 'ST'
let output = []
let ss_cumulative_mrr = 0
let st_cumulative_mrr = 0
let product_ss_cumulative_mrr = {}
let product_st_cumulative_mrr = {}
let total_cumulative_mrr = 0
for (let index = 0; index < mrr_changes.length; index++) {
  let channel_type = channel_types[index]
  let previous_channel_type = index > 0 ? channel_types[index - 1] : null
  let mrr_change = parseFloat(mrr_changes[index].toFixed(5))
  let product = products[index]
  let transaction_id = txn_ids[index]
  let txn_description = txn_descriptions[index]
  let transaction_type = txn_types[index]
  initialize_product_mrr(product_ss_cumulative_mrr, product, txn_description)
  initialize_product_mrr(product_st_cumulative_mrr, product, txn_description)
  let ss_mrr_change = 0
  let st_mrr_change = 0
  let id = ids_list[index]
  let entry_plan_id = entry_plan_ids_list[index]
  let exit_plan_id = exit_plan_ids_list[index]
  let new_monthly_price = new_monthly_price_list[index]
  
  // MRR Change
  if (mrr_change >= 0) {
    if (channel_type == SS_CHANNEL) {
      ss_mrr_change = mrr_change
    } else if (channel_type == ST_CHANNEL) {
      st_mrr_change = mrr_change
    }
  } else if (mrr_change < 0) {
    if (channel_type == SS_CHANNEL) {
      let diff_mrr = add(product_ss_cumulative_mrr[product], mrr_change)
      if (diff_mrr <= 0) {
        ss_mrr_change = -1 * product_ss_cumulative_mrr[product]
        st_mrr_change = diff_mrr
      } else {
        ss_mrr_change = mrr_change
        st_mrr_change = 0
      }
    } else if (channel_type == ST_CHANNEL) {
      let diff_mrr = add(product_st_cumulative_mrr[product], mrr_change)
      if (diff_mrr <= 0) {
        st_mrr_change = -1 * product_st_cumulative_mrr[product]
        ss_mrr_change = diff_mrr
      } else {
        st_mrr_change = mrr_change
        ss_mrr_change = 0
      }
    }
  }
  
  ss_cumulative_mrr = add(ss_cumulative_mrr, ss_mrr_change)  
  st_cumulative_mrr = add(st_cumulative_mrr, st_mrr_change)
  product_ss_cumulative_mrr[product] = add(product_ss_cumulative_mrr[product], ss_mrr_change)
  product_st_cumulative_mrr[product] = add(product_st_cumulative_mrr[product], st_mrr_change)
  total_cumulative_mrr = add(total_cumulative_mrr, mrr_change)
  
  output.push({
    transaction_id,
    id,
    entry_plan_id,
    exit_plan_id,
    new_monthly_price,
    mrr_change,
    transaction_type,
    ss_mrr_change,
    st_mrr_change,
    ss_cumulative_mrr,
    st_cumulative_mrr,
    total_cumulative_mrr,
    total_mrr_change: mrr_change,
    channel_type,
    product_ss_cumulative_mrr: product_ss_cumulative_mrr[product],
    product_st_cumulative_mrr: product_st_cumulative_mrr[product]
  })
}
return output
""";
CREATE OR REPLACE TABLE
  `dwh-staging-electron.datastudio_reports.sales_channel_plan_wise_mrrs` as
WITH
  base_query AS (
  SELECT
    sales_channel.id,
    sales_channel.channel_type,
    sales_channel.transaction_product,
    lower(sales_channel.transaction_description) as transaction_description,
    plans_hist.all_old_plans,
    plans_hist.entry_plan_id,
    plans_hist.exit_plan_id,
    plans_hist.group_id,
    plans_hist.plan_id,
    plans_hist.is_annual,
    plans_hist.is_enterprise,
    plans_hist.new_monthly_price,
    SAFE_CAST(plans_hist.old_monthly_price as float64) as old_monthly_price,
    plans_hist.order_id,
    plans_hist.plan,
    plans_hist.transaction_type as transaction_type,
    plans_hist.txn_date as transaction_date,
    plans_hist.txn_id as transaction_id,
    plans_hist.txn_month as transaction_month,
    plans_hist.type,
    (CASE WHEN plans_hist.mrr_change IS NULL THEN 0 ELSE CAST(plans_hist.mrr_change AS float64) END) AS mrr_change,
    RANK() OVER(ORDER BY sales_channel.id, plans_hist.type, get_priority(plans_hist.transaction_type) DESC ) rank
  FROM
    datastudio_reports.all_plans_history AS plans_hist
  JOIN (
    SELECT
      transaction_id,
      ARRAY_AGG(group_id)[safe_OFFSET(0)] group_id,
      ARRAY_AGG(id)[safe_OFFSET(0)] id,
      ARRAY_AGG(channel_type)[safe_OFFSET(0)] channel_type,
      ARRAY_AGG(transaction_product)[safe_OFFSET(0)] transaction_product,
      ARRAY_AGG(transaction_description)[safe_OFFSET(0)] transaction_description
    FROM
      datastudio_reports.sales_channel_wise_mrrs
    GROUP BY
      transaction_id) AS sales_channel
  ON
    plans_hist.txn_id=sales_channel.transaction_id
    ORDER BY sales_channel.id, plans_hist.type, get_priority(plans_hist.transaction_type) DESC ),
    
    
    group_wise_plan_mrr_split AS (
  SELECT
  group_id,
    get_plan_wise_mrr_split( 
      ARRAY_AGG(group_id ORDER BY base.rank),
      ARRAY_AGG(transaction_id ORDER BY base.rank),
      ARRAY_AGG(transaction_description ORDER BY base.rank),
      ARRAY_AGG(transaction_product ORDER BY base.rank),
      ARRAY_AGG(mrr_change ORDER BY base.rank),
      ARRAY_AGG(channel_type ORDER BY base.rank),
      ARRAY_AGG(id ORDER BY base.rank),
      ARRAY_AGG(entry_plan_id ORDER BY base.rank),
      ARRAY_AGG(exit_plan_id ORDER BY base.rank),
      ARRAY_AGG(new_monthly_price ORDER BY base.rank),
      ARRAY_AGG(transaction_type ORDER BY base.rank)) plan_wise_mrr_split
  FROM
    base_query base
  WHERE
    channel_type IS NOT NULL
  GROUP BY group_id)

SELECT
  base.rank id,
  plan_id,
  base.group_id,
  order_id,
  plan_wise_mrr_split.transaction_id,
  transaction_type,
  Date(transaction_date) as transaction_date,
  CAST(transaction_month as int64) transaction_month,
  base.channel_type,
  plan,
  all_old_plans,
  entry_plan_id,
  exit_plan_id,
  CAST(is_annual as bool) as is_annual,
  CAST(is_enterprise as bool) as is_enterprise,
  new_monthly_price,
  CASE when old_monthly_price is null then 0.0 else old_monthly_price end as old_monthly_price,
  type,
  base.mrr_change,
  ss_mrr_change,
  st_mrr_change,
  ss_cumulative_mrr,
  st_cumulative_mrr,
  total_mrr_change as total_mrr,
  transaction_product,
  transaction_description,
  product_ss_cumulative_mrr,
  product_st_cumulative_mrr
FROM
  group_wise_plan_mrr_split, UNNEST(plan_wise_mrr_split) AS plan_wise_mrr_split
JOIN
  base_query AS base
USING
  (transaction_id, entry_plan_id , exit_plan_id, id, new_monthly_price, transaction_type)

PFA report: <%= @file_name %>

Note: Diffs in report means that there can be possible diff in channel type tagging or any of the mrr fields of sales_channel_plan_wise_mrrs. For eg., manual recon or through sales ops