Improve view DML support

While Citus has greatly improved its support for views referencing distributed tables, users' ability to issue modifications against these views remains essentially non-existant. The relevant places in our planner code _do_ detect the use of a "relation […] [that is] not distributed", so it's not like this is an unhandled case; it's just unimplemented.

# Simple _Updatable Views_

For _simple_ views (essentially those defined using non-repeated direct references to columns of a single table), PostgreSQL has included `INSERT`/`UPDATE`/`DELETE` support since [PostgreSQL 9.3](https://www.postgresql.org/docs/9.3/static/sql-createview.html#SQL-CREATEVIEW-UPDATABLE-VIEWS). For instance:

```sql
CREATE TABLE employees (id integer, name text, salary integer);
-- CREATE TABLE

INSERT INTO employees VALUES (1, 'Bob Maiden', 123456789);
-- INSERT 0 1

CREATE VIEW employee_directory AS SELECT id, name FROM employees;
-- CREATE VIEW

UPDATE employee_directory SET name='Bob Married' WHERE id = 1;
-- UPDATE 1

SELECT * FROM employees;
-- ┌────┬─────────────┬───────────┐
-- │ id │ name │ salary │
-- ├────┼─────────────┼───────────┤
-- │ 1 │ Bob Married │ 123456789 │
-- └────┴─────────────┴───────────┘
-- (1 row)
```

But let's try distributing this table in Citus:

```sql
SELECT create_distributed_table('employees', 'id');
-- NOTICE: 00000: Copying data from local table...
-- LOCATION: CopyLocalDataIntoShards, create_distributed_table.c:1257
-- ┌──────────────────────────┐
-- │ create_distributed_table │
-- ├──────────────────────────┤
-- │ │
-- └──────────────────────────┘
-- (1 row)

UPDATE employee_directory SET name='Bob Married' WHERE id = 1;
-- ERROR: XX000: relation employee_directory is not distributed
-- LOCATION: DistributedTableCacheEntry, metadata_cache.c:711
```

So step one is supporting the plain "updatable views" supported by PostgreSQL since 9.3. I used an `UPDATE` as my example, but the same error occurs even for `COPY`.

# Trigger Workarounds

As in pre-9.3 PostgreSQL, users can often work around Citus' VIEW limitations using a trigger function. For the above case, we could do something like this:

```sql
CREATE OR REPLACE FUNCTION public.insert_row_to_employees()
RETURNS trigger
LANGUAGE plpgsql
AS $irte$
BEGIN
INSERT INTO employees (id, name) VALUES (NEW.id, NEW.name);
RETURN NEW;
END;
$irte$;
-- CREATE FUNCTION

CREATE TRIGGER employee_directory_ins_trg
INSTEAD OF INSERT ON employee_directory
FOR EACH ROW EXECUTE PROCEDURE insert_row_to_employees ();
-- CREATE TRIGGER
```

This works fine, at least for `INSERT` (note that this example continues from above: `employees` remains distributed):

```sql
INSERT INTO employee_directory VALUES (4, 'John');
-- INSERT 0 1

SELECT * FROM employees WHERE id = 4;
-- ┌────┬──────┬────────┐
-- │ id │ name │ salary │
-- ├────┼──────┼────────┤
-- │ 4 │ John │ ∅ │
-- └────┴──────┴────────┘
-- (1 row)
```

Unfortunately, this approach (i.e. using `CREATE TRIGGER [...] INSTEAD OF [...] FOR EACH ROW [...]` on the coordinator) falls apart for `UPDATE` and `DELETE`. In order for `FOR EACH ROW` to work correctly, PostgreSQL executes a scan of the view and calls the trigger function with each returned row. Even if this could be made to work within Citus, it would be _horribly_ slow, pulling rows from workers to coordinator to determine how to update them before just sending them back to workers.

## View-based Shards

One clever approach offers a workaround that preserves Citus' existing behavior of pushing DML statements to workers rather than executing them on the coordinator as some sort of scan: instead of distributing a table and pointing a coordinator view at it, users could create views and triggers on each worker. While this approach creates a lot of objects (functions, triggers, and views for each shard), it allows `UPDATE` and `DELETE` support on stock Citus.

Given two colocated tables and a an empty "view" table…

```sql
CREATE TABLE foo (id integer, name text);
-- CREATE TABLE

CREATE TABLE bar (id integer, title text);
-- CREATE TABLE

CREATE TABLE fake_view (id integer, name text, title text);
-- CREATE TABLE

SELECT create_distributed_table('foo', 'id');
-- ┌──────────────────────────┐
-- │ create_distributed_table │
-- ├──────────────────────────┤
-- │ │
-- └──────────────────────────┘
-- (1 row)

SELECT create_distributed_table('bar', 'id');
-- ┌──────────────────────────┐
-- │ create_distributed_table │
-- ├──────────────────────────┤
-- │ │
-- └──────────────────────────┘
-- (1 row)

SELECT create_distributed_table('fake_view', 'id');
-- ┌──────────────────────────┐
-- │ create_distributed_table │
-- ├──────────────────────────┤
-- │ │
-- └──────────────────────────┘
-- (1 row)
```

For convenience, we begin by pushing the colocation information to all workers…

```sql
SELECT run_command_on_workers('CREATE TABLE shard_colocations ( view_shard text, foo_shard text, bar_shard text )');
-- ┌──────────────────────────────────────────────┐
-- │ run_command_on_workers │
-- ├──────────────────────────────────────────────┤
-- │ (citusdocker_worker_1,5432,t,"CREATE TABLE") │
-- └──────────────────────────────────────────────┘
-- (1 row)

SELECT run_command_on_colocated_placements('fake_view',
'foo',
$$INSERT INTO shard_colocations (view_shard, foo_shard) VALUES ('%s', '%s')$$);
-- ┌──────────────────────────────────────────────────────────┐
-- │ run_command_on_colocated_placements │
-- ├──────────────────────────────────────────────────────────┤
-- │ (citusdocker_worker_1,5432,102072,102008,t,"INSERT 0 1") │
-- | (... ...) |
-- │ (citusdocker_worker_1,5432,102103,102039,t,"INSERT 0 1") │
-- └──────────────────────────────────────────────────────────┘
-- (32 rows)

SELECT run_command_on_colocated_placements('bar',
'fake_view',
$$UPDATE shard_colocations SET bar_shard='%s' WHERE view_shard='%s'$$);
-- ┌────────────────────────────────────────────────────────┐
-- │ run_command_on_colocated_placements │
-- ├────────────────────────────────────────────────────────┤
-- │ (citusdocker_worker_1,5432,102040,102072,t,"UPDATE 1") │
-- | (... ...) |
-- │ (citusdocker_worker_1,5432,102071,102103,t,"UPDATE 1") │
-- └────────────────────────────────────────────────────────┘
-- (32 rows)
```

Now we can drop the shards for our "view" table and replace each with an actual `VIEW` on each worker. In addition, we create a trigger for each of these views…

```sql
SELECT run_command_on_workers($cmd$
DO LANGUAGE plpgsql
$repl_shard$
DECLARE
view_shard text;
foo_shard text;
bar_shard text;
trg_name text;
BEGIN
-- Create trigger function; accepts name of foo, bar shard as arguments
CREATE OR REPLACE FUNCTION modify_foo_bar()
RETURNS trigger AS
$mme$
DECLARE
delete_tmpl CONSTANT text := 'DELETE FROM %I WHERE id=$1';
update_tmpl CONSTANT text := 'UPDATE %I SET %I=$2 WHERE id=$1';
insert_tmpl CONSTANT text := 'INSERT INTO %I (id, %I) VALUES ($1, $2)';
foo_shard text := TG_ARGV[0];
bar_shard text := TG_ARGV[1];
BEGIN
CASE TG_OP
WHEN 'DELETE' THEN
EXECUTE format(delete_tmpl, bar_shard) USING OLD.id;
EXECUTE format(delete_tmpl, foo_shard) USING OLD.id;
RETURN OLD;
WHEN 'UPDATE' THEN
EXECUTE format(update_tmpl, bar_shard, 'title') USING NEW.id, NEW.title;
EXECUTE format(update_tmpl, foo_shard, 'name') USING NEW.id, NEW.name;
RETURN NEW;
WHEN 'INSERT' THEN
EXECUTE format(insert_tmpl, foo_shard, 'name') USING NEW.id, NEW.name;
EXECUTE format(insert_tmpl, bar_shard, 'title') USING NEW.id, NEW.title;
RETURN NEW;
END CASE;
END;
$mme$ LANGUAGE plpgsql;

-- Loop over the triples in our colocation info table
FOR view_shard, foo_shard, bar_shard IN SELECT * FROM shard_colocations LOOP
trg_name := format('%I_mod_trg', view_shard);

-- Drop the fake "view" shard
EXECUTE format('DROP TABLE %I', view_shard);

-- Recreate it as a view referencing its colocated shards
EXECUTE format($$CREATE VIEW %I AS
SELECT f.id, f.name, b.title
FROM %I f, %I b
WHERE f.id = b.id$$, view_shard, foo_shard, bar_shard);

-- Add the appropriate modification trigger
EXECUTE format($$CREATE TRIGGER %I
INSTEAD OF INSERT OR UPDATE OR DELETE
ON %I FOR EACH ROW EXECUTE PROCEDURE modify_foo_bar(%L, %L)$$,
trg_name, view_shard, foo_shard, bar_shard);
END LOOP;

-- We don't need the colocation info anymore
DROP TABLE shard_colocations;
END;
$repl_shard$;
$cmd$);
-- ┌──────────────────────────────────┐
-- │ run_command_on_workers │
-- ├──────────────────────────────────┤
-- │ (citusdocker_worker_1,5432,t,DO) │
-- └──────────────────────────────────┘
-- (1 row)
```

While this is clearly anything but simple, it permits most DML against this more "complex" view, even though the view has two underlying distributed tables…

```sql
INSERT INTO fake_view VALUES (1, 'Bob', 'Programmer'), (2, 'Jane', 'Programmer');
-- INSERT 0 2

SELECT * FROM fake_view;
-- ┌────┬──────┬────────────┐
-- │ id │ name │ title │
-- ├────┼──────┼────────────┤
-- │ 1 │ Bob │ Programmer │
-- │ 2 │ Jane │ Programmer │
-- └────┴──────┴────────────┘
-- (2 rows)

UPDATE fake_view SET title=('Senior ' || title) WHERE id = 2;
-- UPDATE 1

SELECT name, title FROM fake_view;
-- ┌──────┬───────────────────┐
-- │ name │ title │
-- ├──────┼───────────────────┤
-- │ Bob │ Programmer │
-- │ Jane │ Senior Programmer │
-- └──────┴───────────────────┘
-- (2 rows)

DELETE FROM fake_view RETURNING id;
-- ┌────┐
-- │ id │
-- ├────┤
-- │ 1 │
-- │ 2 │
-- └────┘
-- (2 rows)

SELECT COUNT(*) FROM fake_view;
-- ┌───────┐
-- │ count │
-- ├───────┤
-- │ 0 │
-- └───────┘
-- (1 row)
```

I haven't included queries against the table itself here as this section is already running quite long. This approach can obviously _work_, it's just a little brittle and requires a lot of work by the end user, who just wants to have a `VIEW` they can easily run DML against with a trigger, similar to PostgreSQL's existing behavior. An approach similar to this was first mentioned [here](https://github.com/citusdata/citus/issues/442#issuecomment-208926054), and a feature request [here](https://github.com/citusdata/citus/issues/538) asked whether we couldn't track which tables have views as their shards on the workers.

# Citus Improvements

There are several things that could be implemented in order to improve our support for writable views…

## PostgreSQL-like Auto-Updatable Views

For this feature, we'd augment Citus to have the same sort of "auto-updatable" views that PostgreSQL has supported since 9.3, i.e. if a distributed view involves only a single table using simple column references, we plan and execute updates against that view's shards.

### Benefits

* Supports everything PostgreSQL does out of the box
* Gives simple DML support for simple views
* Easy to understand

### Downsides

* Doesn't cover colocated modifications
* Users may still try writing triggers which won't work
* Unclear whether actually desired by anyone

## Submit Multi-Relation Patch to PostgreSQL

I don't know the likelihood of this, but it is possible we could devise a multi-relation updatable view improvement for PostgreSQL and send it upstream. This machinery would likely make it easier for us to propagate DML to multiple distributed relations at once.

### Benefits

* Improves general PostgreSQL experience
* Covers more complex cases
* Less Citus code, possibly

### Downsides

* High bar for patch (covers edge cases, etc.)
* Possibly not accepted as feature
* Must wait for future release

## Use Modifying CTEs

When users have a multi-relation view composed of simple references to columns in each distributed relation, it's possible to translate most DML into a `SELECT` with multiple modifying common-table expressions (`WITH` clauses).

For instance, given our above example (a `foo` table linking `id`s and names, a `bar` table linking `id`s and titles, and a view joining the two tables), we could translate `UPDATE view SET name TO 'Bob' WHERE id = 5 RETURNING *` into:

```sql
WITH affected_rowids AS (
SELECT f.ctid AS fid, b.ctid AS bid
FROM foo f, bar b
WHERE f.id = b.id AND f.id = 1
), updated_foo AS (
UPDATE foo SET name = 'Bob' WHERE ctid IN (SELECT fid FROM affected_rowids)
RETURNING *
), updated_bar AS (
UPDATE bar SET title = 'Lawyer' WHERE ctid IN (SELECT bid FROM affected_rowids) RETURNING *
)
SELECT f.id, f.name, b.title FROM updated_foo f, updated_bar b WHERE f.id = b.id;
```

### Benefits

* No upstream modifications needed
* Flexible (can support view quals and quals on constituent relations)
* Transparent (could _just work_ for many queries)

### Downsides

* Speed (I believe CTEs are an optimization barrier)
* Failure handling (don't want to expose confusing query to end user?)

## View Shards on Workers

As discussed in the above example, it could be possible to create views on the workers and just delegate to those for modifications against views composed of colocated tables.

### Benefits

* Very little code change

### Downsides

* More stateful (cleanup, etc.)
* Tons more cruft on workers
* Difficulty of generating appropriate triggers on each worker

## Statement-Based Triggers

Though traditionally end users choose row-based triggers as the tool for updatable views, statement-based triggers also exist. These are seldom useful for end users, given they only provide a string of the statement being executed, but in our case it might be possible (unclear in the case of `RETURNING`) to parse that string and mutate it into something appropriate for modifications.

### Benefits

* Actually, I don't know what this gives us over the CTE case

### Downsides

* Possibly limitations in capabilities of statement triggers
* OK, I read some documentation and actually I'm not sure this approach works at all, given that statement-based triggers are always supposed to return `NULL`

# Summary

We can probably support multi-relation–based views against colocated distributed relations by transforming modifications into a set of DMLs against each row (being careful to join and apply view quals), or a CTE. The view-shards-on-workers approach is also appealing at the cost of tons of views and triggers on each worker.