Skip to content

Commit 6569711

Browse files
avamingliavamingli
committed
Enable Parallel Hash Left Anti Semi (Not-In) Join(parallel-oblivious).
This is a parallel-oblivious parallel hash join, that each inner side table would be duplicately processed without a shared hash table. We could benefit from parallel if the outer table is large and inner table is relatively small. See [1] below for example DDL and DML. Non-parallel plan: explain(analyze, costs off) select sum(t1.c1) from t1 where c1 not in (select c1 from t2); QUERY PLAN ------------------------------------------------------------------------------------------------------------------------ Finalize Aggregate (actual time=1808.872..1808.875 rows=1 loops=1) -> Gather Motion 3:1 (slice1; segments: 3) (actual time=1745.235..1808.858 rows=3 loops=1) -> Partial Aggregate (actual time=1808.622..1808.625 rows=1 loops=1) -> Hash Left Anti Semi (Not-In) Join (actual time=2.890..1583.005 rows=1667434 loops=1) Hash Cond: (t1.c1 = t2.c1) Extra Text: (seg2) Hash chain length 1.0 avg, 2 max, using 1199 of 524288 buckets. -> Seq Scan on t1 (actual time=0.355..678.531 rows=1667832 loops=1) -> Hash (actual time=2.068..2.069 rows=1200 loops=1) Buckets: 524288 Batches: 1 Memory Usage: 4139kB -> Broadcast Motion 3:3 (slice2; segments: 3) (actual time=1.476..1.772 rows=1200 loops=1) -> Seq Scan on t2 (actual time=0.356..0.499 rows=407 loops=1) Planning Time: 0.454 ms (slice0) Executor memory: 124K bytes. (slice1) Executor memory: 4443K bytes avg x 3x(0) workers, 4443K bytes max (seg0). Work_mem: 4139K bytes max. (slice2) Executor memory: 262K bytes avg x 3x(0) workers, 262K bytes max (seg0). Memory used: 128000kB Optimizer: Postgres query optimizer Execution Time: 1809.517 ms (18 rows) Time: 1810.827 ms (00:01.811) Parallel plan: explain(analyze, costs off) select sum(t1.c1) from t1 where c1 not in (select c1 from t2); QUERY PLAN ------------------------------------------------------------------------------------------------------------------------ Finalize Aggregate (actual time=758.707..758.710 rows=1 loops=1) -> Gather Motion 6:1 (slice1; segments: 6) (actual time=668.747..758.685 rows=6 loops=1) -> Partial Aggregate (actual time=752.479..752.483 rows=1 loops=1) -> Hash Left Anti Semi (Not-In) Join (actual time=3.010..565.127 rows=833732 loops=1) Hash Cond: (t1.c1 = t2.c1) Extra Text: (seg2) Hash chain length 1.0 avg, 2 max, using 1199 of 524288 buckets. -> Parallel Seq Scan on t1 (actual time=0.368..231.049 rows=833932 loops=1) -> Hash (actual time=2.148..2.149 rows=1200 loops=1) Buckets: 524288 Batches: 1 Memory Usage: 4139kB -> Broadcast Motion 3:6 (slice2; segments: 3) (actual time=0.203..1.779 rows=1200 loops=1) -> Seq Scan on t2 (actual time=0.361..0.499 rows=407 loops=1) Planning Time: 0.470 ms (slice0) Executor memory: 124K bytes. (slice1) Executor memory: 4483K bytes avg x 6x(0) workers, 4483K bytes max (seg0). Work_mem: 4139K bytes max. (slice2) Executor memory: 262K bytes avg x 3x(0) workers, 262K bytes max (seg0). Memory used: 128000kB Optimizer: Postgres query optimizer Execution Time: 759.440 ms (18 rows) Time: 760.874 ms [1] Example: create table t1(c1 int, c2 int) using ao_row distributed by (c1); create table t2(c1 int, c2 int) using ao_row distributed by (c1); set enable_parallel = on; set gp_appendonly_insert_files = 2; set gp_appendonly_insert_files_tuples_range = 100; set max_parallel_workers_per_gather = 2; insert into t1 select i, i from generate_series(1, 5000000) i; insert into t2 select i+1, i from generate_series(1, 1200) i; analyze t1; analyze t2; Authored-by: Zhang Mingli avamingli@gmail.com
1 parent fc90fec commit 6569711

3 files changed

Lines changed: 109 additions & 1 deletion

File tree

‎src/backend/optimizer/path/joinpath.c‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2299,7 +2299,6 @@ hash_inner_and_outer(PlannerInfo *root,
22992299
save_jointype != JOIN_UNIQUE_OUTER &&
23002300
save_jointype != JOIN_FULL &&
23012301
save_jointype != JOIN_RIGHT &&
2302-
save_jointype != JOIN_LASJ_NOTIN &&
23032302
save_jointype != JOIN_DEDUP_SEMI &&
23042303
save_jointype != JOIN_DEDUP_SEMI_REVERSE &&
23052304
outerrel->partial_pathlist != NIL &&
@@ -2319,6 +2318,7 @@ hash_inner_and_outer(PlannerInfo *root,
23192318
*/
23202319
if (innerrel->partial_pathlist != NIL &&
23212320
save_jointype != JOIN_UNIQUE_INNER &&
2321+
save_jointype != JOIN_LASJ_NOTIN &&
23222322
enable_parallel_hash)
23232323
{
23242324
cheapest_partial_inner =

‎src/test/regress/expected/gp_parallel.out‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1513,6 +1513,81 @@ select * from t1 order by c2 asc limit 3 offset 5;
15131513

15141514
abort;
15151515
--
1516+
-- Test Parallel Hash Left Anti Semi (Not-In) Join(parallel-oblivious).
1517+
--
1518+
create table t1(c1 int, c2 int) using ao_row distributed by (c1);
1519+
create table t2(c1 int, c2 int) using ao_row distributed by (c1);
1520+
create table t3_null(c1 int, c2 int) using ao_row distributed by (c1);
1521+
set enable_parallel = on;
1522+
set gp_appendonly_insert_files = 2;
1523+
set gp_appendonly_insert_files_tuples_range = 100;
1524+
set max_parallel_workers_per_gather = 2;
1525+
insert into t1 select i, i from generate_series(1, 5000000) i;
1526+
insert into t2 select i+1, i from generate_series(1, 1200) i;
1527+
insert into t3_null select i+1, i from generate_series(1, 1200) i;
1528+
insert into t3_null values(NULL, NULL);
1529+
analyze t1;
1530+
analyze t2;
1531+
analyze t3_null;
1532+
explain(costs off) select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
1533+
QUERY PLAN
1534+
---------------------------------------------------------------------------
1535+
Finalize Aggregate
1536+
-> Gather Motion 6:1 (slice1; segments: 6)
1537+
-> Partial Aggregate
1538+
-> Hash Left Anti Semi (Not-In) Join
1539+
Hash Cond: (t1.c1 = t2.c1)
1540+
-> Parallel Seq Scan on t1
1541+
-> Hash
1542+
-> Broadcast Motion 3:6 (slice2; segments: 3)
1543+
-> Seq Scan on t2
1544+
Optimizer: Postgres query optimizer
1545+
(10 rows)
1546+
1547+
select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
1548+
sum
1549+
----------------
1550+
12500001778200
1551+
(1 row)
1552+
1553+
explain(costs off) select * from t1 where c1 not in (select c1 from t3_null);
1554+
QUERY PLAN
1555+
---------------------------------------------------------------
1556+
Gather Motion 6:1 (slice1; segments: 6)
1557+
-> Hash Left Anti Semi (Not-In) Join
1558+
Hash Cond: (t1.c1 = t3_null.c1)
1559+
-> Parallel Seq Scan on t1
1560+
-> Hash
1561+
-> Broadcast Motion 3:6 (slice2; segments: 3)
1562+
-> Seq Scan on t3_null
1563+
Optimizer: Postgres query optimizer
1564+
(8 rows)
1565+
1566+
select * from t1 where c1 not in (select c1 from t3_null);
1567+
c1 | c2
1568+
----+----
1569+
(0 rows)
1570+
1571+
-- non-parallel results.
1572+
set enable_parallel = off;
1573+
select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
1574+
sum
1575+
----------------
1576+
12500001778200
1577+
(1 row)
1578+
1579+
select * from t1 where c1 not in (select c1 from t3_null);
1580+
c1 | c2
1581+
----+----
1582+
(0 rows)
1583+
1584+
drop table t1;
1585+
drop table t2;
1586+
drop table t3_null;
1587+
--
1588+
-- End of Test Parallel Hash Left Anti Semi (Not-In) Join.
1589+
--
1590+
--
15161591
-- Test alter ao/aocs table parallel_workers options
15171592
--
15181593
begin;

‎src/test/regress/sql/gp_parallel.sql‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -441,6 +441,39 @@ set local enable_parallel = off;
441441
explain(costs off, locus) select * from t1 order by c2 asc limit 3 offset 5;
442442
select * from t1 order by c2 asc limit 3 offset 5;
443443
abort;
444+
445+
--
446+
-- Test Parallel Hash Left Anti Semi (Not-In) Join(parallel-oblivious).
447+
--
448+
create table t1(c1 int, c2 int) using ao_row distributed by (c1);
449+
create table t2(c1 int, c2 int) using ao_row distributed by (c1);
450+
create table t3_null(c1 int, c2 int) using ao_row distributed by (c1);
451+
set enable_parallel = on;
452+
set gp_appendonly_insert_files = 2;
453+
set gp_appendonly_insert_files_tuples_range = 100;
454+
set max_parallel_workers_per_gather = 2;
455+
insert into t1 select i, i from generate_series(1, 5000000) i;
456+
insert into t2 select i+1, i from generate_series(1, 1200) i;
457+
insert into t3_null select i+1, i from generate_series(1, 1200) i;
458+
insert into t3_null values(NULL, NULL);
459+
analyze t1;
460+
analyze t2;
461+
analyze t3_null;
462+
explain(costs off) select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
463+
select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
464+
explain(costs off) select * from t1 where c1 not in (select c1 from t3_null);
465+
select * from t1 where c1 not in (select c1 from t3_null);
466+
-- non-parallel results.
467+
set enable_parallel = off;
468+
select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
469+
select * from t1 where c1 not in (select c1 from t3_null);
470+
drop table t1;
471+
drop table t2;
472+
drop table t3_null;
473+
--
474+
-- End of Test Parallel Hash Left Anti Semi (Not-In) Join.
475+
--
476+
444477
--
445478
-- Test alter ao/aocs table parallel_workers options
446479
--

0 commit comments

Comments
 (0)