Implement Parallel-aware Hash Left Anti Semi (Not-In) Join

src/backend/cdb/cdbpath.c

@@ -2980,13 +2980,6 @@ cdbpath_motion_for_parallel_join(PlannerInfo *root,
 	outer.ok_to_replicate = !outer.has_wts;
 	inner.ok_to_replicate = true;
 
-	/*
-	 * For parallel mode, join is executed by each batches.
-	 * It is hard to tell whether null exists in the whole table.
-	 */
-	if (parallel_aware && jointype == JOIN_LASJ_NOTIN)
-		goto fail;
-
 	switch (jointype)
 	{
 		case JOIN_INNER:

src/backend/executor/nodeHash.c

@@ -324,6 +324,14 @@ MultiExecParallelHash(HashState *node)
 			{
 				bool		hashkeys_null = false;
 
+				/* CBDB_PARALLEL: Siblings must have found null value. */
+				if (pstate->phs_lasj_has_null)
+				{
+					node->hs_hashkeys_null = true;
+					ExecSquelchNode(outerNode);
+					break;
+				}
+
 				slot = ExecProcNode(outerNode);
 				if (TupIsNull(slot))
 					break;
@@ -333,14 +341,40 @@ MultiExecParallelHash(HashState *node)
 										 &hashvalue, &hashkeys_null))
 					ExecParallelHashTableInsert(hashtable, slot, hashvalue);
 				hashtable->partialTuples++;
+
+				if (node->hs_quit_if_hashkeys_null && hashkeys_null)
+				{
+					/* CBDB_PARALLEL:
+					 * If we are LASJ and found NULL value by ourself or sibling processes had
+					 * found NULL values, quit and tell siblings to quit if possible.
+					 *
+					 * It's safe to fetch and set phs_lasj_has_null without lock here and at
+					 * other places. As it's a atomic boolean value. And we should avoid more locks in HashJion Impl.
+					 * If other processes miss it here and some others set it at the same time, just bypass
+					 * and we may get it at the next Hash batch.
+					 * If we missed it across all batches, we will know it when PHJ_BUILD_HASHING_INNER
+					 * ends with the help of build_barrier.
+					 * If we never participated in building hash table, check it when hash table
+					 * creation job is finished.
+					 */
+					pstate->phs_lasj_has_null = true;
+					pg_write_barrier();
+					node->hs_hashkeys_null = true;
+					ExecSquelchNode(outerNode);
+					break;
+				}
 			}
 
+			/* CBDB_PARALLEL: No need to flush tuples if phs_lasj_has_null. */
 			/*
 			 * Make sure that any tuples we wrote to disk are visible to
 			 * others before anyone tries to load them.
 			 */
-			for (i = 0; i < hashtable->nbatch; ++i)
-				sts_end_write(hashtable->batches[i].inner_tuples);
+			if (!pstate->phs_lasj_has_null)
+			{
+				for (i = 0; i < hashtable->nbatch; ++i)
+					sts_end_write(hashtable->batches[i].inner_tuples);
+			}
 
 			/*
 			 * Update shared counters.  We need an accurate total tuple count
@@ -366,9 +400,23 @@ MultiExecParallelHash(HashState *node)
 				 * skew).
 				 */
 				pstate->growth = PHJ_GROWTH_DISABLED;
+				/* In case we didn't find null values ourself. */
+				if (pstate->phs_lasj_has_null)
+				{
+					node->hs_hashkeys_null = true;
+					return;
+				}
 			}
 	}
 
+	/* In case we didn't participate in PHJ_BUILD_HASHING_INNER */
+	pg_memory_barrier();
+	if (pstate->phs_lasj_has_null)
+	{
+		node->hs_hashkeys_null = true;
+		return;
+	}
+
 	/*
 	 * We're not yet attached to a batch.  We all agree on the dimensions and
 	 * number of inner tuples (for the empty table optimization).
@@ -3779,7 +3827,12 @@ ExecHashTableDetachBatch(HashJoinTable hashtable)
 		sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
 
 		/* Detach from the batch we were last working on. */
-		if (BarrierArriveAndDetach(&batch->batch_barrier))
+		/*
+		 * CBDB_PARALLEL: Parallel Hash Left Anti Semi (Not-In) Join(parallel-aware)
+		 * If phs_lasj_has_null is true, that means we have found null when building hash table,
+		 * there were no batches to detach.
+		 */
+		if (!hashtable->parallel_state->phs_lasj_has_null && BarrierArriveAndDetach(&batch->batch_barrier))
 		{
 			/*
 			 * Technically we shouldn't access the barrier because we're no

src/backend/executor/nodeHashjoin.c

@@ -2065,6 +2065,7 @@ ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
 
 	BarrierInit(&pstate->sync_barrier, pcxt->nworkers);
 	BarrierInit(&pstate->batch0_barrier, pcxt->nworkers);
+	pstate->phs_lasj_has_null = false;
 
 	/* Set up the space we'll use for shared temporary files. */
 	SharedFileSetInit(&pstate->fileset, pcxt->seg);

src/backend/optimizer/path/joinpath.c

@@ -2318,7 +2318,6 @@ hash_inner_and_outer(PlannerInfo *root,
 			 */
 			if (innerrel->partial_pathlist != NIL &&
 				save_jointype != JOIN_UNIQUE_INNER &&
-				save_jointype != JOIN_LASJ_NOTIN &&
 				enable_parallel_hash)
 			{
 				cheapest_partial_inner =

src/include/executor/hashjoin.h

@@ -286,6 +286,7 @@ typedef struct ParallelHashJoinState
 	Barrier		grow_buckets_barrier;
 	Barrier		sync_barrier;
 	Barrier		batch0_barrier;
+	volatile 	bool	phs_lasj_has_null; 	/* LASJ has found null value, identify early quit */
 	pg_atomic_uint32 distributor;	/* counter for load balancing */
 
 	SharedFileSet fileset;		/* space for shared temporary files */

src/test/regress/expected/gp_parallel.out

@@ -1728,10 +1728,12 @@ abort;
 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);
 create table t3_null(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;
+begin;
+set local enable_parallel = on;
+set local gp_appendonly_insert_files = 2;
+set local gp_appendonly_insert_files_tuples_range = 100;
+set local max_parallel_workers_per_gather = 2;
+set local enable_parallel_hash = off;
 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;
 insert into t3_null select i+1, i from generate_series(1, 1200) i;
@@ -1779,7 +1781,7 @@ select * from t1 where c1 not in (select c1 from t3_null);
 (0 rows)
 
 -- non-parallel results.
-set enable_parallel = off;
+set local enable_parallel = off;
 select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
       sum       
 ----------------
@@ -1791,13 +1793,89 @@ select * from t1 where c1 not in (select c1 from t3_null);
 ----+----
 (0 rows)
 
+end;
 drop table t1;
 drop table t2;
 drop table t3_null;
 --
 -- End of Test Parallel Hash Left Anti Semi (Not-In) Join.
 --
 --
+-- Test Parallel-aware Hash Left Anti Semi (Not-In) Join.
+--
+begin;
+create table t1(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+create table t2(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+create table t3_null(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+set local enable_parallel = on;
+set local max_parallel_workers_per_gather = 2;
+insert into t1 select i, i from generate_series(1, 500000) i;
+insert into t2 select i, i+1 from generate_series(1, 500000) i;
+insert into t3_null select i, i+1 from generate_series(1, 500000) i;
+insert into t3_null values(NULL, NULL);
+analyze t1;
+analyze t2;
+analyze t3_null;
+explain(costs off) select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+                                     QUERY PLAN                                     
+------------------------------------------------------------------------------------
+ Finalize Aggregate
+   ->  Gather Motion 6:1  (slice1; segments: 6)
+         ->  Partial Aggregate
+               ->  Parallel Hash Left Anti Semi (Not-In) Join
+                     Hash Cond: (t1.c1 = t2.c2)
+                     ->  Parallel Seq Scan on t1
+                     ->  Parallel Hash
+                           ->  Parallel Broadcast Motion 6:6  (slice2; segments: 6)
+                                 ->  Parallel Seq Scan on t2
+ Optimizer: Postgres query optimizer
+(10 rows)
+
+select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+ sum 
+-----
+   1
+(1 row)
+
+explain(costs off) select * from t1 where c1 not in (select c2 from t3_null);
+                               QUERY PLAN                               
+------------------------------------------------------------------------
+ Gather Motion 6:1  (slice1; segments: 6)
+   ->  Parallel Hash Left Anti Semi (Not-In) Join
+         Hash Cond: (t1.c1 = t3_null.c2)
+         ->  Parallel Seq Scan on t1
+         ->  Parallel Hash
+               ->  Parallel Broadcast Motion 6:6  (slice2; segments: 6)
+                     ->  Parallel Seq Scan on t3_null
+ Optimizer: Postgres query optimizer
+(8 rows)
+
+select * from t1 where c1 not in (select c2 from t3_null);
+ c1 | c2 
+----+----
+(0 rows)
+
+-- non-parallel results.
+set local enable_parallel = off;
+select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+ sum 
+-----
+   1
+(1 row)
+
+select * from t1 where c1 not in (select c2 from t3_null);
+ c1 | c2 
+----+----
+(0 rows)
+
+drop table t1;
+drop table t2;
+drop table t3_null;
+end;
+--
+-- End of Test Parallel-aware Hash Left Anti Semi (Not-In) Join.
+--
+--
 -- Test alter ao/aocs table parallel_workers options
 --
 begin;

src/test/regress/sql/gp_parallel.sql

@@ -491,10 +491,12 @@ abort;
 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);
 create table t3_null(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;
+begin;
+set local enable_parallel = on;
+set local gp_appendonly_insert_files = 2;
+set local gp_appendonly_insert_files_tuples_range = 100;
+set local max_parallel_workers_per_gather = 2;
+set local enable_parallel_hash = off;
 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;
 insert into t3_null select i+1, i from generate_series(1, 1200) i;
@@ -507,16 +509,49 @@ select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
 explain(costs off) select * from t1 where c1 not in (select c1 from t3_null);
 select * from t1 where c1 not in (select c1 from t3_null);
 -- non-parallel results.
-set enable_parallel = off;
+set local enable_parallel = off;
 select sum(t1.c1) from t1 where c1 not in (select c1 from t2);
 select * from t1 where c1 not in (select c1 from t3_null);
+end;
 drop table t1;
 drop table t2;
 drop table t3_null;
 --
 -- End of Test Parallel Hash Left Anti Semi (Not-In) Join.
 --
 
+--
+-- Test Parallel-aware Hash Left Anti Semi (Not-In) Join.
+--
+begin;
+create table t1(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+create table t2(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+create table t3_null(c1 int, c2 int) with(parallel_workers=2) distributed by (c1);
+set local enable_parallel = on;
+set local max_parallel_workers_per_gather = 2;
+insert into t1 select i, i from generate_series(1, 500000) i;
+insert into t2 select i, i+1 from generate_series(1, 500000) i;
+insert into t3_null select i, i+1 from generate_series(1, 500000) i;
+insert into t3_null values(NULL, NULL);
+analyze t1;
+analyze t2;
+analyze t3_null;
+explain(costs off) select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+explain(costs off) select * from t1 where c1 not in (select c2 from t3_null);
+select * from t1 where c1 not in (select c2 from t3_null);
+-- non-parallel results.
+set local enable_parallel = off;
+select sum(t1.c1) from t1 where c1 not in (select c2 from t2);
+select * from t1 where c1 not in (select c2 from t3_null);
+drop table t1;
+drop table t2;
+drop table t3_null;
+end;
+--
+-- End of Test Parallel-aware Hash Left Anti Semi (Not-In) Join.
+--
+
 --
 -- Test alter ao/aocs table parallel_workers options
 --