Procella: Unifying serving and analytical data at YouTube
Biswapesh Chattopadhyay Priyam Dutta Weiran Liu Ott Tinn
Andrew Mccormick Aniket Mokashi Paul Harvey Hector Gonzalez
David Lomax Sagar Mittal Roee Ebenstein Nikita Mikhaylin Hung-ching Lee
Xiaoyan Zhao Tony Xu Luis Perez Farhad Shahmohammadi Tran Bui
Neil McKay Selcuk Aya Vera Lychagina Brett Elliott
Google LLC

procella-paper@google.com
ABSTRACT• Reporting and dashboarding: Video creators, content owners, and various internal stakeholders at
YouTube need access to detailed real time dashboards
to understand how their videos and channels are performing. This requires an engine that supports executing tens of thousands of canned queries per second
with low latency (tens of milliseconds), while queries
may be using filters, aggregations, set operations andjoins. The unique challenge here is that while our data
volume is high (each data source often contains hundreds of billions of new rows per day), we require nearreal-time response time and access to fresh data.

Large organizations like YouTube are dealing with exploding
data volume and increasing demand for data driven applications. Broadly, these can be categorized as: reporting anddashboarding, embedded statistics in pages, time-series monitoring, and ad-hoc analysis. Typically, organizations build
specialized infrastructure for each of these use cases. This,
however, creates silos of data and processing, and results in
a complex, expensive, and harder to maintain infrastructure.
At YouTube, we solved this problem by building a new
SQL query engine – Procella. Procella implements a superset of capabilities required to address all of the four use cases
above, with high scale and performance, in a single product.
Today, Procella serves hundreds of billions of queries per
day across all four workloads at YouTube and several other
Google product areas.

• Embedded statistics: YouTube exposes many realtime statistics to users, such as likes or views of a video,
resulting in simple but very high cardinality queries.
These values are constantly changing, so the system
must support millions of real-time updates concurrently
with millions of low latency queries per second.

PVLDB Reference Format:
Biswapesh Chattopadhyay, Priyam Dutta, Weiran Liu, Ott Tinn,
Andrew Mccormick, Aniket Mokashi, Paul Harvey, Hector Gonzalez, David Lomax, Sagar Mittal, Roee Ebenstein, Hung-ching Lee,
Xiaoyan Zhao, Tony Xu, Luis Perez, Farhad Shahmohammadi,
Tran Bui, Neil McKay, Nikita Mikhaylin, Selcuk Aya, Vera Lychagina, Brett Elliott. Procella: Unifying serving and analytical
data at YouTube. PVLDB, 12(12): 2022-2034, 2019.
DOI: https://doi.org/10.14778/3352063.3352121

1.

• Monitoring: Monitoring workloads share many properties with the dashboarding workload, such as relatively simple canned queries and need for fresh data.
The query volume is often lower since monitoring is typically used internally by engineers. However, there is aneed for additional data management functions, suchas automatic downsampling and expiry of old data, andadditional query features (for example, efficient approximation functions and additional time-series functions).

INTRODUCTION

YouTube, as one of the world’s most popular websites,
generates trillions of new data items per day, based on billionsof videos, hundreds of millions of creators and fans, billionsof views, and billions of watch time hours. This data is used
for generating reports for content creators, for monitoring
our services health, serving embedded statistics such as video
view counts on YouTube pages, and for ad-hoc analysis.
These workloads have different set of requirements:

This work is licensed under the Creative Commons AttributionNonCommercial-NoDerivatives 4.0 International License. To view a copyof this license, visit http://creativecommons.org/licenses/by-nc-nd/4.0/. For
any use beyond those covered by this license, obtain permission by emailing
info@vldb.org. Copyright is held by the owner/author(s). Publication rights
licensed to the VLDB Endowment.
Proceedings of the VLDB Endowment, Vol. 12, No. 12
ISSN 2150-8097.
DOI: https://doi.org/10.14778/3352063.3352121• Ad-hoc analysis: Various YouTube teams (data scientists, business analysts, product managers, engineers)need to perform complex ad-hoc analysis to understandusage trends and to determine how to improve the product. This requires low volume of queries (at most tens
per second) and moderate latency (seconds to minutes)
of complex queries (multiple levels of aggregations, set
operations, analytic functions, joins, unpredictable patterns, manipulating nested / repeated data, etc.) overenormous volumes (trillions of rows) of data. Query patterns are highly unpredictable, although some standard
data modeling techniques, such as star and snowflake
schemas, can be used.
Historically, YouTube (and other similar products at
Google and elsewhere) have relied on different storage andquery backends for each of these needs: Dremel [33] for ad-hoc

2022Realtime
Data Flow
Batch
Data Flow
Query Flow

Batch
Process• Data needed to be loaded into multiple systems using different Extract, Transform, and Load (ETL)
processes, leading to significant additional resource
consumption, data quality issues, data inconsistencies,
slower loading times, high development and maintenance cost, and slower time-to-market.

• Rich API: Procella supports an almost complete implementation of standard SQL, including complex multistage joins, analytic functions and set operations, withseveral useful extensions such as approximate aggregations, handling complex nested and repeated schemas,
user defined functions, and more.

Compaction
ServerRealtime
Data– Vertical scaling is challenging. Each server runs
many tasks, each potentially with different resource needs. To improve overall fleet utilization,
it is better to run many small tasks than a small
number of large tasks.
– Borg master can often bring down machines formaintenance, upgrades, etc. Well behaved tasks
thus must learn to recover quickly from being
evicted from one machine and brought up in adifferent machine. This, along with the absenceof local storage, makes keeping large local state
impractical, adding to the reasons for having alarge number of smaller tasks.
– A typical Borg cluster will have thousands of inexpensive machines, often of varying hardware configuration, each with a potentially different mix oftasks with imperfect isolation. Performance of atask, thus, can be unpredictable. This, combinedwith the factors above, means that any distributed
system running on Borg must deploy sophisticatedstrategies for handling badly behaving tasks, random task failures, and periodic unavailability due
to evictions.

• High Performance: Procella uses state-of-the-art
query execution techniques to enable efficient executionof high volume (millions of QPS) of queries with very
low latency (milliseconds).
• Data Freshness: Procella supports high volume, lowlatency data ingestion in both batch and streaming
modes, the ability to work directly on existing data,and native support for lambda architecture [35].

Procella is designed to run on Google infrastructure.
Google has one of world’s most advanced distributed systems
infrastructure, with several notable features, which have hada large impact on Procella’s design:• Disaggregated storage: All durable data must be
stored in Colossus. No local storage is available to thesystem. Colossus, while being almost infinitely scalable,
differs from local storage in a few important ways:

Ingestion
Server(IgS)

Registration
Server (RS)• Shared compute: All servers must run on Borg, ourdistributed job scheduling and container infrastructure.
This has several important implications:• High Scalability: Procella separates compute (running on Borg [42]) and storage (on Colossus [24]), enabling high scalability (thousands of servers, hundredsof petabytes of data) in a cost efficient way.

Google Infrastructure

Data Server(DS)– Data is immutable. A file can be opened andappended to, but cannot be modified. Further,once finalized, the file cannot be altered at all.
– Common metadata operations such as listing andopening files can have significantly higher latency(especially at the tail) than local file systems, because they involve one or more RPCs to the Colossus metadata servers.
– All durable data is remote; thus, any read or write
operation can only be performed via RPC. As
with metadata, this results in higher cost andlatency when performing many small operations,especially at the tail.

To solve these problem, we built Procella, a new distributed
query engine. Procella implements a superset of features
required for the diverse workloads described above:

2.1

Metadata
Store

Query
ClientFigure 1: Procella System Architecture.

• Several of the underlying components had performance,
scalability and efficiency issues when dealing with data
at YouTube scale.

ARCHITECTURE

Root
Server (RS)Distributed non-volatile ﬁle system (Colossus)• Since each internal system uses different languages
and API’s, migrating data across these systems, toenable the utilization of existing tools, resulted in reduced usability and high learning costs. In particular,
since many of these systems do not support full SQL,
some applications could not be built by using somebackends, leading to data duplication and accessibility
issues across the organization.

2.

Metadata
Server(MDS)

Remotememoryanalysis and internal dashboards, Mesa [27] and Bigtable [11]for external-facing high volume dashboards, Monarch [32]for monitoring site health, and Vitess [25] for embedded
statistics in pages. But with exponential data growth andproduct enhancements came the need for more features, better performance, larger scale and improved efficiency, making
meeting these needs increasingly challenging with existing
infrastructures. Some specific problems we were facing were:

Figure 1 illustrates the overall Procella architecture. Ourarchitecture is structured from multiple components, each2023

executes in a distributed manner. When a component does
not have any executing instance, the functionality it provides
is unavailable (for example, even when data servers are down,
ingestion and registration can function).

2.2

Procella’s Data

2.2.1

Data Storage

Like most databases, Procella data is logically organized
into tables. Each table’s data is stored across multiple files(also referred to as tablets or partitions). Procella uses its
own columnar format, Artus (Section 3.2) for most data, butalso supports querying data stored in other formats suchas Capacitor [34]. All durable data is stored in Colossus,
allowing Procella to decouple storage from compute; i.e,the servers processing the data can be scaled to handle
higher traffic independently of the size of the underlying
data. Additionally, the same data can be served by multiple
instances of Procella.

2.2.2

Metadata Storage

Like many modern analytical engines [18, 20], Procelladoes not use the conventional BTree style secondary indexes,
opting instead for light weight secondary structures such aszone maps, bitmaps, bloom filters, partition and sort keys [1].
The metadata server serves this information during query
planning time. These secondary structures are collected
partly from the file headers during file registration, by theregistration server, and partly lazily at query evaluation timeby the data server. Schemas, table to file mapping, stats,zone maps and other metadata are mostly stored in themetadata store (in Bigtable [11] and Spanner [12]).

2.2.3

Table management

Table management is through standard DDL commands(CREATE, ALTER, DROP, etc.) sent to the registration
server (RgS), which stores them in the metadata store. The
user can specify column names, data types, partitioning
and sorting information, constraints, data ingestion method(batch or realtime) and other options to ensure optimal tablelayout. For real-time tables, the user can also specify how to
age-out, down-sample or compact the data; this is importantfor monitoring applications and for supporting a lambda
architecture where batch data replaces real time data at aregular cadence.
Once a table object is created in the system, the table data
can be populated, or ingested, in batch mode or realtime
mode. Both of the ingestion modes differ in the optimizations
they use.

2.2.4

Batch ingestion

Users generate data using offline batch processes (such asa MapReduce [15]) and register the data by making a DDL
RPC to the RgS. This is the most common method employedby automated pipelines that refresh data at a regular cadence,such as hourly or daily. During the data registration step,
the RgS extracts mapping of table to files and secondary
structures (described in section 2.2.2 above) from file headers.
We avoid scanning data during the registration step to enable
fast data registration. However, we may utilize Procella dataservers to lazily generate expensive secondary structures
if the required indexing information is not present in thefile headers, for example, if bloom filters are missing inexternally generated data. There are no prerequisites ondata organization for registering data with Procella [40]. In
practice, however, for best performance, data should be laidout optimally using partitioning, sorting, etc.
The RgS is also responsible for sanity checks during thetable and data registration step. It validates backwards
compatibility of schemas, prunes and compacts complex
schemas, ensures that file schemas are compatible with thetable schemas registered by the user, etc.

2.2.5

Realtime ingestion

The ingestion server (IgS) is the entry point for real-timedata into the system. Users can stream data into it using
a supported streaming mechanism such as RPC or PubSub.
The IgS receives the data, optionally transforms it to alignwith the table structure and appends it to a write-ahead
log on Colossus. In parallel, it also sends the data to thedata servers according to the data partitioning scheme forthe table. The data is (temporarily) stored in the memory
buffer of the data server for query processing. The buffers arealso regularly checkpointed to Colossus to help in recovery if
and after crash or eviction occurs, but this is best-effort anddoes not block query access to the data. The IgS (optionally) sends the data to multiple data servers for redundancy.
Queries access all copies of the buffered data and use themost complete set.
The write-ahead log is compacted in the background bycompaction tasks that provide durable ingestion.
Having the data follow two parallel paths allows it to beavailable to queries in a dirty-read fashion in seconds or evensub-second, while being eventually consistent with slower
durable ingestion. Queries combine data from the in-memory
buffers and the on-disk tablets, taking from the buffers only
data that is not yet durably processed. Serving from thebuffers can be turned off to ensure consistency, at the costof additional data latency.

2.2.6

Compaction

The compaction server periodically compacts and repartitions the logs written by the IgS into larger partitioned
columnar bundles for efficient serving by the data servers.
During the compaction process, the compaction server canalso apply user defined SQL based logic (specified duringtable registration) to reduce the size of the data by filtering, aggregation, aging out old data, keeping only the latest
value, etc. Procella allows a fairly rich SQL based logic tobe applied during the compaction cycle, thus giving more
control to the user to manage their real-time data.
The compaction server updates the metadata store(through the RgS) after each cycle, removing metadata about
the old files and inserting metadata about the newly generated files. The old files are subsequently deleted by abackground process.

2.3

Query Lifecycle

Clients connect to the Root Server (RS) to issue SQL
queries. The RS performs query rewrites, parsing, planningand optimizations to generate the execution plan. To thisend, it uses metadata such as schema, partitioning and index
information from the Metadata Server (MDS) to prune thefiles to be read, as detailed in Section 3.4. It then orchestrates
the query execution as it goes through the different stages

2024

enforcing timing/data dependencies and throttling. To address the needs of executing complex distributed query plans(timing/data dependencies, diverse join strategies, etc.), theRS builds a tree composed of query blocks as nodes and data
streams as edges (Aggregate, Execute Remotely, Stagger
Execution, etc.) and executes it accordingly. This enables
functionality such as shuffle (that requires timing dependency), uncorrelated subquery, subquery broadcast joins(data dependency) and multi-level aggregation. This graphstructure allows several optimizations by inserting custom
operations into the tree based on query structure. Once the
RS receives the final results, it sends the response back to
the client, along with statistics, error and warning messages,
and any other information requested by the client.
The Data Server (DS) receives plan fragments from the
RS or another DS and does most of the heavy lifting, such as
reading the required data (from local memory, remote anddistributed files in Colossus, remote memory using RDMA, or
another DS), executing the plan fragment and sending the results back to the requesting RS or DS. As in most distributed
SQL engines, Procella aggressively pushes compute close to
the data. The plan generator ensures that filters, project expressions, aggregations (including approximate aggregations
such as TOP, UNIQUE, COUNT DISTINCT and QUANTILE), joins, etc. are pushed down to the DS to the extent
possible. This allows the DS to use the encoding-native
functions for these operators for optimal performance. Data
servers use Stubby [41] to exchange data with other dataservers and RDMA for shuffle (Procella reuses the BigQuery
shuffle library [4]).

3.

Note that since Colossus files are essentially immutableonce closed, cache consistency is simply a matter ofensuring that file names are not reused.
• Metadata caching: Procella scales metadata storage by using a distributed storage system (Bigtable or
Spanner) to store, and a distributed metadata service
to serve metadata. However, this means that, often,
metadata operations such as fetching table to file namemappings, schemas, and constraints can become a bottleneck. To avoid this, the metadata servers cache thisinformation in a local LRU cache.
• Affinity scheduling: Caches are more effective wheneach server caches a subset of the data. To ensure this,Procella implements affinity scheduling to the dataservers and the metadata servers to ensure that operations on the same data / metadata go to the same serverwith high probability. This means that each server isonly responsible for serving a small subset of the data /metadata, which significantly improves cache hit ratio,
dramatically reducing the time spent fetching remote
data. An important aspect of Procella’s scheduling isthat the affinity is loose, i.e. the request can go to adifferent server (for example, if the primary is down orslow). When this happens, the cache hit probability
is lower, but since the data / metadata itself is stored
in a reliable durable storage (Bigtable, Spanner, orColossus), the request still completes successfully. This
property is crucial for high availability in the serving
path in the face of process evictions, overloaded machines, and other issues associated with running in alarge shared cluster.

OPTIMIZATIONS

Procella employs several techniques to achieve high query
performance for various query patterns (lookups, reporting,ad-hoc queries, and monitoring). This section documents
several of these techniques.

3.1

Caching

Procella achieves high scalability and efficiency by segregating storage (in Colossus) from compute (on Borg). However,this imposes significant overheads for reading or even opening
files, since multiple RPCs are involved for each. Procellaemploys multiple caches to mitigate this networking penalty:• Colossus metadata caching: To avoid file open calls
to the Colossus name server, the data servers cache
the file handles. The file handles essentially store themapping between the data blocks and the Colossus
servers that contain the data. This eliminates one ormore RPC roundtrips on file opens.

The caching schemes are designed such that when there
is sufficient memory, Procella essentially becomes a fully
in-memory database. In practice, for our reporting instance,
only about 2% of the data can fit in memory, but accesspatterns and cache affinity ensures that we get 99%+ filehandle cache hit rate and 9̃0% data cache hit rate.

3.2

Data format

The first implementation of Procella used Capacitor as thedata format, which was primarily designed for large scans
typical in ad-hoc analysis workloads. Since Procella aims tocover several other use cases requiring fast lookups and range
scans (such as serving embedded statistics on high traffic
pages), we built a new columnar file format called Artus,
which is designed for high performance on both lookups andscans. Specifically, Artus:• Uses custom encodings, avoiding generic compression
algorithms like LZW. This ensures that it can seek tosingle rows efficiently without needing to decompress
blocks of data, making it more suitable for small point
lookups and range scans.

• Header caching: The header (or sometimes, thefooter) of columnar files contain various column metadata such as start offset, column size, and minimum
and maximum values. Procella data servers cache theheaders in a separate LRU cache, thus avoiding moreround trips to Colossus.

• Does multi-pass adaptive encoding; i.e. it does a firstpass over the data to collect lightweight information(e.g. number of distinct values, minimum and maximum, sort order, etc.) and uses this information todetermine the optimal encoding to use for the data.
Artus uses a variety of methods to encode data: dictionary encoding with a wide variety of dictionary and• Data caching: The DS caches columnar data in a
separate cache. The Procella data format, Artus, isdesigned so that data has the same representation inmemory and on disk, making cache population fairly
cheap. In addition, the DS caches derived informationsuch as output of expensive operations and bloom filters.

2025

indexer types, run-length, delta, etc. to achieve compression within 2x of strong general string compression(e.g. ZSTD) while still being able to directly operateon the data. Each encoding has estimation methods
for how small and fast it will be on the data supplied.
Once the user has specified their objective function (relative value of size versus speed), Artus will use thesemethods to automatically choose the optimal encodings
for the provided data.

Table 1: Queries for Artus benchmark
Query Id Query
Q1
SUM(views) where pk = X: pure lookup, 1
matching row
Q2
SUM(views) where fk = X: moderate filter,
5000 matching rows
Q3
SUM(views): full scan, all 250k matching
Q4
SUM(views), fk GROUP BY fk: full scan withgroup by

• Chooses encodings that allow binary search for sorted
columns, allowing fast lookups in O (logN ) time.
Columns also support O (1) seeks to a chosen row number, which means that a row with K columns matchinga given primary key can be found in O (logN + K). For
columns that allow direct indexing, such as packing
integers into a fixed number of bits, providing theseO (1) seeks is trivial, whereas for other columns, suchas run-length-encoded columns, we must maintain additional ‘skip block’ information that records the internal
state of the column every B rows, allowing us to ‘skip’
into the encoded data as if we had scanned throughit until that point. Note that technically the seeks onthese types of columns are O (B) rather than O (1) as
we must iterate forward from the last skip block. The
values used for B are generally quite small, such as 32
or 128, and represent a tradeoff between compression
size and lookup speed. Fast row lookups are criticalfor lookup queries and distributed lookup joins, where
we treat the right side table as a distributed hash map.
• Uses a novel representation for nested and repeateddata types that is different from the method originally implemented in ColumnIO [33] and subsequently
adopted by Capacitor, Parquet and others. We visualize a table’s schema as a tree of fields. We store aseparate column on disk for each of these fields (unlike
rep/def, which only stores leaf fields). When ingestinga record, each time a parent field exists, we note thenumber of times each of its children occur. For optional
fields, this is either 0 or 1. For repeated fields, thisnumber is non-negative. Of particular interest hereis that we do not record any information about fields
whose parent does not exist; we are thus able to store
sparse data more efficiently than rep/def, which always
records data in each of its leaf fields. When reconstructing records, we essentially reverse the ingestion process,using the occurence information to tell us how many elements we need to copy. An important implementation
detail is that by storing this occurence information ina cumulative manner separately from the actual values
in the children we are able to still have O (1) seeks evenon nested and repeated data.
• Directly exposes dictionary indices, Run Length Encoding (RLE) [2] information, and other encoding
information to the evaluation engine. Artus also implements various common filtering operations natively
inside its API. This allows us to aggressively push suchcomputations down to the data format, resulting inlarge performance gains in many common cases.
• Records rich metadata in the file and column header.
Apart from the schema of the data, we also encode sorting, minimum and maximum values, detailed encoding

FormatSize (KB)

Table 2: Artus Data Size
Capacitor ArtusDisk Artusmemory
2406
2060
2754

information, bloom filters, etc. making many common
pruning operations possible without the need to readthe actual data in the column. In practice we often
range-partition our files on a primary key, allowing us
to use the metadata to prune file access based on thequery filters.
• Supports storing inverted indexes. Inverted indices arecommonly used at Google and elsewhere to speed up
keyword search queries [7,38]. While we could use themfor that purpose here, the main usage in Procella so far
has been in experiment analysis. In experiment analysis, we have a small array of low-cardinality integers ineach record, and we want to know which records containa particular integer within their array, i.e. ‘WHERE
123 IN ArrayOfExperiments’, and then do further analysis on them. Evaluating this filter is normally thedominant cost in a query, because each array contains
hundreds of elements. We avoid this cost by precomputing the inverted indices (the row offsets of records
whose arrays contain each value), storing them on disk
alongside regular columns, and then simply loading
the appropriate indices based on the query. By storingthese indices as Roaring bitmaps [10] we are able toeasily evaluate typical boolean filters (i.e. ‘WHERE
123 in ArrayOfExperiments OR 456 in ArrayOfExperiments’) efficiently, without having to go through thenormal evaluation pathway. In current production use
cases we have found that experiment analysis queries
have end to end latencies reduced by ∼ 500x orders ofmagnitude when we apply this technique.
Figure 2 presents results of a simple benchmark comparing
Artus with Capacitor on typical query patterns on a typicalYouTube Analytics dataset. The numbers represent the bestvalue observed in 5 runs of the test framework. The queries
were run on a single file with ˜250k rows sorted on the video
id and fully loaded into memory. Table 1 lists the queries
used for the benchmark.
Table 2 shows Artus uncompressed in-memory size, as wellas its LZW-compressed on-disk size, compared to Capacitor,
for the data set used for the benchmark.

3.3

Evaluation Engine

High performance evaluation is critical for low latency
queries. Many modern analytical systems today achieve thisby using LLVM to compile the execution plan to native2026

Figure 2: Artus vs Capacitor Performance.

Figure 4: Superluminal performance on Artus & Capacitor.
the open source Supersonic engine using TPC-H query #1.
The filter clause was removed to fairly compare only theaggregation performance, and the data was loaded into inmemory arrays to eliminate file format bias. The results (perCPU core) are in Figure 3. We also compared the performance on different formats, notably Capacitor, Artus and
raw C++ arrays. The results (per CPU core) are in Figure
4. As evidenced from the numbers:• Superluminal significantly improves on the legacy row
based engine (almost twice as fast on Capacitor, andalmost five times as fast on Artus, as shown in Figure4). However, operating on raw C++ arrays is still
significantly faster, albeit at the cost of much higher
memory usage 1 .

Figure 3: Superluminal vs Supersonic.
code at query time. However, Procella needs to serve bothanalytical and high QPS serving use cases, and for the latter,the compilation time can often become the bottleneck. The
Procella evaluation engine, Superluminal, takes a different
approach. Specifically, Superluminal:• Makes extensive use of C++ template metaprogramming for compile time code generation. This allows
utilizing many optimization techniques, such as partial
application, to be applied to a single function implementation automatically, while avoiding large virtual
call overheads.
• Processes data in blocks to take advantage of vectorized
computation [37] and cache-aware algorithms. Blocksizes are estimated to fit in the L1 cache memory forbest cache friendliness. The code is carefully written to
enable the compiler to auto-vectorize the execution [8].
• Operates on the underlying data encodings natively,
and preserves them during function application wherever possible. The encodings are either sourced fromthe file format or generated during execution as needed.
• Processes structured data in a fully columnar fashion.
No intermediatte representations are materialized.
• Dynamically combines filters and pushes them down theexecution plan, all the way to the scan node, allowing
the system to only scan the exact rows that are required
for each column independently.
We benchmarked the aggregation performance of Superluminal against our first genertion row based evaluator and• On C++ arrays, Superluminal is almost 5 times fasterthan Supersonic, as shown in Figure 3; however, a very
carefully hand-optimized native C++ implementation
specific to the problem is about 50% faster. In our
experience, a naive C++ implementation backed by astandard library hash map is an order of magnitude
slower than Superluminal.

3.4

Partitioning & Indexing

Procella supports multi-level partitioning and clustering.
Typically, most fact tables are date partitioned and withineach date, clustered by multiple dimensions. Dimension
tables would typically be partitioned and sorted by the dimension key. This enables Procella to quickly prune tabletsthat do not need to be scanned, and to perform co-partitioned
joins, avoiding large shuffles. This is a critical feature, especially for our external reporting instance which needs toanswer thousands of queries over tens of petabytes of datawith milliseconds latency.
The MDS is responsible for efficient storage and retrieval ofthis information. For high scalability, MDS is implemented
as a distributed service, with affinity scheduling for cache
efficiency, as described in Section 3.1. In-memory structures
used for caching the metadata are transformed from theirstorage format in Bigtable by using prefix, delta, RLE, andother encodings. This ensures that Procella can handle avery large volume of metadata (thousands of tables, billions
of files, trillions of rows) in a memory efficient way.
1
Artus files were about an order of magnitude smaller than
the raw data for the TPC-H dataset.

2027

Table 3: MDS Performance
Tablets
Tablets
Percentile Latency (ms)PrunedScanned50
61037455790
11863975237
95
151009479307
99
109
32799020
3974
99.9
336
37913412
7185
99.99
5030
38048271
11125

Table 3 illustrates the efficacy of the MDS tablets/partition
pruning for our external reporting instance. The pruning
itself is performed based on the filtering criteria within thequery, e.g. ids based on hashing, date ranges, etc.
Once the MDS has effectively pruned the tablets fromthe plan, the individual leaf requests are sent to the dataserver. The data server uses bloom filters, min/max values,
inverted indices, and other file level metadata to minimize
disk access based on the query filters. This information iscached lazily in an LRU cache in each data server. This
allows us to entirely skip evaluation for almost half of thedata server requests in our main reporting instance.

3.5
3.5.1

requests to ensure that only the minimum required datais fetched. Procella can execute such lookup joins withhigh performance and efficiency because the underlying
columnar format, Artus, supports fast lookups, without
the need to first transform the data into in-memoryhash table or similar structure.

3.5.2

In modern distributed scale out systems which operate on
cheap shared hardware, individual machines can often behave
unreliably. This makes achieving low tail latency difficult.
Various techniques exist to address this [14]. Procella employs
similar techniques to achieve low tail latencies for its queries,
but adapts them to its architecture. For example:• Since all Procella data is present in Colossus, any fileis accessible by any data server. The RS employs aneffective backup strategy to exploit this. Specifically,
it maintains quantiles of DS response latency dynamically while executing a query, and if a request takes
significantly longer than the median, sends a backup
RPC to a secondary data server.
• The RS limits the rates of, and batches, requests to the
DS for large queries to avoid overwhelming the samedata server with too many requests.

Distributed operations• The RS attaches a priority to each request it sends
to the DS – typically, this is set to high for smaller
queries and low for larger queries. The DS in turn,
maintains separate threads for high and low priority
requests. This ensures faster response for the smaller
queries, so that one very large query cannot slow down
the entire system.

Distributed Joins

Joins are often a challenge for distributed query execution
engines [6, 9, 21]. Procella has several join strategies thatcan be controlled explicitly using hints or implicitly by the
optimizer based on data layout and size:• Broadcast: This is the simplest join where one sideis small enough so it can be loaded into the memory ofeach DS running the query. Small dimensions such as
Country and Date often use this strategy.
• Co-partitioned: Often, the fact and dimension tables
can be partitioned on the same (join) key. In thesecases, even though the overall size of the RHS is large,
each DS only needs to load a small subset of the datain order to do the join. This technique is commonlyused for the largest dimension of a star schema (e.g.
the Video dimension table).
• Shuffle: When both sides are large and neither ispartitioned on the join key, data is shuffled on the join
key to a set of intermediate servers.
• Pipelined: When the RHS is a complex query, but
likey to result in a small set, the RHS is executed first
and the result is inlined and sent to the LHS shards,
resulting in a broadcast-like join.
• Remote lookup: In many situations, the dimension
table (build side) may be large but partitioned on thejoin key, however the fact table (probe side) is not. In
such cases, the DS sends remote RPCs to the DS serving
the build side tablets to get the required keys and values
for the joins. The RPC cost can often be high, so caremust be taken to apply all possible filters and batch
the keys into as few RPCs as possible. We also pushdown the projections and filters to the build side RPC

Addressing Tail Latency

3.5.3

Intermediate Merging

For queries that perform heavy aggregations, the final
aggregation often becomes the bottleneck as it needs toprocess large amounts of data in a single node. To avoid this,we add an intermediate operator at the input of the final
aggregation which buffers the data and dynamically spins
up additional threads to perform intermediate aggregations,
if the final aggregator cannot keep up with the responses
from the leaf data servers. This significantly improves theperformance of such queries, as can be seen in figure 5.

3.6
3.6.1

Query Optimization
Virtual Tables

A common technique used to maximize performance of low
latency high QPS queries (needed for high volume reporting)is materialized views [3]. The core idea is to generate multiple
aggregates of the underlying base table and choose the rightaggregate at query time (either manually or automatically).
Procella uses virtual tables to this end, which employs asimilar approach to this problem with some additional features to ensure optimal query performance. Specifically, theProcella virtual table implementation supports:• Index-aware aggregate selection: choose the righttable(s) based not just on size but also on data organization such as clustering, partitioning, etc. by matching
the filter predicates in the query to the table layout,
thus, minimizing data scans.

2028

Figure 5: Intermediate merging performance for different T OP (K) settings with increasing number ofintermediate aggregation threads.
• Stitched queries: stitch together multiple tables toextract different metrics from each using UNION ALL, if
they all have the dimensions in the query.
• Lambda architecture awareness: stitch together
multiple tables with different time range availabilities
using UNION ALL. This is useful for two purposes. First,
for stitching together batch and realtime data wherebatch data is more accurate and complete, but arrives
later than realtime data. Second, for making batch
data consistently available at a specific cutoff pointsooner by differentiating between batch tables thatmust be available to cover all dimension combinationsvs. ones that are just useful for further optimization.

insert “collection sites” to all shuffle operations. A shuffle
operation can be explained as a map-reduce. During themap phase we collect all rows that share a key into the sameshard. During the reduce phase we perform computationon the shard. For example, for a group by, we map rows
with the same group key to the same shard and then reduce
the shard to compute the aggregates; for a join, we map therows from the left and right tables with matching keys to thesame shard and then reduce the shard to produce the joined
rows. A shuffle defines a natural materialization point thatfacilitates the collection of stats. The stats are then used
to decide how to partition the data including the number of
reducers and the partition function. Stats are also used torewrite the plan for the un-executed portion of the plan.
Adaptive Aggregation In this case we have two queryfragments; a leaf fragment that computes partial aggregation
on each partition of the data, and a root fragment that
aggregates the partial results. We estimate the number ofrecords that the first fragment will emit by computing thepartial aggregate on a subset of the partitions and deciding
the number of shards to satisfy a target number of rows per
shard (e.g. 1 million).
Adaptive Join. When we encounter a join we estimate
the keys that will participate in the join from the LHS andRHS. We use a data structure that summarizes the keys foreach partition. The data structure holds the count of keys,
the min and max values per partition, and a bloom filter if
the number of keys is less than a few million. In most cases
this data structure is collected from the entire input, not justfrom a sample. We support the following join optimizations:• Broadcast. When one of the sides is small (build-side)(e.g. less than 100,000 records) we broadcast it (using
RDMA) to every partition of the probe-side. In thiscase the plan is rewritten to avoid the shuffle.

• Join awareness: the virtual table layer understands
star joins and can automatically insert joins in thegenerated query when a dimensional attribute is chosenthat is not de-normalized into the fact table.

3.6.2• Pruning. When the keys on one of the sides (filterside) can be summarized with a bloom filter of size
O(10M B) with a false positive rate of about 10%; we
use this bloom filter to prune the other side (pruneside). The map operation only writes records from theprune-side that pass the filter, we only map records
with a high probability of joining (based on the bloom
filter false positive ratio).

Query Optimizer

Procella has a query optimizer that makes use of staticand adaptive query optimization techniques [16]. At querycompile time we use a rule based optimizer that applies
standard logical rewrites (always beneficial) such as filter
push down, subquery decorrelation, constant folding, etc.
At query execution time we use adaptive techniques to select/tune physical operators based on statistics (cardinality,
distinct count, quantiles) collected on a sample of the actual
data used in the query.
Adaptive techniques have enabled powerful optimizations
hard to achieve with traditional cost-based optimizers, while
greatly simplifying our system, as we do not have to collectand maintain statistics on the data (especially hard wheningesting data at a very high rate) and we do not have tobuild complex estimation models that will likely be useful
only for a limited subset of queries [19, 31]. We instead getstatistics from the actual data as it flows through the queryplan, our stats are thus equally accurate throughout the plan,in traditional optimizers estimates are more accurate at theleaves of the plan than closer to the root.
Our adaptive optimizer adds stats “collection sites” to thequery plan and decides how to transform the un-executed
portion of the plan based on this information. Currently we• Pre-shuffled. If one side of the join (partitioned-side)
is already partitioned on the join key, we use the map
operation to partition only the other side map-side tomatch the partition-side partition. For example, if thepartition-side has partitions [min,max] = [1,5] [10,50][150,200] we map records from the map-side to matchthese intervals (we may overpartition if the map-side islarge). The partition-side could initially have hundredsof partitions, but after applying filters only these threeranges of values will actually participate in the join.
This optimization is very frequently applicable in fact
to dimension joins as dimension tables are partitionedon the join key. Note that if both left and right sides
are partitioned on the join key, at compile time, we
select a lookup join strategy (i.e. the shuffle is avoided).
• Full Shuffle. If no other join optimization applies, we
map records from the left and right sides of the join. We
automatically determine the number of shards basedon the sum of records from the left and right sides.

2029• When new data is registered, the registration server
notifies the data servers (both primary and secondary)that new data is available, so that it can be loaded inmemory immediately, instead of being lazily loaded on
first request. This ensures that there is no remote disk
access in the serving path, even if a few leaf servers are
down. This ends up using more RAM but since we are
dealing with relatively small volumes of data, this is areasonable trade-off.

Adaptive Sorting. When handling an order by statement, we first issue a query to estimate the number of rowsthat need to be sorted and use this information to determine
a target number of shards n; We then issue a second query
to estimate n quantiles that are used to range partition thedata during the map operation. During the reduce operation
we locally sort the data in each shard. Adaptive sorting
allows us to reuse our shuffle infrastructure to order large
datasets efficiently.
Limitations of Adaptive optimization. Adaptive optimization works well for large queries where we only add
around 10% of overhead for stats collection. But for small
queries (i.e. queries that execute in O(10ms)), the overheadof adaptive optimization can be large (around 2X penalty).
For low latency queries we allow the user to specify query
hints that fully define the execution strategy without theadaptive optimizer. We have found that for ad-hoc queries,
users are willing to tolerate a slow down of fast queries (e.g.
10ms to 20ms) in exchange for better performance on large
queries and overall much better worst-case behavior.
Another limitation of our current adaptive framework isthat it is not yet used for join ordering. We are working on
an extension to the framework that uses a standard dynamic
programming algorithm that collects table level statistics
with adaptive queries. This can be a good compromise
between the number of the executed stats queries and theaccuracy of the estimates.

3.7• The MDS module is compiled into the Root Server (RS).
This saves RPC communication overheads between
these sub-systems, allowing the RS to be efficient athigh QPS.
• all metadata are fully preloaded and kept up to dateasynchronously to avoid having to remotely access metadata at query time and thus suffer from higher taillatencies. The system is set up to have slightly stale
metadata still give the right answers. The stats instances have a relatively small number of tables so thisoverhead is justifiable.
• Query plans are aggressively cached to eliminate parsing and planning overhead. This is very effective since
the stats query patterns are highly predictable.
• The RS batches all requests for the same key and sends
them to a single pair of primary and secondary dataservers. This minimizes the number of RPCs required
to serve simple queries. This means that single key
queries have exactly 2 outgoing RPCs where we use
the fastest response to minimize tail latencies.

Data Ingestion

For optimal query processing performance, it is importantto create datasets that are suitably partitioned, clustered,and sorted. At the same time, integrating loading into thequery engine reduces scalability and flexibility for teams
who want to import large (often petabytes) of existing data
quickly into Procella. To address this, Procella provides
an offline data generation tool. The tool takes the input
and output schemas and data mapping, and executes anoffline MapReduce based pipeline to create data in Procellaoptimized format and layout. Since the tool can run on cheap
low priority batch resources on any data center, it vastly
improves the scalability of ingestion of large data sets into
Procella, and ensures that expensive production resources
are only used for serving queries.
The tools can apply several optimizations during data
generation such as uniform and suitably sized partitioning,
aggregation, sorting, placing data for hot tables in SSD,
replicating, encrypting, etc. Note that the use of the tool is
optional; Procella can accept data generated by any tool aslong as it is in a supported format.

3.8• The root and data server tasks are monitored for unusually high error rates and latencies compared to theother tasks of the same server in the same data center. The problematic outlier tasks are automatically
moved to other machines. This is useful because of theimperfect isolation on Borg.
• Most of the expensive optimizations and operations
such as adaptive joins and shuffles are disabled to avoid
related overheads and production risks.

4.

Serving Embedded Statistics

Procella powers various embedded statistical counters, suchas views, likes, and subscriptions, on high traffic pages suchas YouTube’s watch and channel pages. The (logical) query issimply: SELECT SUM(views) FROM Table WHERE video id= N, and the data volumes are relatively small: up to a
few billion rows and a small number of columns per row.
However, each Procella instance needs to be able to serve
over a million QPS of such queries with millisecond latency.
Further, the values are being rapidly updated and we need to
be able to apply the updates in near real-time. Procella solves
this problem by running these instances in “stats serving”
mode, with specialized optimizations for such workloads:

PERFORMANCE

Procella is designed for high performance and scalability on
flexible workloads over shared infrastructure. In this section,
we present benchmarks and production serving numbers
for three most common workloads, namely, ad-hoc analysis,real-time reporting, and serving stats counters.

4.1

Ad-hoc analysis

For ad-hoc analysis, we chose to use the TPC-H queries.
TPC-H is a widely used and published benchmark for
database systems that represents a typical normalized
database and analytical workload.
We ran the TPC-H 1T (SF: 1000) and 10T (SF: 10000)
queries on an internal Procella instance with 3000 cores and
20 TB RAM, running on Borg. Data was generated to ourfile format (Artus), partitioned, and sorted, and stored in
Colossus. Caches were warmed up prior to executing the

2030

96

22.91122.07523.06322.53722.05

22.14522.13123.2564

17.66

176.6169

67.58

68.18969.60768.20369.63768.17266.654

67.34

7.47

74.74

03

5.535

29.56

62.81864.08563.91529.175

6.385

27.6383

1.68

16.78

48

156.849

792.052771.465771.689771.928

162.538

158.019390.3752

15.96

159.6305

88.184

1.845

36.33

36.30135.15481.36788.24948.7476

0.90

8.95

21

570.714790.354839.119826.699838.822554.234

571.393698.208

16.00

159.95113.5887

70.783146.81462.09367.45772.366101.406

78.93776.725711.36

22863.503423.542443.19442.126438.486

860.006

866.783696.6681

64.81

82

466.133

47.34

50.14929.881151.871450.176459.951301.3434

648.05
Table 4:YTA Instance Scale
3.19
31.92

36

163.472

163.389137.206

161.54178.718159.866

157.711154.39339.60

396.0249

46.59348.65

39.64444.031110.29645.25853.31995135.674

126.305146.512137.778389.605133.224128.94231

27.31528.48825.08425.264356.60725.64428.88368

41.79936.42119.80728.22629.40732.49737.44653

35.94322.32174.95265.40683.87936.10734.96247694.057

661.674

674.863668.044

126.244700.043664.36159.981

288.50Queries executed
2.10
20.97
173.2003
Queries on24.98
realtime249.83
data
56.1077
Leaf plans 2.17
executed 21.72
36.8926
10.73
107.25
Rows scanned
40.7546
8.68
86.76
Rows returned
611.8779
50.14
501.3846.60128.94
89.37
Peak scan rate
per query
20.4225
3.25
32.48
Schema size
156.321218.83
188.28
Tables180.9596
42.18
421.7542.548315

38.09669.96427.59137.25529.84

35.21516

23.34324.13222.88426.2096.271

26.39923.05835

146.912130.476

142.401

119.065

212.232

141.989

127.18143

205.092224.344

193.673

49.72876.19

203.184221.53925

143.896116.987

117.337342.20854.721129.566

116.33139.7651

28.85

45

83.92527793

74.29340251

82.40078015

82.1180227

88.64690544

89.82867515

85.38232639

92.89258739

9.60

95.96

49

4024.061

3985.686

3990.346

4044.409

4028.778

4008.771

3982.814

3999.7948

381.883,818.78

Figure 6: TPC-H Scalability - 1T Expected Execution Time vs. 10T Execution Times.

Figure 7: TPC-H 10T Per Query Throughput Execution times.
benchmarks. For 10T, we ran a single stream (power run)as well as ten parallel streams (throughput run) 2 .
Figure 6 compares the execution time of individual TPC-H
queries over 1T and 10T of data using the same instance. As
shown, the 1T run had a geomean of about 2 seconds, while
the power run on 10T had a geomean of about 10 seconds.
The system, thus, scales well with data volume. The latency
scaled sub-linearly, mainly because the 10T run was able toutilize more parallelism.
Figure 7 captures Procella performance for the TPC-H 10T
throughput run with ten parallel streams. As shown, Procellascales well under load, i.e the performance degrades quitesmoothly as the number of parallel queries increases. There
was a fair amount of variance, however, with some queries (e.g.
Q7) degrading much more than expected, while some queries(e.g. Q2) performing significantly better than expected. The
main bottleneck we established was inter-node data transfer
overhead (RPC, serialization & deserialization, etc.). Thisis somewhat expected of systems that scale horizontally byusing many (hundreds or thousands of) relatively small tasks
instead of a small number (ten or less) of large tasks.

4.2

Real-time reporting

Procella is used to serve many reporting and dashboarding
solutions both internally and externally. Some of the external
2
Note that these benchmarks are not fully compliant with
the TPC-H rules, e.g. we hand tuned some queries for best
performance in our environment. Thus, the results are notdirectly comparable to published TPC-H data but provide
an insight on the raw query processing speed of the system.

1.5+ billion per day
700+ million per day
250+ billion per day
80+ quadrillion per day
100+ billion per day
100+ billion rows per second
200+ metrics and dimensions
300+

Table 5: YTA Instance Performance
Property
50%
99%
99.9%
E2E Latency
25 ms 412 ms 2859 ms
Rows Scanned 16.5M
265M
961M
MDS Latency
6 ms 119 ms
248 msDS Latency
0.7 ms
73 ms
220 msQuery Size
300B
4.7KB
10KB

applications are YouTube Analytics, YouTube Music Insights,
Firebase Performance Analytics and Play Developer Console.
In this section, we present results from the most popularapplication, which is youtube.com/analytics, known internally as YouTube Analytics (YTA). YTA runs more than abillion queries per day over tens of petabytes of data withtens of thousands of unique query patterns (dimensional
breakdowns, filters, joins, aggregations, analytical functions,
set operators, approximate functions, etc.). However, because each creator is restricted to their own data, each query
is typically constrained to a small slice of the overall dataset.
The data needs to be provided in real time; typically end to
end delay is under a minute, but we also need to support
large historical reports covering years of data. There is also
an interface to do bulk extraction of data; the largest of suchreports produce over a hundred gigabytes of output.
For redundancy, we maintain five instances of Procella forthis use-case, with at least three fully serving. Each instancehas about 6000 cores and 20 TB of RAM.
Table 4 provides a glimpse into the scale of the Procellainstance that serves YTA (and a few other smaller applications). Table 5 captures the distribution of latency vs various
aspects of the system.
As evidenced, Procella performs and scales very well whenhandling high volume SQL queries of simple to moderatecomplexity, on both real time and batch data, providing subsecond response for the 99%+ of the requests from creators,
app developers and other external users of the system.

4.3

Embedded Statistics

The stats Procella instances power various embedded statistical counters, such as views, likes, and subscriptions, serving
millions of QPS (hundreds of billions of queries per day) withmillisecond latency from more than ten data centers aroundthe world. Table 6 captures the latency profile in productionof the statistics serving instances on a typical day.
As evidenced, in this configuration, Procella achieves performance comparable to NoSQL key value stores while supporting SQL based queries on the data.

2031

which can be trivially parallelized in a distributed tree
execution architecture. The predominant use-case forPowerDrill is fast analysis of logs data and poweringinternal dashboards.

Table 6: Statistics Serving Latency
Percentile Latency
50%
1.6 ms90%
2.3 ms95%
2.6 ms99%
3.3 ms
99.9% 14.8 ms
99.99% 21.7 ms

5.

RELATED WORK

5.1

Google Technologies• Dremel is an exa-scale columnar SQL query engine
that is optimized for large complex ad-hoc queries.
Dremel works on many backends such as ColumnIO,
Capacitor, and Bigtable. The Dremel engine powers
Google’s BigQuery. Procella shares many properties
with Dremel such as the RPC protocol, the SQL parser,
low level storage (both use Colossus), compute (bothuse Borg), etc. However, there are significant design differences, for example, extensive use of stateful caching(Dremel is mostly stateless), separate instances where
each is optimized for different use cases (Dremel is a
global shared service across all users), use of indexable
columnar formats optimized for lookups (Capacitor,
unlike Artus, does not have indexes), etc. These makeProcella suitable for many additional workloads (e.ghigh QPS reporting and lookup queries)• Mesa is a peta-scale data storage and query system
built at Google. Mesa is designed mainly for storing andquerying aggregated stats tables, with a predominantly
delta based low latency (measured in minutes) data
ingestion model. Mesa does not itself support SQL;
instead, F1 provides a flexible SQL interface on top of
Mesa storage, using the lower level Mesa API.
• F1 [39] includes a federated distributed query engine(similar to Dremel) that runs on top of many Google
backends (Mesa, ColumnIO, Bigtable, Spanner, etc.)
and provides support for scalable distributed SQL
queries. F1 aims to support OLTP, ad-hoc, and reporting use-cases, but takes a fundamentally different approach from Procella, namely query federation.
Specifically, F1 aims to decouple the execution engine from storage and instead exploits properties of
the underlying storage subsystem optimally to serve
diverse workloads. For example, OLTP workloads typically query F1 and Spanner databases. Low latency
reporting applications use queries over F1 and Mesa.
Analytical and ad hoc queries run over data in various
sources, including Mesa, F1, and Capacitor files on
Colossus. Procella, on the other hand, aims to serve
diverse workloads (excluding OLTP) using the samestorage and data format, by tightly coupling storage,
data format, and execution into a single system.
• PowerDrill [28] is a columnar in-memory distributed
SQL query engine optimized primarily for serving a relatively small QPS of large but relatively simple queries• Spanner is a distributed SQL database typically usedby OLTP applications. Spanner shares many of thesame underlying design elements (and hence similar
performance characteristics) as Bigtable while providing better data consistency, atomic transactions, and a
SQL based query interface. Spanner provides externalconsistency with realtime data and is often used assource-of-truth data store; however it does not aim toserve the ad-hoc or real-time reporting use-cases.

5.2
5.2.1

External Technologies
Ad-hoc analysis

Presto [22] is an open source query engine originally developed at Facebook that is similar in design to Dremel.
Presto is now also available as a cloud service from Amazon
as Athena. Spark SQL [5], an open source SQL engine
that runs on the Spark framework, has also gained significant
popularity, especially in organizations with large investment
in Spark. Snowflake [13], a popular analytical database,segregates storage and compute, while adding several otherfeatures such as local caching and centralized metadata management, similar to Procella. Redshift [26] is a popular
distributed database available from Amazon that tightly couples storage and compute on the same VMs, though it ispossible to also access external data in S3 through Spectrum. These systems excel at low QPS ad-hoc analysis anddashboarding, but do not offer high QPS low latency serving
solutions, or high bandwidth real time streaming ingestion.

5.2.2

Real time reporting

Both Druid [43] and Pinot [17] share some of Procella’s
characteristics such as columnar storage, mixed streaming
and batch ingestion modes, distributed query execution,
support for tablet and column pruning, etc. that enable
them to support low latency high QPS serving use cases.
ElasticSearch has some structured query capabilities suchas filtering and aggregation using a custom API and has
been used in several places to power low latency real time
dashboards. These engines have limited SQL support andare overall unsuitable for ad-hoc analysis. Companies like
Amplitude and Mixpanel have built custom backends tosatisfy the performance requirements of high volume low
latency real time reporting. Apache Kylin, attempts tosolve this problem differently by building aggregates fromdata cubes and storing them in a key value store such as
HBase for higher performance for pre-defined query patterns
that do not need large scans.

5.2.3

Monitoring

Stackdriver and Cloudwatch are two popular cloudmonitoring services - they have their own custom backends
and expose only a limited API to their users. Facebook hasdeveloped Gorilla [36], a real-time time-series monitoring
database, now open-sourced as Beringei [23]. Beringei
supports a custom time series API. InfluxDB [30] is a
popular open source choice for time series real time analytics
with flexible queries. OpenTSDB is an open source timeseries database popular for monitoring purposes that uses2032

• Embedded statistics: Several high traffic stats counters on YouTube such as views, likes, and subscriptions are powered by Procella, totalling over 100 billion
queries per day.

a key value store such as HBase to store metrics keyedby time and user specified dimension keys and provides acustom API to extract time series data for charting andmonitoring purposes. These systems do not offer efficient
execution of complex queries over large data volumes, andare thus unsuitable for ad-hoc analysis.

5.2.4

Serving Statistics

Key value stores such as HBase are often used as backends
for serving high QPS of updates and lookups simultaneously.
They are often deployed with a caching layer. This method,
however, typically does not provide flexible update modes
such as delta mode for stats or HyperLogLog [29] for uniques,
nor flexible SQL based query APIs that are required foranalytical processing.

6.

FUTURE WORK

Procella is a relatively young product, and there are anumber of areas we are actively working on:• As Procella gains wider adoption inside YouTube and
Google, various usability aspects such as better tooling,
better isolation and quota management for hybrid loads,
better reliability, availability in more data centers, better documentation, better monitoring and such are
gaining importance. The team is increasingly spending
more effort to make it easier for users to adopt Procella.
• We continue to spend significant engineering effort on
performance, efficiency, and scale – for example ondata format (Artus), evaluation engine (Superluminal),structured operators and distributed operators suchas joins. We are actively working on improving the
optimizer, focused mainly on adaptive techniques toachieve optimal and predictable performance on large
complex queries.
• We continue to work on enhancing SQL support, and on
various extensions such as full text search, time-series,
geo-spatial, and graph queries.

7.

• Ad-hoc analysis: Procella is increasingly being
adopted at YouTube for large complex ad-hoc analysis use cases over petabyte scale tables.
• Monitoring: Procella powers various real time
anomaly detection, crash analysis, experiment analysis, and server health related dashboards at Google.
Procella thus enables YouTube to have a unified data stackthat is highly scalable, performant, feature-rich, efficient, andusable for all types of data driven applications. Procella has
now been in production for multiple years. Today, it isdeployed in over a dozen data centers and serves hundredsof billions of queries per day over tens of petabytes of data,
covering all four use cases, at YouTube and several other
Google products.

8.

ACKNOWLEDGEMENTS

We would like to thank the Dremel team, especially Mosha
Pasumansky, for working closely with us on using the Capacitor format and enabling reuse of the Dremel interfaces.
Additionally, We would like to thank the Google SQL teamfor developing and sharing the SQL parser and the test suite.
We thank the YouTube management, especially Joel Truher,
John Harding, Christian Kleinerman, Vaishnavi Shashikant,
Eyal Manor, and Scott Silver for supporting the Procellaproject. We also thank Xing Quan for supporting the product requirements of Procella. Finally, we thank the many
reviewers who helped review this paper, especially Andrew
Fikes, Jeff Naughton and Jeff Shute.

9.

REFERENCES[1] D. Abadi, P. Boncz, and S. o. Harizopoulos. The design
and implementation of modern column-oriented
database systems. Foundations and Trends R in
Databases, 5(3):197–280, 2013.
[2] D. Abadi, Madden, et al. Integrating compression andexecution in column-oriented database systems. In
SIGMOD, pages 671–682. ACM, 2006.
[3] S. Agrawal, S. Chaudhuri, and V. R. Narasayya.
Automated Selection of Materialized Views andIndexes in SQL Databases. In VLDB, VLDB ’00, pages
496–505, San Francisco, CA, USA, 2000. Morgan
Kaufmann Publishers Inc.
[4] H. Ahmadi. In-memory query execution in Google
BigQuery, 2016.
[5] M. Armbrust, R. S. Xin, et al. Spark SQL: Relational
data processing in Spark. In SIGMOD, pages
1383–1394, 2015.
[6] C. Barthels, I. Müller, et al. Distributed Join
Algorithms on Thousands of Cores. PVLDB,
10(5):517–528, 2017.
[7] T. A. Bjørklund, J. Gehrke, and Øystein Torbjørnsen.
A Confluence of Column Stores and Search Engines:
Opportunities and Challenges, 2016.
[8] B. Bramas. Inastemp: A Novel Intrinsics-as-Template
Library for Portable SIMD-Vectorization. Scientific
Programming, 2017.

CONCLUSIONSIn this paper, we presented Procella, a SQL query engine that successfully addresses YouTube’s need for a single
product to serve a diverse set of use cases, such as real-time
reporting, serving statistical counters, time-series monitoring,
and ad-hoc analysis, all at very high scale and performance.
We presented the key challenges of building such as system
on Google’s infrastructure, described the overall architectureof the system, and several optimizations we implemented
to achieve these goals. Many Procella components (suchas Artus and Superluminal) and design elements (such asaffinity scheduling and layered caching) are being adopted
by other Google products such as BigQuery, F1, and logs.
Many of these techniques, we believe, are applicable to othersimilar distributed platforms such as AWS and Azure.
Procella has been successfully applied at Google to solve
problems belonging to all four categories:• Reporting: Virtually all of external reporting and internal dashboards at YouTube are powered by Procella(billions of queries per day). Procella is also powering similar external facing reporting applications for
Google Play and Firebase, among others.

2033[9] N. Bruno, Y. Kwon, and M.-C. Wu. Advanced Join
Strategies for Large-scale Distributed Computation.
PVLDB, 7(13):1484–1495, 2014.
[10] S. Chambi, Lemire, et al. Better bitmap performancewith roaring bitmaps. Software: practice andexperience, 46(5):709–719, 2016.
[11] F. Chang, J. Dean, et al. Bigtable: A distributed
storage system for structured data. TOCS, 26(2):4,
2008.
[12] J. C. Corbett, J. Dean, et al. Spanner: Google’s
globally distributed database. TOCS, 31(3):8, 2013.
[13] B. Dageville, T. Cruanes, et al. The Snowflake Elastic
Data Warehouse. In SIGMOD, SIGMOD ’16, pages
215–226, New York, NY, USA, 2016. ACM.
[14] J. Dean and L. A. Barroso. The Tail at Scale.
Communications of the ACM, 56:74–80, 2013.
[15] J. Dean and S. Ghemawat. MapReduce: Simplified
data processing on large clusters. Communications of
the ACM, 51(1):107–113, 2008.
[16] A. Deshpande, Z. Ives, and V. Raman. Adaptive Query
Processing. Found. Trends databases, 1(1):1–140, 2007.
[17] K. G. Dhaval Patel, Xaing Fu and P. N. Naga.
Real-time Analytics at Massive Scale with Pinot, 2014.
[18] R. Ebenstein and G. Agrawal. Dsdquery dsi-querying
scientific data repositories with structured operators. In
2015 IEEE International Conference on Big Data (Big
Data), pages 485–492. IEEE, 2015.
[19] R. Ebenstein and G. Agrawal. Distriplan: An
optimized join execution framework for geo-distributed
scientific data. In Proceedings of the 29th International
Conference on Scientific and Statistical Database
Management, page 25. ACM, 2017.
[20] R. Ebenstein, G. Agrawal, J. Wang, J. Boley, and
R. Kettimuthu. Fdq: Advance analytics over realscientific array datasets. In 2018 IEEE 14th
International Conference on e-Science (e-Science),
pages 453–463. IEEE, 2018.
[21] R. Ebenstein, N. Kamat, and A. Nandi. Fluxquery: An
execution framework for highly interactive query
workloads. In Proceedings of the 2016 International
Conference on Management of Data, pages 1333–1345.
ACM, 2016.
[22] Facebook Inc. Presto: Distributed SQL Query Engine
for Big Data, 2015.
[23] Facebook Inc. Beringei: A high-performance time series
storage engine, 2016.
[24] A. Fikes. Storage Architecture and Challenges, 2010.
[25] Google, Inc. Vitess: Database clustering system forhorizontal scaling of MySQL, 2003.
[26] A. Gupta, D. Agarwal, et al. Amazon redshift and thecase for simpler data warehouses. In SIGMOD,
SIGMOD ’15, pages 1917–1923, New York, NY, USA,2015. ACM.
[27] A. Gupta, F. Yang, et al. Mesa: Geo-Replicated, Near
Real-Time, Scalable Data Warehousing. PVLDB,
7(12):1259–1270, 2014.
[28] A. Hall, O. Bachmann, et al. Processing a Trillion Cells
per Mouse Click. PVLDB, 5:1436–1446, 2012.
[29] S. Heule, M. Nunkesser, and A. Hall. HyperLogLog inPractice: Algorithmic Engineering of a State of The
Art Cardinality Estimation Algorithm. In EDBT, pages
683–692, Genoa, Italy, 2013.
[30] InfluxData Inc. InfluxDB: The Time Series Database in
the TICK Stack, 2013.
[31] Lohman, Guy. Is query optimization a “solved”
problem?, 2014.
[32] R. Lupi. Monarch, Google’s Planet Scale Monitoring
Infrastructure, 2016.
[33] S. Melnik, A. Gubarev, J. Long, G. Romer,
S. Shivakumar, M. Tolton, and T. Vassilakis. Dremel:Interactive Analysis of Web-Scale Datasets. PVLDB,
3(1):330–339, 2010.
[34] Mosha Pasumansky. Inside Capacitor, BigQuery’s
next-generation columnar storage format, 2016.
[35] Nathan Marz. Lambda Architecture, 2013.
[36] T. Pelkonen, S. Franklin, et al. Gorilla: A Fast,
Scalable, In-memory Time Series Database. PVLDB,
8(12):1816–1827, 2015.
[37] O. Polychroniou, A. Raghavan, and K. A. Ross.
Rethinking SIMD Vectorization for In-Memory
Databases. In SIGMOD, SIGMOD ’15, pages
1493–1508, New York, NY, USA, 2015. ACM.
[38] I. I. Prakash Das. Part 1: Add Spark to a Big Data
Application with Text Search Capability, 2016.
[39] B. Samwel, J. Cieslewicz, et al. F1 Query: Declarative
Querying at Scale. PVLDB, 11(12):1835–1848, 2018.
[40] S. Singh, C. Mayfield, S. Mittal, S. Prabhakar,
S. Hambrusch, and R. Shah. Orion 2.0: native support
for uncertain data. In Proceedings of the 2008 ACM
SIGMOD international conference on Management of
data, pages 1239–1242. ACM, 2008.
[41] Varun Talwar. gRPC: a true internet-scale RPC
framework is now 1.0 and ready for production
deployments, 2016.
[42] A. Verma, L. Pedrosa, M. R. Korupolu,
D. Oppenheimer, E. Tune, and J. Wilkes. Large-scale
cluster management at Google with Borg. In EuroSys,
page 18, Bordeaux, France, 2015. ACM.
[43] F. Yang, E. Tschetter, et al. Druid: A Real-time
Analytical Data Store. In SIGMOD, SIGMOD ’14,
pages 157–168, New York, NY, USA, 2014. ACM.

2034