Subquery pushdown changes in 6.2

[onderkalaci]

Why are we implementing it? (sales eng)

• This feature is only intended for real-time use case.
• Before this change, Citus had some limited support for multi-shard subqueries. The implementation that comes with 6.2 doesn't implement all kinds of subqueries either. But, it removes quite a few limitations of the existing implementation and providing much broader SQL support.
• Who doesn't want more SQL coverage? Although we intended a specific use-case described below, having broader SQL coverage could help other use cases as well.

What are the typical use cases?

• Main motivation is to be able to run multi-shard behavioral analytics - segmentation type of queries.

  • It can answer queries that groups/segments users and or events. For example:
  • Find customers who have done X, Y and Z in the specified order (i.e., Funnel queries)
  • Find customers who have done X and haven’t done Y, and satisfy other customer specific criteria
  • Find me all users / events ...
  • Many other queries that examine behaviors of users by classifying the users using subqueries.
  • This regression test file is mostly intended for documentation purposes. Almost each query has an explanation. That should give you a better understanding of analysis/segmentation queries that Citus supports.

• What makes subqueries useful for writing such queries? The answer is that each group/segment can be expressed via a subquery. And, using subquery JOINs, it becomes easy to write very powerful analytics queries.

Communication goals (e.g. detailed howto vs orientation)

Good locations for content in docs structure

I think this should go into a separate section in the Docs. But, we should probably check with Sumedh and Ozgun.

How does this work? (devs)

• A very critical note here is that we call this feature as Subquery Pushdown.
• Pushdown means Citus does minor modifications on the query
  • Almost sends the query to the workers as it is, simply adding shard_ids
  • Running the master query
• There are two major requirements to enable pushdown
  • All tables should be co-located
    • Each table should be joined on its partition key with at least one another table on partition key via equality operator
    • See all the examples in the regression test file has partition key equality at least one another table
• The above helps us to form a partition key equality among all the tables
• All the required data is present on the workers, no need to move data/re-partition shards etc.
• Remember that there are other types of subqueries that we cannot pushdown

Detailed list of requirements for subquery pushdown

• All tables should be co-located (i.e., no reference table support yet)
• JOINS:
  • Each table should be joined on its partition key with at least one another table on partition key via equality operator.
  • If the JOIN is INNER/OUTER/LATERAL joins, partition key equality is very obvious. For example:
    • In the query, we see that ON clause has ON a.user_id = b.user_id where user_id is the partition key for both tables.
      • Similarly, in this query, each subquery is joined on user_id via LATERAL joines
  • If the query has subqueries in WHERE clause, we have two ways of pushing down the queries:
    1. Correlated subqueries with IN, NOT IN, EXISTS, NOT EXISTS, operator expressions such as (>, <, =, ALL, ANY etc.). Correlated subqueries mean that there is a reference in the WHERE clause from the FROM clause tables/subqueries. The reference/correlation should be again on partition keys and using equality operator. See a typical examples Operators, EXISTS/ NOT EXISTS) .
       2. Non-correlated subqueries with WHERE partition_key IN (SELECT partition_key ..) . Some examples for IN queries.
          • Important: Note that Postgres internally converts these kinds of non-correlated queries into correlated queries. So, I think it makes sense to clarify whether this type of queries are classified as correlated in the literature and document it accordingly.

• Aggregations:

  • If the query has an aggregate, it has to have a GROUP BY clause. In other words, Citus would error out if any of the subqueries is in the following format: SELECT aggregate(column) FROM table
    • If there exists GROUP BY, the GROUP BY clause has to include the partition key.

• Set Operations:

  • Citus supports UNION and UNION ALL clauses in subqueries. Note that in this example the UNION is not in a subquery, thus not supported by Citus yet. But, here the same query is wrapped into a UNION, thus Citus supports.
  • Each leaf query (i.e., child queries of UNION / UNION ALL queries) should have partition key on their target lists.
  • The partition keys should be in the same position in the target list.
  • For example, see this example. There are two leaf queries of the UNION query. Each leaf query has the partition key on the target list (i.e., user_id) at the same position (i.e., both is the first element in the target list). For example, see this unsupported query example where the 5th query returns 2 * user_id, which is not the partition key.

Side Effect: Expanded View Support

• Citus already supports Views for router queries (i.e., covers multi-tenant use case)
• With this change, Citus expands the View support with a wide range of multi-shard real-time queries
  • Postgres sees views as subqueries
  • If the query that is being used to create the View can be planned by the subquery pushdown planner, then Citus supports that View
    • Joins, subqueries, aggregations on the partition key
    • Also joins between views, tables, subqueries etc. is supported

Example sql

1. Generic SQLs
2. Subqueries in WHERE clause
3. UNION/UNION ALL queries that are below a JOIN query
   • Just a private note here: These are equivalent of Heap's production queries with our regression test schema
4. UNION/UNION ALL queries that are NOT below a JOIN query (i.e., Top Level Unions) -- See Corner cases below
5. Expanded View Support

Corner cases, gotchas

UNIONs

• The most annoying limitation is on the UNION / UNION ALL queries. The limitation that I'm going to explain exists just because we didn't have time to implement it. There is nothing related to correctness or any other thing. We're going to tackle this limitation in the next release.

  • When a UNION or UNION ALL query is below a JOIN in the query tree, we support arbitrary subqueries as the leaf queries. (Here arbitrary means that any query that subquery pushdown can plan/execute). *, For example, see this example. As you can see, the UNION query is below the JOIN in the query tree. In this case, the UNION leaf queries can be any subquery that Citus supports.
  • When a UNION or UNION ALL query is NOT below a JOIN in the query tree, i.e., Top Level Unions, we have a very limited SQL coverage. The leaf queries cannot have arbitrary subqueries. They cannot have any JOINs/subqueries in the query. Instead, they could only be simple projection/aggregation queries.
    • See a typical example here
• Critical Note on UNION support documentation: I think we could consider not documenting UNIONs in this release given that explaining the above limitations could be painful.

Views

• Note that aggregations on non-partition keys cannot be pushed down to the workers. So, for example, this query can be planned by Citus, however, it is not supported by VIEWs.
• Similarly aggregates without GROUP BY are not supported. See example here
• Although we have the above limitation, I think we should be documenting the expanded VIEWsupport given that it allows a very wide support.

Limits

• Citus supports LIMITs on the outermost query such as this one
• However, by default, Citus errors out on LIMITs in the subqueries such as here.
  • Important: This limitation could be relaxed by setting subquery_pushdown TO ON. However, that could lead to return wrong results depending on the data distribution/query (i.e., ORDER BYs in the query). The user has to be aware of that. I think we should not document this publicly.
    • Even if the subquery_pushdown flag is set to ON, if the subqueries have a LIMIT, the outermost query has to have a LIMIT as well.
• OFFSET clause is prohibited in the subqueries.

DISTINCT/DISTINCT ON

• DISTINCT and DISTINCT ON should include partition key such as here

Are there relevant blog posts or outside documentation about the concept/feature?

Team meeting slides are here.

Link to relevant commits and regression tests if applicable

We had two PRs, main PR and the following PR. I already shared the related regression test files in Example sql section.

[sumedhpathak]

Removed 6.2 milestone. Will reevaluate in our upcoming release.