From 2a95ae2cea0502bbfa7cf8aa84a08f135798b813 Mon Sep 17 00:00:00 2001 From: lizheng920625 Date: Wed, 27 Jan 2021 10:41:59 +0800 Subject: [PATCH] update the doc for dynamic filters --- hetu-docs/en/admin/dynamic-filters.md | 106 ++++++++++++++++++++++++-- hetu-docs/zh/admin/dynamic-filters.md | 100 +++++++++++++++++++++++- 2 files changed, 197 insertions(+), 9 deletions(-) diff --git a/hetu-docs/en/admin/dynamic-filters.md b/hetu-docs/en/admin/dynamic-filters.md index 025117667..b3224bad2 100644 --- a/hetu-docs/en/admin/dynamic-filters.md +++ b/hetu-docs/en/admin/dynamic-filters.md @@ -1,19 +1,19 @@ # Dynamic Filtering -This section describes the openlookeng dynamic filtering features. Dynamic filtering is suitable for highly selective join scenarios, that is, most of the tables on the probe side are filtered out after reading because they do not match the join conditions. +This section describes the openLooKeng dynamic filtering features. Dynamic filtering is suitable for highly selective join scenarios, i.e., most of the tables on the probe side are filtered out after reading because they do not match the join conditions. -Openlookeng generates dynamic filter conditions based on join conditions and data read from build side during query run, and is applied to the table scan stage of probe side table as an additional filter condition, to reduce the data volume of probe table participating in join operation and effectively reduce IO read and network transmission. +openLooKeng generates dynamic filter conditions based on join conditions and data read from build side during query run, and is applied to the table scan stage of probe side table as an additional filter condition, to reduce the data volume of probe table participating in join operation and effectively reduce IO read and network transmission. ## scenarios -Openlookeng dynamic filtering is currently applicable to 'inner join', 'semi-join' and 'right join' scenarios, only can be applied to 'hive connector', 'DC connector' and 'Memory connector'. +The dynamic filtering is primarily used to optimize the highly selective join scenarios (including dynamic partition pruning for large partitioned tables and row filtering for non-partitioned tables). openLooKeng dynamic filtering is currently applicable to `inner join`, `semi join` and `right join` scenarios, only can be applied to `Hive connector`, `DC connector` and `Memory connector`. ## Usage -Openlookeng's dynamic filtering feature depends on the distributed cache component. Please refer to the section [Configuring HA](../installation/deployment-ha.md) to configure 'hazelcast'. -In '/ etc/ config.properties "The following parameters need to be configured. +openLooKeng dynamic filtering feature depends on the distributed cache component. Please refer to the section [Configuring HA](../installation/deployment-ha.md) to configure `hazelcast`. +In `/etc/config.properties`, yhe following parameters need to be configured. ``` properties enable-dynamic-filtering=true dynamic-filtering-data-type=BLOOM_FILTER -dynamic-filtering-max-per-driver-size='100MB' +dynamic-filtering-max-per-driver-size=100MB dynamic-filtering-max-per-driver-row-count=10000 dynamic-filtering-bloom-filter-fpp=0.1 ``` @@ -38,3 +38,97 @@ The above attributes are described below: - `hive.dynamic-filter-partition-filtering`: Filter out hive splits early based on partition value using dynamic filter, default to false. - `hive.dynamic-filtering-row-filtering-threshold`: Filter out hive rows early if the dynamic filter size is below the threshold, default to 2000. +## Query Plans +The following example shows a query using the dynamic filter, labeled as **dynamicFilter**. We can use the explain command to see whether the dynamic filter works and can also get it form the webUI with the live plan. + +``` sql +create table table1 (id integer, year varchar); +create table table2 (id integer, total integer); +insert into table1 values (1, '2019'), (2, '2020'), (3, '2021'); +insert into table2 values (1, 100), (2, 200); +``` + +Inner join: + +``` sql +explain select t1.id, t1.year from table1 t1, table2 t2 where t1.id = t2.id and t2.total = 200; +Query Plan +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- +Output[id, year] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ RemoteExchange[GATHER] +│ Layout: [year:varchar, id:integer] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ InnerJoin[("id" = "id_0")][$hashvalue, $hashvalue_9] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +│ Distribution: PARTITIONED +│ dynamicFilterAssignments = {id_0 -> 238} +├─ RemoteExchange[REPARTITION][$hashvalue] +│ │ Layout: [id:integer, year:varchar, $hashvalue:bigint] +│ │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +│ └─ ScanFilterProject[table = memory:0, filterPredicate = true, dynamicFilter = {238 -> "id"}] +│ Layout: [id:integer, year:varchar, $hashvalue_8:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +│ $hashvalue_8 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id"), 0)) +│ year := 1 +│ id := 0 +└─ LocalExchange[HASH][$hashvalue_9] ("id_0") +│ Layout: [id_0:integer, $hashvalue_9:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ RemoteExchange[REPARTITION][$hashvalue_10] +│ Layout: [id_0:integer, $hashvalue_10:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ ScanFilterProject[table = memory:5, filterPredicate = ("total" = 200)] +Layout: [id_0:integer, $hashvalue_11:bigint] +Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +$hashvalue_11 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id_0"), 0)) +total := 1 +id_0 := 0 +``` + +Semi join: + +``` sql +explain select * from table1 where id in (select id from table2); +Query Plan +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- +Output[id, year] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ RemoteExchange[GATHER] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ FilterProject[filterPredicate = "expr_6"] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}/{rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ Project[] +│ Layout: [id:integer, year:varchar, expr_6:boolean] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ SemiJoin[id = id_1][$hashvalue, $hashvalue_16] +│ Layout: [id:integer, year:varchar, $hashvalue:bigint, expr_6:boolean] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +│ Distribution: PARTITIONED +│ dynamicFilterId: 279 +├─ RemoteExchange[REPARTITION][$hashvalue] +│ │ Layout: [id:integer, year:varchar, $hashvalue:bigint] +│ │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +│ └─ ScanFilterProject[table = memory:0, filterPredicate = true, dynamicFilter = {279 -> "id"}] +│ Layout: [id:integer, year:varchar, $hashvalue_15:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +│ $hashvalue_15 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id"), 0)) +│ year := 1 +│ id := 0 +└─ LocalExchange[SINGLE] () +│ Layout: [id_1:integer, $hashvalue_16:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ RemoteExchange[REPARTITION - REPLICATE NULLS AND ANY][$hashvalue_17] +│ Layout: [id_1:integer, $hashvalue_17:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ ScanProject[table = memory:5] +Layout: [id_1:integer, $hashvalue_18:bigint] +Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +$hashvalue_18 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id_1"), 0)) +id_1 := 0 +``` diff --git a/hetu-docs/zh/admin/dynamic-filters.md b/hetu-docs/zh/admin/dynamic-filters.md index 6f06b6336..7c4e59d31 100644 --- a/hetu-docs/zh/admin/dynamic-filters.md +++ b/hetu-docs/zh/admin/dynamic-filters.md @@ -4,17 +4,17 @@ openLooKeng在查询运行时，依靠join条件以及build侧读出的数据，生成动态过滤条件，并作为额外的过滤条件应用到probe侧表的table scan阶段，从而减少参与join操作的probe表的数据量，有效地减少IO读取与网络传输。 ## 适用场景 -openLooKeng动态过滤当前适用于`inner join`，`semi join` 以及`right join`场景，适用于`Hive connector`，`DC connector`以及`Memory connector`。 +openLooKeng动态过滤主要应用于高选择性的join场景（包含针对分区表的分区裁剪以及非分区表的行过滤）。openLooKeng动态过滤当前适用于`inner join`，`semi join` 以及`right join`场景，适用于`Hive connector`，`DC connector`以及`Memory connector`。 ## 使用 openLooKeng动态过滤特性依赖于分布式缓存组件，请参考[Configuring HA](../installation/deployment-ha.md)章节配置`Hazelcast`。 -在`/etc/config.properties‘需要配置如下参数 +在`/etc/config.properties`需要配置如下参数 ``` properties enable-dynamic-filtering=true dynamic-filtering-data-type=BLOOM_FILTER -dynamic-filtering-max-per-driver-size='100MB' +dynamic-filtering-max-per-driver-size=100MB dynamic-filtering-max-per-driver-row-count=10000 dynamic-filtering-bloom-filter-fpp=0.1 ``` @@ -38,3 +38,97 @@ hive.dynamic-filtering-row-filtering-threshold=5000 - `hive.dynamic-filter-partition-filtering`：使用动态过滤条件根据分区值进行预先过滤，默认值是false。 - `hive.dynamic-filtering-row-filtering-threshold`：如果动态过滤条件大小低于阈值，则应用行过滤，默认值是2000。 +## 执行计划 +下面的例子展示了SQL语句如何应用动态过滤条件，在执行计划中标记为**dynamicFilter**。 可以使用explain命令查看动态过滤是否应用，也可以在webUI中的liveplan查看当前执行是否应用动态过滤。 + +``` sql +create table table1 (id integer, year varchar); +create table table2 (id integer, total integer); +insert into table1 values (1, '2019'), (2, '2020'), (3, '2021'); +insert into table2 values (1, 100), (2, 200); +``` + +Inner join: + +``` sql +explain select t1.id, t1.year from table1 t1, table2 t2 where t1.id = t2.id and t2.total = 200; +Query Plan +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- +Output[id, year] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ RemoteExchange[GATHER] +│ Layout: [year:varchar, id:integer] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ InnerJoin[("id" = "id_0")][$hashvalue, $hashvalue_9] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +│ Distribution: PARTITIONED +│ dynamicFilterAssignments = {id_0 -> 238} +├─ RemoteExchange[REPARTITION][$hashvalue] +│ │ Layout: [id:integer, year:varchar, $hashvalue:bigint] +│ │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +│ └─ ScanFilterProject[table = memory:0, filterPredicate = true, dynamicFilter = {238 -> "id"}] +│ Layout: [id:integer, year:varchar, $hashvalue_8:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +│ $hashvalue_8 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id"), 0)) +│ year := 1 +│ id := 0 +└─ LocalExchange[HASH][$hashvalue_9] ("id_0") +│ Layout: [id_0:integer, $hashvalue_9:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ RemoteExchange[REPARTITION][$hashvalue_10] +│ Layout: [id_0:integer, $hashvalue_10:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ ScanFilterProject[table = memory:5, filterPredicate = ("total" = 200)] +Layout: [id_0:integer, $hashvalue_11:bigint] +Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +$hashvalue_11 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id_0"), 0)) +total := 1 +id_0 := 0 +``` + +Semi join: + +``` sql +explain select * from table1 where id in (select id from table2); +Query Plan +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- +Output[id, year] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ RemoteExchange[GATHER] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ FilterProject[filterPredicate = "expr_6"] +│ Layout: [id:integer, year:varchar] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}/{rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ Project[] +│ Layout: [id:integer, year:varchar, expr_6:boolean] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +└─ SemiJoin[id = id_1][$hashvalue, $hashvalue_16] +│ Layout: [id:integer, year:varchar, $hashvalue:bigint, expr_6:boolean] +│ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} +│ Distribution: PARTITIONED +│ dynamicFilterId: 279 +├─ RemoteExchange[REPARTITION][$hashvalue] +│ │ Layout: [id:integer, year:varchar, $hashvalue:bigint] +│ │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +│ └─ ScanFilterProject[table = memory:0, filterPredicate = true, dynamicFilter = {279 -> "id"}] +│ Layout: [id:integer, year:varchar, $hashvalue_15:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +│ $hashvalue_15 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id"), 0)) +│ year := 1 +│ id := 0 +└─ LocalExchange[SINGLE] () +│ Layout: [id_1:integer, $hashvalue_16:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ RemoteExchange[REPARTITION - REPLICATE NULLS AND ANY][$hashvalue_17] +│ Layout: [id_1:integer, $hashvalue_17:bigint] +│ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} +└─ ScanProject[table = memory:5] +Layout: [id_1:integer, $hashvalue_18:bigint] +Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B} +$hashvalue_18 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("id_1"), 0)) +id_1 := 0 +```