[ https://issues.apache.org/jira/browse/HIVE-22666?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Krisztian Kasa updated HIVE-22666: ---------------------------------- Status: Open (was: Patch Available) > Introduce TopNKey operator for PTF Reduce Sink > ---------------------------------------------- > > Key: HIVE-22666 > URL: https://issues.apache.org/jira/browse/HIVE-22666 > Project: Hive > Issue Type: Improvement > Reporter: Krisztian Kasa > Assignee: Krisztian Kasa > Priority: Major > Attachments: HIVE-22666.1.patch, HIVE-22666.2.patch, > HIVE-22666.3.patch, HIVE-22666.3.patch, HIVE-22666.4.patch, > HIVE-22666.4.patch, HIVE-22666.4.patch > > > {code} > EXPLAIN EXTENDED > SELECT s_state, ranking > FROM ( > SELECT s_state AS s_state, > rank() OVER (PARTITION BY s_state ORDER BY ss_net_profit) AS ranking > FROM testtable_n1000) tmp1 > WHERE ranking <= 3; > {code} > {code} > STAGE DEPENDENCIES: > Stage-1 is a root stage > Stage-0 depends on stages: Stage-1 > STAGE PLANS: > Stage: Stage-1 > Tez > #### A masked pattern was here #### > Edges: > Reducer 2 <- Map 1 (SIMPLE_EDGE) > #### A masked pattern was here #### > Vertices: > Map 1 > Map Operator Tree: > TableScan > alias: testtable_n1000 > Statistics: Num rows: 10 Data size: 940 Basic stats: > COMPLETE Column stats: COMPLETE > GatherStats: false > Reduce Output Operator > key expressions: s_state (type: string), ss_net_profit > (type: double) > null sort order: az > sort order: ++ > Map-reduce partition columns: s_state (type: string) > Statistics: Num rows: 10 Data size: 940 Basic stats: > COMPLETE Column stats: COMPLETE > tag: -1 > TopN: 4 > TopN Hash Memory Usage: 0.1 > auto parallelism: true > Execution mode: vectorized, llap > LLAP IO: no inputs > Path -> Alias: > #### A masked pattern was here #### > Path -> Partition: > #### A masked pattern was here #### > Partition > base file name: testtable_n1000 > input format: org.apache.hadoop.mapred.TextInputFormat > output format: > org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat > properties: > COLUMN_STATS_ACCURATE > {"BASIC_STATS":"true","COLUMN_STATS":{"s_state":"true","ss_net_profit":"true"}} > bucket_count -1 > bucketing_version 2 > column.name.delimiter , > columns s_state,ss_net_profit > columns.comments > columns.types string:double > #### A masked pattern was here #### > name default.testtable_n1000 > numFiles 1 > numRows 10 > rawDataSize 80 > serialization.ddl struct testtable_n1000 { string > s_state, double ss_net_profit} > serialization.format 1 > serialization.lib > org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > totalSize 90 > #### A masked pattern was here #### > serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > > input format: org.apache.hadoop.mapred.TextInputFormat > output format: > org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat > properties: > COLUMN_STATS_ACCURATE > {"BASIC_STATS":"true","COLUMN_STATS":{"s_state":"true","ss_net_profit":"true"}} > bucket_count -1 > bucketing_version 2 > column.name.delimiter , > columns s_state,ss_net_profit > columns.comments > columns.types string:double > #### A masked pattern was here #### > name default.testtable_n1000 > numFiles 1 > numRows 10 > rawDataSize 80 > serialization.ddl struct testtable_n1000 { string > s_state, double ss_net_profit} > serialization.format 1 > serialization.lib > org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > totalSize 90 > #### A masked pattern was here #### > serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > name: default.testtable_n1000 > name: default.testtable_n1000 > Truncated Path -> Alias: > /testtable_n1000 [testtable_n1000] > Reducer 2 > Execution mode: vectorized, llap > Needs Tagging: false > Reduce Operator Tree: > Select Operator > expressions: KEY.reducesinkkey0 (type: string), > KEY.reducesinkkey1 (type: double) > outputColumnNames: _col0, _col1 > Statistics: Num rows: 10 Data size: 3620 Basic stats: > COMPLETE Column stats: COMPLETE > PTF Operator > Function definitions: > Input definition > input alias: ptf_0 > output shape: _col0: string, _col1: double > type: WINDOWING > Windowing table definition > input alias: ptf_1 > name: windowingtablefunction > order by: _col1 ASC NULLS LAST > partition by: _col0 > raw input shape: > window functions: > window function definition > alias: rank_window_0 > arguments: _col1 > name: rank > window function: GenericUDAFRankEvaluator > window frame: ROWS PRECEDING(MAX)~FOLLOWING(MAX) > isPivotResult: true > Statistics: Num rows: 10 Data size: 3620 Basic stats: > COMPLETE Column stats: COMPLETE > Filter Operator > isSamplingPred: false > predicate: (rank_window_0 <= 3) (type: boolean) > Statistics: Num rows: 3 Data size: 1086 Basic stats: > COMPLETE Column stats: COMPLETE > Select Operator > expressions: _col0 (type: string), rank_window_0 (type: > int) > outputColumnNames: _col0, _col1 > Statistics: Num rows: 3 Data size: 270 Basic stats: > COMPLETE Column stats: COMPLETE > File Output Operator > compressed: false > GlobalTableId: 0 > #### A masked pattern was here #### > NumFilesPerFileSink: 1 > Statistics: Num rows: 3 Data size: 270 Basic stats: > COMPLETE Column stats: COMPLETE > #### A masked pattern was here #### > table: > input format: > org.apache.hadoop.mapred.SequenceFileInputFormat > output format: > org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat > properties: > columns _col0,_col1 > columns.types string:int > escape.delim \ > > hive.serialization.extend.additional.nesting.levels true > serialization.escape.crlf true > serialization.format 1 > serialization.lib > org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > serde: > org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > TotalFiles: 1 > GatherStats: false > MultiFileSpray: false > Stage: Stage-0 > Fetch Operator > limit: -1 > Processor Tree: > ListSink > {code} > In this case the topN value (3+1) will be pushed to the ReduceSink (Reduce > Output Operator) operator in Map 1 > https://github.com/apache/hive/blob/520aa19b20381bfd2ed25c835443c013f6e6ebb9/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java#L257 > ReduceSink operator uses PTFTopNHash to get the topN rows for each partition > key (s_state) value. > The goals of this jira are: > - implement supporting partitioning in TopNKeyOperator > - enable push down of partitioned TopNKeyOperator -- This message was sent by Atlassian Jira (v8.3.4#803005)