Why is my group by pipelined much slower than a group by hash?

abk

Hi all,
I was working on optimizing a slow query that involves joining a 6B rows table with a 22M rows table and then doing a group by. I made new projections that got my join to be merge and my group by to be pipelined and avoiding global resegments but my query got much slower (as in, instead of 30 mins it became so slow I killed it after a few hours).

I've reproduced this behavior on a simpler query that only does a group by. When I let vertica use the optimized projection it comes up with a group by pipelined and it's slow enough that I killed it after 30+ mins. If I use the same query and force it to use a group by hash, the query completes in under 2 mins.

I did notice that the pipelined query uses much fewer resources but I was also expecting the query to run faster. Am I doing something wrong? Is the group by hash in this case supposed to be faster than pipelined and if so, why?

Here is a minimal version of the table, projection and queries that shows the problem, as well as the EXPLAIN VERBOSE for both.

Edit: This is using vertica 9.01

Any insights?
Thanks!

Table:
CREATE TABLE group_by_test_table
(
id IDENTITY ,
col1 varchar(80) DEFAULT NULL,
col2 varchar(80) DEFAULT NULL,
col3 varchar(80) DEFAULT NULL,
user_id int DEFAULT NULL::int,
created_time timestamp DEFAULT NULL::timestamp
);

Projection I made for the optimizer to use a GROUP BY PIPELINED and avoid RESEGMENTs:
CREATE PROJECTION group_by_test_table_projection
(
id,
col1,
col2,
col3,
user_id,
created_time
)
AS
SELECT group_by_test_table.id,
group_by_test_table.col1,
group_by_test_table.col2,
group_by_test_table.col3,
group_by_test_table.user_id,
group_by_test_table.created_time
FROM group_by_test_table
ORDER BY group_by_test_table.col2,
group_by_test_table.col3,
group_by_test_table.col1,
group_by_test_table.user_id
SEGMENTED BY hash(group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1) ALL NODES KSAFE 1;

Query for group by pipelined (that hadn't yet finished after 30+ mins):
select col2, col3, col1, count(distinct user_id) from group_by_test_table group by 1, 2, 3 order by 1, 2, 3

Query for group by hash (that completes in under 2 mins):
select col2, col3, col1, count(distinct user_id) from group_by_test_table group by /*+ GBYTYPE(HASH) */ 1, 2, 3 order by 1, 2, 3;

EXPLAIN VERBOSE for first (slow) query:
QUERY PLAN DESCRIPTION:

---

Opt Vertica Options

---

PLAN_OUTPUT_SUPER_VERBOSE

explain verbose select col2, col3, col1, count(distinct user_id) from group_by_test_table group by 1, 2, 3 order by 1, 2, 3

Access Path:
Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
LDISTRIB_UNSEGMENTED
+-GROUPBY PIPELINED [Cost: 14070115.000000, Rows: 1907699.000000 Disk(B): 0.000000 CPU(B): 518894128.000000 Memory(B): 518894128.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 2)
| Aggregates: count(DISTINCT group_by_test_table.user_id)
| Group By: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1
| Execute on: All Nodes
| Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
| LDISTRIB_UNSEGMENTED
| +---> GROUPBY PIPELINED (LOCAL RESEGMENT GROUPS) [Cost: 14066314.000000, Rows: 1907699.000000 Disk(B): 0.000000 CPU(B): 1847469740608.000000 Memory(B): 518894128.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 3)
| | Group By: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id
| | Execute on: All Nodes
| | Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
| | LDISTRIB_SEGMENTED
| | +---> STORAGE ACCESS for group_by_test_table [Cost: 9553356.000000, Rows: 6792168164.000000 Disk(B): 0.000000 CPU(B): 0.000000 Memory(B): 0.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 4)
| | | Column Cost Aspects: [ Disk(B): 54413885440.000000 CPU(B): 0.000000 Memory(B): 1684457704672.000000 Netwrk(B): 0.000000 Parallelism: 10.000000 ]
| | | Projection: group_by_test_table_projection_b0
| | | Materialize: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id
| | | Execute on: All Nodes
| | | Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
| | | LDISTRIB_SEGMENTED

EXPLAIN VERBOSE for second (fast) query:
QUERY PLAN DESCRIPTION:

---

Opt Vertica Options

---

PLAN_OUTPUT_SUPER_VERBOSE

explain verbose select col2, col3, col1, count(distinct user_id) from group_by_test_table group by /*+ GBYTYPE(HASH) */ 1, 2, 3 order by 1, 2, 3

Access Path:
Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1)
LDISTRIB_UNSEGMENTED
+-GROUPBY HASH (SORT OUTPUT) (LOCAL RESEGMENT GROUPS) [Cost: 14102308.000000, Rows: 1907699.000000 Disk(B): 549417312.000000 CPU(B): 518894128.000000 Memory(B): 1068311440.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 2)
| Aggregates: count(DISTINCT group_by_test_table.user_id)
| Group By: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1
| Execute on: All Nodes
| Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1)
| LDISTRIB_SEGMENTED
| +---> GROUPBY HASH (LOCAL RESEGMENT GROUPS) [Cost: 14082411.000000, Rows: 1907699.000000 Disk(B): 549417312.000000 CPU(B): 1847469740608.000000 Memory(B): 1068311440.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 3)
| | Group By: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id
| | Execute on: All Nodes
| | Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
| | LDISTRIB_SEGMENTED
| | +---> STORAGE ACCESS for group_by_test_table [Cost: 9553356.000000, Rows: 6792168164.000000 Disk(B): 0.000000 CPU(B): 0.000000 Memory(B): 0.000000 Netwrk(B): 0.000000 Parallelism: 10.000000] [OutRowSz (B): 272] (PATH ID: 4)
| | | Column Cost Aspects: [ Disk(B): 54413885440.000000 CPU(B): 0.000000 Memory(B): 1684457704672.000000 Netwrk(B): 0.000000 Parallelism: 10.000000 ]
| | | Projection: group_by_test_table_projection_b0
| | | Materialize: group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id
| | | Execute on: All Nodes
| | | Sort Key: (group_by_test_table.col2, group_by_test_table.col3, group_by_test_table.col1, group_by_test_table.user_id)
| | | LDISTRIB_SEGMENTED

emoreno

Hi,
In that version of vertica probably the group by pipe is Single threaded. In 9.1 we introduce an other operator resegment merge that should multi thread the group by pipe.
You can see the number of threads in the profile counting distinct operator Id in that path.
Hope thishelps.
Eugenia

emoreno

abk

I see, thanks for the explanation. For now I'll stick with the group by hash and see when I can upgrade to 9.1 to test that.

emoreno

Sounds good. And in the join, check the query events sometimes you are not able to to push the group by under the join because late materialization. Just something to look and try.

emoreno

One more thing . You can use direct query to save the plans with the hints so you don't have to write your query always with them.

abk

Thanks again for all the help! I'll take a look at those.

abk

Hi!

Not sure if correct forum etiquette is to update my post with a follow-up or create a new post so let me know if I should make a new one

We've upgraded our vertica cluster to 9.2.1 and I still observe the same behavior: this time a query on the same table of 7billion rows with a group by and 3 sums finishes in about 8 minutes if I let the query planner use a GROUP BY PIPELINED and about 50 seconds if I just add a /*+ GBYTYPE(HASH) */ to the exact same query.

I've counted distinct operators for the group by path and do see more than 1 distinct ids, however I don't see any mention of the "resegment merge" operator in the plan or profile but not sure if I'm supposed to.

Is there anything else I can do to confirm if this new operator is being used or force it to be used? When looking at threads used through the management console's query monitor I do see the hash query using many times more threads.