[
https://issues.apache.org/jira/browse/HIVE-30221?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denys Kuzmenko updated HIVE-30221:
----------------------------------
Description:
A vectorized ReduceSink serializes every row (BinarySortable key, LazyBinary
value) and writes one Tez record per row. On TPC-DS q78 at 10 TB, 77% of LLAP
CPU goes to the exchange
(ReduceSink 44%, Tez writer 33%). Trino spends 4-6% on the same rows: its
PagePartitioner appends a page's rows to a page per partition and ships pages.
On an unsorted edge the rows need no
key order, so they can travel the same way.
*Fix* ({{hive.vectorized.execution.columnar.shuffle.enabled}}, default false)
* VectorReduceSinkColumnarOperator gathers the rows of each partition in
column vectors and writes one record per block.
** A block holds 32 MB / (partitions x row bytes) rows, kept within [64,
1024].
** A batch is appended in slices of a block's rows. Each slice takes two
passes: slots per partition, then a typed scatter per column. Trino's row-wise
append works the same way below 2 x
partitions rows per page.
* A row goes to the partition the row path gives it, computed by the row
path's code.
** The operator is a VectorReduceSinkUniformHashOperator and builds the key
series of the ReduceSink it replaces.
** A key gets the Murmur3 hash of its BinarySortable bytes, or the null key's
hash when all its columns are null, modulo the edge's physical outputs.
** A block's HiveKey hash code is its partition, so a reducer's inputs
partition alike whether they arrive columnar or not.
* Block format: per column, a null flag per row and then the values.
** Supported values: long, double, timestamp and bytes.
** A decimal travels as its scale plus the unscaled longs up to 18 digits,
otherwise as BigInteger bytes, and can be read into either decimal vector.
* ReduceRecordSource decodes a block into the key and value columns of the
reducer's batch. Bytes are referenced in place; a reducer that keeps its
batches gets a copy.
* Vectorizer, after a TezWork is vectorized, picks the qualifying edges.
** The edge must be a CUSTOM_SIMPLE_EDGE into the reducer's vectorized input:
the big table of a dynamically partitioned hash join, or the input of an
unsorted GROUP BY.
** Its ReduceSink must be uniform-hash, have no TopN, and carry only key and
value column types a block supports (LONG, DOUBLE, BYTES, DECIMAL, DECIMAL_64,
TIMESTAMP).
** The small table of a dynamically partitioned hash join keeps the row path,
because its hash table loader reads serialized keys.
*Measured*
Cluster (TPC-DS 10 TB Parquet, 10 x r5d, same session, flag off vs on, the 44
queries with a columnar edge, min of 3):
* Total: 3,162.7 -> 2,885.4 s (-8.8%), with identical results.
* Faster: q78 -103.4 s, q65 -51.7 s, q5 -36.8 s, q14 -31.7 s, q97 -30.8 s.
* Slower: q75 +16.2 s, q64 +6.4 s.
* On 24 x r6id: q78 -40.2 s, q65 -15.2 s, q95 -5.7 s.
These runs used an earlier revision that handled fixed-width keys only,
before the key series. The other key types are not measured on the cluster yet.
Microbenchmarks, ns per row. The figures leave out the per-record Tez write,
which only the row path pays.
||Edge||Row path write||Columnar write||Row path read||Columnar read||
|q78 value shape, 1,009 partitions|66-68|19-22, plus 34 for the key series
hash|-|-|
|q11 year_total: 7 string keys and an int, a decimal value, 253
partitions|214-258|364-424|180-185|13-19|
For string-heavy keys the write CPU is about even, because the columnar path
serializes the keys for their hash and then copies them into the block.
Depends on HIVE-30223: its unsorted GROUP BY edge carries most of the
measured gain, and columnar_shuffle.q turns it on. HIVE-30133 follows; it
covers repeating partition columns and
empty-key ReduceSinks such as the Iceberg V3 DML delete edge.
> Vectorization: a ReduceSink of an unsorted edge ships its rows as columnar
> blocks
> ----------------------------------------------------------------------------------
>
> Key: HIVE-30221
> URL: https://issues.apache.org/jira/browse/HIVE-30221
> Project: Hive
> Issue Type: Improvement
> Reporter: Denys Kuzmenko
> Priority: Major
>
> A vectorized ReduceSink serializes every row (BinarySortable key, LazyBinary
> value) and writes one Tez record per row. On TPC-DS q78 at 10 TB, 77% of LLAP
> CPU goes to the exchange
> (ReduceSink 44%, Tez writer 33%). Trino spends 4-6% on the same rows: its
> PagePartitioner appends a page's rows to a page per partition and ships
> pages. On an unsorted edge the rows need no
> key order, so they can travel the same way.
>
>
>
>
>
> *Fix* ({{hive.vectorized.execution.columnar.shuffle.enabled}}, default
> false)
>
> * VectorReduceSinkColumnarOperator gathers the rows of each partition in
> column vectors and writes one record per block.
>
> ** A block holds 32 MB / (partitions x row bytes) rows, kept within [64,
> 1024].
>
> ** A batch is appended in slices of a block's rows. Each slice takes two
> passes: slots per partition, then a typed scatter per column. Trino's
> row-wise append works the same way below 2 x
> partitions rows per page.
>
>
> * A row goes to the partition the row path gives it, computed by the row
> path's code.
>
> ** The operator is a VectorReduceSinkUniformHashOperator and builds the key
> series of the ReduceSink it replaces.
>
> ** A key gets the Murmur3 hash of its BinarySortable bytes, or the null
> key's hash when all its columns are null, modulo the edge's physical outputs.
>
> ** A block's HiveKey hash code is its partition, so a reducer's inputs
> partition alike whether they arrive columnar or not.
>
> * Block format: per column, a null flag per row and then the values.
>
>
> ** Supported values: long, double, timestamp and bytes.
>
>
> ** A decimal travels as its scale plus the unscaled longs up to 18 digits,
> otherwise as BigInteger bytes, and can be read into either decimal vector.
>
> * ReduceRecordSource decodes a block into the key and value columns of the
> reducer's batch. Bytes are referenced in place; a reducer that keeps its
> batches gets a copy.
> * Vectorizer, after a TezWork is vectorized, picks the qualifying edges.
>
>
> ** The edge must be a CUSTOM_SIMPLE_EDGE into the reducer's vectorized
> input: the big table of a dynamically partitioned hash join, or the input of
> an unsorted GROUP BY.
> ** Its ReduceSink must be uniform-hash, have no TopN, and carry only key
> and value column types a block supports (LONG, DOUBLE, BYTES, DECIMAL,
> DECIMAL_64, TIMESTAMP).
> ** The small table of a dynamically partitioned hash join keeps the row
> path, because its hash table loader reads serialized keys.
> *Measured*
>
>
> Cluster (TPC-DS 10 TB Parquet, 10 x r5d, same session, flag off vs on, the
> 44 queries with a columnar edge, min of 3):
>
> * Total: 3,162.7 -> 2,885.4 s (-8.8%), with identical results.
>
>
> * Faster: q78 -103.4 s, q65 -51.7 s, q5 -36.8 s, q14 -31.7 s, q97 -30.8 s.
>
>
> * Slower: q75 +16.2 s, q64 +6.4 s.
>
>
> * On 24 x r6id: q78 -40.2 s, q65 -15.2 s, q95 -5.7 s.
>
>
> These runs used an earlier revision that handled fixed-width keys only,
> before the key series. The other key types are not measured on the cluster
> yet.
>
>
>
> Microbenchmarks, ns per row. The figures leave out the per-record Tez
> write, which only the row path pays.
>
> ||Edge||Row path write||Columnar write||Row path read||Columnar read||
>
>
> |q78 value shape, 1,009 partitions|66-68|19-22, plus 34 for the key series
> hash|-|-|
>
> |q11 year_total: 7 string keys and an int, a decimal value, 253
> partitions|214-258|364-424|180-185|13-19|
>
> For string-heavy keys the write CPU is about even, because the columnar
> path serializes the keys for their hash and then copies them into the block.
>
> Depends on HIVE-30223: its unsorted GROUP BY edge carries most of the
> measured gain, and columnar_shuffle.q turns it on. HIVE-30133 follows; it
> covers repeating partition columns and
> empty-key ReduceSinks such as the Iceberg V3 DML delete edge.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
