flows.sql

create table flows(date DateTime, src_id UInt64, dst_id UInt64, bytes UInt64)
Engine = MergeTree() order by date;

create table item_dict_t(id UInt64, attr String) Engine=MergeTree order by id;
insert into item_dict_t values(1, '1'),(2,'2');

CREATE DICTIONARY item_dict ( id UInt64, attr String ) 
PRIMARY KEY id SOURCE(CLICKHOUSE(HOST 'localhost' PORT 9000 
TABLE item_dict_t DB 'dw' USER 'default')) 
LIFETIME(MIN 0 MAX 0) LAYOUT(HASHED());

CREATE MATERIALIZED VIEW mv_stats_daily
(
    `day` Date CODEC(DoubleDelta, LZ4),
    `id` UInt32 CODEC(DoubleDelta, LZ4),
    `attr` String,
    `total_src_bytes` SimpleAggregateFunction(sum, Float64) CODEC(Gorilla, LZ4),
    `total_dst_bytes` SimpleAggregateFunction(sum, Float64) CODEC(Gorilla, LZ4)
)
ENGINE = AggregatingMergeTree()
PARTITION BY day
ORDER BY (id) AS
SELECT
    toDate(date) AS day,
    id,
    dictGet('dw.item_dict', 'attr', id) AS attr,
    sum(src_bytes) AS total_src_bytes,
    sum(dst_bytes) AS total_dst_bytes
FROM flows
ARRAY JOIN
    [src_id, dst_id] AS id,
    [bytes, 0] AS src_bytes,
    [0, bytes] AS dst_bytes
GROUP BY
    day,
    id,
    attr
    
insert into flows values(now(), 1,2, 100);
insert into flows values(now(), 1,1, 100);
insert into flows values(now(), 3,3, 100);

select * from mv_stats_daily final

┌────────day─┬─id─┬─attr─┬─total_src_bytes─┬─total_dst_bytes─┐
│ 2019-12-11 │  1 │ 1    │             200 │             100 │
│ 2019-12-11 │  2 │ 2    │               0 │             100 │
│ 2019-12-11 │  3 │      │             100 │             100 │
└────────────┴────┴──────┴─────────────────┴─────────────────┘