How to Avoid Data Access Bottlenecks When Using Trino 

Everyone knows that Trino is a powerful tool for accessing and processing data anywhere — in the cloud or on premises, in a data lake or a traditional database, in tables or open format files. To achieve this, Trino uses both generic and specific connectors for almost any data source, which makes it an extremely powerful tool for bringing together data across an organization.

Digging deeper into Trino performance

If the goal is simply to access data, and performance is not a concern, you don’t have to read any further. Trino is probably the right tool for you, and you don’t have to think too hard about how it works, or how to set up the data sources so that they will work well with Trino. However, if you care about performance, you need to understand a little more about how Trino works and where bottlenecks may emerge. 

This blog post is designed to help you identify and potentially eliminate some of these performance bottlenecks. 

 

Trino is primarily an execution engine

Trino itself is a data processing and query execution engine. It does not store data — it only processes and queries it. This distinguishes Trino from standard “database management systems” such as PostgreSQL, MySQL, or Oracle that perform both functions — the storage and the processing.

The difference in functionality leads to a difference in architectural design. Traditional systems that do both storage and processing can take shortcuts in the data processing code since they know exactly how data is stored. Some of these shortcuts lead to performance improvements for the end user. Others simply make the code development faster, but do not impact the end user in any way. 

For example, a database table can be stored on disk row-by-row or column-by-column. A query execution engine for a traditional column-store system will be designed to take advantage of columnar storage by implementing operators that are designed for column-compressed data. In contrast, an execution engine for a traditional row-store will be designed to avoid table scans whenever possible, since such operations end up reading all columns from a table — even those irrelevant for a particular query. However, an execution engine designed to work with any type of storage needs to be able to handle both row-oriented and column-oriented data inputs. They either need to implement optimizations for both types of inputs, or otherwise need to convert columns to rows on the fly — at a performance cost to the end user. 

The more types of data storage that could exist, the more complexity in the codebase for an execution engine designed to work with any type of storage. Many generic query execution engines have collapsed under this complexity. However, Trino’s connector based architecture has withstood the test of time and allowed it to cope in the face of this complexity. 

 

Trino’s connector-based architecture

The core idea of Trino’s connector-based architecture is that some facets of query execution are the same for all types of storage, whereas other facets need to be specific for particular types of storage and storage systems. 

Storage-specific information

Information about the location and type of statistics that are maintained about a dataset varies significantly across data sources. These statistics are critical inputs to the query optimizer that runs during the first step of query execution to decide on a plan for how to execute the query. To optimize queries, the optimizer uses statistics about the data being queried in order to estimate the cost of different plan options and choose the best plan. Different data sources may have very different ways of tracking these statistics and making them available. The description of the availability of statistics and how to find them are implemented in the connector interface. 

Other important data source-specific functionality in the database connector are: 

• Table functions, which vary significantly across systems 
• Role and grant management
• Schema management
• Data and schema modification 

Types of Trino connectors

There are two types of connectors in Trino: generic and specific. Specific connectors are designed for particular data sources and are able to pass along detailed knowledge regarding these data sources to Trino’s execution engine. These types of connectors are able to take full advantage of Trino’s connector-based architecture to often yield performance similar to systems that natively perform both storage and processing. In contrast, generic connectors are designed to work with multiple types of data sources and generally fail to allow Trino to take advantage of the specifics of how data is stored. 

 

Performance problems when using connectors

The most common source of data access bottlenecks in Trino stem from the use of generic connectors. For example, the generic JDBC connector will allow Trino to access data stored in any system that supports a JDBC interface (basically any database system). Trino’s execution engine issues SQL queries to the underlying database system and extracts the results via this JDBC interface, and then performs any necessary subsequent query processing.  

JDBC and tuple-at-a-time

JDBC is known to be an extremely slow interface for passing data between systems. Data is passed a tuple-at-a-time, in a single thread, as the downstream system requests the results set from a SQL query. It was originally designed for returning a small number of records to the client at the very end of query processing. However, Trino performs a large amount of query processing itself, and the data passed to it is at a much earlier stage in query processing that may still be in the order of gigabytes, terabytes, or more. Passing such large amounts of data via JDBC is almost certainly going to result in a huge bottleneck.  

Furthermore, since JDBC returns data tuple-at-a-time, even if data is originally stored column-by column, these columns need to be stitched together into rows during the JDBC communication process. This prevents Trino from potentially being able to take advantage of the many optimizations that exist for column-oriented data. 

 

The ghastly nature of the JDBC performance bottleneck

Unfortunately, it is extremely hard to fix this bottleneck. Adding more resources or nodes to the Trino cluster will not help because the data is being pulled by a single thread. Adding more resources to the data source also will not help because the bottleneck is not there either. Even improving the network will not help because although the transfer is where the bottleneck is, it is not the network that is the problem, but rather the JDBC protocol. 

It is not uncommon to see data sent at a rate of 10MB per second via a JDBC interface. When the rest of the system is capable of processing data at multiple gigabytes per second, this bottleneck is slowing down the system by a factor of 100s to 1000s — two to three orders of magnitude (or more)! 

Trino attempts to alleviate the JDBC bottleneck

To limit the impact of this bottleneck, Trino will attempt to push down query operators that reduce the size of the dataset that needs to be passed between systems. For example, if a query has a predicate (e.g., that only returns data from a particular location, or for a particular user) it will usually attempt to push the predicate down to the underlying storage system. Similarly, it will attempt to push down some simple types of data aggregations, top-N or limit expressions, or even an occasional join. However, joins across multiple data sources and even complex query expressions within the same data source will not get pushed down. The more raw data that gets sent to Trino over JDBC, the worse the data transfer bottleneck.  

System-specific connectors sometimes avoid this bottleneck

In truth, this JDBC bottleneck is not limited to the generic connectors. Many of the system-specific connectors also use JDBC to connect to the underlying system. And indeed, many of them will suffer from similar performance problems. However, system-specific connectors may have additional tools sometimes help to avoid or alleviate this bottleneck:

Partition-aware extraction

The most important of these tools is data partitioning. Data partitioning involves dividing the table into parts (“partitions”) based on a table attribute such as ID, time, or location. Data partitioning is nearly always used in distributed or parallel systems that span across many servers or “nodes”, but are also frequently used in single-node systems to speed up filters on the partitioning attribute and avoid full table scans, or to divide work across different CPU cores.  

System-specific connectors are able to expose details about data partitioning to Trino. Depending on the quality of the connector, this may allow Trino to pull different partitions in parallel. This makes a huge difference in the data transfer bottleneck. Even if JDBC is used to perform the transfer, now instead of a single-threaded JDBC connection, several JDBC connections can work in parallel to transfer data — often increasing throughput of the transfer by an order of magnitude or more. 

I will discuss the performance improvements possible from partition-aware extraction in much more detail in my next post.

Increased processing pushdown 

System-specific connectors are also often able to push a wider range of query processing operations down to the underlying system. This further alleviates the transfer bottleneck by decreasing the data that needs to be transferred. 

 

Practical advice on what to do if you see these performance issues

If you run into these performance issues with either generic or specific connectors, what should you do? Below are my step-by-step suggestions for what to do next:

1. If you are using Trino to access data in multiple data sources, you need to figure out which data sources are causing the problem. Usually, the problem does not occur for data stored in open data formats in a data lake or HDFS storage. Rather, database system data sources are typically the culprit. For those queries running unacceptably slow, where is the data being accessed? 
2. Check to see what connector is being used to access data in that data source. The details on how to do this vary a little depending on what distribution of Trino you are using, but usually, this information will be in the catalog.
3. If you are using a generic connector, there’s not much you can do. If no database-specific connector for the data source you are using is available in your Trino distribution, you may want to check other distributions. For example, Starburst Enterprise and Galaxy provide additional options beyond open source Trino. 
4. If you are using a database-specific connector, you should check to see if your data is already partitioned in the data source. If not, there could be an easy fix by simply issuing a command to the data source asking it to partition it data. Once it is partitioned there, a good database-specific connector will automatically be able to take advantage of this partitioning. However, not all connectors are created equal. If you are finding that the one in your current distribution is unable to take advantage of data partitioning:
   1. This is another reason to look around at other Trino distributions. Again, in many cases, Starburst Enterprise and Galaxy provide significantly improved connectors over what is available in open source (such as its Oracle and Teradata connectors). 
   2. Alternatively, it could be because your partitioning scheme is not fine-grained enough — i.e. you don’t have enough partitions. You might want to try partitioning on a different attribute to see if that improves anything. 
5. Another approach is to try using more aggressive filters in your queries. Since Trino usually pushes down filters to the underlying data source, the stronger the filter, the less data needs to be extracted from the underlying database. Sometimes it is actually faster to issue 10 separate queries, each of which select 1/10 of the tuples, than a single query that selects all tuples, since each of these 10 queries can create separate JDBC connections that work in parallel.
6. If all the above options fail to improve the situation, at some point it might be time to give up trying to use Trino to access that data source directly. Rather, a good best practice is to create periodic snapshots of the data source and store them as Parquet files (or other open data formats) in a data lake. Trino rarely hits data access bottlenecks when running over data lake data, so this will likely solve the problem (at the cost of needing to replicate data into the lake).
   1. Starburst Enterprise provides a helpful tool to automate this best practice via Table Scan Redirections. 

 

Conclusion

Data access bottlenecks are not uncommon in Trino, especially if JDBC is being used to access data in the data source from Trino. It is important to be aware that such bottlenecks may appear so that you can more easily identify and alleviate the bottlenecks using some of the approaches discussed in this blog post. At the end of the day, Trino is an extremely powerful tool to access data anywhere. But not all access points are created equal — Trino will have a much easier time accessing data in some sources over others, especially at Terabyte (or above) scale. 

In my next post, I will discuss the technical reasons for the JDBC bottleneck in more detail and show some practical performance differences (with real experimental numbers) that emerge from the use of different generic and system-specific connectors to access datasets in underlying database systems.