RWM Console cluster: risingwave-adib.adib-rw.svc.cluster.local

CREATE MATERIALIZED VIEW insights.flow_values_mv
WITH (
  backfill_order=FIXED(olap.transaction_types_dm -> olap.transactions_dm, olap.assets_dm -> olap.transactions_dm, asset_service.foreign_exchange_rates_eod_ft -> olap.transactions_dm, olap.fee_transactions_dm -> olap.transactions_dm, olap.income_transactions_dm -> olap.transactions_dm)
) AS
WITH mapped_sources AS (
  SELECT
    transaction_id,
    source_account_id,
    source_asset_id
  FROM olap.fee_transactions_dm
  WHERE
    NOT source_account_id IS NULL AND NOT source_asset_id IS NULL
  UNION ALL
  SELECT
    transaction_id,
    source_account_id,
    source_asset_id
  FROM olap.income_transactions_dm
  WHERE
    NOT source_account_id IS NULL AND NOT source_asset_id IS NULL
), legs AS (
  SELECT
    t.transaction_id,
    t.account_id,
    t.asset_id,
    t.transaction_valuation_date,
    t.currency_code,
    t.net_value,
    tt.type AS transaction_type
  FROM olap.transactions_dm AS t
  JOIN olap.transaction_types_dm AS tt
    ON tt.transaction_type_id = t.transaction_type_id
  WHERE
    tt.type IN ('TRADE', 'TRANSFER')
  UNION ALL
  SELECT
    t.transaction_id,
    t.account_id,
    t.asset_id,
    t.transaction_valuation_date,
    t.currency_code,
    t.net_value,
    tt.type AS transaction_type
  FROM olap.transactions_dm AS t
  JOIN olap.transaction_types_dm AS tt
    ON tt.transaction_type_id = t.transaction_type_id AND tt.type IN ('INCOME', 'FEE')
  JOIN mapped_sources AS ms
    ON ms.transaction_id = t.transaction_id
  UNION ALL
  SELECT
    t.transaction_id,
    t.account_id AS account_id,
    ms.source_asset_id AS asset_id,
    t.transaction_valuation_date,
    t.currency_code,
    -t.net_value AS net_value,
    tt.type AS transaction_type
  FROM olap.transactions_dm AS t
  JOIN olap.transaction_types_dm AS tt
    ON tt.transaction_type_id = t.transaction_type_id AND tt.type IN ('INCOME', 'FEE')
  JOIN mapped_sources AS ms
    ON ms.transaction_id = t.transaction_id
)
SELECT
  l.transaction_id,
  l.account_id,
  l.asset_id,
  atag.account_group_id,
  l.transaction_valuation_date AS dim_transaction_date,
  l.transaction_type,
  l.currency_code AS transaction_currency,
  hv.type AS position_type,
  atag.base_currency AS group_currency,
  COALESCE(a.type, 'UNSPECIFIED') AS asset_type,
  a.issue_currency_code AS asset_currency,
  l.net_value,
  l.net_value * COALESCE(fx_group.rate, CASE WHEN l.currency_code = atag.base_currency THEN 1 ELSE NULL END) AS net_value_group,
  l.net_value * COALESCE(fx_system.rate, CASE WHEN l.currency_code = 'SAR' THEN 1 ELSE NULL END) AS net_value_system
FROM legs AS l
JOIN insights.account_to_account_groups_mv AS atag
  ON atag.account_id = l.account_id
  AND l.transaction_valuation_date >= atag.effective_start_date
  AND (
    atag.effective_end_date IS NULL
    OR l.transaction_valuation_date < atag.effective_end_date
  )
  AND (
    atag.opening_date IS NULL OR l.transaction_valuation_date >= atag.opening_date
  )
  AND atag.source_entity_type <> 'user'
LEFT JOIN olap.assets_dm AS a
  ON a.asset_id = l.asset_id
LEFT JOIN olap.holding_values_raw_ft AS hv
  ON hv.account_id = l.account_id
  AND hv.asset_id = l.asset_id
  AND hv.dim_value_date = l.transaction_valuation_date
LEFT JOIN asset_service.foreign_exchange_rates_eod_ft AS fx_group
  ON fx_group.source_currency_code = l.currency_code
  AND fx_group.target_currency_code = atag.base_currency
  AND fx_group.date = l.transaction_valuation_date
LEFT JOIN asset_service.foreign_exchange_rates_eod_ft AS fx_system
  ON fx_system.source_currency_code = l.currency_code
  AND fx_system.target_currency_code = 'SAR'
  AND fx_system.date = l.transaction_valuation_date
WHERE
  (
    NOT fx_group.rate IS NULL OR l.currency_code = atag.base_currency
  )
  AND (
    NOT fx_system.rate IS NULL OR l.currency_code = 'SAR'
  )