Hive 邏輯優化器

Optimizer

PointLookupOptimizer

作用：把符合條件的 OR 表達式轉為 IN。
參數hive.optimize.point.lookup 設置是否開啟 PointLookupOptimizer，默認為 true.
參數 hive.optimize.point.lookup.min 控制多少個 OR 表達式轉為 IN，默認 31。
例如以下 sql, or 有3個 child，分別是 web_site_sk =1，web_site_sk =2， web_site_sk =3。這些 child 都必須是 = 。并且判斷的字段是同一字段 web_site_sk。web_site_sk =1 or web_site_sk =2 or web_site_sk =3 可以轉為 `web_site_sk in （1，2，3）``

set hive.optimize.point.lookup.min=2;
select web_county,count(1) cnt 
from web_site 
where web_site_sk = 1 or web_site_sk = 2 or web_site_sk =3 
group by web_county;

判斷 web_site_sk = 2 改成 2 = web_site_sk 也是可以的。

下邊的用 or 關聯的多個字段可以說可以轉化的。

set hive.optimize.point.lookup.min=2;
explain select web_county,count(1) cnt 
from web_site 
where  (web_site_sk = 1 and web_open_date_sk=1) or 
(web_site_sk = 2 and web_open_date_sk=2) or
(web_site_sk = 3 and web_open_date_sk=3)
group by web_county;

生成的執行計劃如下：

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_7]Group By Operator [GBY_5] (rows=1 width=109)Output:["_col0","_col1"],aggregations:["count(VALUE._col0)"],keys:KEY._col0<-Map 1 [SIMPLE_EDGE]SHUFFLE [RS_4]PartitionCols:_col0Group By Operator [GBY_3] (rows=1 width=109)Output:["_col0","_col1"],aggregations:["count(1)"],keys:web_countySelect Operator [SEL_2] (rows=1 width=117)Output:["web_county"]Filter Operator [FIL_8] (rows=1 width=117)predicate:(struct(web_open_date_sk,web_site_sk)) IN (const struct(1L,1L), const struct(2L,2L), const struct(3L,3L))TableScan [TS_0] (rows=32 width=117)tpcds_hdfs_orc_3@web_site,web_site,Tbl:COMPLETE,Col:COMPLETE,Output:["web_site_sk","web_open_date_sk","web_county"]

struct 字段不能轉化為 in

drop table if exists complex_table;
create table complex_table(c_int int,c_struct struct<name:string,age:int>
)
row format delimited
FIELDS TERMINATED BY '\t'
collection items terminated by ','
map keys terminated by ':'
lines terminated by '\n'
stored as textfile;insert into complex_table values(1, named_struct("name","Alice", "age",18)
);select distinct c_int from complex_table where 
(c_struct.name="Alice" and  c_struct.age=18) or 
(c_struct.name="Alice" and  c_struct.age=19) or 
(c_struct.name="Alice" and  c_struct.age=20);

PartitionColumnsSeparator

作用：把分區字段從 in 提取出來。如 STRUCT(T1.a, T1.b, T2.b, T2.c) IN (STRUCT(1, 2, 3, 4) , STRUCT(2, 3, 4, 5))，T1.a, T1.b, T2.c 是分區字段，T2.b 不是分區字段。處理之后刪除額外的斷言 STRUCT(T1.a, T1.b) IN (STRUCT(1, 2), STRUCT(2, 3))
AND
STRUCT(T2.c) IN (STRUCT(4), STRUCT(5))
這些額外的斷言用來分區過濾。一旦分區過濾完成，分區條件會從執行計劃中去除。

例如以下語句僅列出了3個分區的，因為 ws_sold_date_sk 是分區字段。

explain extended select distinct ws_item_sk from web_sales 
where (ws_sold_date_sk=2452642 and ws_sold_time_sk=1) or (ws_sold_date_sk=2452641 and ws_sold_time_sk=2) or (ws_sold_date_sk=2452640 and ws_sold_time_sk=3);

PredicateTransitivePropagate（沒找到觸發 sql）

以下語句能走到邏輯，但是沒有改執行計劃

select sum(ws_net_paid) from web_sales join(select * from web_site where web_site_sk < 10) web_site where ws_sold_date_sk=2452640 and web_site_sk = 1;

ConstantPropagate

從 root 到 child 的順序遍歷 DAG，對于每個條件表達式，做以下處理：

折疊表達式，如果表達式是 UDF 并且所以的參數是常數。

Filter 中包含 true 的表達式去掉。

explain select count(1) from web_site where 1=1 and web_site_sk<10;

可以看到 1=1 總是為 true，可以去掉。

Filter Operator [FIL_8] (rows=10 width=8)predicate:(web_site_sk < 10L)

Filter 中包含 false 的表達式可以短路計算。

explain select count(1) from web_site where 1=1 or web_site_sk<10;

可以看到 1=1 總是為 true，所以所有的表達式都不需要計算。

null 條件等于 false

以下兩個 sql 的結果都為 0

select count(1) from web_site where null;

select count(1) from web_site where false;

表達式的傳遞
如以下語句中 web_site_sk=5 可以向上傳遞, 和 web_site_sk < 10 組在一起。

explain select * from (select * from web_site where web_site_sk < 10) t where web_site_sk=5;

最終的 Filter 如下

Filter Operator [FIL_6]predicate:((web_site_sk < 10L) and (web_site_sk = 5L))

SyntheticJoinPredicate

explain logical 
select sum(ws_net_paid) net_paid 
from  web_sales 
where ws_web_site_sk in(select web_site_sk from web_site where web_site_sk < 10 ) 
and ws_sold_date_sk=2452642;

輸出結果如下：
web_site 的 TablScan 后 Filter 增加了 (web_site_sk) IN (RS[7])

predicate: ((web_site_sk < 10L) and (web_site_sk) IN (RS[7])) (type: boolean)

web_sales 之后增加了 Filter 如下：

alias: web_salesFilter Operator (FIL_20)predicate: (ws_web_site_sk is not null and (ws_web_site_sk) IN (RS[9])) (type: boolean)

explain logical 
select sum(ws_net_paid) net_paid 
from  web_sales join web_site on ws_web_site_sk=web_site_sk
wherews_sold_date_sk=2452642;

生成的執行計劃如下：

Explain
LOGICAL PLAN:
web_sales TableScan (TS_0)alias: web_salesFilter Operator (FIL_16)predicate: (ws_web_site_sk is not null and (ws_web_site_sk) IN (RS[5])) (type: boolean)// ...
web_site TableScan (TS_1)alias: web_siteFilter Operator (FIL_17)predicate: (web_site_sk is not null and (web_site_sk) IN (RS[3])) (type: boolean)// ...

SortedDynPartitionOptimizer

動態分區排序優化，啟用時，在插入記錄之前，按分區字段，或者 bucket 字段運行排序，保證一個 reducer 僅有一個 writer，可以減少 reducer 的內存壓力。

set hive.stats.autogather=false;
create table profile(c1 string) stored as textfile;
load data local inpath '/etc/profile' overwrite into table profile;
create table p_profile(c1 string) partitioned by (len int);

沒有經過 sort 優化

set hive.optimize.sort.dynamic.partition=false;
explain logical insert overwrite table  p_profile select c1,length(c1) from profile;

輸出如下：

LOGICAL PLAN:
profile TableScan (TS_0)alias: profileSelect Operator (SEL_1)expressions: c1 (type: string), length(c1) (type: int)outputColumnNames: _col0, _col1File Output Operator (FS_2)compressed: falsetable:input format: org.apache.hadoop.mapred.TextInputFormatoutput format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormatserde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDename: test.p_profile

經過 sort 優化

set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.optimize.sort.dynamic.partition=true;
explain logical insert overwrite table  p_profile select c1,length(c1) from profile;

多了排序

Explain
LOGICAL PLAN:
profile TableScan (TS_0)alias: profileSelect Operator (SEL_1)expressions: c1 (type: string), length(c1) (type: int)outputColumnNames: _col0, _col1Reduce Output Operator (RS_3)key expressions: _col1 (type: int)sort order: +Map-reduce partition columns: _col1 (type: int)value expressions: _col0 (type: string)Select Operator (SEL_4)expressions: VALUE._col0 (type: string), KEY._col1 (type: int)outputColumnNames: _col0, _col1File Output Operator (FS_2)compressed: falseDp Sort State: PARTITION_SORTEDtable:input format: org.apache.hadoop.mapred.TextInputFormatoutput format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormatserde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDename: test.p_profile

SortedDynPartitionTimeGranularityOptimizer

專門為 FileSink 并且FileSink 為 org.apache.hadoop.hive.druid.io.DruidOutputFormat 可以優化。

PartitionPruner(簡稱 ppr) & PartitionConditionRemover(簡稱 PCR)

select sum(ws_net_paid) sum_ws_net_paid
from web_sales 
where ws_sold_date_sk >= 2452640 and ws_sold_date_sk <=2452642 and ws_net_paid>1.0 ;

PartitionPruner 把 TableScan 后面的 Filter 中的 predicate 信息放入 opToPartList 中。

HashMap<TableScanOperator, PrunedPartitionList> opToPartList;

PartitionConditionRemover 從 opToPartList 取出 ts 對應的,進行 PartitionPruner.prune 得到 partsList。調用 opToPartList.put(ts, partsList);

如果 Filter 中，僅包含分區字段的條件，則然后刪除 TableScan 后的Filter。否則刪除 Filter 中關于分區字段的判斷。

ListBucketingPruner

drop table if exists list_bucket_test;
CREATE TABLE list_bucket_test (key int, value int) partitioned by (dt string)SKEWED BY (key) ON (1,2) STORED AS DIRECTORIES;insert overwrite table list_bucket_test partition(dt=20250519) values(1,1),(1,2),(2,3),(2,4),(3,5);

list_bucket_test 表目錄下有3個目錄，分別是 key=1,key=2 和 HIVE_DEFAULT_LIST_BUCKETING_DIR_NAME。

drwxr-xr-x  4 houzhizhen  staff  128  5 19 16:32 HIVE_DEFAULT_LIST_BUCKETING_DIR_NAME
drwxr-xr-x  4 houzhizhen  staff  128  5 19 16:32 key=1
drwxr-xr-x  4 houzhizhen  staff  128  5 19 16:32 key=2

set hive.optimize.listbucketing=true;
select sum(value) from list_bucket_test where dt=20250519 and key=1;

ListBucketingPruner 把 TableScanOperator->(Partition-> “GenericUDFOPEqual(Column[key], Const int 1)”) 信息放入 ParseContext 的以下字段。

Map<TableScanOperator, Map<String, ExprNodeDesc>> opToPartToSkewedPruner;

value 是 dt=20250519 -> “GenericUDFOPEqual(Column[key], Const int 1)”

然后在 GenMapRedUtils 里，根據 opToPartToSkewedPruner 信息生成 Partition 需要處理的 path.

GroupByOptimizer

Group by 優化，如果 group key 包括所有的 bucketing key 和 sort key，并且順序相同，那么 group by 可以在map 中進行。


drop table if exists test ;
CREATE TABLE test (bkey int,skey int, value int) partitioned by(pkey int)
clustered by(bkey) sorted by (bkey,skey asc) into 8 buckets;insert overwrite table test partition(pkey=5)values(1,1,1),(1,2,3),(2,3,3),(2,3,4);
-- 數據量少，不轉成 Fetcher
set hive.fetch.task.conversion=none;
explain select bkey,skey, sum(value) from test where pkey=5 group by bkey,skey ;

輸出如下，可以看到，在 map 端完成了group by 操作，沒有 reduce，這樣可能降低并行度。

Explain
Stage-0Fetch Operatorlimit:-1Stage-1Map 1File Output Operator [FS_7]Group By Operator [GBY_3] (rows=4 width=16)Output:["_col0","_col1","_col2"],aggregations:["sum(value)"],keys:bkey, skeySelect Operator [SEL_2] (rows=4 width=12)Output:["bkey","skey","value"]TableScan [TS_0] (rows=4 width=12)tpcds_hdfs_orc_3@test,test,Tbl:COMPLETE,Col:COMPLETE,Output:["bkey","skey","value"]

ColumnPruner

后根遍歷所有的 Operator，僅僅保留子操作用到的列。

explain logical 
select sum(ws_net_paid) net_paid 
from  web_sales 
where ws_sold_date_sk=2452642;

如沒有優化的 TableScan，是所有的列，之后的 Select 也是。優化后，僅包含子操作需要的列。

CountDistinctRewriteProc

這個優化僅僅針對 tez 引擎生效。

count（distinct)只能有一個字段。并且僅能有一個 count distinct.

可以生效示例：

explain  
select count(distinct ws_web_site_sk)
from  web_sales;

生成的執行計劃如下：

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)
Reducer 3 <- Reducer 2 (CUSTOM_SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 3 vectorizedFile Output Operator [FS_19]Group By Operator [GBY_18] (rows=1 width=8)Output:["_col0"],aggregations:["count(VALUE._col0)"]<-Reducer 2 [CUSTOM_SIMPLE_EDGE] vectorizedPARTITION_ONLY_SHUFFLE [RS_17]Group By Operator [GBY_16] (rows=1 width=8)Output:["_col0"],aggregations:["count(_col0)"]Group By Operator [GBY_15] (rows=360000188 width=934)Output:["_col0"],keys:KEY._col0<-Map 1 [SIMPLE_EDGE] vectorizedSHUFFLE [RS_14]PartitionCols:_col0Group By Operator [GBY_13] (rows=720000376 width=934)Output:["_col0"],keys:ws_web_site_skSelect Operator [SEL_12] (rows=720000376 width=934)Output:["ws_web_site_sk"]TableScan [TS_0] (rows=720000376 width=934)tpcds_bos_parquet_1000@web_sales,web_sales,Tbl:COMPLETE,Col:NONE,Output:["ws_web_site_sk"]

Count Distinct 的執行過程：
對于 mr 引擎.

多個 Map task 執行 distinct ws_web_site_sk，輸出去重后的 ws_web_site_sk。
然后啟動一個 reducer，拉取所有 Map Task 的數據，進行最終的 count(distinct ws_web_site_sk) 處理。如果每個 distinct ws_web_site_sk 數量比較多，可能 OOM。

對于 tez 引擎

多個 Map task 執行 distinct ws_web_site_sk, 根據 ws_web_site_sk 為 partition key，把數據分給多個 reducer。
多個 reducer 做 count(distinct ws_web_site_sk) 處理。把數值輸出。
一個 reducer 對第 2 步的數值進行相加。

不能生效示例1:

explain 
select count(distinct ws_web_site_sk),
count(distinct ws_sold_time_sk)
from  web_sales;

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)
Reducer 3 <- Reducer 2 (CUSTOM_SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 3 vectorizedFile Output Operator [FS_21]Group By Operator [GBY_20] (rows=1 width=16)Output:["_col0","_col1"],aggregations:["count(VALUE._col0)","count(VALUE._col1)"]<-Reducer 2 [CUSTOM_SIMPLE_EDGE] vectorizedPARTITION_ONLY_SHUFFLE [RS_19]Group By Operator [GBY_18] (rows=1 width=16)Output:["_col0","_col1"],aggregations:["count(_col0)","count(_col1)"]Select Operator [SEL_17] (rows=720000376 width=934)Output:["_col0","_col1"]Group By Operator [GBY_16] (rows=720000376 width=934)Output:["_col0","_col1","_col2"],keys:KEY._col0, KEY._col1, KEY._col2<-Map 1 [SIMPLE_EDGE] vectorizedSHUFFLE [RS_15]PartitionCols:_col0, _col1, _col2Group By Operator [GBY_14] (rows=1440000752 width=934)Output:["_col0","_col1","_col2"],keys:_col0, _col1, 0LSelect Operator [SEL_13] (rows=720000376 width=934)Output:["_col0","_col1"]TableScan [TS_0] (rows=720000376 width=934)tpcds_bos_parquet_1000@web_sales,web_sales,Tbl:COMPLETE,Col:NONE,Output:["ws_web_site_sk","ws_sold_time_sk"]

不能生效示例2:

explain 
select count(distinct ws_web_site_sk, ws_sold_time_sk)
from  web_sales;

生成的執行計劃如下：

Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_6]Group By Operator [GBY_4] (rows=1 width=16)Output:["_col0"],aggregations:["count(DISTINCT KEY._col0:0._col0, KEY._col0:0._col1)"]<-Map 1 [SIMPLE_EDGE]SHUFFLE [RS_3]Group By Operator [GBY_2] (rows=720000376 width=934)Output:["_col0","_col1","_col2"],aggregations:["count(DISTINCT ws_web_site_sk, ws_sold_time_sk)"],keys:ws_web_site_sk, ws_sold_time_skSelect Operator [SEL_1] (rows=720000376 width=934)Output:["ws_sold_time_sk","ws_web_site_sk"]TableScan [TS_0] (rows=720000376 width=934)tpcds_bos_parquet_1000@web_sales,web_sales,Tbl:COMPLETE,Col:NONE,Output:["ws_sold_time_sk","ws_web_site_sk"]

SkewJoinOptimizer

僅僅對 MR 有效。

drop table if exists skew_web_sales;
CREATE TABLE `skew_web_sales`(`ws_sold_time_sk` bigint, `ws_ship_date_sk` bigint, `ws_item_sk` bigint, `ws_bill_customer_sk` bigint, `ws_bill_cdemo_sk` bigint, `ws_bill_hdemo_sk` bigint, `ws_bill_addr_sk` bigint, `ws_ship_customer_sk` bigint, `ws_ship_cdemo_sk` bigint, `ws_ship_hdemo_sk` bigint, `ws_ship_addr_sk` bigint, `ws_web_page_sk` bigint, `ws_web_site_sk` bigint, `ws_ship_mode_sk` bigint, `ws_warehouse_sk` bigint, `ws_promo_sk` bigint, `ws_order_number` bigint, `ws_quantity` int, `ws_wholesale_cost` decimal(7,2), `ws_list_price` decimal(7,2), `ws_sales_price` decimal(7,2), `ws_ext_discount_amt` decimal(7,2), `ws_ext_sales_price` decimal(7,2), `ws_ext_wholesale_cost` decimal(7,2), `ws_ext_list_price` decimal(7,2), `ws_ext_tax` decimal(7,2), `ws_coupon_amt` decimal(7,2), `ws_ext_ship_cost` decimal(7,2), `ws_net_paid` decimal(7,2), `ws_net_paid_inc_tax` decimal(7,2), `ws_net_paid_inc_ship` decimal(7,2), `ws_net_paid_inc_ship_tax` decimal(7,2), `ws_net_profit` decimal(7,2),`ws_sold_date_sk` bigint) 
skewed by (ws_web_site_sk) on (1);
insert overwrite table skew_web_sales select * from web_sales where ws_web_site_sk=1;
insert into table skew_web_sales select * from web_sales where ws_sold_date_sk=2452642 and ws_web_site_sk=3 limit 1;

set hive.optimize.skewjoin.compiletime=true;
set hive.execution.engine=mr;
explain  
select ws_web_site_sk,web_class,sum(ws_net_paid) 
from  web_site join  skew_web_sales
on skew_web_sales.ws_web_site_sk=web_site.web_site_sk 
group by ws_web_site_sk,web_class;

生成的執行計劃如下：

Explain
LOGICAL PLAN:
skew_web_sales TableScan (TS_1)alias: skew_web_salesStatistics: Num rows: 126827 Data size: 15217112 Basic stats: COMPLETE Column stats: COMPLETEFilter Operator (FIL_24)predicate: (ws_web_site_sk is not null and (ws_web_site_sk = 1L)) (type: boolean)Statistics: Num rows: 63414 Data size: 7608672 Basic stats: COMPLETE Column stats: COMPLETEReduce Output Operator (RS_5)key expressions: ws_web_site_sk (type: bigint)sort order: +Map-reduce partition columns: ws_web_site_sk (type: bigint)Statistics: Num rows: 63414 Data size: 7608672 Basic stats: COMPLETE Column stats: COMPLETEvalue expressions: ws_net_paid (type: decimal(7,2))Join Operator (JOIN_6)condition map:Inner Join 0 to 1keys:0 web_site_sk (type: bigint)1 ws_web_site_sk (type: bigint)outputColumnNames: _col7, _col42, _col58Statistics: Num rows: 63414 Data size: 13379346 Basic stats: COMPLETE Column stats: COMPLETEUnion (UNION_27)Statistics: Num rows: 94130 Data size: 19859414 Basic stats: COMPLETE Column stats: COMPLETEGroup By Operator (GBY_8)aggregations: sum(_col58)keys: _col42 (type: bigint), _col7 (type: varchar(50))mode: hashoutputColumnNames: _col0, _col1, _col2Statistics: Num rows: 2 Data size: 422 Basic stats: COMPLETE Column stats: COMPLETEReduce Output Operator (RS_9)key expressions: _col0 (type: bigint), _col1 (type: varchar(50))sort order: ++Map-reduce partition columns: _col0 (type: bigint), _col1 (type: varchar(50))Statistics: Num rows: 2 Data size: 422 Basic stats: COMPLETE Column stats: COMPLETEvalue expressions: _col2 (type: decimal(17,2))Group By Operator (GBY_10)aggregations: sum(VALUE._col0)keys: KEY._col0 (type: bigint), KEY._col1 (type: varchar(50))mode: mergepartialoutputColumnNames: _col0, _col1, _col2Statistics: Num rows: 2 Data size: 422 Basic stats: COMPLETE Column stats: COMPLETEFile Output Operator (FS_12)compressed: falseStatistics: Num rows: 2 Data size: 422 Basic stats: COMPLETE Column stats: COMPLETEtable:input format: org.apache.hadoop.mapred.SequenceFileInputFormatoutput format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormatserde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
subquery1:skew_web_sales TableScan (TS_18)alias: skew_web_salesStatistics: Num rows: 126827 Data size: 15217112 Basic stats: COMPLETE Column stats: COMPLETEFilter Operator (FIL_26)predicate: (ws_web_site_sk is not null and (not (ws_web_site_sk = 1L))) (type: boolean)Statistics: Num rows: 63413 Data size: 7608552 Basic stats: COMPLETE Column stats: COMPLETEReduce Output Operator (RS_20)key expressions: ws_web_site_sk (type: bigint)sort order: +Map-reduce partition columns: ws_web_site_sk (type: bigint)Statistics: Num rows: 63413 Data size: 7608552 Basic stats: COMPLETE Column stats: COMPLETEvalue expressions: ws_net_paid (type: decimal(7,2))Join Operator (JOIN_21)condition map:Inner Join 0 to 1keys:0 web_site_sk (type: bigint)1 ws_web_site_sk (type: bigint)outputColumnNames: _col7, _col42, _col58Statistics: Num rows: 30716 Data size: 6480068 Basic stats: COMPLETE Column stats: COMPLETEUnion (UNION_27)Statistics: Num rows: 94130 Data size: 19859414 Basic stats: COMPLETE Column stats: COMPLETE
subquery1:web_site TableScan (TS_15)alias: web_siteStatistics: Num rows: 32 Data size: 3168 Basic stats: COMPLETE Column stats: COMPLETEFilter Operator (FIL_25)predicate: (web_site_sk is not null and (not (ws_web_site_sk = 1L))) (type: boolean)Statistics: Num rows: 1 Data size: 99 Basic stats: COMPLETE Column stats: COMPLETEReduce Output Operator (RS_17)key expressions: web_site_sk (type: bigint)sort order: +Map-reduce partition columns: web_site_sk (type: bigint)Statistics: Num rows: 1 Data size: 99 Basic stats: COMPLETE Column stats: COMPLETEvalue expressions: web_class (type: varchar(50))Join Operator (JOIN_21)condition map:Inner Join 0 to 1keys:0 web_site_sk (type: bigint)1 ws_web_site_sk (type: bigint)outputColumnNames: _col7, _col42, _col58Statistics: Num rows: 30716 Data size: 6480068 Basic stats: COMPLETE Column stats: COMPLETE
web_site TableScan (TS_0)alias: web_siteStatistics: Num rows: 32 Data size: 3168 Basic stats: COMPLETE Column stats: COMPLETEFilter Operator (FIL_23)predicate: (web_site_sk is not null and (ws_web_site_sk = 1L)) (type: boolean)Statistics: Num rows: 32 Data size: 3168 Basic stats: COMPLETE Column stats: COMPLETEReduce Output Operator (RS_3)key expressions: web_site_sk (type: bigint)sort order: +Map-reduce partition columns: web_site_sk (type: bigint)Statistics: Num rows: 32 Data size: 3168 Basic stats: COMPLETE Column stats: COMPLETEvalue expressions: web_class (type: varchar(50))Join Operator (JOIN_6)condition map:Inner Join 0 to 1keys:0 web_site_sk (type: bigint)1 ws_web_site_sk (type: bigint)outputColumnNames: _col7, _col42, _col58Statistics: Num rows: 63414 Data size: 13379346 Basic stats: COMPLETE Column stats: COMPLETE

SamplePruner

示例：

drop table if exists t1 ;
create table t1(c1 string) stored as textfile;
load data local inpath '/etc/profile' overwrite into table t1;drop table if exists t2;
create external table t2(c1 string) CLUSTERED BY (c1) into 8 buckets;
insert overwrite table t2 select * from t1;
set hive.execution.engine=mr;
set hive.fetch.task.conversion=none;
explain select * from t2 TABLESAMPLE(BUCKET 3 OUT OF 8 ON c1) s;

SamplePruner 生效條件:

t2 是外部表。
CLUSTERED BY 字段和 select 中 TABLESAMPLE 的 ON 字段一樣。如果是on rand 不行。
創建表中into 8 buckets 和檢索中 OUT OF 8 的數字要一樣。
檢索中的 (BUCKET 3 OUT OF 8 ON c1) 的 3 代表是第3個文件。

MapJoinProcessor

用戶指定使用 MapJoin 的 hint。這種方法已經不再使用，現在是基于統計信息自動把 join 轉為 mapjoin.

set hive.ignore.mapjoin.hint=false;
explain select /*+mapjoin(web_site) */ sum(ws_net_paid) from web_sales join web_site on web_site_sk=ws_web_site_sk 
where ws_sold_date_sk=2452640 ;

BucketMapJoinOptimizer

使用條件:

mr 引擎
sql 中 mapjoin hint 生效。
hive.optimize.bucketmapjoin 為 true，默認為 false.

BucketMapJoinOptimizer,SortedMergeBucketMapJoinOptimizer

mr 引擎生效，并且默認不啟用。

BucketingSortingReduceSinkOptimizer

對于 insert overwrite table T2 select * from T1;
如果 T1 和 T2 都 bucket 和 sort key 相同，并且 bucket 的數量相同，那么就不用 reduer 任務。

UnionProcessor

如果 UNION 的兩個子查詢都是 map-only，把此信息存入 unionDesc/UnionOperator。如果其中的一個子查詢是 map-reduce 的作業，在 UNION 之前加入 FS。

UNION 的兩個子查詢都是 map-only 的示例.
union 和 union all 之間的區別。

union 在 operator union 后，有 group by 和 reduce sink，因為需要去重。
union all 在 operator union 后，沒有 group by 和 reduce sink，因為不需要去重。

explain logical select ws_web_site_sk from web_sales where ws_sold_date_sk=2452640
union
select ws_web_site_sk from web_sales where ws_sold_date_sk=2452641;

如果其中的一個子查詢是 map-reduce 的作業示例：

explain logical select ws_web_site_sk from web_sales where ws_sold_date_sk=2452640
union
select distinct ws_web_site_sk from web_sales where ws_sold_date_sk=2452641;

在 MapReduceCompiler 使用

BucketVersionPopulator

根據 hash 算法的不同，數據分到 N 個bucket 時會不一致。
Hive 根據 bucketingVersion 確定使用哪個 hash 算法。
在每個 Reduce Sink 之后，可以使用不同的 Bucketing version。因為 full shuffle 可以重新按新的 hash 算法分配數據。

hive 使用表的 bucket version 寫入數據。
如果從表讀數據，會考慮表的 bucketing_version 字段。

ReduceSinkDeDuplication

如果相鄰的兩個 reduce sink 有共同的 partition/sort 字段，并且字段的順序相同，可以合并為一個 reduce。

例如，以下sql 的兩處 ‘parition by’ 字段不一樣.

explain select 
avg(sum(web_tax_percentage)) over(partition by web_city)avg_monthly_sales,rank() over(partition by web_county, web_cityorder by web_county) rn
from web_site
group by web_county, web_city;

生成的語法樹如下：

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)
Reducer 3 <- Reducer 2 (SIMPLE_EDGE)
Reducer 4 <- Reducer 3 (SIMPLE_EDGE)

以下sql 生成的

explain select 
avg(sum(web_tax_percentage)) over(partition by web_county, web_city)avg_monthly_sales,rank() over(partition by web_county, web_cityorder by web_county) rn
from web_sitegroup by web_county, web_city;

生成的 vertex 如下, 比上一個 sql 少一個 reduce，因為 partition by 相同

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)
Reducer 3 <- Reducer 2 (SIMPLE_EDGE)

NonBlockingOpDeDupProc

相鄰的兩個投影（SEL）操作合并為一個投影（SEL）操作，相鄰的兩個過濾（FIL）操作合并為一個過濾（FIL）操作。

IdentityProjectRemover

刪除不必要的投影（SEL）操作。如果投影（SEL）僅僅是forward，沒有計算，如 select x+1 這種帶計算，并且沒有給列重命名，則可以去除。

如

explain select web_city 
from web_site  
where web_city > '1' 
group by web_city;

生成的 SQL 語法樹為, 沒有 select

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_7]Group By Operator [GBY_5] (rows=1 width=91)Output:["_col0"],keys:KEY._col0<-Map 1 [SIMPLE_EDGE]SHUFFLE [RS_4]PartitionCols:_col0Group By Operator [GBY_3] (rows=1 width=91)Output:["_col0"],keys:web_cityFilter Operator [FIL_8] (rows=10 width=91)predicate:(web_city > '1')TableScan [TS_0] (rows=32 width=91)tpcds_hdfs_orc_3@web_site,web_site,Tbl:COMPLETE,Col:COMPLETE,Output:["web_city"]

如沒有 filter

explain select web_city 
from web_site  
group by web_city;

在 TS 后，有一個 select.

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_6]Group By Operator [GBY_4] (rows=2 width=91)Output:["_col0"],keys:KEY._col0<-Map 1 [SIMPLE_EDGE]SHUFFLE [RS_3]PartitionCols:_col0Group By Operator [GBY_2] (rows=2 width=91)Output:["_col0"],keys:web_citySelect Operator [SEL_1] (rows=32 width=91)Output:["web_city"]TableScan [TS_0] (rows=32 width=91)tpcds_hdfs_orc_3@web_site,web_site,Tbl:COMPLETE,Col:COMPLETE,Output:["web_city"]

如

explain select upper(web_city) d_web_city
from web_site  
where web_city > '1' 
group by web_city;

在 File Output Operator 之前，多了一個 select 操作

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_7]Select Operator [SEL_6] (rows=1 width=144)Output:["_col0"]Group By Operator [GBY_5] (rows=1 width=91)Output:["_col0"],keys:KEY._col0<-Map 1 [SIMPLE_EDGE]SHUFFLE [RS_4]PartitionCols:_col0Group By Operator [GBY_3] (rows=1 width=91)Output:["_col0"],keys:web_cityFilter Operator [FIL_8] (rows=10 width=91)predicate:(web_city > '1')TableScan [TS_0] (rows=32 width=91)tpcds_hdfs_orc_3@web_site,web_site,Tbl:COMPLETE,Col:COMPLETE,Output:["web_city"]

GlobalLimitOptimizer

set hive.limit.optimize.enable=true;
select * from web_sales limit 10;

如 web_sales 有很多分區，每個分區下很多文件，查詢10條數據，沒必要打開所有文件，可能第1個文件就有10條記錄，可以減少收入。

CorrelationOptimizer

僅僅對 MR 引擎生效。參考論文：YSmart: Yet Another SQL-to-MapReduce Translator(Rubao Lee, Tian Luo, Yin Huai, Fusheng Wang, Yongqiang He, and Xiaodong Zhang)

LimitPushdownOptimizer

explain extended select sum(web_tax_percentage)
from web_site  
group by web_city limit 1;

order by: 把 limit 推到最后的 reduce sink。

explain select ws_web_site_sk
from web_sales  
order by ws_web_site_sk limit 1;

StatsOptimizer

有些 query 的結果，可以從 stats 信息中，直接獲取。
先生成表的統計信息。

analyze table web_site compute statistics for columns;

執行 sql:

explain select min(web_site_sk) from web_site;

從統計信息獲取結果的執行計劃如下，僅有一個 Fetch

Explain
Stage-0Fetch Operatorlimit:1

不能從統計信息獲取結果示例

explain select min(web_site_sk) from web_site where web_site_sk <> 2;

此 SQL 的執行計劃如下：

Explain
Vertex dependency in root stage
Reducer 2 <- Map 1 (CUSTOM_SIMPLE_EDGE)Stage-0Fetch Operatorlimit:-1Stage-1Reducer 2File Output Operator [FS_7]Group By Operator [GBY_5] (rows=1 width=8)Output:["_col0"],aggregations:["min(VALUE._col0)"]<-Map 1 [CUSTOM_SIMPLE_EDGE]PARTITION_ONLY_SHUFFLE [RS_4]Group By Operator [GBY_3] (rows=1 width=8)Output:["_col0"],aggregations:["min(web_site_sk)"]Filter Operator [FIL_8] (rows=32 width=8)predicate:(web_site_sk <> 2L)TableScan [TS_0] (rows=32 width=8)tpcds_hdfs_orc_3@web_site,web_site,Tbl:COMPLETE,Col:COMPLETE,Output:["web_site_sk"]

AnnotateWithStatistics, AnnotateWithOpTraits

當 explain 的時候，并且是 mr 引擎時，在邏輯執行計劃優化（Optimizer）時執行。
TEZ 在物理執行計劃優化時執行。
AnnotateWithStatistics 給各 Operator 設置 stats（統計信息）。
AnnotateWithOpTraits 設置 opTraits，適用于 bucket 表.如兩個 bucket 表做 join。
OpTraits 定義如下，

public OpTraits(List<List<String>> bucketColNames, int numBuckets,List<List<String>> sortColNames, int numReduceSinks) {this.bucketColNames = bucketColNames;this.numBuckets = numBuckets;this.sortColNames = sortColNames;this.numReduceSinks = numReduceSinks;}

explain
select web_site.web_city, sum(ws_ext_tax) ws_ext_tax_sum
from web_site join web_sales on web_site.web_site_sk = web_sales.ws_web_site_sk
where ws_sold_date_sk=2452642 and ws_web_site_sk =1 
group by web_site.web_city;

SimpleFetchOptimizer

對于單表簡單的操作（沒有 group by, 沒有 distinct，單表），不啟動分布式任務，直接在 fetch task里讀取表返回，可以加快執行速度。
hive.fetch.task.conversion 的值
none :禁用
minimal : 支持 SELECT *, 在分區字段過濾, LIMIT
more : SELECT, FILTER, LIMIT only (支持 TABLESAMPLE 和虛擬字段)

hive.fetch.task.conversion.threshold：控制適用這個優化的數據量，默認1G。

TablePropertyEnrichmentOptimizer

默認關閉，除了表的 properties 外，可以獲取表的SerDe 的屬性信息，然后都放到表的properties 中。
Serde 由參數 hive.optimize.update.table.properties.from.serde.list 控制，默認只有 AvroSerDe。

HiveOpConverterPostProc

以下3個條件都符合才執行，returnPathEnabled 默認 false，所以會不執行。

final boolean cboEnabled = HiveConf.getBoolVar(pctx.getConf(), HiveConf.ConfVars.HIVE_CBO_ENABLED);final boolean returnPathEnabled = HiveConf.getBoolVar(pctx.getConf(), HiveConf.ConfVars.HIVE_CBO_RETPATH_HIVEOP);final boolean cboSucceeded = pctx.getContext().isCboSucceeded();