Support MERGE with local tables

[tejeswarm]

All the tables (target, source or any CTE present) in the SQL statement are local i.e., a merge-sql with a combination of Citus local and Non-Citus tables (regular Postgres tables) should work and give the same result as Postgres MERGE on regular tables. Catch and throw an exception (not-yet-supported) for all other scenarios during Citus-planning phase.

[onderkalaci]

Some initial observations:

1. INSERTs could happen on a random shard, we cannot seem to pushdown INSERT as is. Can you see a way to restrict the INSERT to the shards that are used in the MERGE command? Like the things we do for triggers?

CREATE TABLE target (tid integer, balance integer)
  WITH (autovacuum_enabled=off);
SELECT create_distributed_table('target', 'tid');

CREATE TABLE source (sid integer, delta integer) -- no index
  WITH (autovacuum_enabled=off);
SELECT create_distributed_table('source', 'sid');


-- insert just 1 row to sourc so that we can test UNMATCHED on target
INSERT INTO source VALUES (1, 10);

-- so, now the INSERT with tid:2000 will happen on a random shard
MERGE INTO target t 
USING SOURCE AS s ON t.tid = s.sid 
WHEN NOT MATCHED THEN 
	INSERT VALUES (2000,2000);

SELECT * FROM target ;
┌──────┬─────────┐
│ tid  │ balance │
├──────┼─────────┤
│ 2000 │    2000 │
└──────┴─────────┘
(1 row)


-- but, we cannot find tid = 2000, because we are searching for a wrong shard
SELECT * FROM target WHERE tid = 2000;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
└─────┴─────────┘
(0 rows)

Time: 1.700 ms

2. Less important, can be disallowed, we cannot allow UPDATE dist_keys.

TRUNCATE source, target;

-- insert some rows that do match
INSERT INTO source VALUES (1,1);
INSERT INTO target VALUES (1,1);

-- OOPS, we changed the dist key, we should disallow this
MERGE INTO target t
USING source AS s
ON t.tid = s.sid
WHEN MATCHED THEN
	UPDATE SET tid = 0;

SELECT * FROM target;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
│   0 │       1 │
└─────┴─────────┘
(1 row)


-- we cannot fiind
SELECT * FROM target WHERE tid = 0;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
└─────┴─────────┘
(0 rows)

There are probably more cases like (2) where we need to be more careful, like we cannot allow UPDATE .. SET nextval('sequence') as the sequences should be evaluated on the coordinator. But, as we resolve bigger issues as (1), we can gradually fine tune things like (2), if that makes sense to you as well.

[tejeswarm]

Some initial observations:

1. INSERTs could happen on a random shard, we cannot seem to pushdown INSERT as is. Can you see a way to restrict the INSERT to the shards that are used in the MERGE command? Like the things we do for triggers?

CREATE TABLE target (tid integer, balance integer)
  WITH (autovacuum_enabled=off);
SELECT create_distributed_table('target', 'tid');

CREATE TABLE source (sid integer, delta integer) -- no index
  WITH (autovacuum_enabled=off);
SELECT create_distributed_table('source', 'sid');


-- insert just 1 row to sourc so that we can test UNMATCHED on target
INSERT INTO source VALUES (1, 10);

-- so, now the INSERT with tid:2000 will happen on a random shard
MERGE INTO target t 
USING SOURCE AS s ON t.tid = s.sid 
WHEN NOT MATCHED THEN 
	INSERT VALUES (2000,2000);

SELECT * FROM target ;
┌──────┬─────────┐
│ tid  │ balance │
├──────┼─────────┤
│ 2000 │    2000 │
└──────┴─────────┘
(1 row)


-- but, we cannot find tid = 2000, because we are searching for a wrong shard
SELECT * FROM target WHERE tid = 2000;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
└─────┴─────────┘
(0 rows)

Time: 1.700 ms

2. Less important, can be disallowed, we cannot allow UPDATE dist_keys.

TRUNCATE source, target;

-- insert some rows that do match
INSERT INTO source VALUES (1,1);
INSERT INTO target VALUES (1,1);

-- OOPS, we changed the dist key, we should disallow this
MERGE INTO target t
USING source AS s
ON t.tid = s.sid
WHEN MATCHED THEN
	UPDATE SET tid = 0;

SELECT * FROM target;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
│   0 │       1 │
└─────┴─────────┘
(1 row)


-- we cannot fiind
SELECT * FROM target WHERE tid = 0;
┌─────┬─────────┐
│ tid │ balance │
├─────┼─────────┤
└─────┴─────────┘
(0 rows)

There are probably more cases like (2) where we need to be more careful, like we cannot allow UPDATE .. SET nextval('sequence') as the sequences should be evaluated on the coordinator. But, as we resolve bigger issues as (1), we can gradually fine tune things like (2), if that makes sense to you as well.

Very much valid and good scenarios, but with the new strategy of Phase I supporting local tables, this is not an issue. but I will track them and address in Phase II

[codecov]

Codecov Report

Merging #6365 (58213b3) into main (5268d0a) will decrease coverage by 0.02%.
The diff coverage is 86.15%.

@@            Coverage Diff             @@
##             main    #6365      +/-   ##
==========================================
- Coverage   92.92%   92.89%   -0.03%     
==========================================
  Files         259      259              
  Lines       55528    55570      +42     
==========================================
+ Hits        51597    51623      +26     
- Misses       3931     3947      +16