Greenplum vs Citus. Part 1

Citus works with several types of tables:

An example of an entry in pg_dist_partition:

In this case, the name of the logical table (former "regular" PostgreSQL local table) is contained in the logicalrelid field. For this table, the distribution is performed by hash (the partmethod column) and the partkey column contains service information on the distribution key.

In Greenplum, such division does not exist. For append optimized tables, there are so-called aux tables, but they are not hidden in any way and are visible to users and/or administrator of a database. As a result, there is no need for explicit transformation and implicit copying of the original data by calling any special functions.

Tables managed by Citus have the following types:

Citus implements so-called multi-level approach to planning queries. Citus integrates into the planning process by using the callback functions available to extensions: planner_hook, set_rel_pathlist_hook, get_relation_info_hook, set_join_pathlist_hook.

The entry point for planning from the Citus perspective is the distributed_planner implementation. For the Citus planner to be included in the process, at least one Citus table should be used in a query. Whether a table corresponds to this requirement is determined by the pg_dist_partition internal system table.

The planner is divided into several levels, which are activated depending on the structure (one could even say, on the conditional "complexity") of a query:

Each next level implements more complex or high-level (grouping, sorting, etc.) approaches to query planning.

Fast path planner

Fast path planner is limited to planning queries that essentially affect a single shard (or use reference tables).

Also, a query:

This planner type is focused on the OLTP load, when a query can be redirected to a single shard. In this planning option, standard_planner is not called. According to the documentation, developers wanted to avoid unnecessary overhead associated with planning by the standard PostgreSQL planner. This option is well suited for schemes using sharding by some key for so-called Multi-tenant applications. For example, when all operations with tables are limited to a certain shard that corresponds to the user id.

Router planner

Router planner extends fast path planner with the ability to query multiple tables, but still within the boundaries of a single shard, according to filtering conditions on dist_key. As described, Citus stores the actual table data in separate shard tables, which are generally not visible to users. At this stage, the planner replaces the original distributed table visible to users with shard tables, taking into account their distribution across the nodes. This type of planner can already plan subqueries and CTE.

For example, for the case where the a and b tables are sharded by a single column c1, the following query can be planned as follows:

What you can pay attention to:

In the line marked with the number 3, you can see Seq Scan on the actual physical table public.b_102332 (shard data). As we mentioned earlier, this table is generally not visible to users, but it appears in the plan as one of 32 shards of the b table.

Since the data in the b table is sharded by the c1 column and the b.c1 = 1 condition is specified in the filter, the planner can use router planner (fast path planner cannot be used because the query contains a subquery) and redirects the query to only one shard.

In line 2, you can see that this part of the query is planned as a subplan (SubPlan 1) and, from the perspective of the actual execution of the query, can be performed by one task (line 1, Task Count: 1). We will talk about tasks (Task) in more detail in the section on the next type of planner — recursive planning.

If we compare the same query in Greenplum, we can see the following similarities and differences:

Greenplum does not distinguish this scenario in terms of redirection to a single segment — the plan includes Gather Motion 32:1 (line marked with the number 1), which collects data from segments from underlying nodes. Due to the filter (line 6), the amount of data can be reduced in this case, but formally the plan will not differ for any number of rows. The tables in Greenplum for this query are distributed on a.c1 and b.c1, and this allows the planner, after collecting data from the Seq Scan node (line 5), to immediately calculate the count(), b.c1 aggregation function (line 4), then build a hash table from this data (line 3), and perform a local (within the segment) Hash Left Join (line 2).

In fact, the Citus plan seems to be more optimal for this scenario, since a query is immediately redirected to the required shard (which is 1/32th of the total data amount of the entire table by default), without waiting for other nodes. Of course, this affects the execution time:

However, the situation changes significantly in the case of such a query (in this case, a more advanced recursive planner is already being used, which will be discussed in the next section.):

Greenplum has a more advanced plan, just look at the lines marked with the numbers 1 and 3:

The first significant difference is that Greenplum calculated the COUNT aggregate and grouped the results by the b.c1 column (lines 2 and 3). The second difference is that GPORCA pushed the filter condition (b.c1 > 19990000) deeper into the plan (line 4), which significantly reduced the data selection from the b table. At the same time, Hash Left Join was used for joining (line 1) — actually, passing through the a table with the condition (a.c1 > 19990000), we get the desired result from the hash table (line 3).

What Citus does: in fact, the Seq Scan node of sequential reading on the a table calls the SubPlan 1 subplan (line 1) for each row, and the filter condition (c1 > 100) of reading on the b table (line 3) leads to a full read of the b table (discarding the first 100 rows does not count). This happens 321 times, based on the value of the loops=321 counter (line 2), for each of the 32 shards.

It is not difficult to imagine that this would lead to a significant difference in execution time. Even for a small number of rows in the a and b tables (21 million rows each), the difference in time was:

If tables are not sharded by the same column (in my case, I recreated the tables and sharded one of them by a.c1 and the second one by b.c2), then the attempt to execute such a query will fail:

Greenplum, in turn, does not experience any difficulties here — as we said earlier, the plan will be generally similar to the original one (for the case of distribution by one column), but the Redistribute Motion node is added (the line marked with the number 1):

To summarize the fast path planner and router planner planners:

The following advanced planner of the recursive planner type removes these limitations.

Recursive planner

Recursive planner implements:

This type of planner tries to consider each possible part of a query for its isolated execution in the pushdown version. The essence of this action is as follows: conditionally, the planner receives a query tree, which is analyzed in depth. During planning, all subqueries and CTE are checked for their ability to be executed directly on the worker nodes. If this is impossible for a given subquery/CTE, then the regular PostgreSQL planner (planner) is called for it. The result is added to the list of subquery plans. An important point is that these subplans are launched before the distributed plan is processed. The results of processing these subplans are written to temporary files on the disk. We will consider this process in detail a little later.

With this approach, developers solve the main problem — the ability to support most types of SQL queries (yes, there are exceptions: for example, the lack of support for GROUPING SETS, CUBE, ROLLUP).

By receiving and materializing the intermediate results of a subquery on each node/shard, Citus can attach this data to distributed tables using any key.

It is worth noting that, unlike Greenplum, recursive and modifying CTE are not supported unless a query is limited to a single shard:

Let’s look at an example of how this planner works on a couple of queries:

Recursive planner decides to plan a subquery (line marked with the number 1) using a subplan (line with the number 2), collect the result of executing the subplan on one node, and calculate the final aggregate function based on the intermediate data (line 7).

I propose to analyze why the planner makes such decisions and how the intermediate storage of results is carried out.

According to the pushdown rules for subqueries, the presence of LIMIT requires that a subquery be formatted as a separate subplan. If we consider the plan from the bottom up, the lines from Seq Scan (line 6) are fed to the input of the Sort node (line 5), then the LIMIT condition is applied. These actions are performed in parallel on all shards of the table (by default, equal to 32). Thus, at the output of these plan nodes, LIMIT N rows come from each shard, sorted by a.c3. However, in order to ensure that a user receives exactly the LIMIT N rows as expected, the planner has to sort the data from all shards before applying the final LIMIT condition (lines 3 and 4). Otherwise, the user might receive results that are different from expectations. These results are cached on the query coordinator node and become available to the rest of the plan through the pg_catalog.read_intermediate_result function. The mechanism of the read_intermediate_result function is described in the Query execution features section.

The Greenplum planner plans the query like this:

Here, it is worth paying attention to the Gather Motion 32:1 step and its parameterization as Merge Key: c3 (lines 2 and 3). The GPORCA planner, knowing that the results from each segment will come in order according to a.c3 (line 4), has the ability to do, in fact, a merge join and apply LIMIT on the results of this set of rows (line 1). This step corresponds formally to the Limit node and Sort under it (lines 3 and 4 in the Citus plan).

The key difference between the plans is that Citus executes Distributed Subplan for each shard (line 2). Within this plan, the total result from the shards is sorted (line 4) and limited by the Limit node (line 3). The results are collected on the query coordinator and materialized as files, which are read by the read_intermediate_result function on the coordinator (line 8).

Greenplum does this within Gather Motion on the fly and shows the best results in practice:

The same type of planner allows you to plan queries with HAVING in subqueries, but it is not able to handle related (correlated) queries, unlike Greenplum:

Greenplum copes with this task without any problems:

Even with fairly simple examples, Citus raises questions about what I need to do with the schema and data distribution in order to ensure that the query will be still executed. The GPORCA planner seems more advanced in practice.

For example, for correlated subqueries:

For the first query, large tables can be partitioned by some fields in addition to sharding.

Logical Planner & Optimizer

A detailed discussion of this planning level is outside the scope of this article. The tasks at this level are partly similar to the tasks solved by groupping_planner in PostgreSQL: these are operations such as GROUP BY, ORDER BY, and LIMIT, as well as aggregating functions.

Join operations are also optimized at this level. The main goal of this optimization is to reduce the number of joins that lead to data redistribution. Apparently, due to the specifics of redistribution based on writing intermediate results to a file and the COPY protocol (more details later), cost-based estimates are not performed.

Citus builds its execution plan based on the implementation of the CustomScan node.

I propose to analyze a query execution process on a simple query, but touching on some interesting points such as executing subplans, storing intermediate results in files and broadcast copying them to nodes.

First, the query itself and its plan:

We already know that such a query leads to the creation and execution of a subplan (line marked 1), the use of the subplan results in the Hash Join (line marked 6), and reading of intermediate results in the inner part of the join (line 7).

Citus is built into the query execution process at the following extension points: ExecutorStart_hook, ExecutorRun_hook, ExplainOneQuery_hook, prev_ExecutorEnd, ExecutorEnd_hook.

Let’s start our countdown from the ExecutorStart_hook implementation of the CitusExecutorRun callback function. For a query like ours, the PostgreSQL engine calls the CitusExecutorRun function in ExecutorRun.

CitusExecutorRun, in addition to auxiliary Citus-specific tasks like skipping constraints check on the coordinator for ALTER TABLE queries (this is delegated to nodes), solves the main task — search for scan state nodes (CitusScanState) that were previously added by one of the planners.

CitusScanState is a "container" structure that includes:

The plan (to be more precise, the PlanState nodes) is traversed in depth, and the found instances of CitusScanState are added to the list.

For our case, this list contains one Custom Scan (Citus Adaptive) node (line 1) and an associated CustomScanState instance. For each of the found nodes and the associated distributed execution plan for distributed tables, locks of all partitions are taken, and all subplans are executed, starting from the top one. In our case, this is the Distributed Subplan 24_1 subplan.

Receiving results on nodes is carried out through the DestReceiver instance. The coordinator node creates the so-called RemoteFileDestReceiver.

Its tasks include:

Once RemoteFileDestReceiver is initialized, Citus begins executing the subplan (line 2). For this purpose, the portal mechanism (Portal) is used. The previously planned part of the HashAggregate query is sent to the portal (line 3). The portal takes as one of its arguments an instance of DestReceiver, which is RemoteFileDestReceiver.

During the processing of the query by the portal, the subquery again gets planned, a regular call of ExecutorStart_hook occurs, which in turn calls the CitusExecutorRun function. By analogy with the main query, it finds and executes the CustomScan node nested in the subquery (line 4).

Further, the plan goes to the input of the ExecutePlan PostgreSQL kernel function. Since the top-level node of the plan is the HashAggregate aggregate function node (line 3, our COUNT(DISTINCT)), the control is transferred to the PostgreSQL kernel (the nodeAgg.c module), which in turn calls the underlying node, which is the CustomScan Citus node.

To perform scanning tasks (that is, directly retrieving rows), the CustomScan implementation uses the so-called AdaptiveExecutor. A full consideration of this module will take another similar article, so, we will consider only the main tasks related to our query.

At the top level, the main task of AdaptiveExecutor is to execute a list of tasks. An example of a task would be a query to a shard (essentially, to a table). In our case, the planner defined 32 tasks (according to the number of shards in the table).

Also, AdaptiveExecutor creates a TupleStore to temporarily store obtained results.

How the read_intermediate_result function works:

The task of read_intermediate_result is to present the intermediate result in COPY format, obtained as a result of a broadcast execution of a subplan, or shuffling (Repartitioning, more on this later) as a set of rows. The file with the intermediate results is parsed and converted into the column formats that the caller passed as parameters. For example, in the case of our query (line 5) read_intermediate_result('24_1'::text, 'binary'::citus_copy_format) intermediate_result(c2 integer) the first argument 24_1 is the identifier of the result row set, the second argument specifies the format ( 'binary' ). The next identifier is used to obtain the name of the file in the pgsql_job_cache directory.

Then, the ReadFileIntoTupleStore function converts rows from the COPY format (PostgreSQL implementation of the copy.c module is used) into Tuplestorestate. Further, the results are presented in the form of tuples.

Where are intermediate results stored:

Intermediate results of query execution are stored in the base/pgsql_job_cache directory on each node. If these results are used in a distributed transaction, then this is the directory base/pgsql_job_cache/<user id>_<coordinator node id>_<transaction number>/. If a distributed transaction is not utilized, this path is base/pgsql_job_cache/<user id>_<process id>/.

The example of location:

How does Citus make a join that is not based on a sharding key when the tables are not distributed using that key?

Consider the following query for the case when the a table is distributed by the a.c1 key, and the b table is distributed by the b.c3 key (to simplify the analysis, we will first distribute each table across two shards):

To perform such a join, the repartitioning mechanism is used. It is based on partitioning into jobs with a single-level hierarchy, where a certain set of jobs is dependent on the main one. Essentially, this set of tasks is a variation of subplans, the results of which are redistributed between nodes according to the keys. The difference from ordinary subplans is that, in tasks, there is no stage of combining results (meaning the merge stage); in subplans, the result, in turn, is always completely redistributed among the nodes (broadcast). Jobs are included in a distributed execution plan and executed in two stages.

At the first stage, each of the 4 shards of both tables (for service, a separate libpq connection is used) receives a query of the form:

The most important things in these queries:

Internally, this results in a call to the worker_partition_query_result function. The task of this function is to obtain the results of executing this query and write them to local files (for the node) according to the sharding scheme and the sharding key. DestReceiver instances are used to receive and write results to intermediate files. For each result file, its own recipient is created. The total number of files for one shard is set by the value of the Merge Task Count parameter, in our case, it equals to 12. When the recipients are ready, the portal, which executes the query, is started (PortalRun).

After performing this operation for 1 of the 4 shards, the contents of the pgsql_job_cache directory for the corresponding distributed transaction will be something like this:

Thus, for one table that participates in the repartitioning process, the number of intermediate files for this stage will be equal to Map Task Count (Table Shard Count) * Merge Task Count.

You can notice that for the same table you can see the following directory contents for 2 nodes:

Based on the file names and their content, we can conclude that the set of data does not intersect and each directory contains its own portion of data, which at the next stage is redistributed to make a local join using keys.

At the second stage, the fetch_intermediate_results function is used, the task of which is to read the intermediate results from other nodes and save these results as intermediate local ones. Before this, data is copied to the intermediate data directory to allow local joins. For example, after a call like this:

For the table above, the contents of the intermediate directory will be:

To connect to other nodes, the libpq protocol and the COPY command are used. At this stage, the join is actually performed using already local data.

Then, the read_intermediate_result function, already familiar to us, uses these results.

Schematically, the described process looks like the follows.

This brings us to the interesting and complex topic of visibility of changes by different transactions, locks, and distributed transactions themselves. The topic is comprehensive, with a lot of nuances. As part of this article, I would like to show the main differences using a couple of queries as an example.

To begin with, the general architecture of Citus is in this part:

Citus uses the standard libpq protocol and, accordingly, libpq connections for connections between nodes. Connections are cached and reused. To achieve a better parallelization when executing commands, multiple connections to one node are used, if possible.

An important nuance is that with such an organization of data access, for example, for transactions involving several tables after a write operation to a shard, all transactions working with shards of this group must use the same connection.

Otherwise, other commands within the same transaction would not be able to see the data that has not yet been committed. A similar requirement arises when writing to a reference table, after which read commands from this table are executed, including join operations with distributed tables (distributed). This is not always possible, in general. The documentation provides an example of such a case if GUC citus.multi_shard_modify_mode is not set to sequential:

The DELETE operation (line marked 2) within a general multi-command transaction (line 1) affects multiple shards. If this operation is performed in parallel, multiple connections will be involved. Thus, given the requirement for reference tables, it becomes impossible to use one connection for the join command (line 3).

In general, Citus has to carefully handle connections in many aspects: on the one hand, control the number of parallel connections so as not to overload the nodes, on the other hand, comply with data visibility rules and avoid deadlocks. The last two aspects are covered by a separate Placement connection tracking module. Its tasks include tracking which groups of shards were affected by the DML/DDL/SELECT operations of a given transaction and deciding which connections to use for further operations.

The module follows these rules:

For these purposes, Greenplum implements a mechanism for reading and writing processes within a group of related processes (gangs). When creating connections between QE and QD, the latter specifies a parameter that determines whether the process will be a reader or a writer. Only one writing process is allowed per group of related processes, and it will also perform writing tasks as part of a distributed transaction. Thus, only this process can change the state of the database and, as a rule, it is also responsible for managing locks within a transaction (LockHolder). To me, such a clear separation seems to be a more transparent approach.

Regarding protocols for interaction between nodes, in Greenplum, the libpq protocol is patched, in particular, for distributed processing (mainly changes concern the interaction of QD and QE). The interface for processing commands received via libpq has been expanded, for example, with commands such as:

Several types of interconnect are available: tcp, ufpifc, and ic proxy. The central element of such interaction are the Motion nodes including the Gather Motion, Redistribute Motion, and Broadcast Motion variations, which are responsible for transferring data between segments.

Thus, Citus relies entirely on standard libpq, while Greenplum uses its own variant of the libpq protocol to send queries to segments, while interconnect is responsible for moving data. Its own implementation of the protocol means difficulties with infusing changes of the PostgreSQL upstream version, but this concerns the general disadvantage of Greenplum, as it is significantly tuned for PostgreSQL MPP.

If comparing the Motion nodes with their closest conceptual analogues — the fetch_intermediate_results, read_intermediate_result, and worker_partition_query_result implementations described above, in Greenplum there is no data transfer among files from node to node, use of the COPY command, etc. The motivation for the decision made by the creators of Citus is clear — maximum use of the tools available in PostgreSQL. The effectiveness of such an implementation is an open question. It seems to me that sending Motion nodes data directly "to the network" is a more efficient approach.

From my perspective, as analogues of query execution parallelization, the closest analogue of Greenplum slices (which are independently executed parts of the plan, the "bridge" between which are various kinds of Motion nodes) are tasks (Tasks) and jobs (Jobs).

For writing transactions that affect shards on different nodes, Citus uses PostgreSQL implementation of two-phase commit (2PC) with some modifications for a distributed environment (related to HA and auto-recovery of 2PC transactions after a failure).

The approach itself and the features of its implementation in Citus are described in a lot of articles. It is based on the PREPARE TRANSACTION, COMMIT PREPARED, and ROLLBACK PREPARED statements that PostgreSQL offers for external transaction managers. I propose to compare and highlight the advantages of implementing a similar, but, in fact, more advanced mechanism in Greenplum.

In my opinion, the main drawback of the implementation of Citus distributed transactions is the lack of guarantees of compliance with visibility rules for committed transactions within the entire cluster.

To demonstrate the possible consequences of such a situation, I added a delay to the block of code that sends COMMIT PREPARED commands to the working nodes according to the 2PC protocol implementation. The sequence of queries is as follows:

Then, we send the INSERT queries:

But since within COMMIT (line marked with the number 2) between these two transactions we have added an artificial (but possible in reality) delay, then it becomes possible to receive the following data in a parallel session:

The problem is in the line marked 1, where we see that the SELECT query returned the line (3, 'foo', 3) in a parallel session, but did not return (4, 'foo', 4).

Thus, the obvious conclusion follows — Citus, as a distributed DBMS, according to the consistency model of distributed systems, falls into the Read Uncommitted classification.

The reason for this summary is that Citus does not have the concept of a single snapshot (meaning Snapshot for the MVCC implementation in PostgresSQL) for distributed transactions.

Previously prepared transactions on individual nodes are applied at different times. Thus, some subsequent transaction will be able to read data that has been applied on one node, but has not yet been applied to another.

In fact, this violates the principle of atomicity — the user will see only part of the committed transaction data. In my view, this may be a significant disadvantage of Citus.

In Greenplum, this situation is not allowed. The system core tracks information about active distributed transactions. To do this, the distributed transaction identifier (gxid) is stored in the shared process memory of each backend, similar to the standard PostgreSQL local transaction identifiers (xid). This identifier is assigned if the transaction requires 2PC (i.e. the transaction is distributed). The query coordinator (QD) is responsible for generating and assigning the gxid value.

As mentioned above, Greenplum has expanded the set of commands that are processed by backends. In particular, the 'M' and 'T' commands. Handlers of these commands also accept serialized information for distributed transaction processing from QD. A global (distributed) snapshot also includes this data.

Greenplum extends the PostgreSQL structure for the snapshot storage to store the global snapshot information. Further, this snapshot is used to check the visibility of the rows. As stated earlier, only one process in gang can be a writer, so the reader processes assume that the writer process received the snapshot. This also solves problems with the visibility of rows affected by the writing part of the transaction (i.e. something that Citus has to solve by manipulating which libpq connections can be used).

A simplified model for checking row visibility using global snapshots is as follows.

If the xmin identifier of this row matches to a transaction that is applied (we will leave the visibility rules for not yet applied and frozen transactions outside the brackets in this case), then Greenplum checks the visibility taking into account the visibility snapshot for the distributed transaction.

Using the query example that was described earlier, Greenplum checks whether this transaction is distributed and whether it needs to be checked against a global visibility snapshot. In our case, the distributed transaction does not have a completed status, so this row is not visible to a query in another session.

Citus

Citus follows these rules for executing and obtaining the resulting value of functions:

This last point can lead to potential problems. For example, for the following query:

Citus creates the following plan:

According to the rules of function calls, it evaluates the condition in WHERE (line marked 4) for each shard (line marked 1). Thus, the following situation is theoretically possible:

Let’s say the distributed table d stores 2 rows. On the first shard Shard 1 the attribute value is 0, on the second shard Shard 2 the value is 1. The reference table stores the same 2 rows.

Since, according to the execution plan, a select from table r is performed on each shard according to the coincidence of the calculated value round(random()), i.e. it is either 0 or 1. With some probability, it is possible to receive such tuples at the output of the Seq Scan node (line 3): the first Shard 1 received only a row with the key 1, the second — with the key 0.

However, the planner did pushdown of the Hash Join connection to each shard (line 2). Obviously, in this combination of keys, users are likely to get a result that they would never get, for example, for regular PostgreSQL!

Since the "superposition" of the r projections for both shards is the set { 1, 0 }, which would be matched by keys with the rows 0 and 1 of the distributed table d, in our case it gave an empty set after joining results from shards:

Greenplum

In turn, Greenplum plans the query to match the "result identity" criteria to the undistributed storage. An important point is that it builds once a select with the volatile function random() within the query to the replicated one (analogue of the Citus reference table) and distributes it into segments through Redistribute Motion (line marked with number 2).

In this way, Greenplum can ensure the correctness of the Hash Join (line 1), thereby corresponding to the possible select results of an undistributed DBMS:

Since the results of Seq Scan (line 3) will be identical for all segments, each segment will be able to match each one using its own key and give the correct result:

As a result, in Citus, in such a combination of used tables, there is some probability of obtaining incorrect sampling results from the point of view of compliance with the undistributed DBMS.