This implements MERGE phase-III

[tejeswarm]

DESCRIPTION: Support MERGE-SQL, if tables are distributed, co-located, joined on dist-column.

This primarily has 3 commits.
44c387b --> Phase-II commit, which was reverted (This commit was already reviewed and pushed)
4fe2a48 --> Phase-III commit to expand it to multi-shard
d12e429 --> Fixes a bug

Fixes: #6672 #6674 #6676

Phase-III
Support pushdown query where all the tables in the merge-sql are Citus-distributed, co-located, and both
the source and target relations are joined on the distribution column. This will generate multiple tasks
which execute independently after pushdown.

SELECT create_distributed_table('t1', 'id');
SELECT create_distributed_table('s1', 'id', colocate_with => ‘t1’);

MERGE INTO t1
USING s1
ON t1.id = s1.id
        WHEN MATCHED THEN
                UPDATE SET val = s1.val + 10
        WHEN MATCHED THEN
                DELETE
        WHEN NOT MATCHED THEN
                INSERT (id, val, src) VALUES (s1.id, s1.val, s1.src)

*The only exception for both the phases II and III is, UPDATEs and INSERTs must be done on the same shard-group
as the joined key; for example, below scenarios are NOT supported as the key-value to be inserted/updated is not
guaranteed to be on the same node as the id distribution-column.

MERGE INTO target t
USING source s ON (t.customer_id = s.customer_id)
WHEN NOT MATCHED THEN - -
     INSERT(customer_id, …) VALUES (<non-local-constant-key-value>, ……);

OR this scenario where we update the distribution column itself

MERGE INTO target t
USING source s On (t.customer_id = s.customer_id)
WHEN MATCHED THEN
     UPDATE SET customer_id = 100;

Subquery/CTE support: We have only very limited subquery/CTE support until we have recursive planner support for MERGE. Complex subqueries as source are not supported at this time, for example

EXPLAIN MERGE INTO target sda USING (SELECT * FROM SOURCE where z NOT IN (SELECT z FROM target WHERE id > 5 )) sdn ON
sda.id = sdn.id
WHEN NOT matched THEN
INSERT (id, z)
VALUES (sdn.id, 5);
ERROR:  MERGE command is only supported when distributed tables are joined on their distribution column

MERGE INTO target_set
USING (SELECT 2 as s3, source_set.* FROM (SELECT * FROM source_set LIMIT 1) as foo LEFT JOIN source_set USING( s1)) AS foo
ON target_set.t1 = foo.s1
WHEN MATCHED THEN UPDATE SET t2 = t2 + 1
WHEN NOT MATCHED THEN INSERT VALUES(s1, s3);
psql:sets:35: ERROR:  cannot push down this subquery
DETAIL:  Limit clause is currently unsupported when a subquery references a column from another query

[codecov]

Codecov Report

Merging #6696 (d75fab3) into main (20a5f3a) will decrease coverage by 0.05%.
The diff coverage is 87.93%.

@@            Coverage Diff             @@
##             main    #6696      +/-   ##
==========================================
- Coverage   93.21%   93.16%   -0.05%     
==========================================
  Files         259      260       +1     
  Lines       55926    56073     +147     
==========================================
+ Hits        52131    52242     +111     
- Misses       3795     3831      +36     

[onderkalaci]

I think directionally pushdown planner sounds the first right step. My current feedback is mostly around cleaning-up the PR (e.g., create a new file, open small PRs to main branch if possible etc).

Also, we should consider allowing recursive planning on the "source" part of the merge command, which makes this a lot more flexible.

[onderkalaci]

@tejeswarm can you also consider getting this in with some tests: #6673

[onderkalaci]

I think we are getting closer, I need to do one more round of review, I have not checked the tests yet

[onderkalaci]

I think my only main comment left is about function evaluation. I'd prefer to be a lot more restrictive (at least with this PR) and focus on that later.

[onderkalaci]

I'd probably squash few of the commits, like:
1 and 2 into a single commit
5 and 6 into the first commit.