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

CREATE MATERIALIZED VIEW insights.abor_settlement_positions_mv
WITH (
  backfill_order=FIXED(asset_service.assets_dm -> insights.transactions_merged_mv)
) AS
WITH eligible AS (
  SELECT
    atag.account_group_id,
    atag.source_entity_type,
    t.transaction_id,
    t.asset_id,
    t.transaction_settlement_date AS dim_settlement_date,
    t.currency_code,
    CASE WHEN a.type = 'LIABILITY' THEN 'LIABILITY' ELSE 'ASSET' END AS position_type,
    CASE
      WHEN a.type = 'CASH'
      THEN t.net_value
      ELSE COALESCE(t.quantity, CAST(0 AS DECIMAL))
    END AS settlement_quantity_delta,
    t.net_value AS settlement_value_delta
  FROM insights.transactions_merged_mv AS t
  LEFT JOIN asset_service.assets_dm FOR SYSTEM_TIME AS OF PROCTIME() AS a
    ON a.id = t.asset_id
  JOIN insights.account_to_account_groups_settlement_binned_mv AS atag
    ON atag.account_id = t.account_id
    AND atag.dim_settlement_month = CAST(DATE_TRUNC('MONTH', t.transaction_settlement_date) AS DATE)
    AND t.transaction_settlement_date >= atag.effective_start_date
    AND (
      atag.effective_end_date IS NULL
      OR t.transaction_settlement_date < atag.effective_end_date
    )
    AND (
      atag.opening_date IS NULL OR t.transaction_settlement_date >= atag.opening_date
    )
    AND atag.source_entity_type = 'account'
  WHERE
    NOT t.transaction_settlement_date IS NULL
), daily AS (
  SELECT
    account_group_id,
    source_entity_type,
    dim_settlement_date,
    asset_id,
    CASE WHEN GROUPING(position_type) = 1 THEN 'POSITION' ELSE position_type END AS position_type,
    currency_code,
    SUM(settlement_quantity_delta) AS settlement_quantity_delta,
    SUM(settlement_value_delta) AS settlement_value_delta,
    COUNT(transaction_id) AS transaction_count
  FROM eligible
  GROUP BY
    GROUPING SETS (
      (
        account_group_id,
        source_entity_type,
        dim_settlement_date,
        asset_id,
        position_type,
        currency_code
      ),
      (
        account_group_id,
        source_entity_type,
        dim_settlement_date,
        asset_id,
        currency_code
      )
    )
)
SELECT
  account_group_id,
  source_entity_type,
  dim_settlement_date,
  asset_id,
  position_type,
  currency_code,
  settlement_quantity_delta,
  settlement_value_delta,
  SUM(settlement_quantity_delta) OVER w AS settled_quantity,
  SUM(settlement_value_delta) OVER w AS settled_value,
  transaction_count
FROM daily
WINDOW w AS (
  PARTITION BY account_group_id, asset_id, position_type, currency_code
  ORDER BY dim_settlement_date
)