What a Citus extension can and cannot do

Citus is a PostgreSQL extension that adds sharding capabilities on top of the base system, and like any such solution it draws lines where native behavior stops. The interesting question is not whether advanced SQL works at all, but which features survive the transition to a sharded cluster and how to work around those that do not.

The features below — database naming, bulk loading, ordered sets, window functions, grouping sets, triggers and ALTER TABLE — are the ones most commonly needed in practice.

Use the postgres database

Database management is no longer as straightforward as in a plain PostgreSQL deployment: Citus workers are bound to a specific database, since PostgreSQL treats a database as a strict separation. Most Citus examples therefore use the postgres database for demonstrations.

In practice, schemas should be preferred over multiple databases. Large-scale analytics workloads rarely need several databases on the same infrastructure anyway, and multiple clusters can be built in Kubernetes with Patroni for high availability.

Just use the postgres database and you'll be OK.

Bulk loading with COPY

For fast data loads, the PostgreSQL COPY command remains the primary tool, and it is available in Citus.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

postgres=# CREATE TABLE t_oil (

region text,

country text,

year int,

production int,

consumption int

);

CREATE TABLE

postgres=# SELECT create_distributed_table('t_oil', 'country');

create_distributed_table

--------------------------

(1 row)

postgres=# COPY t_oil

FROM PROGRAM

'curl /secret/oil_ext.txt';

COPY 644

Citus routes the tuples to the correct shard automatically and keeps the process fully transparent. Bulk loading with COPY in Citus works just fine.

Ordered sets and the median

Averages often hide what matters. With three people earning 1 USD, 3 USD and 1,000,000 USD, the average sits above 300,000 USD even though two of them are starving. The median — the middle value of an ordered data set, here 3 — is the more meaningful figure, with 50% of values below it and 50% above. Computing it in SQL requires sorting the data, which is what aggregates such as the median need:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

postgres=# SELECT  country,

   avg(production),

   percentile_disc(0.5)

WITHIN GROUP (ORDER BY production)

FROM    t_oil

GROUP BY country

ORDER BY avg(production) DESC NULLS LAST;

       country        |          avg          | percentile_disc

----------------------+-----------------------+-----------------

USA                  | 9141.3478260869565217 |            9159

Saudi Arabia         | 7641.8260869565217391 |            8820

Iran                 | 3631.6956521739130435 |            3730

Mexico               | 2359.5217391304347826 |            2930

Canada               | 2123.2173913043478261 |            1967

Kuwait               | 2083.6956521739130435 |            2182

United Arab Emirates | 1936.0434782608695652 |            1937

Iraq                 | 1780.4130434782608696 |            1899

Qatar                |  609.8695652173913043 |             461

Oman                 |  586.4545454545454545 |             625

Syria                |  335.9767441860465116 |             341

Yemen                |  307.2800000000000000 |             346

Other Middle East    |   74.2173913043478261 |              53

Israel               |                       |                

(14 rows)

The execution plan for the query:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

postgres=# explain analyze

SELECT  country,

   avg(production),

   percentile_disc(0.5)

WITHIN GROUP (ORDER BY production)

FROM    t_oil

GROUP BY country

ORDER BY avg(production) DESC NULLS LAST;

                                                                  QUERY PLAN                                                                  

-----------------------------------------------------------------------------------------------------------------------------------------------

Sort  (cost=12406.82..12656.82 rows=100000 width=68) (actual time=11.035..11.037 rows=14 loops=1)

   Sort Key: remote_scan.avg DESC NULLS LAST

   Sort Method: quicksort  Memory: 25kB

   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=68) (actual time=11.012..11.014 rows=14 loops=1)

         Task Count: 32

         Tuple data received from nodes: 384 bytes

         Tasks Shown: One of 32

         ->  Task

               Tuple data received from node: 81 bytes

               Node: host=localhost port=6004 dbname=postgres

               ->  GroupAggregate  (cost=8.28..9.71 rows=3 width=42) (actual time=0.097..0.133 rows=3 loops=1)

                     Group Key: country

                     ->  Sort  (cost=8.28..8.63 rows=138 width=10) (actual time=0.068..0.077 rows=138 loops=1)

                           Sort Key: country

                           Sort Method: quicksort  Memory: 32kB

                           ->  Seq Scan on t_oil_102255 t_oil  (cost=0.00..3.38 rows=138 width=10) (actual time=0.008..0.028 rows=138 loops=1)

                   Planning Time: 0.025 ms

                   Execution Time: 0.147 ms

Planning Time: 1.153 ms

Execution Time: 11.069 ms

(20 rows)

Ordered set aggregates can be used with Citus DB.

Window functions

Time-series analysis frequently requires the difference between a row and its predecessor — for example, to determine whether production has risen or fallen:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

postgres=# SELECT

year,

production,

production - lag(production) OVER (ORDER BY year) AS diff,

avg(production) OVER (ORDER BY year

ROWS BETWEEN 2 PRECEDING AND 2 FOLLOWING)

FROM  t_oil

WHERE country = 'USA'

AND year > 2000

;

year | production | diff |          avg          

------+------------+------+-----------------------

2001 |       7669 |      | 7565.0000000000000000

2002 |       7626 |  -43 | 7480.7500000000000000

2003 |       7400 | -226 | 7363.6000000000000000

2004 |       7228 | -172 | 7198.0000000000000000

2005 |       6895 | -333 | 7042.2000000000000000

2006 |       6841 |  -54 | 6909.0000000000000000

2007 |       6847 |    6 | 6917.6000000000000000

2008 |       6734 | -113 | 7041.2000000000000000

2009 |       7271 |  537 | 7091.2500000000000000

2010 |       7513 |  242 | 7172.6666666666666667

(10 rows)

Window functions and analytics do run in Citus. The caveat is the sharded environment itself: window functions usually need sorted input.

Citus supports window functions, but your data should be aligned in a way that the system can provide sorted input to achieve a decent level of efficiency.

Grouping sets need an emulation

Reporting projects inevitably call for aggregations at more than one level at once — say, average production per country plus an overall average across all rows. Advanced SQL expresses this with ROLLUP:

1

2

3

4

5

6

SELECT  country, avg(production)

FROM    t_oil

GROUP BY ROLLUP (country)

ORDER BY avg(production) DESC NULLS LAST;

ERROR:  could not run distributed query with GROUPING SETS, CUBE, or ROLLUP

HINT:  Consider using an equality filter on the distributed table's partition column.

ROLLUP groups by "country" and by "nothing" at the same time, leaving two grouping criteria:

1

2

3

4

5

6

SELECT  country, avg(production)

FROM    t_oil

GROUP BY GROUPING SETS ((), (country))

ORDER BY avg(production) DESC NULLS LAST;

ERROR:  could not run distributed query with GROUPING SETS, CUBE, or ROLLUP

HINT:  Consider using an equality filter on the distributed table's partition column.

Neither feature exists in Citus, so the workaround is to model grouping sets with UNION ALL:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

SELECT  country,

        avg(production)

FROM    t_oil

GROUP BY country

UNION ALL

SELECT  NULL,

        avg(production)

FROM    t_oil

ORDER BY country;

       country        |          avg          

----------------------+-----------------------

Canada               | 2123.2173913043478261

Iran                 | 3631.6956521739130435

Iraq                 | 1780.4130434782608696

Israel               |                      

Kuwait               | 2083.6956521739130435

Mexico               | 2359.5217391304347826

Oman                 |  586.4545454545454545

Other Middle East    |   74.2173913043478261

Qatar                |  609.8695652173913043

Saudi Arabien        | 7641.8260869565217391

Syria                |  335.9767441860465116

USA                  | 9141.3478260869565217

United Arab Emirates | 1936.0434782608695652

Yemen                |  307.2800000000000000

                      | 2607.5139860139860140

(15 rows)

The cost is reading the data twice — once for the country list and once for the overall average. Even so, emulating grouping sets with UNION ALL in a column store will still perform better than using grouping sets with a row store.

Triggers are unsupported on distributed tables

Writing a basic trigger reveals the catch immediately:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

CREATE FUNCTION trig_func()

RETURNS trigger AS

$

BEGIN

NEW.production := round(NEW.production, -1);

RAISE NOTICE 'rounded production: %', NEW.production;

RETURN NEW;

END;

$ LANGUAGE 'plpgsql';

CREATE TRIGGER mytrig

BEFORE INSERT

ON t_oil

FOR EACH ROW

EXECUTE PROCEDURE trig_func();

ERROR:  triggers are not supported on distributed tables

Triggers cannot be used on distributed Citus tables.

Restrictions on ALTER TABLE

Schema changes are unavoidable over time, and ALTER TABLE works only within limits:

1

2

3

postgres=# ALTER TABLE t_oil

ADD COLUMN data_verified boolean DEFAULT true;

ALTER TABLE

Adding columns is simple and presents no problem. Dropping them is more delicate:

1

2

postgres=# ALTER TABLE t_oil DROP COLUMN country;

ERROR:  cannot execute ALTER TABLE command involving partition column

If the partitioning column is involved, Citus raises an error and PostgreSQL does not drop the column — expected behavior, since the operation would destroy the entire setup. Renaming the sharding column, by contrast, is handled cleanly by Citus and PostgreSQL:

1

2

postgres=# ALTER TABLE t_oil RENAME COLUMN country TO data_country;

ALTER TABLE

Restrictions apply mainly when the sharding column is touched.

Database naming in Citus

The recommended approach is to collapse everything into a single database and rely on schemas:

1

2

3

4

testbox:citus hs$ createdb somename

NOTICE:  Citus partially supports CREATE DATABASE for distributed databases

DETAIL:  Citus does not propagate CREATE DATABASE command to workers

HINT:  You can manually create a database and its extensions on workers.