Understanding Trino Architecture: Unleashing the Power of Distributed SQL Queries

This article assumes that you have Trino installed and configured. We will now delve into the internal workings of Trino to provide a comprehensive understanding of its architecture and query-processing capabilities.

Coordinator and Workers in a Cluster

When installing Trino, you use only a single machine to run everything. For the desired scalability and performance, this deployment is not suitable.

Trino is a distributed SQL query engine resembling massively parallel processing (MPP) style databases and query engines. Rather than relying on vertical scaling of the server running Trino, it is able to distribute all processing across a cluster of servers in a horizontal fashion. This means that you can add more nodes to gain more processing power.

Leveraging this architecture, the Trino query engine is able to process SQL queries on large amounts of data in parallel across a cluster of computers, or nodes. Trino runs a single-server process on each node. Multiple nodes running Trino, configured to collaborate with each other, make up a Trino cluster.

Let's see how the communication within the cluster happens between the coordinator and the workers, as well as from one worker to another.

The coordinator talks to workers to assign work, update status, and fetch the top-level result set to return to the users. The workers talk to each other to fetch data from upstream tasks, running on other workers. The workers retrieve result sets from the data source.

A coordinator is a Trino server that handles incoming queries and manages the workers to execute the queries. A worker is a Trino server responsible for executing tasks and processing data.

The discovery service typically runs on the coordinator and allows workers to register to participate in the cluster.

Note: All communication and data transfer between clients, coordinators, and workers uses REST-based interactions over HTTP/HTTPS.

Query Planning

Let's try to understand the Trino Query planner by an example.

SELECT
(
SELECT name
FROM region r
WHERE regionkey = n.regionkey
) AS region_name,
n.name AS nation_name,
sum(totalprice) orders_sum
FROM nation n, orders o, customer c
WHERE n.nationkey = c.nationkey
AND c.custkey = o.custkey
GROUP BY n.nationkey, regionkey, n.name ORDER BY orders_sum DESC
LIMIT 5;

SELECT
(
SELECT name
FROM region r
WHERE regionkey = n.regionkey
) AS region_name,
n.name AS nation_name,
sum(totalprice) orders_sum
FROM nation n, orders o, customer c
WHERE n.nationkey = c.nationkey
AND c.custkey = o.custkey
GROUP BY n.nationkey, regionkey, n.name ORDER BY orders_sum DESC
LIMIT 5;

• A SELECT query using three tables in the FROM clause implicitly defines a CROSS
• JOIN between the nation, orders, and customer tables
• A WHERE condition to retain the matching rows from the nation, orders, and customer tables
• An aggregation using GROUP BY regionkey to aggregate values of orders for each nation
• A subquery, (SELECT name FROM region WHERE regionkey = n.regionkey), to pull the region name from the region table; note that this query is correlated as if it was supposed to be executed independently for each row of the containing result set
• An ordering definition, ORDER BY orders_sum DESC, to sort the result before returning
• A limit of five rows is defined to return only nations with the highest order sums and filter out all others.

Step 1: Parsing and Analysis

Before a query can be planned for execution, it needs to be parsed and analyzed.

Trino performs 3 key steps

a. Identifying tables used in a query

b. Identifying columns used in a query

c. Identifying references to fields within ROW values

Step 2: Initial Query Planning

A query plan can be viewed as a program that produces query results. Recall that SQL is a declarative language: the user writes a SQL query to specify the data they want from the system. Unlike an imperative program, the user does not specify how to process the data to get the result. This part is left to the query planner and optimizer to determine the sequence of steps to process the data for the desired result.

This sequence of steps is often referred to as the query plan. Theoretically, an exponential number of query plans could yield the same query result

The performance of the plans varies dramatically, and this is where the Trino planner and optimizer try to determine the optimal plan. Plans that always produce the same results are called equivalent plans.

 Limit[5]
  - Sort[orders_sum DESC]
    - LateralJoin[2]
      - Aggregate[by nationkey...; orders_sum := sum(totalprice)]
       - Filter[c.nationkey = n.nationkey AND c.custkey = o.custkey]
          - CrossJoin
            - CrossJoin
              - TableScan[nation]
              - TableScan[orders]
            - TableScan[customer]
      - EnforceSingleRow[region_name := r.name]
        - Filter[r.regionkey = n.regionkey]
          - TableScan[region]

 Limit[5]
  - Sort[orders_sum DESC]
    - LateralJoin[2]
      - Aggregate[by nationkey...; orders_sum := sum(totalprice)]
       - Filter[c.nationkey = n.nationkey AND c.custkey = o.custkey]
          - CrossJoin
            - CrossJoin
              - TableScan[nation]
              - TableScan[orders]
            - TableScan[customer]
      - EnforceSingleRow[region_name := r.name]
        - Filter[r.regionkey = n.regionkey]
          - TableScan[region]

Let's now consider the computational complexity of this query plan.

If N, O, C, and R represent the number of rows in nation, orders, customer, and region tables, respectively, we can observe the following:

• TableScan[orders] reads the orders table, returning O rows, so its complexity is Ω(O). Similarly, the other two TableScans return N and C rows; thus their complexity is Ω_N_ and ΩC, respectively.
• CrossJoin above TableScan[nation] and TableScan[orders] combine the data from nation and orders tables; therefore, its complexity is Ω(N × O).
• The CrossJoin above combines the earlier CrossJoin, which produced N × O rows, with TableScan[customer] so with data from the customer table, therefore its complexity is Ω(N × O × C).
• TableScan[region] at the bottom has complexity Ω(R). However, because of the LateralJoin, it is invoked N times, with N as the number of rows returned from the aggregation. Thus, in total, this operation incurs Ω(R × N) computational cost.
• The Sort operation needs to order a set of N rows, so it cannot take less time than proportional to N × log(N).

Enough of algebraic formulas. It's time to see what this means in practice! Let's consider an example of a popular shopping site with 100 million customers from 200 nations who placed 1 billion orders in total. The CrossJoin of these two tables needs to materialize 20 quintillion (20,000,000,000,000,000,000) rows. For a moderately beefy 100-node cluster, processing 1 million rows a second on each node, it would take over 63 centuries to compute the intermediate data for our query.

Of course, Trino does not even try to execute such a naive plan. A naive plan has its role, though. The initial plan serves as a bridge between two worlds: the world of SQL language and its semantic rules, and the world of query optimizations. The role of query optimization is to transform and evolve the initial plan into an equivalent plan that can be executed as fast as possible, at least in a reasonable amount of time, given the finite resources of the Trino cluster.

Trino's Optimization Rule

Predicate pushdown is probably the single most important optimization and easiest to understand. Its role is to move the filtering condition as close to the source of the data as possible. As a result, data reduction happens as early as possible during query execution. In our case, it transforms a Filter into a simpler Filter and an InnerJoin above the same CrossJoin condition, leading to a plan shown

...
- Aggregate[by nationkey...; orders_sum := sum(totalprice)]
  - Filter[c.nationkey = n.nationkey AND c.custkey = o.custkey]  // original filter
    - CrossJoin
      - CrossJoin
        - TableScan[nation]
        - TableScan[orders]
      - TableScan[customer]
...
...
- Aggregate[by nationkey...; orders_sum := sum(totalprice)]
  - Filter[c.nationkey = n.nationkey]
    - InnerJoin[o.custkey = c.custkey]
      - CrossJoin
        - TableScan[nation]
        - TableScan[orders]
      - TableScan[customer]

...
- Aggregate[by nationkey...; orders_sum := sum(totalprice)]
  - Filter[c.nationkey = n.nationkey AND c.custkey = o.custkey]  // original filter
    - CrossJoin
      - CrossJoin
        - TableScan[nation]
        - TableScan[orders]
      - TableScan[customer]
...
...
- Aggregate[by nationkey...; orders_sum := sum(totalprice)]
  - Filter[c.nationkey = n.nationkey]
    - InnerJoin[o.custkey = c.custkey]
      - CrossJoin
        - TableScan[nation]
        - TableScan[orders]
      - TableScan[customer]

The "bigger" join that was present is now converted into InnerJoin on an equality condition. Without going into details, let's assume for now that such a join can be efficiently implemented in a distributed system, with computational complexity equal to the number of produced rows. This means that predicate pushdown replaced an "at least" Ω(N × O × C) CrossJoin with a Join that is "exactly" Θ(N × O).

DuckDB and Trino for a Modern Data Stack

When you head to the racetrack, you bring a high-performance car. You don't take your family-friendly, but very versatile SUV. While we may not usually think of ducks as high-performance, that is exactly what DuckDB is for — fast queries on large data sets. It is the racecar of databases — you can't fit anything in it, it seats two, and your golf clubs might not even fit in the back, but it is perfect for the racetrack. The features it lacks don't detract from its strengths. This minimalism is a central design principle in the containerized world of the modern data stack.

However, it's essential to know DuckDB's limits. DuckDB isn't designed for transactional workloads, which makes it inappropriate for high-concurrency write scenarios. DuckDB offers two concurrency options: one process can both read and write to the database, or multiple processes can read only simultaneously.

Another aspect to consider is the absence of user management features, rendering DuckDB more suited for individuals rather than collaborative team environments, which may require user access controls and permissions.

Despite these limitations, DuckDB shines in analytical tasks, provided they align with its strengths and intended use.

How about we use it with Trino?

To facilitate collaborative workloads, we can employ Trino in conjunction with a DuckDB source connector. Please note that as of now, Trino does not offer a native DuckDB source connector; however, we can utilize the Iceberg connector as an alternative.

Conclusion: Unlocking the Full Potential of Trino

In this deep dive into Trino's internal architecture, we've uncovered the intricate mechanisms that power this versatile distributed SQL query engine. Assuming you've got Trino installed and configured, you now have a clearer understanding of how it processes queries, optimizes performance, handles fault tolerance, and ensures security. Trino's ability to seamlessly connect to various data sources through connectors like Iceberg opens up a world of possibilities for data-driven collaboration and analytics.

As the world of data continues to evolve, Trino stands as a valuable tool for organizations seeking to harness the power of their data. With its ever-growing community and the flexibility to extend its capabilities, Trino remains at the forefront of modern data analytics.

We encourage you to further explore Trino, experiment with its advanced features, and engage with the vibrant Trino community. As you continue your journey in the realm of data analytics, may Trino serve as a robust and reliable companion in your quest for insights and innovation.