Watch the replay of Citus 11.2 Release Event livestream—with demos + Q&A! 🎦
Watch the replay of Citus 11.2 Release Event livestream—with demos + Q&A! 🎦
Updates to this version:
Welcome to the 11.2 release of the Citus database. This "Updates" page dives deep into many of the changes in Citus 11.2 open source extension to PostgreSQL, including these headline features:
ALTER TABLE without having to specify a nameRead the release blog post: If you’re looking for a good big-picture overview about what’s new in Citus 11.2, Marco's blog post is the place to start.
Citus 11.2 release event: A few of the Citus database engineers on our team at Microsoft livestreamed a “Release Event” to informally chat about what’s new in the 11.2 release—and to show you a some live demos of what you can now do, including the Patroni HA integration with Citus. Watch the replay on YouTube.
As always, we welcome issues, feedback, & stars on our GitHub repo: If you ever run into issues or have any feedback, you can submit issues on the Citus repo on GitHub. And if you like Citus, why not let the world know with a GitHub star ⭐ on the repo?
The recent Patroni 3.0 release includes support for Citus. Now you can use Patroni to add High Availability (HA) to your Citus cluster, just like you could with Patroni for regular Postgres before. Citus 11.2 fixes a few issues that we found while adding this Citus support to Patroni. So by upgrading to 11.2 you'll have an even more seamless HA experience. The main thing that we improved, is that we now transparently reconnect when we detect that a cached connection to a worker got disconnected while we were not using it. Before 11.2 you could get errors because of disconnected cached connections for a while, after a failover.
If you want to try out the Citus support for Patroni you can use the docker-compose example or the Kubernetes example. And if you want to know more how it works under the hood and how to configure it yourself, then you should take a look at the Patroni documentation about Citus support.
Prior to Citus 11.2, a subset of the outer joins between Citus tables were not supported. In other words, while the following were already supported before Citus 11.2:
-- both sides are reference or distributed tables
<reference table> LEFT/RIGHT/FULL JOIN <reference table>
<distributed table> LEFT/RIGHT/FULL JOIN <distributed table>
-- the reference table is on the the inner side and the distributed table is on the outer side
<distributed table> LEFT JOIN <reference table>
<reference table> RIGHT JOIN <distributed table>
The following were not supported in the older Citus releases. So with the Pull Request #6512, Citus 11.2 closes a gap in terms of its outer join support by adding support for the outer joins where the reference table is on the outer side and the distributed table is on the inner side of the join clause:
<reference table> LEFT JOIN <distributed table>
<distributed table> RIGHT JOIN <reference table>
<reference table> FULL JOIN <distributed table>
<distributed table> FULL JOIN <reference table>
This means that now you can perform outer joins between your facts tables and dimension tables to find out unreferenced rows in your facts table:
CREATE TABLE cities (city_code serial, city_name text); -- reference table
CREATE TABLE users (user_id serial, user_handle text, user_city_code int); -- distributed by user_id column
-- Let's find out the cities where no users signed from
SELECT city_name
FROM cities
LEFT JOIN users
ON (city_code=user_city_code)
WHERE user_id IS NULL;
This is not limited to having a reference table on the outer side of the join clause but you can imagine that the reference table in above examples could be replaced by complex subqueries, CTE references or function calls—a table like query expression that Citus needs to replicate to all nodes during query execution—as in below example:
CREATE TABLE users (user_id serial, user_handle text, user_city_code int); -- distributed by user_id column
-- Let's find out the cities where no users signed from
WITH cities_cte AS (
SELECT * FROM (
VALUES (42, 'a nice city name'),
(43, 'and another nice city')
) cities_subquery (city_code, city_name)
)
SELECT city_name
FROM cities_cte
LEFT JOIN users
ON (city_code=user_city_code)
WHERE user_id IS NULL;
One thing to know about this feature is that it relies on pull-push execution and all performance considerations for pull-push execution apply this feature as well.
Citus already supported adding constraints to citus tables using the ALTER TABLE table_name ADD CONSTRAINT constraint_name ... syntax. As of 11.2, we support adding table constraints without having to specify a constraint name. The following constraints can be created using Postgres' syntax for automatic naming. All the supported options by Citus when adding constraints are also applicable to the nameless syntax.
-- create a distributed table in which we will apply all nameless constraints
CREATE TABLE dist_table (col_1 int, col_2 int);
SELECT create_distributed_table('dist_table', 'col_1');
-- create another distributed table to use for the FOREIGN KEY constraint
CREATE TABLE dist_table_2 (col_3 int, col_4 int);
SELECT create_distributed_table('dist_table_2', 'col_3');
-- Add nameless PRIMARY KEY, UNIQUE, EXCLUDE and CHECK to dist_table
ALTER TABLE dist_table ADD PRIMARY KEY(col_1);
ALTER TABLE dist_table ADD UNIQUE(col_1, col_2);
ALTER TABLE dist_table ADD EXCLUDE (col_1 WITH =);
ALTER TABLE dist_table ADD CHECK (col_1 > col_2);
-- Add nameless FOREIGN KEY constraint to dist_table_2
ALTER TABLE dist_table_2 ADD FOREIGN KEY(col_3) REFERENCES dist_table(col_1);
-- display the tables to check the automatically generated constraint names
\d dist_table;
Table "public.dist_table"
Column | Type | Collation | Nullable | Default
--------+---------+-----------+----------+---------
col_1 | integer | | not null |
col_2 | integer | | |
Indexes:
"dist_table_pkey" PRIMARY KEY, btree (col_1)
"dist_table_col_1_col_2_key" UNIQUE CONSTRAINT, btree (col_1, col_2)
"dist_table_col_1_excl" EXCLUDE USING btree (col_1 WITH =)
Check constraints:
"dist_table_check" CHECK (col_1 > col_2)
Referenced by:
TABLE "dist_table_2" CONSTRAINT "dist_table_2_col_3_fkey" FOREIGN KEY (col_3) REFERENCES dist_table(col_1)
\d dist_table_2;
Table "public.dist_table_2"
Column | Type | Collation | Nullable | Default
--------+---------+-----------+----------+---------
col_3 | integer | | |
col_4 | integer | | |
Foreign-key constraints:
"dist_table_2_col_3_fkey" FOREIGN KEY (col_3) REFERENCES dist_table(col_1)
Citus 11.2 now supports Identity Columns on Citus tables, a feature requested by many of our users. With this release, you can now use the GENERATED AS IDENTITY syntax in a column definition with Citus tables. This feature automatically assigns unique values to a column across multiple nodes in a Citus cluster.
Citus supports the full syntax of this column constraint: column_name type GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY[ ( sequence_option ) ]
The IDENTITY constraint utilizes the SEQUENCE object, allowing you to specify options for the system-generated values.
The GENERATED ALWAYS constraint requires PostgreSQL to automatically generate a value for the identity column in all cases. Attempting to insert or update a value in the column will result in an error from PostgreSQL.
On the other hand, the GENERATED BY DEFAULT constraint instructs PostgreSQL to generate a value for the identity column, but it also allows you to manually provide a value for insert or update operations. In this case, PostgreSQL will use the manually supplied value instead of the system-generated one.
You can find more information on identity columns in PostgreSQL in the following documentation.
CREATE TABLE companies (
company_id BIGINT GENERATED ALWAYS AS IDENTITY (START WITH 10 INCREMENT BY 10),
company_name VARCHAR NOT NULL
);
SELECT create_distributed_table('companies', 'company_id');
-- Values for the identity column are automatically generated
INSERT INTO companies(company_name) VALUES ('CompanyName1');
INSERT INTO companies(company_name) VALUES ('CompanyName2');
SELECT * FROM companies;
company_id | company_name
------------+--------------
10 | CompanyName1
20 | CompanyName2
(2 rows)
-- Manually supplied value for the identity column which has GENERATED ALWAYS expression will result in error
INSERT INTO companies(company_id, company_name) VALUES (1, 'CompanyName');
ERROR: cannot insert a non-DEFAULT value into column "company_id"
DETAIL: Column "company_id" is an identity column defined as GENERATED ALWAYS.
HINT: Use OVERRIDING SYSTEM VALUE to override.
CREATE TABLE companies (
company_id BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 10 INCREMENT BY 10),
company_name VARCHAR NOT NULL
);
SELECT create_reference_table('companies', 'company_id');
-- Manually supplied value for the identity column will be used instead of the system-generated one.
INSERT INTO companies(company_id, company_name) VALUES (1, 'CompanyName1');
-- Values for the identity column are automatically generated
INSERT INTO companies(company_name) VALUES ('CompanyName2');
SELECT * FROM companies;
company_id | company_name
------------+--------------
1 | CompanyName1
10 | CompanyName2
(2 rows)
ALTER TABLE statement is unsupported.ALTER TABLE statements that involve changing the identity column are unsupported.These limitations may cause errors and related error messages provided.
In Citus 11.1, we introduced running rebalancer in the background and improved observability of shard rebalancer. In this release we introduce some more functions that report status of background rebalance operations.
This feature is introduced via Pull request #6576 and later improved via Pull request #6683.
This function lists all the background jobs in the cluster. As of Citus 11.2, only rebalance operations can be listed here, but we plan to add more background job types in upcoming releases.
SELECT * FROM citus_job_list();
+--------+----------+-----------+---------------------------------+-------------------------------+-------------------------------+
| job_id | state | job_type | description | started_at | finished_at |
+--------+----------+-----------+---------------------------------+-------------------------------+-------------------------------+
| 1 | finished | rebalance | Rebalance all colocation groups | 2023-01-31 14:34:38.6235+03 | 2023-01-31 14:46:46.557483+03 |
| 2 | failed | rebalance | Rebalance all colocation groups | 2023-01-31 15:02:00.968141+03 | 2023-01-31 15:34:23.375007+03 |
| 3 | running | rebalance | Rebalance all colocation groups | 2023-01-31 16:02:05.512076+03 | (null) |
+--------+----------+-----------+---------------------------------+-------------------------------+-------------------------------+
This function shows the status of a single job that is supplied as an argument to it. The result data type contains all the columns from citus_job_list with one addition: a detail column of jsonb type. This detail column contains information on the tasks associated with the job.
Let's first take a look at the structure of this jsonb column for a rebalance job:
task_state_counts gives an overview of all the jobs. You can understand how much progress is made for the background job by monitoring the state changes over time.tasks is an array of details on background tasks that are associated with the background job. To avoid repetition, we only show details on tasks that are relevant at that time. It will be more evident when we share some examples below. LSN contains the details on the current LSN values on the source and target node, and the LSN lag between the two nodes. You can monitor the progress of the replication for a particular task here.size contains the details on the sizes of the shard that is being moved as part of this task. You can compare the values over time and monitor progress.hosts contains information on the source and target nodes for the shard move.phase indicates the current phase of the task. As of Citus 11.2 a running task goes through the following phases: Not Started Yet, Setting Up, Copying Data, Catching Up, Creating Constraints, Final Catchup, Creating Foreign Keys, Completing, and finally Completed.state shows the state of the task. Right now we only show tasks in running and error, as well as the tasks that run into a problem and are being retried.command shows the SQL query that is run by the task.message shows the last log message of the SQL command.retried shows the number of times this task is retried in the past. If you see a nonzero retried value, consider checking out the message field to understand why there was an error in the last time this task was run.task_id is the identifier for this task.Let's go through some examples to see how this works.
Right after starting a rebalance we see only see that one task is in runnable state, and another is in blocked state. As of 11.2, we do the shard moves for colocation groups sequentially and hence the blocked task will wait until the runnable task finishes.
SELECT jsonb_pretty(details) FROM citus_job_status(3);
+----------------------------+
| jsonb_pretty |
+----------------------------+
| { +|
| "tasks": [ +|
| ], +|
| "task_state_counts": {+|
| "blocked": 1, +|
| "runnable": 1 +|
| } +|
| } |
+----------------------------+
(1 row)
After a while, runnable background task switches to running state, and we can see the details on the task. Below you can see that it is now in Copying
Data phase and the shard size on the target is moving closer to that of the source over time.
SELECT jsonb_pretty(details) FROM citus_job_status(3);
+-------------------------------------------------------------------------------------------+
| jsonb_pretty |
+-------------------------------------------------------------------------------------------+
| { +|
| "tasks": [ +|
| { +|
| "LSN": { +|
| "lag": null, +|
| "source": "0/13C845E0", +|
| "target": null +|
| }, +|
| "size": { +|
| "source": "95 MB", +|
| "target": "73 MB" +|
| }, +|
| "hosts": { +|
| "source": "localhost:9702", +|
| "target": "localhost:9701" +|
| }, +|
| "phase": "Copying Data", +|
| "state": "running", +|
| "command": "SELECT pg_catalog.citus_move_shard_placement(102008,2,1,'auto')",+|
| "message": "", +|
| "retried": 0, +|
| "task_id": 11 +|
| } +|
| ], +|
| "task_state_counts": { +|
| "blocked": 1, +|
| "running": 1 +|
| } +|
| } |
+-------------------------------------------------------------------------------------------+
(1 row)
If there were updates to the shards during this rebalance operation, we might wait for replication to finish and the target node to catchup. Below you can see that the sizes of the tables are not the same, and there is a replication lag of 101 MB.
SELECT jsonb_pretty(details) FROM citus_job_status(3);
+-------------------------------------------------------------------------------------------+
| jsonb_pretty |
+-------------------------------------------------------------------------------------------+
| { +|
| "tasks": [ +|
| { +|
| "LSN": { +|
| "lag": "101 MB", +|
| "source": "0/1A1A2000", +|
| "target": "0/13C845E0" +|
| }, +|
| "size": { +|
| "source": "153 MB", +|
| "target": "95 MB" +|
| }, +|
| "hosts": { +|
| "source": "localhost:9702", +|
| "target": "localhost:9701" +|
| }, +|
| "phase": "Final Catchup", +|
| "state": "running", +|
| "command": "SELECT pg_catalog.citus_move_shard_placement(102008,2,1,'auto')",+|
| "message": "", +|
| "retried": 0, +|
| "task_id": 11 +|
| } +|
| ], +|
| "task_state_counts": { +|
| "blocked": 1, +|
| "running": 1 +|
| } +|
| } |
+-------------------------------------------------------------------------------------------+
(1 row)
After one shard move operation is over, we can see that the last task we used to monitor is now finished. Consequently the blocked task is now running. You can see the changes in several fields:
task_state_counts changed from the last time.task_id is now 12 instead of 11. This shows that we started to see details on a new task.SELECT jsonb_pretty(details) FROM citus_job_status(3);
+-------------------------------------------------------------------------------------------+
| jsonb_pretty |
+-------------------------------------------------------------------------------------------+
| { +|
| "tasks": [ +|
| { +|
| "LSN": { +|
| "lag": null, +|
| "source": "0/1DCC0900", +|
| "target": null +|
| }, +|
| "size": { +|
| "source": "181 MB", +|
| "target": "124 MB" +|
| }, +|
| "hosts": { +|
| "source": "localhost:9702", +|
| "target": "localhost:9701" +|
| }, +|
| "phase": "Copying Data", +|
| "state": "running", +|
| "command": "SELECT pg_catalog.citus_move_shard_placement(102009,2,1,'auto')",+|
| "message": "", +|
| "retried": 0, +|
| "task_id": 12 +|
| } +|
| ], +|
| "task_state_counts": { +|
| "done": 1, +|
| "running": 1 +|
| } +|
| } |
+-------------------------------------------------------------------------------------------+
(1 row)
If we wait until the rebalance completes, will see all the tasks associated with the rebalance job is now complete:
SELECT jsonb_pretty(details) FROM citus_job_status(3);
+----------------------------+
| jsonb_pretty |
+----------------------------+
| { +|
| "tasks": [ +|
| ], +|
| "task_state_counts": {+|
| "done": 2 +|
| } +|
| } |
+----------------------------+
(1 row)
This function shows details on the last rebalance operation on the cluster. The result data type is similar to citus_job_status(). It is easier to use as you do not need to provide the job_id parameter.
SELECT * FROM citus_rebalance_status();
+-[ RECORD 1 ]+-------------------------------------------------+
| job_id | 6 |
| state | finished |
| job_type | rebalance |
| description | Rebalance all colocation groups |
| started_at | 2023-02-02 02:46:54.254831+03 |
| finished_at | 2023-02-02 02:47:29.745273+03 |
| details | {"tasks": [], "task_state_counts": {"done": 2}} |
+-------------+-------------------------------------------------+
In this release, in addition to all great improvements in the previous releases, we have added lsn and status information to get_rebalance_progress function. This will help you monitor the progress of ongoing shard rebalance operations.
The LSN fields help you to keep track of state change used for synchronization coordination, by comparing source_lsn and target_lsn. When they are equal, it means that the subscription has caught up with the publication, i.e the target is synchronized with the source. The data is obtained from publication and subscription objects info, which are used for the shard move/copy.
This feature is introduced via Pull request #6364
Another new field is status. Which simply tells you the status of a shard move/copy. It can be one of the below list of status messages:
An example get_rebalance_progress output, check the three rightmost columns:
┌───────────┬────────────┬─────────┬────────────┬────────────┬────────────┬────────────┬────────────┬──────────┬───────────────────┬───────────────────┬────────────────┬────────────┬────────────┬───────────────┐
│ sessionid │ table_name │ shardid │ shard_size │ sourcename │ sourceport │ targetname │ targetport │ progress │ source_shard_size │ target_shard_size │ operation_type │ source_lsn │ target_lsn │ status │
├───────────┼────────────┼─────────┼────────────┼────────────┼────────────┼────────────┼────────────┼──────────┼───────────────────┼───────────────────┼────────────────┼────────────┼────────────┼───────────────┤
│ 14060 │ t1 │ 102008 │ 11345920 │ localhost │ 9701 │ localhost │ 9702 │ 1 │ 11345920 │ 11345920 │ move │ 89677216 │ 89677040 │ Final Catchup │
└───────────┴────────────┴─────────┴────────────┴────────────┴────────────┴────────────┴────────────┴──────────┴───────────────────┴───────────────────┴────────────────┴────────────┴────────────┴───────────────┘
(1 row)
Note that these new fields are provided for logical replication. For blocking writes, lsn fields are not provided, and the list of status messages are not all but a subset of the above.
Previously, each task was sequentially executed by background task worker, which causes independent tasks to not benefit from possible concurrency. You can see details in rebalancer-background to know more about background task execution.
As of Citus 11.2, via Pull request #6459, Citus can execute independent tasks concurrently as much as possible. Concurrency limit can be configured via a GUC citus.max_background_task_executors by user. You can consider it as soft limit since hard limit is determined by Postgres itself via the GUC max_worker_processes.
Although we have introduced the concurrent background task execution, citus_rebalance_start cannot yet utilize the capabilities. We aim to incorporate the concurrent background task execution with shard moves in the future releases.
Previously we had citus_job_wait to wait for a job to reach desired status. With Pull request #6475, we also introduce a new udf, which is named as citus_task_wait, to wait until a background task reaches desired status.
In the below example, you'll see how we wait until a task reaches desired statuses:
-- create a sample table for testing purpose
CREATE TABLE test(key int);
SELECT create_distributed_table('test', 'key');
INSERT INTO test SELECT 1 FROM generate_series(1, 100000000) i;
INSERT INTO test SELECT 3 FROM generate_series(1, 100000000) i;
-- start a rebalance operation
SELECT citus_rebalance_start('by_disk_size',shard_transfer_mode := 'force_logical') AS jobid \gset
-- obtain the first task's id for the rebalance job
SELECT task_id AS taskid FROM pg_dist_background_task WHERE job_id = :jobid ORDER BY task_id LIMIT 1 \gset
-- wait for the task to reach 'running' state
SELECT citus_task_wait(:taskid, desired_status := 'running');
-- wait for the task to reach 'done' state
SELECT citus_task_wait(:taskid, desired_status := 'done');
In PostgreSQL 15.0, the PG community released a new feature, MERGE SQL, that can conditionally insert, update, or delete rows of a table. Sample merge command looks like below where all modifications are done on the target table.
CREATE TABLE source
(
order_id INT,
customer_id INT,
order_center VARCHAR,
order_time timestamp
);
CREATE TABLE target
(
customer_id INT,
last_order_id INT,
order_center VARCHAR,
order_count INT,
last_order timestamp
);
SELECT citus_add_local_table_to_metadata('target');
SELECT citus_add_local_table_to_metadata('source');
MERGE INTO target t
USING source s ON (t.customer_id = s.customer_id)
WHEN MATCHED AND t.order_center = 'value' THEN
-- Remove the customer in target
DELETE
WHEN MATCHED THEN
-- Existing customer, update the order count and last_order_id in target
UPDATE SET order_count = t.order_count + 1, last_order_id = s.order_id
WHEN NOT MATCHED THEN
-- New entry, record it in target
INSERT (customer_id, last_order_id, order_center, order_count, last_order)
VALUES (customer_id, s.order_id, s.order_center, 123, s.order_time);
Update in May 2023: MERGE support for co-located distributed tables was added in Citus 11.3.
As this is a complex SQL statement, we decided to support this in Citus in a phased manner. In this phase we support MERGE SQL command if 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 tables 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.
DROP OWNED BY command is used to drop all the objects within the current database that are owned by one of the specified roles.
This feature is now supported in Citus, and was developed by an external contributor, TsinghuaLucky912 via Pull request #6519.
Below we show a basic example where role role_1 owns a distributed and a reference table. DROP OWNED BY role_1 command drops all the objects owned by the role.
CREATE SCHEMA sc1;
CREATE ROLE role_1 WITH LOGIN;
GRANT ALL ON SCHEMA sc1 TO role_1;
SET ROLE role_1;
-- these tables are owned by role_1
CREATE TABLE sc1.dist (key int);
SELECT create_distributed_table('sc1.dist', 'key');
CREATE TABLE sc1.ref (key int);
SELECT create_reference_table('sc1.ref');
-- show that both tables are owned by role_1
SELECT DISTINCT relowner::regrole FROM pg_class WHERE relname IN ('dist', 'ref');
┌──────────┐
│ relowner │
├──────────┤
│ role_1 │
└──────────┘
(1 row)
-- return back to the original role
RESET ROLE;
-- all objects owned by role_1
-- are dropped from the Citus cluster
DROP OWNED BY role_1;
-- show that the objects are dropped
select count(*) from pg_class where relname IN ('dist', 'ref');
┌───────┐
│ count │
├───────┤
│ 0 │
└───────┘
(1 row)
By default, when multiple nodes are involved in a transaction, Citus always sets the remote transaction's isolation level with BEGIN TRANSACTION ISOLATION READ COMMITTED.
As of Citus 11.0, via Pull request #4945, Citus can propagate SET TRANSACTION commands to the involving nodes in a transaction. With Citus 11.2 release, via Pull request #6483, we are expanding the support for setting the transaction properties by propagating the SET TRANSACTION as well.
Note that this does not mean that Citus supports comprehensive repeatable read / serializable semantics for all workloads. By propagating the level Citus' semantics will be mostly correct for workloads without multi-shard queries (e.g. multi-tenant). See Marco's detailed comment on this subject.
In the below example, you'll see that we start the transacton on REPEATABLE READ isolation level:
-- create a sample table for testing purpose
CREATE TABLE test(key int);
SELECT create_distributed_table('test', 'key');
INSERT INTO test VALUES (1);
-- start the transaction on REPEATABLE READ
BEGIN ISOLATION LEVEL REPEATABLE READ;
-- should reflect isolation level of current transaction
SELECT current_setting('transaction_isolation') FROM test WHERE key = 1;
┌─────────────────┐
│ current_setting │
├─────────────────┤
│ repeatable read │
└─────────────────┘
(1 row)
END;
Citus 11.2 also supports syntaxes like START TRANSACTION ISOLATION LEVEL REPEATABLE READ, BEGIN READ ONLY or BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE DEFERRABLE.
citus_drain_node() to allow draining the specified worker onlyUPDATE with RETURNING clauses