Propagate DDL commands to workers through 2PC

[lithp]

(update to issue to track DDL propagation through 2PC here)

Citus 5.0 propagates Alter Table and Create Index commands to worker nodes. We implemented this feature using Citus' current replication model. We also decided to switch to using 2PC (or pg_paxos) once the metadata propagation changes were implemented.

One drawback to the current approach is when ALTER TABLE ... SET NOT NULL fails it marks shards inactive:

postgres=# ALTER TABLE ddl_testing ALTER COLUMN address SET NOT NULL;
WARNING:  could not receive query results from [snip]
DETAIL:  Client error: column "address" contains null values
WARNING:  could not apply command on shard 102025 on [snip]
DETAIL:  Shard placement will be marked as inactive.
ALTER TABLE

If the problem was caused by the user and not by a failed node we probably shouldn't mark the placements inactive. We should instead somehow error out like we do when you try to create a duplicate index.

[ozgune]

@lithp -- we're planning on improving DDL propagation logic to make sure that we don't have partial failures.

Do you think #25 or #265 captures the issue mentioned here?

[lithp]

Yep, #25, or better DDL propogation, relates to this issue. Added a comment over there.

[ozgune]

This issue is also related to #19. We're prioritizing this issue assuming that we could use a shared infrastructure for #19 and this one (#513).

[marcocitus]

I think the existing 2PC infrastructure for COPY and master_modify_multiple_shards (see multi_transaction.c) is suitable for this. Metadata propagation for masterless will also reuse some of that infrastructure.

[metdos]

Also related to inactive shards problem here #480.

[ozgune]

I'm copy/pasting an internal email thread as additional context to this issue.

Marco: Speculation: A DDL command might have been blocked on something and then failed when the node crashed, which currently causes it to mark the placement as inactive, since it's no longer in sync with the other shards and an operator needs to step in to apply the command manually. In 5.2, we'll have DDL via 2PC, which partially that problem.

Lukas: You are right - I was indeed creating an index around that time. It might have been a DDL command that was running, not (just) COPY.

I'm not sure on logs, since the server was replaced after the crash, and we don't store historic per-node logs elsewhere right now (afaik, Daniel can confirm).

[aamederen]

After discussions with @marcocitus and @sumedhpathak, we have decided our approach to this issue.

The resolution of this issue could in the future also address other DDL related issues such as #480, #131, #356, #357, #265 and #192.

In the solution, we will focus on 2PC while also handling #480. Besides,

• We will open a new connection per shard, instead of per worker, for considering future work about Improve error messages on failed unique indexes #480.
• We will use xact handler for abort/commit mechanism, since it will make it easier to handle Make shard creation atomic #265 in the future.

[marcocitus]

We are making 2 changes to the way DDL commands are used.

The first is that we will prevent DDL in a transaction block, since that would likely lead to various issues in combination with other commands. Currently we allow DDL Commands in transaction blocks, though actually doing so would be very dangerous. It is probably a 1 week task to properly support multi-statement DDL transactions.

The second is that it will error out if max_prepared_transactions is not set on the workers, since the DDL propagation always uses 2PC. We initially considered reusing multi_shard_commit_protocol which defaults to 1pc, but recovery from commit failures is difficult in that case. In MX we have a function to recover from 2PC failures, which would take 2-3 days to integrate.

This means you cannot CREATE INDEX on a distributed table without setting max_prepared_transactions on the workers to >0

Any strong objections to these changes?