[
https://issues.apache.org/jira/browse/HIVE-30133?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Denys Kuzmenko updated HIVE-30133:
----------------------------------
Description:
On the columnar shuffle (hive.vectorized.execution.columnar.shuffle.enabled)
a ReduceSink of an unsorted edge gathers its rows per partition and ships them
as column blocks. Two gaps:
* a column that repeats across a batch is flattened, so every row carries its
value again;
* an empty-key ReduceSink partitioned by other columns
(VectorReduceSinkObjectHashOperator) never takes the columnar shuffle. The
DISTRIBUTE BY FILE__PATH of an Iceberg V3 merge-on-read
DELETE, UPDATE or MERGE is one: every deleted row is serialized with its
FILE__PATH, PARTITION__SPEC__ID and PARTITION__KEY into a Tez record of its
own.
When every partition column of a batch repeats, all its rows go to one
reducer, so the batch can travel as one block with each repeating column
written once.
*Fix*
* Block format: per-column flags HAS_NULLS and REPEATING; a repeating column
travels as its one value; the selected rows of a batch encode without a gather.
* A batch whose rows all go to one partition (64 rows or more) leaves as a
block of its own; the other rows gather per partition as before.
* VectorReduceSinkColumnarObjectHashOperator: the columnar shuffle of an
empty-key ReduceSink with partition columns and no buckets. Rows go to the
partitions the row path gives them, hashed
once per batch when every partition column repeats.
* Row path: the ObjectHash ReduceSink computes the partition hashCode once
per batch when every partition column repeats.
*Measured* (local, the V3 DELETE ReduceSink: 150 B FILE__PATH, spec id,
partition key, position; 1,024-row batches)
||Path||ns per row||Value bytes per row||Tez records||
|row path, master|103.0|168|one per row|
|row path, hash once per batch|46.9|168|one per row|
|columnar shuffle|0.30|8.4|one per batch|
Reducer decode on the columnar shuffle: 0.16 ns/row. Not timed on the
cluster; the gain scales with the share of rows a DML writes.
was:
VectorReduceSinkObjectHashOperator extracts and hashes the partition columns
for every row. When every partition column repeats across the batch, every row
gets the same hashCode, so it is now computed once per batch. The bucket
number and _bucket_number are still computed per row.
Iceberg V3 merge-on-read DELETE distributes by FILE__PATH, which repeats within
a batch. In a microbenchmark of the operator: 107.7 -> 53.4 ns/row for a
repeating 150-byte string key, -24% for a repeating bigint, unchanged when the
key does not repeat.
> Columnar shuffle: a batch with repeating partition columns travels as one
> block
> -------------------------------------------------------------------------------
>
> Key: HIVE-30133
> URL: https://issues.apache.org/jira/browse/HIVE-30133
> Project: Hive
> Issue Type: Bug
> Components: Iceberg integration
> Reporter: Denys Kuzmenko
> Priority: Major
> Labels: pull-request-available
>
> On the columnar shuffle
> (hive.vectorized.execution.columnar.shuffle.enabled) a ReduceSink of an
> unsorted edge gathers its rows per partition and ships them as column blocks.
> Two gaps:
> * a column that repeats across a batch is flattened, so every row carries
> its value again;
>
> * an empty-key ReduceSink partitioned by other columns
> (VectorReduceSinkObjectHashOperator) never takes the columnar shuffle. The
> DISTRIBUTE BY FILE__PATH of an Iceberg V3 merge-on-read
> DELETE, UPDATE or MERGE is one: every deleted row is serialized with its
> FILE__PATH, PARTITION__SPEC__ID and PARTITION__KEY into a Tez record of its
> own.
>
>
>
> When every partition column of a batch repeats, all its rows go to one
> reducer, so the batch can travel as one block with each repeating column
> written once.
>
>
>
> *Fix*
>
>
> * Block format: per-column flags HAS_NULLS and REPEATING; a repeating
> column travels as its one value; the selected rows of a batch encode without
> a gather.
> * A batch whose rows all go to one partition (64 rows or more) leaves as a
> block of its own; the other rows gather per partition as before.
>
> * VectorReduceSinkColumnarObjectHashOperator: the columnar shuffle of an
> empty-key ReduceSink with partition columns and no buckets. Rows go to the
> partitions the row path gives them, hashed
> once per batch when every partition column repeats.
>
>
> * Row path: the ObjectHash ReduceSink computes the partition hashCode once
> per batch when every partition column repeats.
>
>
>
>
> *Measured* (local, the V3 DELETE ReduceSink: 150 B FILE__PATH, spec id,
> partition key, position; 1,024-row batches)
>
> ||Path||ns per row||Value bytes per row||Tez records||
>
>
> |row path, master|103.0|168|one per row|
>
>
> |row path, hash once per batch|46.9|168|one per row|
>
>
> |columnar shuffle|0.30|8.4|one per batch|
>
>
> Reducer decode on the columnar shuffle: 0.16 ns/row. Not timed on the
> cluster; the gain scales with the share of rows a DML writes.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
