SET citus.next_shard_id TO 310000;
SET citus.log_multi_join_order to true;
SET client_min_messages TO LOG;
CREATE TABLE multi_outer_join_left
(
	l_custkey integer not null,
	l_name varchar(25) not null,
	l_address varchar(40) not null,
	l_nationkey integer not null,
	l_phone char(15) not null,
	l_acctbal decimal(15,2) not null,
	l_mktsegment char(10) not null,
	l_comment varchar(117) not null
);
SELECT create_distributed_table('multi_outer_join_left', 'l_custkey', 'hash');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

CREATE TABLE multi_outer_join_right
(
	r_custkey integer not null,
	r_name varchar(25) not null,
	r_address varchar(40) not null,
	r_nationkey integer not null,
	r_phone char(15) not null,
	r_acctbal decimal(15,2) not null,
	r_mktsegment char(10) not null,
	r_comment varchar(117) not null
);
SELECT create_distributed_table('multi_outer_join_right', 'r_custkey', 'hash');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

CREATE TABLE multi_outer_join_right_reference
(
	r_custkey integer not null,
	r_name varchar(25) not null,
	r_address varchar(40) not null,
	r_nationkey integer not null,
	r_phone char(15) not null,
	r_acctbal decimal(15,2) not null,
	r_mktsegment char(10) not null,
	r_comment varchar(117) not null
);
SELECT create_reference_table('multi_outer_join_right_reference');
 create_reference_table
---------------------------------------------------------------------

(1 row)

CREATE TABLE multi_outer_join_third
(
	t_custkey integer not null,
	t_name varchar(25) not null,
	t_address varchar(40) not null,
	t_nationkey integer not null,
	t_phone char(15) not null,
	t_acctbal decimal(15,2) not null,
	t_mktsegment char(10) not null,
	t_comment varchar(117) not null
);
SELECT create_distributed_table('multi_outer_join_third', 't_custkey', 'hash');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

CREATE TABLE multi_outer_join_third_reference
(
	t_custkey integer not null,
	t_name varchar(25) not null,
	t_address varchar(40) not null,
	t_nationkey integer not null,
	t_phone char(15) not null,
	t_acctbal decimal(15,2) not null,
	t_mktsegment char(10) not null,
	t_comment varchar(117) not null
);
SELECT create_reference_table('multi_outer_join_third_reference');
 create_reference_table
---------------------------------------------------------------------

(1 row)

\set customer_1_10_data :abs_srcdir '/data/customer-1-10.data'
\set customer_11_20_data :abs_srcdir '/data/customer-11-20.data'
\set customer_1_15_data :abs_srcdir '/data/customer-1-15.data'
\set client_side_copy_command '\\copy multi_outer_join_left FROM ' :'customer_1_10_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_left FROM ' :'customer_11_20_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_right FROM ' :'customer_1_15_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_right_reference FROM ' :'customer_1_15_data' ' with delimiter '''|''';'
:client_side_copy_command
-- Make sure we do not crash if one table has no shards
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_third b ON (l_custkey = t_custkey);
 min | max
---------------------------------------------------------------------
   1 |  20
(1 row)

SELECT
	min(t_custkey), max(t_custkey)
FROM
	multi_outer_join_third a LEFT JOIN multi_outer_join_right_reference b ON (r_custkey = t_custkey);
 min | max
---------------------------------------------------------------------
     |
(1 row)

-- Third table is a single shard table with all data
\set customer_1_30_data :abs_srcdir '/data/customer-1-30.data'
\set client_side_copy_command '\\copy multi_outer_join_third FROM ' :'customer_1_30_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_third_reference FROM ' :'customer_1_30_data' ' with delimiter '''|''';'
:client_side_copy_command
-- Regular outer join should return results for all rows
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 |  20
(1 row)

-- Since this is a broadcast join, we should be able to join on any key
SELECT
	count(*)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b ON (l_nationkey = r_nationkey);
 count
---------------------------------------------------------------------
    28
(1 row)

-- Anti-join should return customers for which there is no row in the right table
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b ON (l_custkey = r_custkey)
WHERE
	r_custkey IS NULL;
 min | max
---------------------------------------------------------------------
  16 |  20
(1 row)

-- Partial anti-join with specific value
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b ON (l_custkey = r_custkey)
WHERE
	r_custkey IS NULL OR r_custkey = 5;
 min | max
---------------------------------------------------------------------
   5 |  20
(1 row)

-- This query is an INNER JOIN in disguise since there cannot be NULL results
-- Added extra filter to make query not router plannable
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b ON (l_custkey = r_custkey)
WHERE
	r_custkey = 5 or r_custkey > 15;
 min | max
---------------------------------------------------------------------
   5 |   5
(1 row)

-- Apply a filter before the join
SELECT
	count(l_custkey), count(r_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b
	ON (l_custkey = r_custkey AND r_custkey = 5);
 count | count
---------------------------------------------------------------------
    20 |     1
(1 row)

-- Apply a filter before the join (no matches right)
SELECT
	count(l_custkey), count(r_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b
	ON (l_custkey = r_custkey AND r_custkey = -1 /* nonexistant */);
 count | count
---------------------------------------------------------------------
    20 |     0
(1 row)

-- Apply a filter before the join (no matches left)
SELECT
	count(l_custkey), count(r_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right_reference b
	ON (l_custkey = r_custkey AND l_custkey = -1 /* nonexistant */);
 count | count
---------------------------------------------------------------------
    20 |     0
(1 row)

-- Right join should be disallowed in this case
SELECT
	min(r_custkey), max(r_custkey)
FROM
	multi_outer_join_left a RIGHT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 |  15
(1 row)

-- Reverse right join should be same as left join
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_right_reference a RIGHT JOIN multi_outer_join_left b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 |  20
(1 row)

-- Turn the right table into a large table
\set customer_21_30_data :abs_srcdir '/data/customer-21-30.data'
\set client_side_copy_command '\\copy multi_outer_join_right FROM ' :'customer_21_30_data' ' with delimiter '''|''';'
:client_side_copy_command
-- Shards do not have 1-1 matching. We should error here.
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 |  20
(1 row)

-- empty tables
TRUNCATE multi_outer_join_left;
TRUNCATE multi_outer_join_right;
-- reload shards with 1-1 matching
\set customer_subset_11_20_data :abs_srcdir '/data/customer-subset-11-20.data'
\set client_side_copy_command '\\copy multi_outer_join_left FROM ' :'customer_subset_11_20_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_left FROM ' :'customer_21_30_data' ' with delimiter '''|''';'
:client_side_copy_command
\set customer_subset_21_30_data :abs_srcdir '/data/customer-subset-21-30.data'
\set client_side_copy_command '\\copy multi_outer_join_right FROM ' :'customer_11_20_data' ' with delimiter '''|''';'
:client_side_copy_command
\set client_side_copy_command '\\copy multi_outer_join_right FROM ' :'customer_subset_21_30_data' ' with delimiter '''|''';'
:client_side_copy_command
-- multi_outer_join_third is a single shard table
-- Regular left join should work as expected
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
  11 |  30
(1 row)

-- Since we cannot broadcast or re-partition, joining on a different key should error out
SELECT
	count(*)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_nationkey = r_nationkey);
ERROR:  complex joins are only supported when all distributed tables are co-located and joined on their distribution columns
-- Anti-join should return customers for which there is no row in the right table
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey)
WHERE
	r_custkey IS NULL;
 min | max
---------------------------------------------------------------------
  23 |  29
(1 row)

-- Partial anti-join with specific value (5, 11-15)
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey)
WHERE
	r_custkey IS NULL OR r_custkey = 15;
 min | max
---------------------------------------------------------------------
  23 |  29
(1 row)

-- This query is an INNER JOIN in disguise since there cannot be NULL results (21)
-- Added extra filter to make query not router plannable
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey)
WHERE
	r_custkey = 21 or r_custkey < 10;
 min | max
---------------------------------------------------------------------
  21 |  21
(1 row)

-- Apply a filter before the join
SELECT
	count(l_custkey), count(r_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b
	ON (l_custkey = r_custkey AND r_custkey = 21);
 count | count
---------------------------------------------------------------------
    17 |     1
(1 row)

-- Right join should be allowed in this case
SELECT
	min(r_custkey), max(r_custkey)
FROM
	multi_outer_join_left a RIGHT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
  11 |  30
(1 row)

-- Reverse right join should be same as left join
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_right a RIGHT JOIN multi_outer_join_left b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
  11 |  30
(1 row)

-- Mix of outer joins on partition column
SELECT
	l1.l_custkey
FROM
	multi_outer_join_left l1
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_right r2 ON (l1.l_custkey  = r2.r_custkey)
	RIGHT JOIN multi_outer_join_left l2 ON (r2.r_custkey = l2.l_custkey)
ORDER BY 1
LIMIT 1;
 l_custkey
---------------------------------------------------------------------
        11
(1 row)

-- add an anti-join
SELECT
	l1.l_custkey
FROM
	multi_outer_join_left l1
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_right r2 ON (l1.l_custkey  = r2.r_custkey)
	RIGHT JOIN multi_outer_join_left l2 ON (r2.r_custkey = l2.l_custkey)
WHERE
	r1.r_custkey is NULL
ORDER BY 1
LIMIT 1;
 l_custkey
---------------------------------------------------------------------

(1 row)

-- Three way join 2-2-1 (local + broadcast join) should work
SELECT
	l_custkey, r_custkey, t_custkey
FROM
	multi_outer_join_left l1
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_third_reference t1 ON (r1.r_custkey  = t1.t_custkey)
ORDER BY l_custkey, r_custkey, t_custkey;
 l_custkey | r_custkey | t_custkey
---------------------------------------------------------------------
        11 |        11 |        11
        12 |        12 |        12
        14 |        14 |        14
        16 |        16 |        16
        17 |        17 |        17
        18 |        18 |        18
        20 |        20 |        20
        21 |        21 |        21
        22 |        22 |        22
        23 |           |
        24 |        24 |        24
        25 |           |
        26 |        26 |        26
        27 |        27 |        27
        28 |        28 |        28
        29 |           |
        30 |        30 |        30
(17 rows)

-- Right join with single shard right most table should work
SELECT
	l_custkey, r_custkey, t_custkey
FROM
	multi_outer_join_left l1
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
	RIGHT JOIN multi_outer_join_third_reference t1 ON (r1.r_custkey  = t1.t_custkey)
ORDER BY l_custkey, r_custkey, t_custkey;
LOG:  join order: [ "multi_outer_join_left" ]
LOG:  join order: [ "multi_outer_join_right" ]
 l_custkey | r_custkey | t_custkey
---------------------------------------------------------------------
        11 |        11 |        11
        12 |        12 |        12
        14 |        14 |        14
        16 |        16 |        16
        17 |        17 |        17
        18 |        18 |        18
        20 |        20 |        20
        21 |        21 |        21
        22 |        22 |        22
        24 |        24 |        24
        26 |        26 |        26
        27 |        27 |        27
        28 |        28 |        28
        30 |        30 |        30
           |           |         1
           |           |         2
           |           |         3
           |           |         4
           |           |         5
           |           |         6
           |           |         7
           |           |         8
           |           |         9
           |           |        10
           |           |        13
           |           |        15
           |           |        19
           |           |        23
           |           |        25
           |           |        29
(30 rows)

-- Right join with single shard left most table should work
SELECT
	t_custkey, r_custkey, l_custkey
FROM
	multi_outer_join_third_reference t1
	RIGHT JOIN multi_outer_join_right r1 ON (t1.t_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_left l1 ON (r1.r_custkey  = l1.l_custkey)
ORDER BY t_custkey, r_custkey, l_custkey;
 t_custkey | r_custkey | l_custkey
---------------------------------------------------------------------
        11 |        11 |        11
        12 |        12 |        12
        13 |        13 |
        14 |        14 |        14
        15 |        15 |
        16 |        16 |        16
        17 |        17 |        17
        18 |        18 |        18
        19 |        19 |
        20 |        20 |        20
        21 |        21 |        21
        22 |        22 |        22
        24 |        24 |        24
        26 |        26 |        26
        27 |        27 |        27
        28 |        28 |        28
        30 |        30 |        30
(17 rows)

-- Make it anti-join, should display values with l_custkey is null
SELECT
	t_custkey, r_custkey, l_custkey
FROM
	multi_outer_join_third_reference t1
	RIGHT JOIN multi_outer_join_right r1 ON (t1.t_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_left l1 ON (r1.r_custkey  = l1.l_custkey)
WHERE
	l_custkey is NULL
ORDER BY t_custkey, r_custkey, l_custkey;
 t_custkey | r_custkey | l_custkey
---------------------------------------------------------------------
        13 |        13 |
        15 |        15 |
        19 |        19 |
(3 rows)

-- Cascading right join with single shard left most table
SELECT
	t_custkey, r_custkey, l_custkey
FROM
	multi_outer_join_third_reference t1
	RIGHT JOIN multi_outer_join_right r1 ON (t1.t_custkey = r1.r_custkey)
	RIGHT JOIN multi_outer_join_left l1 ON (r1.r_custkey  = l1.l_custkey)
ORDER BY 1,2,3;
 t_custkey | r_custkey | l_custkey
---------------------------------------------------------------------
        11 |        11 |        11
        12 |        12 |        12
        14 |        14 |        14
        16 |        16 |        16
        17 |        17 |        17
        18 |        18 |        18
        20 |        20 |        20
        21 |        21 |        21
        22 |        22 |        22
        24 |        24 |        24
        26 |        26 |        26
        27 |        27 |        27
        28 |        28 |        28
        30 |        30 |        30
           |           |        23
           |           |        25
           |           |        29
(17 rows)

-- full outer join should work between reference tables
SELECT
	t_custkey, r_custkey
FROM
	(SELECT * FROM multi_outer_join_third_reference r1
	 FULL JOIN multi_outer_join_right_reference r2 ON (r1.t_custkey = r2.r_custkey)
	) AS foo
	INNER JOIN multi_outer_join_right USING (r_custkey)
ORDER BY 1,2;
 t_custkey | r_custkey
---------------------------------------------------------------------
        11 |        11
        12 |        12
        13 |        13
        14 |        14
        15 |        15
(5 rows)

-- full outer join should work with 1-1 matched shards
SELECT
	l_custkey, r_custkey
FROM
	multi_outer_join_left l1
	FULL JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
ORDER BY 1 DESC, 2 DESC;
 l_custkey | r_custkey
---------------------------------------------------------------------
           |        19
           |        15
           |        13
        30 |        30
        29 |
        28 |        28
        27 |        27
        26 |        26
        25 |
        24 |        24
        23 |
        22 |        22
        21 |        21
        20 |        20
        18 |        18
        17 |        17
        16 |        16
        14 |        14
        12 |        12
        11 |        11
(20 rows)

-- full outer join + anti (right) should work with 1-1 matched shards
SELECT
	l_custkey, r_custkey
FROM
	multi_outer_join_left l1
	FULL JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
WHERE
	r_custkey is NULL
ORDER BY 1 DESC, 2 DESC;
 l_custkey | r_custkey
---------------------------------------------------------------------
        29 |
        25 |
        23 |
(3 rows)

-- full outer join + anti (left) should work with 1-1 matched shards
SELECT
	l_custkey, r_custkey
FROM
	multi_outer_join_left l1
	FULL JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
WHERE
	l_custkey is NULL
ORDER BY 1 DESC, 2 DESC;
 l_custkey | r_custkey
---------------------------------------------------------------------
           |        19
           |        15
           |        13
(3 rows)

-- full outer join + anti (both) should work with 1-1 matched shards
SELECT
	l_custkey, r_custkey
FROM
	multi_outer_join_left l1
	FULL JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
WHERE
	l_custkey is NULL or r_custkey is NULL
ORDER BY 1 DESC, 2 DESC;
 l_custkey | r_custkey
---------------------------------------------------------------------
           |        19
           |        15
           |        13
        29 |
        25 |
        23 |
(6 rows)

-- full outer join should error out for mismatched shards
SELECT
	l_custkey, t_custkey
FROM
	multi_outer_join_left l1
	FULL JOIN multi_outer_join_third t1 ON (l1.l_custkey = t1.t_custkey)
ORDER BY 1 DESC, 2 DESC;
 l_custkey | t_custkey
---------------------------------------------------------------------
           |        19
           |        15
           |        13
           |        10
           |         9
           |         8
           |         7
           |         6
           |         5
           |         4
           |         3
           |         2
           |         1
        30 |        30
        29 |        29
        28 |        28
        27 |        27
        26 |        26
        25 |        25
        24 |        24
        23 |        23
        22 |        22
        21 |        21
        20 |        20
        18 |        18
        17 |        17
        16 |        16
        14 |        14
        12 |        12
        11 |        11
(30 rows)

-- inner join  + single shard left join should work
SELECT
	l_custkey, r_custkey, t_custkey
FROM
	multi_outer_join_left l1
	INNER JOIN multi_outer_join_right r1 ON (l1.l_custkey = r1.r_custkey)
	LEFT JOIN multi_outer_join_third_reference t1 ON (r1.r_custkey  = t1.t_custkey)
ORDER BY 1 DESC, 2 DESC, 3 DESC;
 l_custkey | r_custkey | t_custkey
---------------------------------------------------------------------
        30 |        30 |        30
        28 |        28 |        28
        27 |        27 |        27
        26 |        26 |        26
        24 |        24 |        24
        22 |        22 |        22
        21 |        21 |        21
        20 |        20 |        20
        18 |        18 |        18
        17 |        17 |        17
        16 |        16 |        16
        14 |        14 |        14
        12 |        12 |        12
        11 |        11 |        11
(14 rows)

-- inner (broadcast) join  + 2 shards left (local) join should work
SELECT
	l_custkey, t_custkey, r_custkey
FROM
	multi_outer_join_left l1
	INNER JOIN multi_outer_join_third_reference t1 ON (l1.l_custkey = t1.t_custkey)
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey  = r1.r_custkey)
ORDER BY 1 DESC, 2 DESC, 3 DESC;
 l_custkey | t_custkey | r_custkey
---------------------------------------------------------------------
        30 |        30 |        30
        29 |        29 |
        28 |        28 |        28
        27 |        27 |        27
        26 |        26 |        26
        25 |        25 |
        24 |        24 |        24
        23 |        23 |
        22 |        22 |        22
        21 |        21 |        21
        20 |        20 |        20
        18 |        18 |        18
        17 |        17 |        17
        16 |        16 |        16
        14 |        14 |        14
        12 |        12 |        12
        11 |        11 |        11
(17 rows)

-- inner (local) join  + 2 shards left (dual partition) join
SELECT
	t_custkey, l_custkey, r_custkey
FROM
	multi_outer_join_third_reference t1
	INNER JOIN multi_outer_join_left l1 ON (l1.l_custkey = t1.t_custkey)
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey  = r1.r_custkey)
ORDER BY
    t_custkey, l_custkey, r_custkey;
 t_custkey | l_custkey | r_custkey
---------------------------------------------------------------------
        11 |        11 |        11
        12 |        12 |        12
        14 |        14 |        14
        16 |        16 |        16
        17 |        17 |        17
        18 |        18 |        18
        20 |        20 |        20
        21 |        21 |        21
        22 |        22 |        22
        23 |        23 |
        24 |        24 |        24
        25 |        25 |
        26 |        26 |        26
        27 |        27 |        27
        28 |        28 |        28
        29 |        29 |
        30 |        30 |        30
(17 rows)

-- inner (local) join  + 2 shards left (dual partition) join
SELECT
	l_custkey, t_custkey, r_custkey
FROM
	multi_outer_join_left l1
	INNER JOIN multi_outer_join_third_reference t1 ON (l1.l_custkey = t1.t_custkey)
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey  = r1.r_custkey)
ORDER BY 1 DESC, 2 DESC, 3 DESC;
 l_custkey | t_custkey | r_custkey
---------------------------------------------------------------------
        30 |        30 |        30
        29 |        29 |
        28 |        28 |        28
        27 |        27 |        27
        26 |        26 |        26
        25 |        25 |
        24 |        24 |        24
        23 |        23 |
        22 |        22 |        22
        21 |        21 |        21
        20 |        20 |        20
        18 |        18 |        18
        17 |        17 |        17
        16 |        16 |        16
        14 |        14 |        14
        12 |        12 |        12
        11 |        11 |        11
(17 rows)

-- inner (broadcast) join  + 2 shards left (local) + anti join should work
SELECT
	l_custkey, t_custkey, r_custkey
FROM
	multi_outer_join_left l1
	INNER JOIN multi_outer_join_third_reference t1 ON (l1.l_custkey = t1.t_custkey)
	LEFT JOIN multi_outer_join_right r1 ON (l1.l_custkey  = r1.r_custkey)
WHERE
	r_custkey is NULL
ORDER BY 1 DESC, 2 DESC, 3 DESC;
 l_custkey | t_custkey | r_custkey
---------------------------------------------------------------------
        29 |        29 |
        25 |        25 |
        23 |        23 |
(3 rows)

-- Test joinExpr aliases by performing an outer-join.
SELECT
	t_custkey
FROM
	(multi_outer_join_right r1
	LEFT OUTER JOIN multi_outer_join_left l1 ON (l1.l_custkey = r1.r_custkey)) AS
    test(c_custkey, c_nationkey)
    INNER JOIN multi_outer_join_third_reference t1 ON (test.c_custkey = t1.t_custkey)
ORDER BY 1 DESC;
 t_custkey
---------------------------------------------------------------------
        30
        28
        27
        26
        24
        22
        21
        20
        19
        18
        17
        16
        15
        14
        13
        12
        11
(17 rows)

-- Outer joins with subqueries on distribution column
SELECT
  l1.l_custkey,
  count(*) as cnt
FROM (
  SELECT l_custkey, l_nationkey
  FROM multi_outer_join_left
  WHERE l_comment like '%a%'
) l1
LEFT JOIN (
  SELECT r_custkey, r_name
  FROM multi_outer_join_right
  WHERE r_comment like '%b%'
) l2 ON l1.l_custkey = l2.r_custkey
GROUP BY l1.l_custkey
ORDER BY cnt DESC, l1.l_custkey DESC
LIMIT 20;
 l_custkey | cnt
---------------------------------------------------------------------
        30 |   1
        29 |   1
        28 |   1
        27 |   1
        25 |   1
        24 |   1
        23 |   1
        22 |   1
        21 |   1
        20 |   1
        18 |   1
        17 |   1
        16 |   1
        14 |   1
        12 |   1
        11 |   1
(16 rows)

-- Add a shard to the left table that overlaps with multiple shards in the right
\set customer_1_data_file :abs_srcdir '/data/customer.1.data'
\set client_side_copy_command '\\copy multi_outer_join_left FROM ' :'customer_1_data_file' ' with delimiter '''|''';'
:client_side_copy_command
-- All outer joins should error out
SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a LEFT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 | 1000
(1 row)

SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a RIGHT JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
  11 |  30
(1 row)

SELECT
	min(l_custkey), max(l_custkey)
FROM
	multi_outer_join_left a FULL JOIN multi_outer_join_right b ON (l_custkey = r_custkey);
 min | max
---------------------------------------------------------------------
   1 | 1000
(1 row)

SELECT
	t_custkey
FROM
	(multi_outer_join_right r1
	LEFT OUTER JOIN multi_outer_join_left l1 ON (l1.l_custkey = r1.r_custkey)) AS
    test(c_custkey, c_nationkey)
    INNER JOIN multi_outer_join_third t1 ON (test.c_custkey = t1.t_custkey)
ORDER BY 1;
 t_custkey
---------------------------------------------------------------------
        11
        11
        12
        12
        13
        14
        14
        15
        16
        16
        17
        17
        18
        18
        19
        20
        20
        21
        21
        22
        22
        24
        24
        26
        26
        27
        27
        28
        28
        30
        30
(31 rows)

-- simple test to ensure anti-joins work with hash-partitioned tables
CREATE TABLE left_values(val int);
SET citus.shard_count to 16;
SET citus.shard_replication_factor to 1;
SELECT create_distributed_table('left_values', 'val');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

COPY left_values from stdin;
CREATE TABLE right_values(val int);
SELECT create_distributed_table('right_values', 'val');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

COPY right_values from stdin;
SELECT
    *
FROM
    left_values AS l
    LEFT JOIN right_values AS r ON l.val = r.val
WHERE
    r.val IS NULL
ORDER BY 1 DESC, 2 DESC;
 val | val
---------------------------------------------------------------------
   5 |
   1 |
(2 rows)