Amazon Athena Execution Plans - Additional Operations

Estimated Reading Time: 7 minutes
Last Updated: 2024-12-05

Disclaimer: The details described in this post are the results of my own research and investigations. While some effort has been expended in ensuring their accuracy - with ubiquitous references to source material - I cannot guarantee that accuracy. The views and opinions expressed on this blog are my own and do not necessarily reflect the views of any organization I am associated with, past or present.

Earlier, we cast a light over a distributed execution plan, defining the terminology as we worked through the fragments. In this post, we look to extend our investigation to other operations and terms that we might find in a plan.

Values

First up we have:

• Value [] - indicates that one or more rows will be produced containing literal values.

Here is an example of such a query:

SELECT 1
UNION ALL 
SELECT 2 
UNION ALL 
SELECT 3;

And the execution plan:

This simple query uses 3 Values operations to produce 3 separate rows containing 1, 2 and 3. The query is so simple that there is only one fragment and it is of type SINGLE, meaning that this computation takes place on a single node - the coordinator node. Note also, that in contrast to the query in the previous post, the statistics are provided because Athena already knows the relative CPU, memory and network cost involved.

Watch what happens though if we change the UNION ALL clauses to UNION (which requires the result set to be deduplicated):

SELECT 1
UNION 
SELECT 2 
UNION
SELECT 3;

Suddenly the simple plan on the coordinator node has exploded to multiple fragments. Given that all of these fragments are of type SINGLE, it remains unclear whether these operations take place on the coordinator or on workers as well. If nothing else, it gives us a warning of the extra steps required to use UNION over UNION ALL.

AssignUniqueId, MarkDistinct, FilterProject

Next we have a query that uses a correlated scalar sub-query, resulting in a few new operations:

SELECT sp.product_id, 
       (SELECT sc.category_name FROM aept_db.supermarket_categories sc WHERE sc.category_name = sp.category) AS product_category_name
FROM aept_db.supermarket_products sp;

In Fragment 0, we have operations^[1]:

• AssignUniqueId[] - adds to each row of the dataset a unique identifier.
  In the query above, a unique id of type bigint is assigned to each row of the data scanned from aept_db.supermarket_products. Notice that the unique id is projected and used within a LeftJoin to data from aept_db.supermarket_categories (output from Fragment 1)
• MarkDistinct[distinct = [unique:bigint], marker = is_distinct] - evaluates if a row has the same unique id (from AssignUniqueId) as another row, adding a column is_distinct = false if the row shares a unique id with another row, true otherwise. MarkDistinct can also be used as a distinct aggregation strategy set via the Trino optimizer. See this link for more information about the various strategies that can be employed

Let’s briefly break this apart: the correlated sub-query

(SELECT sc.category_name FROM aept_db.supermarket_categories sc WHERE sc.category_name = sp.category)
FROM aept_db.supermarket_products sp;

gets converted into a join by the query engine (pseudo-code):

aept_db.supermarket_products LEFT JOIN aept_db.supermarket_categories

The correlated sub-query must be scalar; it can only produce 1 value from the join. So, when the AssignUniqueId operator added a unique id to every row of aept_db.supermarket_products, if aept_db.supermarket_categories had N matches for the condition sc.category_name = sp.category, then the unique_id would be duplicated after the join N-1 times. Thus, when the MarkDistinct operator executes, the unique_id would appear more than once, and would ‘mark’ the is_distinct column as false for those rows.

This becomes important in the operation:

• FilterProject[filterPredicate = (CASE “is_distinct” WHEN true THEN true ELSE CAST(fail(28, VARCHAR ‘Scalar sub-query has returned multiple rows’) AS boolean) END), projectLocality = LOCAL, protectedBarrier = NONE] - a combination of the Filter and Project operations.
  
  Here, if any is_distinct value is false, then the query will fail with the error ‘Scalar sub-query has returned multiple rows’

TopNRanking, Window and TopN

We previously saw that the Aggregate operator appeared when we executed a SUM function in the previous post. What about Window functions such as ROW_NUMBER, LEAD, LAG and MAX?

Here we have a simple query to get the maximum total_amount value for each transaction_date using a ROW_NUMBER function.

SELECT t_outer.transaction_date,
       t_outer.total_amount AS max_total_amount
FROM (
      SELECT t.transaction_date,
             t.total_amount,
             ROW_NUMBER() OVER (PARTITION BY t.transaction_date ORDER BY t.total_amount DESC) AS rn
      FROM aept_db.supermarket_transactions t
     ) t_outer 
WHERE t_outer.rn = 1
ORDER BY t_outer.transaction_date DESC;

We are introduced to the operator:

• TopNRanking[partitionBy = [transaction_date], orderBy = [total_amount DESC NULLS LAST], limit = 1, hash = [$hashvalue_1]] - indicates that a ranking function is to be executed on the table and that at most N rows need to be returned. In this case, the table is partitioned by transaction_date and ordered by total_amount DESC NULLS LAST.

Note: I was unable to find a good reference to confirm the above definition but it’s unlikely to be significantly wrong

The limit = 1 (the number N of rows to be returned) ties in with the predicate

WHERE t_outer.rn = 1

It appears that the TopNRanking operator will appear as long as there is an upper bound to the number of rows required. For example, the following predicates will invoke the TopNRanking operation:

1) WHERE t_outer.rn = 2
2) WHERE t_outer.rn <= 10
3) WHERE t_outer.rn >= 25 AND t_outer.rn <= 50

However, change the predicate so it no longer has an upper bound on the returned data:

WHERE t_outer.rn > 10

and suddenly our TopNRanking is replaced with:

• Window[partitionBy = [transaction_date], orderBy = [total_amount DESC NULLS LAST], hash = [$hashvalue]] - indicates that a Window function is to be applied to the dataset. In this case, the window partitioning is on transaction_date and the data are ordered by total_amount DESC NULLS LAST.

The TopNRanking operator seems to appear whenever there is an upper bound on some sequential labelling of rows, through Window functions such as:

• RANK
• DENSE_RANK

The Window operator is more common. Here is an example with the MAX function.

SELECT t.transaction_date,
       t.total_amount,
       MAX(t.total_amount) OVER (PARTITION BY t.transaction_date ORDER BY t.total_amount DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS max_amount_for_date
FROM aept_db.supermarket_transactions t
ORDER BY t.transaction_date DESC;

Note the third line of the Window operator block. The function max is referenced there, indicating that it will apply on the window, with ROWS UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING.


Now, we have seen the TopNRanking node, but there is also a TopN node which appears in a more straightforward scenario:

SELECT td.* 
FROM aept_db.supermarket_transaction_details td
ORDER BY td.total_price DESC
LIMIT 10;

We have^[2]:

• TopN[count = 10, orderBy = [total_price DESC NULLS LAST]] - indicates that the first N rows will be retrieved based on the global ordering specified for the dataset. In this case, the data are sorted by total_price DESC NULLS LAST, and only the first 10 rows are required
• TopNPartial[count = 10, orderBy = [total_price DESC NULLS LAST]] - indicates that the first N rows will be retrieved on the global ordering of a subset of the total dataset. In this case, the data are sorted by total_price DESC NULLS LAST, and only the first 10 rows are required

The TopNPartial operation executes on one or more worker nodes to get the top N rows within a subset of data on each node whereas the TopN operation executes on the final projected dataset.

UNNEST

The last operation we will explore today is the UNNEST operator.

Let us create some fake array data and then unnest it:

WITH q_array1 AS (
                  SELECT 'id_101' AS id,
                         ARRAY[1,2,3] AS items
                 )
SELECT qa1.id,
       item
FROM q_array1 qa1, UNNEST(items) t(item)

In this example, the UNNEST clause translates in the plan into a CrossJoin Unnest:

• CrossJoin Unnest[replicate = [expr:varchar(6)], unnest = [expr_0:array(integer)]] - indicates that a cross join will occur between a table and a dataset generated by unnesting an array.

Note: I was unable to find a good reference for this either, but this is clearly what it’s doing

The Values[] section labels the literal ‘id_101’ as expr against the array of integers. The data are then exchanged between tasks in the node before the UNNEST function is applied. The expr is marked for replication on each row and the unnest is executed on each entry in the array.

Conclusion

There are many different types of operators/operations/nodes that can achieve a variety of outcomes within the Athena/Trino query engine. We have only reviewed a few of the most important, but if you are interested in finding out more, then this link and this link to the Presto and Trino Github code may be of interest. Just be wary that not all operators listed in these repositories will necessarily be available or used within Amazon Athena.

References

[1] - Trino Software Foundation. Trino Documentation - Supporting MERGE. https://trino.io/docs/current/develop/supporting-merge.html
[2] - PrestoDB/Presto Github - Plan Nodes. https://github.com/prestodb/presto/wiki/Plan-nodes