异步物化视图可以更好地支持多表关联的场景，减少对于数据导入影响，可用于复杂查询加速、数据加工、查询匹配等场景。本文介绍了异步物化视图的语法，如何创建及配置异步物化视图，也提供了异步物化视图的最佳实践。

使用前请务必检查视图定义是否符合异步视图使用的最佳场景：

• 视图定义 SQL 中使用的基表推荐离线天级更新表，不推荐实时表，否则会导致频繁触发刷新任务，增加刷新代价，给系统负载造成负担。
• 视图定义 SQL 要保证分区对齐：视图目的表分区是由基表中一张表分区数据计算得来，否则如果基表有变化就会触发全量刷新，代价很高。
• 多表 join 最合适的场景：事实表天级离线更新，维表基本不更新，对于事实表不能对齐分区或者维表频繁变动的，会频繁全量刷新，加重系统负载。
• 视图刷新没有重试机制，失败只能到下个调度周期调用，建议设置分钟级调度间隔，增加重试机会。
• 调度间隔并不是验证按照设置的时间间隔，跟任务执行时长也有关系，任务执行时长超过间隔会推迟到下个调度周期执行，不能强依赖调度间隔，期望准时调度。
• ByteHouse 云数仓版异步视图的参数设置较为复杂，涉及到各种计算组（VW）设置，否则会走到默认计算组，造成 server 负载加重，请参考VW资源隔离章节谨慎配置。

异步视图功能 #

异步视图功能如下：

• 复杂查询加速： 异步物化视图存储了基于基表特定查询语句的预计算结果。对基表执行复杂查询时，可以直接复用预计算结果，避免重复计算，进而提高查询性能。查询的频率越高或查询语句越复杂，性能增益就会越很明显。
• 数据加工：通过异步物化视图可以对底层数据进行ELT做轻量级的数据加工，直接在bytehouse引擎内容实现数据的转化，减少在不同的数据处理系统切换，降低使用门槛和维护成本。
• 查询匹配：异步视图会通过基表数据刷新版本来判断读取的是明细表还是物化表，保证数据查询的一致性。

异步物化视图优势 #

• 性能与实时性的平衡：通过异步更新机制，在保证数据新鲜度的同时最小化对写入性能的影响。
• 复杂查询优化：预计算和存储复杂查询结果，显著提升分析性能。
• 资源利用效率：通过智能调度更新操作，优化系统资源使用。
• 数据一致性：在近实时更新和最终一致性之间找到平衡点，支持视图改写查询一致性
• 支持复杂物化查询：支持多表复杂的视图定义，支持嵌套视图，嵌套物化视图

分区增量刷新策略 #

单表场景较为简单，多表关联的场景下如何尽可能减少刷新的代价，方案是限定场景，寻找相对不变数据与经常变动的数据，找到增量数据刷新，减少刷新代价。
数据建模往往涉及根据业务语义关联数据，不同的数据关联方式会产生不同的建模方法，如星型模式和雪花模式，它们都使用事实表和维度表。一些业务场景需要多个大型事实表，而其他场景涉及复杂的维度表及其之间的复杂的关联关系。事实表变动比较频繁，维表基本不会发生变化，基于这个场景，可以使视图目标表对齐某一个频繁变动的基表，基于分区粒度的刷新策略。

ByteHouse 云数仓版 2.2 及以上版本支持此功能。

相关语法 #

--建异步视图
CREATE MATERIALIZED VIEW [IF NOT EXISTS] [database.]<mv_name> [TO [database.]<target_name>] [REFRESH [ASYNC [START (start_time)] EVERY (INTERVAL refresh_interval) | MANUAL]] AS <query_statement> [SETTINGS <mv_query_settings>] [PROPERTIES <mv_properties>];

--系统命令，启动，停止视图刷新，激活，失效视图
SYSTEM [START VIEW | STOP VIEW | ACTIVATE VIEW | DEACTIVATE VIEW] [database.]<mv_name>;

--手动刷新视图
REFRESH MATERIALIZED VIEW [database.]<mv_name> [ASYNC | SYNC];

--修改视图属性
ALTER TABLE [database.]<mv_name> MODIFY PROPERTY ...;

--修改视图SQL的settings
ALTER TABLE [database.]<mv_name> MODIFY VIEW QUERY SETTING ...;

--修改视图刷新模式
ALTER TABLE [database.]<mv_name> MODIFY REFRESH ...;

说明

• 刷新间隔支持整数单位为天/小时/分钟/秒，例如：INTERVAL 5 SECOND， INTERVAL 10 MINUTE
• 刷新开始时间可选配置，必须使用YYYY-mm-dd HH:MM:SS格式定义，例如 2023-12-05 10:00:00
• 刷新模式支持定时自动触发刷新(ASYNC)，手动触发刷新(MANUAL)
• 由于目的表由原表产生所以只需要限定原表的分区范围就可以自动生成目的表，所以目的表不允许设置TTL进行数据删除，否则会造成数据不一致，建表时也会校验目的表是否有TTL定义

支持场景 #

• 基表类型支持CnchMergeTree表(包括unqiue表），HIVE外表，嵌套物化视图，普通视图
• 单表聚合，支持各种聚合算子，过滤条件
• 单表无聚合，修改排序键，缩减读取字段，过滤条件
• 多表join聚合查询，支持inner join, left join, semi jon , anti join，非等值join，等值join类型
• 物化视图多层级联
• 支持union all，full join刷新，标量子查询，但是不支持视图改写，用于ETL场景

异步视图属性 #

异步视图特有相关属性配置说明，在建表语句PROPERTIES关键字中定义，也可以通过MODIFY PROPERTY命令进行修改。

建表实例 #

DROP DATABASE IF EXISTS async_mv_left_join_test;
CREATE DATABASE async_mv_left_join_test;

-- Create base tables
CREATE TABLE async_mv_left_join_test.orders
(
    order_id UInt32,
    user_id UInt32,
    product_id UInt32,
    order_date Date,
    order_amount Decimal(10,2)
) ENGINE = CnchMergeTree
PARTITION BY order_date
ORDER BY (order_id, user_id);

CREATE TABLE async_mv_left_join_test.users
(
    user_id UInt32,
    username String,
    registration_date Date
) ENGINE = CnchMergeTree
PARTITION BY toYYYYMM(registration_date)
ORDER BY user_id;

CREATE TABLE async_mv_left_join_test.products
(
    product_id UInt32,
    product_name String,
    category String,
    price Decimal(10,2),
    last_updated DateTime
) ENGINE = CnchMergeTree
PARTITION BY toYYYYMM(last_updated)
ORDER BY product_id
UNIQUE KEY product_id;

-- Create materialized view with WITH clause and implicit table
CREATE MATERIALIZED VIEW async_mv_left_join_test.order_summary_mv
ENGINE = CnchMergeTree
PARTITION BY order_date
ORDER BY (category, username, product_name)
REFRESH ASYNC START('2024-09-12 15:43:48') EVERY(INTERVAL 5 SECOND) AS
WITH 
    price_categories AS (
        SELECT
            product_id,
            multiIf(
                price <= 50, 'Low',
                price <= 100, 'Medium',
                'High'
            ) as price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1))as order_date,
    assumeNotNull(u.username) as username,
    p.product_name as product_name,
    assumeNotNull(p.category) as category,
    o.order_amount as order_amount,
    assumeNotNull(pc.price_category) as price_category,
    p.price as product_price
FROM async_mv_left_join_test.orders o
LEFT JOIN async_mv_left_join_test.users u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products p ON o.product_id = p.product_id
LEFT JOIN price_categories pc ON o.product_id = pc.product_id
WHERE o.order_amount > 0 AND p.price > 0 SETTINGS max_memory_usage = 68719476736,
  max_threads = 8,
  virtual_warehouse = 'hdfs-staging-default',
  max_memory_usage_for_user = 242292621312,
  exchange_queue_bytes = 1073741824,
  max_execution_time = 3600 PROPERTIES 
  async_mv_refresh_execution_time = 3600,
  async_mv_enable_refresh_execute_serial = 1,
  async_mv_max_refresh_task_num = 8,
  async_mv_enbale_query_rewrite = 1,
  async_mv_enable_refresh = 1,
  async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai',
  async_mv_refresh_task_bsp_mode = 1;

-- Insert initial test data
INSERT INTO async_mv_left_join_test.orders VALUES (1, 101, 1001, '2023-05-01', 55.00), (2, 102, 1002, '2023-05-02', 85.00), (3, 103, 1003, '2023-05-03', 110.00), (4, 104, 1004, '2023-06-01', 130.00), (5, 105, 1005, '2023-06-02', 95.00); 

INSERT INTO async_mv_left_join_test.users VALUES (101, 'user1', '2023-01-01'), (102, 'user2', '2023-01-02'), (103, 'user3', '2023-01-03'), (104, 'user4', '2023-02-01'), (105, 'user5', '2023-02-02'); 

INSERT INTO async_mv_left_join_test.products VALUES (1001, 'Product A', 'Category1', 50.00, '2023-05-01 10:00:00'), (1002, 'Product B', 'Category2', 80.00, '2023-05-01 11:00:00'), (1003, 'Product C', 'Category1', 120.00, '2023-05-01 12:00:00'), (1004, 'Product D', 'Category2', 150.00, '2023-06-01 10:00:00'), (1005, 'Product E', 'Category1', 95.00, '2023-06-01 11:00:00');

-- enable_materialized_view_rewrite = 1 test whether match views
WITH 
    price_categories AS (
        SELECT
            product_id,
            multiIf(
                price <= 50, 'Low',
                price <= 100, 'Medium',
                'High'
            ) as price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1))as order_date,
    assumeNotNull(u.username) as username,
    p.product_name as product_name,
    assumeNotNull(p.category) as category,
    o.order_amount as order_amount,
    assumeNotNull(pc.price_category) as price_category,
    p.price as product_price
FROM async_mv_left_join_test.orders o
LEFT JOIN async_mv_left_join_test.users u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products p ON o.product_id = p.product_id
LEFT JOIN price_categories pc ON o.product_id = pc.product_id
WHERE o.order_amount > 10 AND p.price > 30 order by order_date, username  settings enable_optimizer = 1, enable_materialized_view_rewrite = 1;
   

-- enable_materialized_view_rewrite = 0 check not match view result
WITH 
    price_categories AS (
        SELECT
            product_id,
            multiIf(
                price <= 50, 'Low',
                price <= 100, 'Medium',
                'High'
            ) as price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1))as order_date,
    assumeNotNull(u.username) as username,
    p.product_name as product_name,
    assumeNotNull(p.category) as category,
    o.order_amount as order_amount,
    assumeNotNull(pc.price_category) as price_category,
    p.price as product_price
FROM async_mv_left_join_test.orders o
LEFT JOIN async_mv_left_join_test.users u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products p ON o.product_id = p.product_id
LEFT JOIN price_categories pc ON o.product_id = pc.product_id
WHERE o.order_amount > 10 AND p.price > 30  order by order_date,username settings enable_optimizer = 1, enable_materialized_view_rewrite = 0;

-- modify property
ALTER TBALE async_mv_left_join_test.order_summary_mv MODIFY PROPERTY async_mv_max_refresh_task_num = 10;

-- modify view query setting
ALTER TBALE async_mv_left_join_test.order_summary_mv MODIFY VIEW QUERY SETTING max_execution_time = 3000;

-- modify refresh 
ALTER TBALE async_mv_left_join_test.order_summary_mv MODIFY REFRESH ASYNC START('2024-09-12 15:43:48') EVERY(INTERVAL 10 SECOND);

-- check table definition
SHOW CREATE TABLE async_mv_left_join_test.order_summary_mv;

DROP DATABASE IF EXISTS async_mv_left_join_test;

异步视图执行流程如下：

创建异步视图 #

用户通过 SQL 语句创建异步物化视图。系统在元数据中记录视图定义、依赖关系等信息。

后台管理进程 #

DaemonManager 作为 bytehouse 的后台进程管理器，负责监控和管理各种系统任务。当 DaemonManager 探测到存在异步视图时，它会在 server 端初始化并启动 RefreshMaterializedViewThread专门用于刷新异步物化视图的后台线程。线程启动后，会根据配置的刷新策略定期执行刷新任务。

刷新任务执行 #

• 获取基表分区变化：检查基表自上次刷新以来的分区变化，识别新增、修改或删除的分区。
• 计算影响范围：根据基表分区变化，分析并确定受影响的目标表分区，考虑视图定义中的过滤条件、聚合操作等，精确定位需要更新的数据范围。
• 构造刷新任务：将大型刷新任务拆分为多个小任务，以便并行处理和资源管理。每个小任务负责刷新特定的分区。
• 任务提交和调度：将刷新任务提交到 bytehouse 的任务队列，系统调度器根据当前负载和优先级安排任务执行。
• 任务执行：执行刷新操作，可能包括数据重计算、聚合、更新等步骤，对于增量更新，只处理变化的数据部分，提高效率。
• 元信息更新：更新物化视图的元信息，包括最新的分区信息、时间戳等。
• 资源管理和优化：实现动态调整刷新频率的机制，根据系统负载和数据变化率自适应。引入优先级队列，确保关键视图能够及时刷新。
• 监控和报告：提供详细的刷新日志和统计信息，方便管理员监控和优化。

校验的目的是需要对视图的SQL进行语法分析，确定目的表的分区是否由其中一个基表计算得到，并且能通过分区值计算得到，才属于可以做分区级异步刷新的视图。

对齐分区校验 #

• 条件1：目的表的分区计算字段需要来源于同一个基表
• 条件2: 目的表分区的计算表达式可以由基表的分区字段计算得到

实例：
目的表分区：

(date, toUInt64(encoded_shop_id + 10))

视图SQL映射：

date_trunc('month', toDate(order.order_date)) as date

multiIf(order.shop_id >= 2702, toUInt64(30000), shop_id <= 2701, toUInt64(40000), toUInt64(50000)) as encoded_shop_id

基表分区：

(toDate(order_date), shop_id)

校验条件1： date 由order_date转化得到，encoded_shop_id 由shop_id转化得到， order_date，shop_id都来自同一个基表order_list
校验条件2： 目的表分区需要 (toDate(order_date), shop_id)两个字段，都是order_list表分区字段

(date_trunc('month', `toDate(order_date)`), toUInt64(multiIf(shop_id >= 2702, CAST(30000, 'UInt64'), shop_id <= 2701, CAST(40000, 'UInt64'), CAST(50000, 'UInt64')) + 10)

分区转换表达式 #

输入：源表的partition 数据 eg. '2023-11-10' or ('2023-11-10', 1)
转化expression ：构造一个partition expression action ，提取有关分区的过滤条件
输出：目的表的partition 数据 eg. '2023-11-10'
实例：
输入 block header： (toDate(order_date), shop_id)
输入值： ('2023-11-10', 1)
转化expression list with filter

(date_trunc('month', `toDate(order_date)`), toUInt64(multiIf(shop_id >= 2702, CAST(30000, 'UInt64'), shop_id <= 2701, CAST(40000, 'UInt64'), CAST(50000, 'UInt64')) + 10), (date_trunc('month', `toDate(order_date)`) > '2023-04-01') AND (date_trunc('month', `toDate(order_date)`) <= '2023-12-01'))

输出block header： (date, toUInt64(encoded_shop_id + 10))
输出值：('2023-11-01', 3000)

分区映射计算 #

原表与目的表分区映射关系除了 1:1 的映射关系，实际上其中还有 1:n、n:1 以及 n:n 的映射关系，需要从基表变化的分区推导出目的表的被覆写的分区列表。

分区映射表达式，是对order_list的date分区，rollup到month的分区，相当于N:1的映射关系，在构造刷新任务的insert overwrite语句是需要确定从基表哪些partition做查询更新target表的分区，下面举例说明：

(date_trunc('month', `toDate(order_date)`), toUInt64(multiIf(shop_id >= 2702, CAST(30000, 'UInt64'), shop_id <= 2701, CAST(40000, 'UInt64'), CAST(50000, 'UInt64')) + 10)

• 检测到对齐基表business.order_list的分区（'2023-12-06', 2703) 版本发生了变化
• 第一次分区映射计算：得到business.order_statistics的分区 ('2023-12-01', 30010) 需要做insert overwrite

由于分区是N:1的关系，business.order_statistics分区是由business.order_list所有2023-12月的分区一起计算得
到，所以需要确定business.order_list中还有哪些其他分区需要作为数据源一起参与计算

• 第二次分区计算: 获取当前business.order_list所有分区列表，通过映射计算得到目的分区列表，与 ('2023-12-01', 30010) 做一次对比，如果值一样则确定为参与计算基表分区
• 最终的映射关系：target <('2023-12-01', 30010)> ----> source <(('2023-12-06', 2703), ('2023-12-17', 2703))>

NOTE：
根据分区直接做分区映射计算会对视图的SQL有一定的限制，缺乏一定的灵活性，这一点在对工具设计上做提示引导用户去修改SQL，同时分区的映射计算可能会造成有些分区会多计算，毕竟没有考虑所有的过滤条件，但是能保证变化的分区能被覆写。

详细执行逻辑 #

• 获取所有基表当前分区版本信息列表
• 获取上次刷新保存基表的分区版本信息列表
• 计算同一张基表分区版本信息的差别集合，包含新增，删除的分区
• 如果不是对齐的基表的分区发生更新，需要刷新所有的物化视图
• 如果是对齐的基表分区发生更新，做分区映射，计算目的分区到基表分区的映射关系
• 根据映射关系构造刷新查询，每个更新目的分区构造一个刷新任务，添加到执行队列等待执行
• 视图刷新调度线程根据设定的时间定期会启动执行刷新任务
• 刷新任务执行完成，提交事务和数据，更新基表更新分区元信息

视图状态表(system.cnch_view_tables) #

为了更好的诊断视图的状态， 把视图相关的信息，partition当前的状态记录在表system.cnch_view_tables

定义 #

CREATE TABLE system.cnch_view_tables
(
    `database` String,
    `table` String,
    `uuid` UUID,
    `vw_name` String,
    `definition` String,
    `base_tables` Array(String),
    `dependent_tables` Array(String),
    `non_dependent_tables` Array(String),
    `target_table` String,
    `partition_diffs` Array(String),
    `latest_visible_partitions` Array(String),
    `previous_partitions` Array(String),
    `refresh_type` String,
    `refresh_start_time` String,
    `refresh_interval` String,
    `active` UInt8,
    `enable_refresh` UInt8,
    `enable_query_rewrite` UInt8,
    `refresh_status` String
)
ENGINE = SystemCnchViewTables 

字段 #

示例 #

查询

select * from system.cnch_view_tables where database = 'async_mv_left_join_test' and table = 'order_summary_mv';

结果

database:                  async_mv_left_join_test
table:                     order_summary_mv
uuid:                      e13047fd-01b2-4a1d-86b4-dda47324b828
vw_name:                   
definition:                CREATE MATERIALIZED VIEW async_mv_left_join_test.order_summary_mv UUID 'e13047fd-01b2-4a1d-86b4-dda47324b828' TO INNER UUID '123eac1f-f19b-4363-8262-9e3063a7992c'
(
    `order_date` UInt32,
    `username` String,
    `product_name` String,
    `category` String,
    `order_amount` Decimal(10, 2),
    `price_category` String,
    `product_price` Decimal(10, 2)
)
ENGINE = CnchMergeTree
PARTITION BY order_date
ORDER BY (category, username, product_name)
REFRESH ASYNC START('2024-09-12 15:43:48') EVERY(INTERVAL 5 SECOND) AS
WITH price_categories AS
    (
        SELECT
            product_id,
            multiIf(price <= 50, 'Low', price <= 100, 'Medium', 'High') AS price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1)) AS order_date,
    assumeNotNull(u.username) AS username,
    p.product_name AS product_name,
    assumeNotNull(p.category) AS category,
    o.order_amount AS order_amount,
    assumeNotNull(pc.price_category) AS price_category,
    p.price AS product_price
FROM async_mv_left_join_test.orders AS o
LEFT JOIN async_mv_left_join_test.users AS u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products AS p ON o.product_id = p.product_id
LEFT JOIN
price_categories AS pc ON o.product_id = pc.product_id
WHERE (o.order_amount > 0) AND (p.price > 0)
SETTINGS max_memory_usage = 68719476736, max_threads = 8, max_memory_usage_for_user = 242292621312, exchange_queue_bytes = 1073741824, max_execution_time = 3600 PROPERTIES active = 1, base_tables = 'async_mv_left_join_test.orders,async_mv_left_join_test.products,async_mv_left_join_test.users', async_mv_refresh_execution_time = 3600, async_mv_enable_refresh_execute_serial = 1, async_mv_max_refresh_task_num = 8, async_mv_enbale_query_rewrite = 1, async_mv_enable_refresh = 1, async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai', async_mv_refresh_task_bsp_mode = 1

base_tables:               ['async_mv_left_join_test.users','async_mv_left_join_test.products','async_mv_left_join_test.orders']
dependent_tables:          ['async_mv_left_join_test.orders']
non_dependent_tables:      ['async_mv_left_join_test.products','async_mv_left_join_test.users']
target_table:              async_mv_left_join_test..inner_id.e13047fd-01b2-4a1d-86b4-dda47324b828
partition_diffs:           ['{"add_partitions": {},"drop_partitions": {}}']
latest_visible_partitions: ['{"version_partitions": {"async_mv_left_join_test.orders": ["partition-\'2023-05-01\', versoin-453920447620186137", "partition-\'2023-05-02\', versoin-453920447620186137", "partition-\'2023-05-03\', versoin-453920447620186137", "partition-\'2023-06-01\', versoin-453920447620186137", "partition-\'2023-06-02\', versoin-453920447620186137"],"async_mv_left_join_test.products": ["partition-202305, versoin-453920448119046158", "partition-202306, versoin-453920448119046158"],"async_mv_left_join_test.users": ["partition-202301, versoin-453920447830163476", "partition-202302, versoin-453920447830163476"]}}']
previous_partitions:       ['{"version_partitions": {"async_mv_left_join_test.users": ["partition-202301, versoin-453920447830163476", "partition-202302, versoin-453920447830163476"],"async_mv_left_join_test.products": ["partition-202305, versoin-453920448119046158", "partition-202306, versoin-453920448119046158"],"async_mv_left_join_test.orders": ["partition-\'2023-05-01\', versoin-453920447620186137", "partition-\'2023-05-02\', versoin-453920447620186137", "partition-\'2023-05-03\', versoin-453920447620186137", "partition-\'2023-06-01\', versoin-453920447620186137", "partition-\'2023-06-02\', versoin-453920447620186137"]}}']
refresh_type:              ASYNC
refresh_start_time:        2024-09-12 15:43:48
refresh_interval:          5 Second
active:                    1
enable_refresh:            1
enable_query_rewrite:      1
refresh_status:            Running

系统状态查看命令 #

改写性能查询 #

查询

SELECT
    materialized_views,
    query_duration_ms,
    (ProfileEvents['MvRewriteElapsedMicroseconds']) / 1000 AS rewrite_duration_ms,
    concat(formatReadableQuantity(round((((ProfileEvents['MvRewriteElapsedMicroseconds']) / 1000) / query_duration_ms) * 100, 2)), '%') AS rewrite_ratio,
    concat(formatReadableQuantity(round(((ProfileEvents['MvPartitionConsistencyCheckElapsedMicroseconds']) / (ProfileEvents['MvRewriteElapsedMicroseconds'])) * 100, 2)), '%') AS part_check_ratio,
    concat(formatReadableQuantity(round(((ProfileEvents['GetPreviousPartMetaElapsedMicroseconds']) / (ProfileEvents['MvRewriteElapsedMicroseconds'])) * 100, 2)), '%') AS get_previous_version_ratio,
    concat(formatReadableQuantity(round(((ProfileEvents['GetCurrentPartMetaElapsedMicroseconds']) / (ProfileEvents['MvRewriteElapsedMicroseconds'])) * 100, 2)), '%') AS get_current_version_ratio,
    concat(formatReadableQuantity(round(((ProfileEvents['SyncBaseTablePartitionsElapsedMicroseconds']) / (ProfileEvents['MvRewriteElapsedMicroseconds'])) * 100, 2)), '%') AS sync_partition_diff_ratio,
    concat(formatReadableQuantity(round(((ProfileEvents['TransformForAsyncElapsedMicroseconds']) / (ProfileEvents['MvRewriteElapsedMicroseconds'])) * 100, 2)), '%') AS transform_part_ratio
FROM system.query_log
WHERE (query_id = 'e1caee81-b346-4d70-84ba-057622a62dff') AND (type = 2)

结果

materialized_views:         ['async_mv_left_join_test.order_summary_mv'] --命中视图
query_duration_ms:          1155  -- 查询耗时
rewrite_duration_ms:        1124.7 -- 视图改写时间
rewrite_ratio:              97.38% -- 视图改写时间占总耗时时间比例
part_check_ratio:           0.92% -- 分区检查占改写时间比例
get_previous_version_ratio: 0.58% -- 获取已经物化分区原信息时间比例
get_current_version_ratio:  0.33% -- 获取当前分区元信息时间比例 
sync_partition_diff_ratio:  0.92% -- 同步分区计算时间比例
transform_part_ratio:       0.00% -- 分区转化时间比例

为了避免视图刷新占用资源过大，对集群整体负载造成影响，需要一些措施来控制刷新负载，通过一下几种方式可以有效平衡资源的使用。

计算组（VW）资源隔离 #

VW资源隔离：单独配置VW进行刷新运算，在建表时可以指定virtual_warehouse，刷新的负载会在指定的vw进行运算

CREATE MATERIALIZED VIEW async_mv_left_join_test.order_summary_mv
ENGINE = CnchMergeTree
PARTITION BY order_date
ORDER BY (category, username, product_name)
REFRESH ASYNC START('2024-09-12 15:43:48') EVERY(INTERVAL 5 SECOND) AS
WITH 
    price_categories AS (
        SELECT
            product_id,
            multiIf(
                price <= 50, 'Low',
                price <= 100, 'Medium',
                'High'
            ) as price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1))as order_date,
    assumeNotNull(u.username) as username,
    p.product_name as product_name,
    assumeNotNull(p.category) as category,
    o.order_amount as order_amount,
    assumeNotNull(pc.price_category) as price_category,
    p.price as product_price
FROM async_mv_left_join_test.orders o
LEFT JOIN async_mv_left_join_test.users u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products p ON o.product_id = p.product_id
LEFT JOIN price_categories pc ON o.product_id = pc.product_id
WHERE o.order_amount > 0 AND p.price > 0 
SETTINGS 
  max_memory_usage = 68719476736,
  max_threads = 8,
  virtual_warehouse = 'hdfs-staging-default', --在视图查询Settings中配置刷新vw
  max_memory_usage_for_user = 242292621312,
  exchange_queue_bytes = 1073741824,
  max_execution_time = 3600 
PROPERTIES 
  async_mv_refresh_execution_time = 3600,
  async_mv_enable_refresh_execute_serial = 1,
  async_mv_max_refresh_task_num = 8,
  async_mv_enbale_query_rewrite = 1,
  async_mv_enable_refresh = 1,
  async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai',
  async_mv_refresh_task_bsp_mode = 1;

BSP(Bulk Synchronous Parallel)执行模式 #

ByteHouse在高性能计算框架的基础上，增加了对bsp模式的支持：可以进行task级别的容错；更细粒度的调度；支持资源感知的调度（BORN：ByteHouse On-premises Resource Negotiator）。​带来的收益有：

• 当query运行中遇到错误时，可以自动重试当前的task，而不是从头进行重试。大大减少重试成本。
• 当query需要的内存巨大，甚至大于单机的内存时，可以通过增加并行度来减少单位时间内内存的占用。只需要调大并行度参数即可，理论上是可以无限扩展的。

在视图的属性上可以配置bsp运行方式async_mv_refresh_task_bsp_mode=1，同时也是默认的执行方式。

CREATE MATERIALIZED VIEW async_mv_left_join_test.order_summary_mv
ENGINE = CnchMergeTree
PARTITION BY order_date
ORDER BY (category, username, product_name)
REFRESH ASYNC START('2024-09-12 15:43:48') EVERY(INTERVAL 5 SECOND) AS
WITH 
    price_categories AS (
        SELECT
            product_id,
            multiIf(
                price <= 50, 'Low',
                price <= 100, 'Medium',
                'High'
            ) as price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1))as order_date,
    assumeNotNull(u.username) as username,
    p.product_name as product_name,
    assumeNotNull(p.category) as category,
    o.order_amount as order_amount,
    assumeNotNull(pc.price_category) as price_category,
    p.price as product_price
FROM async_mv_left_join_test.orders o
LEFT JOIN async_mv_left_join_test.users u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products p ON o.product_id = p.product_id
LEFT JOIN price_categories pc ON o.product_id = pc.product_id
WHERE o.order_amount > 0 AND p.price > 0 
SETTINGS 
  max_memory_usage = 68719476736,
  max_threads = 8,
  virtual_warehouse = 'hdfs-staging',
  max_memory_usage_for_user = 242292621312,
  exchange_queue_bytes = 1073741824,
  max_execution_time = 3600 
PROPERTIES 
  async_mv_refresh_execution_time = 3600,
  async_mv_enable_refresh_execute_serial = 1,
  async_mv_max_refresh_task_num = 8,
  async_mv_enbale_query_rewrite = 1,
  async_mv_enable_refresh = 1,
  async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai',
  async_mv_refresh_task_bsp_mode = 1;

视图匹配改写 #

优化器通过将查询表示为 SPJG 标准形式 (Select-Project-Join-GroupBy)，提取各种表达式信息，通过一系列规则进行改写，与物化视图进行匹配和改写。

改写能力 #

改写限制 #

1. 视图中只能包含 OrderBy / Aggregate / Join / Projection / Filter / TableScan，不能有 Limit / Window 等等。
2. 最多有一个 Aggregate，Aggregate 必须在 Join 之后，即 Subquery 中不能包含 Aggregate，注意查询中 Select Distinct 会转换为 Aggregate
3. 如果有非 inner join (left join / semi join), 必须是 Star-Join。不支持 Cross Join 和 Full Outer Join，例如支持 T1 left join T2 left join T3, 不支持 T1 left join (T2 left join T3)

一致性检查及分区级别改写 #

多表视图每次刷新后会把基表的partition version信息存放到catalog中，用于计算下次刷新的partition diff，确定哪些分区发生了更新，从而做分区级别的刷新。
优化器视图改写需要准确获取哪些partition 原表分区与物化表分区是一致的，哪些是不一致的，在进行视图改写。

• 获取查询SQL使用到基表partition version列表
• 获取基表上次刷新对应partitition version列表

两个versioned partition list进行比较：根据是否是刷新依赖的基表和分区对比结果进行不同的改写逻辑

为了更好的感知查询是否命中视图，通过不同的方式可以看到视图命中的状态，下面分别介绍：

TCP #

直接通过cli，tcp的方式执行查询。

WITH price_categories AS
    (
        SELECT
            product_id,
            multiIf(price <= 50, 'Low', price <= 100, 'Medium', 'High') AS price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1)) AS order_date,
    assumeNotNull(u.username) AS username,
    p.product_name AS product_name,
    assumeNotNull(p.category) AS category,
    o.order_amount AS order_amount,
    assumeNotNull(pc.price_category) AS price_category,
    p.price AS product_price
FROM async_mv_left_join_test.orders AS o
LEFT JOIN async_mv_left_join_test.users AS u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products AS p ON o.product_id = p.product_id
LEFT JOIN price_categories AS pc ON o.product_id = pc.product_id
WHERE (o.order_amount > 0) AND (p.price > 0) settings enable_optimizer = 1, enable_materialized_view_rewrite =1;

JSON FORMAT #

直接通过cli，tcp的方式执行查询，带上format JSON，在返回结果数据中能拿到是否命中视图。

WITH price_categories AS
    (
        SELECT
            product_id,
            multiIf(price <= 50, 'Low', price <= 100, 'Medium', 'High') AS price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1)) AS order_date,
    assumeNotNull(u.username) AS username,
    p.product_name AS product_name,
    assumeNotNull(p.category) AS category,
    o.order_amount AS order_amount,
    assumeNotNull(pc.price_category) AS price_category,
    p.price AS product_price
FROM async_mv_left_join_test.orders AS o
LEFT JOIN async_mv_left_join_test.users AS u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products AS p ON o.product_id = p.product_id
LEFT JOIN price_categories AS pc ON o.product_id = pc.product_id
WHERE (o.order_amount > 0) AND (p.price > 0) format JSON settings enable_optimizer = 1, enable_materialized_view_rewrite =1;

HTTP #

通过http方式，在返回的结果header中，可以看到是否命中视图。

echo "WITH price_categories AS (SELECT product_id, multiIf(price <= 50, 'Low', price <= 100, 'Medium', 'High') AS price_category FROM async_mv_left_join_test.products) SELECT toYYYYMM(addYears(addDays(o.order_date, 2), 1)) AS order_date, assumeNotNull(u.username) AS username, p.product_name AS product_name, assumeNotNull(p.category) AS category, o.order_amount AS order_amount, assumeNotNull(pc.price_category) AS price_category, p.price AS product_price FROM async_mv_left_join_test.orders AS o LEFT JOIN async_mv_left_join_test.users AS u ON o.user_id = u.user_id LEFT JOIN async_mv_left_join_test.products AS p ON o.product_id = p.product_id LEFT JOIN price_categories AS pc ON o.product_id = pc.product_id WHERE (o.order_amount > 0) AND (p.price > 0) format JSON settings enable_optimizer = 1, enable_materialized_view_rewrite =1" | sed 's/[[:space:]]+/ /g' | curl -is -H "X-ClickHouse-User: default" -H "X-ClickHouse-Key: P8rZPviauQIIIQIL" 'http://127.0.0.1:9477' --data-binary @-

QueryLog #

查询结束后，可以在query log中查看materialized_views字段查询是否命中视图。

SELECT *
FROM system.query_log
WHERE (query_id = 'd2c44842-3105-4f58-a43e-6547fd36a1df') AND (type = 2)

QueryPlan #

可以在执行前，通过explain命令获取执行逻辑计划，查看是否命中视图。

EXPLAIN
WITH price_categories AS
    (
        SELECT
            product_id,
            multiIf(price <= 50, 'Low', price <= 100, 'Medium', 'High') AS price_category
        FROM async_mv_left_join_test.products
    )
SELECT
    toYYYYMM(addYears(addDays(o.order_date, 2), 1)) AS order_date,
    assumeNotNull(u.username) AS username,
    p.product_name AS product_name,
    assumeNotNull(p.category) AS category,
    o.order_amount AS order_amount,
    assumeNotNull(pc.price_category) AS price_category,
    p.price AS product_price
FROM async_mv_left_join_test.orders AS o
LEFT JOIN async_mv_left_join_test.users AS u ON o.user_id = u.user_id
LEFT JOIN async_mv_left_join_test.products AS p ON o.product_id = p.product_id
LEFT JOIN price_categories AS pc ON o.product_id = pc.product_id
WHERE (o.order_amount > 0) AND (p.price > 0)
FORMAT JSON
SETTINGS enable_optimizer = 1, enable_materialized_view_rewrite = 1

通过一个创建异步物化视图的实例来说明实际使用过程中最佳实践方法。

业务需求背景 #

大型在线购物平台，提供各种类别的商品，如电子产品、服装、家居用品、图书和运动器材等。
数据分析的业务目标：

• 用户行为跟踪：平台记录用户的各种行为，包括浏览商品（view）、点击商品详情（click）和购买商品。
• 促销活动：平台经常进行各种促销活动，需要特别关注促销商品的表现。
• 多维度分析：管理层和营销团队需要从多个维度（如日期、事件类型、商品类别）来分析平台性能和用户行为。

由此使用异步物化视图来进行数据查询的加速，旨在为电子商务平台提供一个高效、多维度的数据分析工具，支持从战略决策到日常运营的各个方面。它通过预先计算和存储关键指标，大大提高了数据分析的效率和灵活性，使得业务团队能够快速获取洞察，做出数据驱动的决策。
数据基表定义：

-- 数据库创建
DROP DATABASE IF EXISTS async_mv_join_agg_test;
CREATE DATABASE async_mv_join_agg_test;

-- 购买行为事件表
CREATE TABLE async_mv_join_agg_test.events
(
    `event_date` Date,
    `event_type` String,
    `user_id` UInt32,
    `item_id` UInt32,
    `price` Decimal(10, 2),
    `quantity` UInt16,
    `tags` Array(String),
    `session_duration` UInt32
)
ENGINE = CnchMergeTree
PARTITION BY toStartOfMonth(event_date)
ORDER BY (event_type, user_id, item_id);

-- 购买物品维度表
CREATE TABLE async_mv_join_agg_test.items
(
    `item_id` UInt32,
    `category` String,
    `brand` String,
    `is_promo` UInt8
)
ENGINE = CnchMergeTree
PARTITION BY item_id % 10
ORDER BY item_id;

-- 事件行为数据导入
INSERT INTO async_mv_join_agg_test.events 
SELECT
    toDate(now()) - toIntervalDay(364 - (number % 365)) AS event_date,['view', 'click', 'purchase'][1 + (number % 3)] AS event_type,1 + (number % 1000) AS user_id,1 + (number % 10000) AS item_id,round(10 + ((number * 17) % 990), 2) AS price,1 + ((number * 31) % 5) AS quantity,
    arrayMap(x -> ['promo', 'sale', 'new', 'seasonal'][1 + ((number + x) % 4)], range(1 + ((number * 13) % 3))) AS tags,10 + ((number * 23) % 3590) AS session_duration
FROM numbers(1, 10000);

-- 商品数据导入
INSERT INTO async_mv_join_agg_test.items
SELECT
    number AS item_id,
    ['Electronics', 'Clothing', 'Home', 'Books', 'Sports'][1 + (number % 5)] AS category,
    ['BrandA', 'BrandB', 'BrandC', 'BrandD', 'BrandE'][1 + ((number * 7) % 5)] AS brand,
    number % 2 AS is_promo
FROM numbers(1, 10000);

构造异步物化视图 #

第1步: 确定物化查询 #

为电商平台提供了一个全面的视图，涵盖了销售、用户行为、产品性能和营销效果等多个方面，支持数据驱动的决策制定和业务优化，需要构建视图查询以加速分析过程，下面是确定了物化查询。

SELECT
    toStartOfMonth(e.event_date) AS dim_event_month,
    e.event_type,
    i.category,
    count() AS event_count,
    uniqExact(e.user_id) AS unique_users,
    sum(e.price * e.quantity) AS total_revenue,
    min(e.price) AS min_price,
    max(e.price) AS max_price,
    avg(e.price) AS avg_price,
    median(e.price) AS median_price,
    quantile(0.95)(e.price) AS price_95th_percentile,
    sumMap(e.tags, arrayMap(x -> 1, e.tags)) AS tag_counts,
    groupArray(e.item_id) AS sample_items,
    countIf(e.quantity > 1) AS multi_quantity_purchases
FROM async_mv_join_agg_test.events AS e
INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE e.event_type IN ('view', 'click', 'purchase')
  AND e.price > 0
  AND e.price < 10000
  AND i.is_promo = 1
  AND e.session_duration < 3600 
GROUP BY toStartOfMonth(e.event_date), e.event_type, i.category;

第2步: 评估物化查询 #

EXPLAIN VALIDATE_MATERIALIZED_VIEW SQL 可以对建表查询进行评估，explain结果会显示是否能SQL能建视图，不能建视图的原因，建视图的具体语句，并能自动改写相关的聚合算子，返回视图所需要的列类型定义，维度列，聚合列，推荐的分区键，对应的字段名( valid_mv, reason, mv_sql, table_columns, aggregate_function_columns, dimension_columns, partition_columns)。

EXPLAIN VALIDATE_MATERIALIZED_VIEW
SELECT
  toStartOfMonth(e.event_date) AS dim_event_month,
  e.event_type AS event_type,
  i.category AS category,
  countState() AS event_count,
  uniqExactState(e.user_id) AS unique_users,
  sum(e.price * e.quantity) AS total_revenue,
  min(e.price) AS min_price,
  max(e.price) AS max_price,
  avgState(e.price) AS avg_price,
  medianState(e.price) AS median_price,
  quantileState(0.95)(e.price) AS price_95th_percentile,
  sumMap(e.tags, arrayMap(x -> 1, e.tags)) AS tag_counts,
  groupArrayState(e.item_id) AS sample_items,
  countIfState(e.quantity > 1) AS multi_quantity_purchases
FROM
  async_mv_join_agg_test.events AS e
  INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE
  (e.event_type IN ('view', 'click', 'purchase'))
  AND (e.price > 0)
  AND (e.price < 10000)
  AND (i.is_promo = 1)
  AND (e.session_duration < 3600)
GROUP BY
  toStartOfMonth(e.event_date),
  e.event_type,
  i.category;

返回结果：

valid_mv:                   1
reason:
mv_sql:                     SELECT toStartOfMonth(e.event_date) AS dim_event_month, e.event_type AS event_type, i.category AS category, countState() AS event_count, uniqExactState(e.user_id) AS unique_users, sum(e.price * e.quantity) AS total_revenue, min(e.price) AS min_price, max(e.price) AS max_price, avgState(e.price) AS avg_price, medianState(e.price) AS median_price, quantileState(0.95)(e.price) AS price_95th_percentile, sumMap(e.tags, arrayMap(x -> 1, e.tags)) AS tag_counts, groupArrayState(e.item_id) AS sample_items, countIfState(e.quantity > 1) AS multi_quantity_purchases FROM async_mv_join_agg_test.events AS e INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id WHERE (e.event_type IN ('view', 'click', 'purchase')) AND (e.price > 0) AND (e.price < 10000) AND (i.is_promo = 1) AND (e.session_duration < 3600) GROUP BY toStartOfMonth(e.event_date), e.event_type, i.category
table_columns:              {'dim_event_month':'Date','event_type':'String','category':'String','event_count':'AggregateFunction(count)','unique_users':'AggregateFunction(uniqExact, UInt32)','total_revenue':'Decimal(38, 2)','min_price':'Decimal(10, 2)','max_price':'Decimal(10, 2)','avg_price':'AggregateFunction(avg, Decimal(10, 2))','median_price':'AggregateFunction(quantile, Decimal(10, 2))','price_95th_percentile':'AggregateFunction(quantile(0.95), Decimal(10, 2))','tag_counts':'Tuple(Array(String), Array(UInt64))','sample_items':'AggregateFunction(groupArray, UInt32)','multi_quantity_purchases':'AggregateFunction(countIf, UInt8)'}
aggregate_function_columns: {'event_count':'AggregateFunction(count)','unique_users':'AggregateFunction(uniqExact, UInt32)','total_revenue':'Decimal(38, 2)','min_price':'Decimal(10, 2)','max_price':'Decimal(10, 2)','avg_price':'AggregateFunction(avg, Decimal(10, 2))','median_price':'AggregateFunction(quantile, Decimal(10, 2))','price_95th_percentile':'AggregateFunction(quantile(0.95), Decimal(10, 2))','tag_counts':'Tuple(Array(String), Array(UInt64))','sample_items':'AggregateFunction(groupArray, UInt32)','multi_quantity_purchases':'AggregateFunction(countIf, UInt8)'}
dimension_columns:          {'dim_event_month':'Date','event_type':'String','category':'String'}
partition_columns:          ['dim_event_month']
subquery:                   0

第3步: 构建异步视图 #

从第2步获取的信息，可以用于构建异步视图，，还需要根据业务需求确定引擎类型，排序键，分区键，刷新间隔，刷新起始时间，视图刷新工作时间，以及使用刷新资源名。

• 引擎类型：​对于异步视图由于刷新最小粒度时分区，一般会使用CnchMergeTree，没有必要使用CnchAggregatingMergeTree，因为视图分区都是经过聚合的数据，没有在次聚合的必要，一般推荐使用隐式表方式构建视图，可以避免创建目的表，避免确定目标表列名和类型，目标表与视图表绑定在一起，默认删除视图会一起删除目的表。
• 分区键：​为了使异步视图能根据能按照分区为粒度增量构建，异步视图目的表的分区键需要由视图查询中的一个基表的分区键表示，可以是直接使用或者使用表达式转化。示例中选择跟基表一样的分区键dim_event_month
• 排序键：​查询中出现在where中频次较高但是维度基数较小的维度列可以作为排序键首位，示例中选择(event_type, category)
• bucket配置：​为了加速视图匹配后的查询性能，可以配置对target 进行cluster分桶，针对 cluster key 的聚合计算，可以有更少的内存占用和更短的执行时间，两张或者多张表针对 cluster key 的 join 可以获得 co-located join 的优化，极大的降低 shuffle 数据量并得到更短的执行时间，所以实例中可以定义group by 维度列作为cluster key，bucket number 根据计算vw hdfs-staging-default的pod个数的整数倍定义， 定义

CLUSTER BY EXPRESSION cityHash64(event_type) % 8 INTO 8 BUCKETS

• 刷新间隔：​根据实际业务需求，确定刷新间隔时间，建议分钟级别，1分钟\5分钟，有利刷新任务失败重试
• 起始时间：根据实际业务需求，如果设置的时间早于创建时间会立刻启动刷新，或者预定未来启动时间
• 工作时间：如果没有单独vw资源进行视图刷新，为了不影响工作时间视图刷新影响正常查询负载，可以配置视图在低峰期运行，例如：设置async_mv_refresh_work_period = '21:00-23:59/1,2,3,4,5,6,7;00:00-08:59/1,2,3,4,5,6,7|Asia/Shanghai', 如果有单独刷新vw资源，不用配置或者配置全天运行async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai'。
• 视图查询配置：​视图刷新要求稳定，可以接受运行时间更长，使用的资源更低，所以根据刷新资源配置合理的设置查询配置，为了隔离刷新负载，可以配置virtual_warehouse专门用于刷新，推荐的配置如下：

SETTINGS
  max_memory_usage = 68719476736,
  max_threads = 8,
  virtual_warehouse = 'hdfs-staging-default',
  max_memory_usage_for_user = 242292621312,
  exchange_queue_bytes = 1073741824,
  max_execution_time = 3600

• 视图属性：​一般为了减少视图占用资源，会串行运行视图，并行度配置比较低，开启视图刷新和改写，设置bsp运行模式，推荐的配置如下：

PROPERTIES 
  async_mv_enable_refresh_execute_serial = 1,
  async_mv_max_refresh_task_num = 8,
  async_mv_enbale_query_rewrite = 1,
  async_mv_enable_refresh = 1,
  async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai',
  async_mv_refresh_task_bsp_mode = 1;

最终确定视图定义（在定义视图过程中注意别名的命名，可以根据需要进行修改）

CREATE MATERIALIZED VIEW async_mv_join_agg_test.events_items_summary
ENGINE = CnchMergeTree
PARTITION BY dim_event_month
CLUSTER BY EXPRESSION cityHash64(event_type) % 8 INTO 8 BUCKETS
ORDER BY (event_type, category)
REFRESH ASYNC START('2024-11-01 15:43:48') EVERY(INTERVAL 1 MINUTE) AS
SELECT
  toStartOfMonth(e.event_date) AS dim_event_month,
  e.event_type AS event_type,
  i.category AS category,
  countState() AS event_count,
  uniqExactState(e.user_id) AS unique_users,
  sum(e.price * e.quantity) AS total_revenue,
  min(e.price) AS min_price,
  max(e.price) AS max_price,
  avgState(e.price) AS avg_price,
  medianState(e.price) AS median_price,
  quantileState(0.95)(e.price) AS price_95th_percentile,
  sumMap(e.tags, arrayMap(x -> 1, e.tags)) AS tag_counts,
  groupArrayState(e.item_id) AS sample_items,
  countIfState(e.quantity > 1) AS multi_quantity_purchases
FROM
  async_mv_join_agg_test.events AS e
  INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE
  (e.event_type IN ('view', 'click', 'purchase'))
  AND (e.price > 0)
  AND (e.price < 10000)
  AND (i.is_promo = 1)
  AND (e.session_duration < 3600)
GROUP BY
  toStartOfMonth(e.event_date),
  e.event_type,
  i.category
SETTINGS max_memory_usage = 68719476736,
  max_threads = 8,
  virtual_warehouse = 'hdfs-staging-default',
  max_memory_usage_for_user = 242292621312,
  exchange_queue_bytes = 1073741824,
  max_execution_time = 3600 
PROPERTIES 
  async_mv_enable_refresh_execute_serial = 1,
  async_mv_max_refresh_task_num = 8,
  async_mv_enbale_query_rewrite = 1,
  async_mv_enable_refresh = 1,
  async_mv_refresh_work_period = '00:00-23:59/1,2,3,4,5,6,7|Asia/Shanghai',
  async_mv_refresh_task_bsp_mode = 1;

第4步: 配置全局队列 #

为了确保视图刷新任务资源负载保持可控，必须要配置视图刷新任务的全局队列最大并发，保证所有视图提交的任务在限定的任务数下运行，推荐配置如下：

-- 新增匹配rule，query id 含有async_refresh，在highest 队列中进行匹配
ALTER WAREHOUSE `hdfs-staging-default` ADD RULE SET query_id = '.*async_refresh.*' , rule_name = 'async_mv_refresh_rule' WHERE queue_name = 'highest';

-- 更新highest的等待队列长度为30 
ALTER WAREHOUSE `hdfs-staging-default` MODIFY RULE SET query_queue_size = 30 WHERE queue_name = 'highest';

--- 更新视图刷新任务最大并发数为6
ALTER WAREHOUSE `hdfs-staging-default` MODIFY RULE SET max_concurrency = 6 WHERE queue_name = 'highest';

--- 查询是否这是正确
SELECT * FROM system.virtual_warehouse_queue WHERE rule_name = 'async_mv_refresh_rule';

第5步: 监控视图运行 #

• 查看视图状态

通过查询系统表确定视图的运行状态system.cnch_view_tables，可以看到有些分区没有物化，任务的执行状态，相关的视图定义，以及基表，对齐表的信息，目的表会自动构建名称：async_mv_join_agg_test..inner_id.173bf69a-653d-4cb6-9830-d3c71efba0ca``
隐式视图目的表的命名规则：.inner_id. + materialized_view_uuid

select * from system.cnch_view_tables where database = 'async_mv_join_agg_test' and table = 'events_items_summary';

等待partition_diffs 中的add_partitions， drop_partitions 都为空时说明视图已经刷新完毕。

• 查看刷新任务状态

查询最新执行完成的10个任务：

SELECT *
FROM cnch_system.cnch_view_refresh_task_log
WHERE (database = 'async_mv_join_agg_test') AND (view = 'events_items_summary') AND (status >= 2) and event_date = today() 
ORDER BY event_time_microseconds DESC
LIMIT 10;

根据任务QueryId可以在query log中获取进一步的执行状态：

select
  *
from
  cnch(`hdfs-staging-default`, system.query_log)
where
  initial_query_id = '4c4b447b-4011-4a22-b6c2-e418cdc8a2f5_async_refresh_async_mv_join_agg_test.events_items_summary'
  and type >= 2;

查询今天执行任务是否有异常：

SELECT *
FROM cnch_system.cnch_view_refresh_task_log
WHERE (database = 'async_mv_join_agg_test') AND (view = 'events_items_summary') AND notEmpty(exception)
ORDER BY event_time_microseconds DESC
LIMIT 10;

• 验证查询匹配

执行业务的日常查询，业务上一般会使用format JSON在返回的结果中获取是否命中视图，通过关闭视图匹配和开启视图匹配查看查询结果是否是一致的。
enable_materialized_view_rewrite = 1

-- 开启视图匹配(with materialized view rewrite enabled)
SELECT
    e.event_type AS event_type,
    i.category AS category,
    count() AS event_count,
    sum(e.price * e.quantity) AS total_revenue,
    min(e.price) AS min_price,
    max(e.price) AS max_price,
    avg(e.price) AS avg_price
FROM async_mv_join_agg_test.events AS e
INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE (e.event_type IN ('view')) AND (e.price > 0) AND (e.price < 10000) AND (i.is_promo = 1) AND (e.session_duration < 3600)
GROUP BY
    e.event_type,
    i.category
ORDER BY
    e.event_type ASC,
    i.category ASC
LIMIT 10
SETTINGS enable_optimizer = 1, enable_materialized_view_rewrite = 1 format JSON;

enable_materialized_view_rewrite = 0

-- 关闭视图匹配(with materialized view rewrite disabled)
SELECT
    e.event_type AS event_type,
    i.category AS category,
    count() AS event_count,
    sum(e.price * e.quantity) AS total_revenue,
    min(e.price) AS min_price,
    max(e.price) AS max_price,
    avg(e.price) AS avg_price
FROM async_mv_join_agg_test.events AS e
INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE (e.event_type IN ('view')) AND (e.price > 0) AND (e.price < 10000) AND (i.is_promo = 1) AND (e.session_duration < 3600)
GROUP BY
    e.event_type,
    i.category
ORDER BY
    e.event_type ASC,
    i.category ASC
LIMIT 10
SETTINGS enable_optimizer = 1, enable_materialized_view_rewrite = 0 format JSON;

查询结果一致，说明匹配改写成功。

• 查看改写执行计划

EXPLAIN
SELECT
    e.event_type AS event_type,
    i.category AS category,
    count() AS event_count,
    sum(e.price * e.quantity) AS total_revenue,
    min(e.price) AS min_price,
    max(e.price) AS max_price,
    avg(e.price) AS avg_price
FROM async_mv_join_agg_test.events AS e
INNER JOIN async_mv_join_agg_test.items AS i ON e.item_id = i.item_id
WHERE (e.event_type IN ('view')) AND (e.price > 0) AND (e.price < 10000) AND (i.is_promo = 1) AND (e.session_duration < 3600)
GROUP BY
    e.event_type,
    i.category
ORDER BY
    e.event_type ASC,
    i.category ASC
LIMIT 10
SETTINGS enable_optimizer = 1, enable_materialized_view_rewrite = 1

#