Presto 334 adds significant performance improvements for queries accessing nested fields inside struct columns. They have been optimized through the pushdown of dereference expressions. With this feature, the query execution prunes structural data eagerly, extracting the necessary fields.
RowType is a built-in data type of Presto, storing the in-memory
representation of commonly used nested data types of the connectors, eg.
STRUCT type in Hive. Datasets often contain wide and deeply nested structural
columns, i.e. a struct column having hundreds of fields, with the fields being
RowType columns can contain plenty of data, most of the
analytical queries access just a few fields out of it. Without dereference
pushdown, Presto scans the whole column, and shuffles all that data around
before projecting the necessary fields. This suboptimal execution causes higher
CPU usage, higher memory usage and higher query latencies, than required. The
unnecessary operations get even more expensive with wider/deeper structs and
more complex query plans.
LinkedIn’s data ecosystem makes heavy usage of nested columns. It is common to have 2-3 levels of nesting, and up to 50 fields in most of our tracking tables. Because of the query execution inefficiency for nested fields, ETL pipelines were set up at LinkedIn to copy the nested columns as a set of top-level columns corresponding to subfields. This step added overhead in our ingestion process and delayed data availability for analytics. It also caused ORC schemas to be inconsistent with the rest of the infrastructure, making it harder to migrate from existing flows on row-oriented formats.
Similarly, Lyft’s schemas make heavy use of nested data to decompose a ride into its routes, riders, segments, modes, and geo-coordinates. Prior to the performance improvements, analytical queries would either need to be run on clusters with very long timeouts, or the data would have to be flattened before being analyzed, adding an extra ETL step. Not only would this be costly, it would also cause the original schema to diverge in our data warehouse making it more difficult for data scientists to understand.
The dereference pushdown optimization in Presto is having a massive impact on the ingestion story at both LinkedIn and Lyft. Nested data is now being made available faster for consumption with a consistency of structure across all stores, while maintaining performance parity for analytical queries.
Say we have a Hive table
jobs, with a struct-typed column
job_info in the
schema. The column
jobs is wide and deeply nested, i.e.
varchar, requirements ROW(skills array(...), education ROW(...), salary ...) ,
...). Most queries would access a small percentage of data from this struct
using the dereference projection (the
. operation). Consider such a query
SELECT A.appid id, J.job_info.company c FROM applications A JOIN jobs J ON A.jobid = J.jobid LIMIT 100
It should suffice to scan only one field
executing this query. But, without dereference pushdown, Presto scans and
shuffles everything from
job_info, only to project a single field at the end.
Solution: Pushdown of Dereference Expressions
With dereference pushdown, Presto optimizes queries by extracting the sufficient
fields from a
ROW as early as possible. This is enforced by modifying the
query plan through a set of optimizers, and can be broadly divided into two
First, dereference projections are extracted in the query plan and pushed as close to the table scan as possible. This happens independent of what the connector is. Secondly, there is a further improvement for Hive tables. The Hive Connector and ORC/Parquet readers have been optimized to scan only the sufficient subfield columns.
Pushdown of predicates on the subfields is also a crucial optimization. For
example, if a query has filters on subfields (i.e.
a.b > 5), they should be
utilized by ORC/Parquet readers while scanning files. The pushdown helps with
the pruning of files, stripes and row-groups based on column-level statistics.
This optimization is achieved as a byproduct of the above two optimizations.
With the dereference pushdown, queries observe significant performance gains in terms of CPU/memory usage and query runtime, roughly proportional to the relative size of nested columns compared to the accessed fields.
Pushdown in Query Plan
The goal here is to execute dereference projections as early as possible. This usually means performing them right after the table scans.
A projection operation that performs dereferencing on input symbols (i.e.
job_info.company) reduces the amount of data going up the plan tree. Pushing
dereference projections down means that we are pruning data early. It reduces
the amount of data being processed and shuffled in query execution. For the
Q, the query plan looks like the following when dereference
pushdown is enabled.
job_info.company now directly follows the scan of
avoiding the propagation the
Join nodes. Note
that all of
job_info is still being scanned, and pruning it in the reader
requires connector-dependent optimizations.
Pushdown in the Hive Connector
In columnar formats like ORC and Parquet, the data is laid out in a columnar
fashion even for subfields. If we have a column
STRUCT(f1, f2, f3), the
f3 are stored as independent columns. An optimized
query engine should only scan the required fields through its ORC reader,
skipping the rest. This optimization has been added for Hive connector.
Dereference projections above a
TableScanNode are pushed down in the Hive
connector as “virtual” (or “projected”) columns. The query plan is modified to
refer to these new columns. For the query
jobs table would be scanned
differently with this optimization, as shown below. The projection is now
embedded in the Hive connector. Here,
job_info#company can be thought of as
a virtual column representing the subfield
The Hive connector handles the projections before returning columns to Presto’s engine. It provides the required virtual columns to format-specific readers. ORC and Parquet readers optimize their scans based on subfields required, increasing their read throughput. Subfield pruning is not possible for row-oriented format readers (e.g. AVRO). For them, Hive connector performs adaptation to project the required fields.
Pushdown of Predicates on Subfields
Columnar formats store per-column statistics in the data files, which can be
used by the readers for filtering. eg. if a query contains filter
y = 5 for a
y, Presto’s ORC reader can skip ORC stripes and files by
looking at the upper and lower bounds for
y in the statistics.
The same concept of predicate-based pruning can work for filters involving
subfields, since the statistics are also stored for subfield columns. i.e.
Presto’s ORC/Parquet reader should be able to filter based on a constraint like
x.f1 = 5 for more optimal scans. Good news! In the final optimized plan,
predicates on a subfield are pushed down to the hive connector as a constraint
on the corresponding virtual column, and later used for optimizing the scan.
The complete logic is a bit complicated to explain here, but can be illustrated
through the following example.
Given an initial plan with a predicate on a dereferenced field (
x.f1 = 5), a
chain of optimizers transform it to a more optimal plan with reader-level
predicates. In the future, the same optimization will be added to the Parquet
In the final plan, Hive connector knows to scan the column
y and the subfield
x.f1. It also takes advantage of the “virtual” column constraint
x#f1 = 5
for reader-level pruning.
Dereference pushdown improves performance for queries accessing nested fields in multiple ways. First, it increases the read throughput for table scans, reducing the CPU time. The pruning of fields during the scan also means lesser data to process for all downstream operators and tasks. So the early projections result in more optimal execution for any operations that involve shuffle or copy of data. Moreover, for ORC/Parquet, the read performance improves in the case of selective filters on subfields.
Below are some experimental results on a production dataset at LinkedIn which
STRUCT columns, having ~20-30 small subfields in each. The
example queries used in the analysis access only a few subfields. The queries
have been listed as their approximate query shape for the sake of brevity. The
plots compare CPU usage, peak memory usage and averaged query wall time.
CPU usage and peak memory usage show orders-of-magnitude improvement in
presence of dereference pushdown. Query wall times also reduce considerably,
and this improvement is more drastic for the relatively complex
Please note that these are not benchmarks! The performance improvement you’ll
see will vary depending on how many columns are contained in your nested data
versus how many you’ve referenced. At Lyft we saw improvements of
50x for some
The pushdown of dereference expressions can be extended to arrays. i.e.
dereference operations applied after unnesting an array should also get pushed
down to the readers. For example, using our jobs table from before, our
jobs.job_info structure may contain a repeating structure such as
required_skills. With the following query, the entire required_skills
structure would be read even though only a small part of it is being referenced.
SELECT S.description FROM jobs J CROSS JOIN UNNEST (job_info.required_skills) S WHERE S.years_of_experience >= 2
The work for this improvement is being tracked in this issue.
Similar to Hive Connector, connector-level dereference pushdown can be extended to other connectors supporting nested types.
Another future improvement will be the pushdown of predicates on subfields for data stored in Parquet format. Although the pruning of nested fields occurs with Parquet, the predicates are not yet pushed down into the reader.
Pushing down dereference operations in the query provides massive performance gains, especially while operating on large structs. At LinkedIn and Lyft, this feature has shown great impact for analytical queries on nested datasets.