In my previous post, I discussed the enormous performance problems that can emerge when accessing data inside an underlying system from Trino using a JDBC connection. You don’t need to read that previous post before reading this one. The main thing you need to know is that JDBC was not designed to extract large amounts of data from a database system, yet Trino often wants to extract large amounts of data from any data source that it is connected to. If large amounts of data is being extracted over a single JDBC connection, it almost certainly will become a massive performance bottleneck.
As an end user, it is often not immediately obvious whether Trino is accessing data over JDBC or not. But if you are accessing data in a traditional database system such as Oracle, SQL Server, MySQL, or PostgreSQL, JDBC is probably being used.
In this post, I will make the problem, along with some of the mitigation techniques discussed in my previous post, more concrete with real numbers on industry-standard benchmark data. We will see how serious the problem really is, along with how well the mitigation techniques work.
Experimental setup
In order to understand the performance problems caused by JDBC and how it varies across different Trino distributions, we ran some experiments in my lab at the University of Maryland, led by my PhD student Chujun Song. We ran the TPC-H benchmark — a suite of 22 representative queries from a retail data analysis application.
TPC-H consists of 8 tables, corresponding to customers, orders, parts, and other important information found in typical retail applications. In order to understand how JDBC can become a bottleneck, we stored these tables in different locations — some of them in parquet files inside HDFS (that Trino can access without JDBC) and some stored in traditional database systems that Trino accesses over a JDBC connection.
We ran our experiments on CloudLab, c22og2 nodes. HDFS was installed on four such nodes, while the database systems were installed on a single node. Trino was also given 4 nodes to perform its parallel query execution.
As far as Trino distributions, we experimented with the standard open source Trino distribution along with the Presto version of the codebase. Note that Trino and Presto used to be a single project before branching off from each other over five years ago. We also experimented with Starburst Enterprise, running an implementation of Trino.
A detailed look at the JDBC bottleneck
The results of these experiments were quite interesting. There were a surprising number of differences across the distributions that caused significant performance differences (beyond only the JDBC bottleneck). This post would get far too long if we discussed every query here. Instead, we will focus on discussing the JDBC-related results and save discussion of other results for future posts that will address other important performance considerations beyond the JDBC bottleneck.
TPC-H Query 14
Query 14 from the benchmark illustrates the ugliness of the JDBC bottleneck. Query 14 performs a join between the lineitem (from customer orders) and part tables, analyzing the revenue from orders stemming from a particular part promotion during a one-month period of data. The full SQL query is shown below.
SELECT 100.00 * sum(case when p_type like 'PROMO%' then l_extendedprice * (1 - l_discount) else 0 end) / sum(l_extendedprice * (1 - l_discount)) as promo_revenue FROM lineitem, part WHERE l_partkey = p_partkey AND l_shipdate >= date '1995-09-01' AND l_shipdate < date '1995-09-01' + interval '1' month;
In our experiments, the lineitem table was one of the tables stored in HDFS, while the part table was stored in the Oracle database system. Therefore, the lineitem table can be accessed without JDBC, while Trino can only access the part table in Oracle via a JDBC connection through Trino’s Oracle database connector.
We ran this experiment at several different TPH-H scales factors: 1, 10, 100, and 1000. Each scale factor represents the approximate size of the database. The scale factor of 1 corresponds to an approximate 1GB database, while a scale factor of 1000 corresponds to an approximate 1TB database.
At the scale factor of 100, the lineitem table is 52.71GB. It is the largest table in the dataset since it contains all the detailed information about customer orders. The part table is much smaller, consisting of 8.72GB.
Unlike some of the other queries in the benchmark in which the different distributions of Trino chose different execution strategies (to be discussed in future posts), for query 14, each distribution executes the same way. Specifically, they execute this query as a hash join, with the smaller table (part) placed into a hash table partitioned across the four computing nodes, and the larger table (lineitem) scanned, where each lineitem tuple is used to probe the hash table (made from the part dataset) to find join matches.
Thus, both tables need to be scanned as part of this query. 8.72GB needs to be scanned from the part table, and 52.71GB needs to be scanned from the lineitem table. Note that Trino does not push down the predicate on lineitem for this query, so the entire table must be sent to Trino.
It should be obvious which scan should take longer. Clearly, it should be the lineitem table!
However, the JDBC bottleneck rears its ugly head for this query. The lineitem table can be accessed without JDBC and takes 5.11 seconds to scan the 6 billion rows of the lineitem table. The part table needs to be accessed over JDBC and takes 273 seconds to access its 0.2 billion rows.
In other words, lineitem data is 6 times the size of the part data, yet can be accessed 53.4 times faster. This means its extraction rate into Trino is 320 times the extraction rate of the part table. This is because the lineitem table is divided into multiple partitions in HDFS, and Trino uses 2537 parallel connections to extract the lineitem table. In contrast, the part table is extracted over a single JDBC connection. Obviously, this makes the extraction of the part table much slower.
TPC-H Queries 2, 9, 11, 16, 19, and 20
Query 14 was not the only query to be bottlenecked by the scan of the part table. Query 19 also does a join between the lineitem and part table, and differs from Query 14 mainly in its use of a more complicated WHERE clause predicate. This query also took over 300 seconds to run, again due to the bottleneck of reading the part table from Oracle over a JDBC connection.
To confirm that the problem was not specific to Oracle, we stored some of the other TPC-H tables in different database systems. We stored the customers table (which is close to the same size as the part table) in MySQL, and the partsupp table (which is 4 times the size of the part table) in PostgreSQL.
We noticed that queries 2, 9, 11, 16, and 20 were bottlenecked from reading the partsupp table from PostgreSQL. Queries 2, 9, 11, and 20 took approximately 4 times as long as Query 14, since the partsupp table is 4 times as large. Query 16 did not quite take as long since there is a predicate on the partsupp table in the query, which is pushed down to PostgreSQL and therefore less data needs to read over JDBC. Similarly, queries 10, 13, and 22 were bottlenecked from reading the customers table from MySQL.
What should we learn from these results?
The bottom line is that the JDBC bottleneck is a huge problem. Indeed — quite a bit larger than many people realize. The lineitem table is by far the largest table in TPC-H, but yet was multiple orders of magnitude faster to access in our benchmarking efforts because it was possible to access without using a JDBC connection.
This does not mean that Trino should not be used to access data in database systems over a JDBC connection. First of all, the fact that we could run these TPC-H queries at all, with each table stored in a different system — HDFS, Oracle, MySQL, PostgreSQL, Iceberg, etc. is impressive. There are very few systems that would be capable of running such a federated benchmark, and certainly not at the performance that Trino was able to achieve. We ran the same benchmark on a commercial traditional data virtualization system (that we will keep anonymous for legal reasons) that performed an order of magnitude worse than Trino, and in fact failed to run several of the TPC-H queries at large scale.
Second, and more importantly, the JDBC bottleneck is not always present. Sometimes it is possible to avoid it. We discussed some of these techniques in a previous post. However, these techniques may require a different Trino distribution.
JDBC Bottleneck Mitigation with Starburst
The most important of these techniques is to create multiple parallel JDBC connections, all extracting a different partition of the data. This technique works best when:
- The data is already partitioned (it does not matter by which attribute it is partitioned by) in the underlying system, and
- The Trino database connector is knowledgeable about these partitions and knows how to extract them from that system.
For example, Starburst includes upgraded Trino database connectors for several different underlying database systems, including Oracle. Therefore, unlike the open source distribution of Trino, Starburst is able to create multiple parallel JDBC connections to Oracle, each one extracting data from a different partition.
To see the difference in performance that stems from this upgraded connector, we ran the same benchmark discussed above using Starburst Enterprise. Each TPC-H table was stored in the same place as before (lineitem in HDFS, customers in MySQL, part in Oracle, etc.), and Starburst was configured to connect to each underlying system with its optimized connector for that system.
There were several interesting differences between Starburst and open source Trino that we discovered when running this benchmark. Some of the queries were optimized differently, some were processed differently, and there were some instances of Starburst being able to complete queries for which Trino ran out of memory. We will discuss some of these other discoveries in a future post.
However, those queries that we mentioned above were limited by the JDBC bottleneck when extracting data from Oracle performed significantly better when using Starburst’s Oracle connector.
For Query 14, we said above that it took a total of 273 seconds to scan the part table from Trino. The total query time was 277 seconds, since most of the other parts of query processing could be done in parallel while the part table is being read. In contrast, when running the query in Starburst, the total query time took 37.7 seconds. In other words, the query was performed over 7 times faster when using parallel JDBC connections to extract different data partitions from Oracle.
Similarly, for Query 19, it originally took 424 seconds. However, in Starburst, it took 51.9 seconds. This comes out as an 8X speedup from using Starburst (though only 7X can be attributed to the parallel JDBC connections).
In these experiments, there were 20 partitions in Oracle. However, the speedup from accessing these partitions in parallel was 7X instead of 20X. Since Oracle was running on a single machine, there is clearly a limit to how many parallel partition reads it is capable of performing.
To understand how the number of partitions affects the speedup, we ran an additional experiment in which we varied the number of partitions that the part table was partitioned into by Oracle. We varied this number from 1 to 100 partitions. The results of this experiment are shown in the figure below.
This figure indicates that there are definitely diminishing marginal returns with respect to the number of partitions. The biggest jump in performance is going from a single partition to 5 partitions. However, performance continues to improve until around 20 partitions, and then it levels off.
Thus, we can conclude that any amount of partitioning is a big win relative to no partitioning at all. You do not need a large number of partitions to get the performance benefits of parallel JDBC connections. However, there is a limit to how much of a performance improvement you can get if the underlying database system is running only on a single machine.
Nonetheless, if the underlying database system is running on a large cluster of machines, such as in Teradata or Oracle Exadata, the improvement would be expected to be even larger. For example, if Starburst is accessing a 10-machine Teradata cluster, one should expect a factor of 10 extra performance improvement.
Other mitigation strategies
We also ran an experiment in which the partitioning was done manually instead of physically in the underlying system. The part table was stored as a single partition in Oracle; however, it was logically divided into partitions based on its key (partkey) within Trino (see SQL below), so that each range could be extracted separately via a parallel JDBC connection.
This idea is that by expressing the part table as a union of separate queries to the part table, with each query accessing a unique range of the partkey attribute, when the query accesses it, it will end up issuing multiple different select statements. Each of these select statements will be extracted from the underlying system via a separate JDBC connection, thereby achieving our goal of parallelizing this extraction process.
When partitioning the part table in this manual way, we expect to achieve the benefit of parallel JDBC connections that we saw in the previous experiments. However, this puts significantly more pressure on Oracle, since the data is not pre-partitioned by these WHERE clause predicates, so each partition has to be extracted via a new scan of the complete part table.
Indeed, the figure above shows that this manual / logical partitioning is able to get approximately a 3X improvement relative to not partitioning the data at all (i.e., having a single partition as shown on the left side of the figure). However, it is not able to match the performance of having real, physical partitions in Oracle that are much easier for Oracle to extract in parallel.
Conclusions
It is clear from these experiments that the only way to avoid the JDBC bottleneck is to have many parallel JDBC connections to the underlying database system. The best way to do this is to partition the data in the underlying database system and use a database connector that is smart enough to be able to leverage these partitions. This is a major reason to use Starburst over open source Trino. If this is not possible, then creating logical partitions at the Trino level is certainly better than not doing anything at all to avoid this bottleneck.
The bottom line is that Trino should still certainly be used to access data located inside database systems. Although this process is a little more challenging to perform at high performance, avoiding the database system data entirely would be wasting Trino’s renowned ability to access data anywhere. However, we need to take action to make sure that JDBC is not a bottleneck. This could be as simple as running a single one-time command to partition the data in the underlying database. However, if you are not using Starburst, you may need to work a little harder to create logical partitions instead.
