[
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.
Depends on HIVE-30221 (columnar shuffle)
was:
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.
> 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.
> Depends on HIVE-30221 (columnar shuffle)
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
