Video summary
The video addresses a critical industry challenge where data lakes like Apache Iceberg are increasingly serving as the single source of truth for analytical data, yet they struggle to meet the demands of high-concurrency, low-latency workloads required by modern product surfaces such as real-time dashboards and generative agents. Traditionally, organizations have attempted to bridge this gap by copying massive volumes of data from the lake into specialized serving engines like Redis or ClickHouse. However, this approach creates a fragmented architecture with multiple copies of data, leading to expensive sync pipelines, consistency issues, and operational overhead that ultimately defeats the purpose of maintaining a unified data lake. The speaker argues that instead of duplicating data, the solution lies in building query engines capable of serving directly from the lake itself while adhering to strict sub-second latency Service Level Agreements (SLAs).
To achieve this, Apache Pinot introduces a novel architecture that reads directly from Parquet files stored in object storage like S3 without requiring data duplication. The core innovation involves a family of persistent indexes—such as inverted, range, text, and Bloom filters—that are built during ingestion and stored alongside the data segments. These indexes enable highly efficient pruning at the page level rather than just the row group level, allowing the engine to skip irrelevant data blocks entirely before reading any actual bytes. Additionally, Pinot employs smart parallel asynchronous fetching to pre-load relevant data from S3 into memory and utilizes a configurable cache hierarchy where indexes can be pinned to local disks to minimize network latency, offering a flexible trade-off between cost and performance based on specific business needs.
The presentation validates these architectural improvements through rigorous benchmarks comparing Pinot against industry standards like Trino and ClickHouse using a dataset of over 12 billion rows. While engines like Trino and ClickHouse perform well for ad-hoc exploration and batch processing, they fail to maintain low latency under high concurrency when executing complex aggregations on data lakes. In contrast, Pinot demonstrated the ability to execute heavy queries involving map column aggregations and multiple filters in milliseconds by leveraging its indexes to reduce the amount of scanned data from billions of records to near zero. A live demo further illustrated this capability, showing that a query with no matching rows could be answered in under 30 milliseconds by serving directly from the index without touching the underlying Parquet files, whereas an unindexed version took several minutes and scanned gigabytes of data.
In conclusion, the talk establishes that high-concurrency serving is a distinct workload category that requires specialized handling beyond traditional analytics engines or simple data replication. By combining precise index-driven pruning, aggressive parallel fetching, and flexible caching strategies, Apache Pinot allows organizations to keep their historical data in low-cost object storage while still delivering real-time insights directly from the lake. This approach eliminates the need for costly data duplication, reduces infrastructure costs by minimizing CPU and network usage, and provides a unified platform where users can run both fresh real-time queries and historical analysis on the same dataset without changing their application APIs or architectural patterns.
Read the full video transcript
This is Vivek here again. Introduce me.
A lot of Vivek today.
So, I lead the product at StarTree.
Everything related to Iceberg, ingestion
into Pinot, be it AI or any other
solution on top of Pinot, right? So, I'm
here to talk about a a really
interesting religion that is happening
in the industry. On one side, you have a
a workloads where it requires like low
latency and fast and high QPS kind of
queries. And the on the other side, you
have uh data lakes coming up like
Iceberg and Delta Unity and whatnot.
That is actually coming up as a single
store of
truth, single source of truth where
you're storing all kind of analytical
data, right? So, the these two kinds of
these two kinds of workloads do not
actually match up or did not actually
match before. So, what I'm here to talk
about is how the teams are stuck between
these two realities and what what we are
doing to fill it. The three parts of the
agenda for today. The first one is the
problem. How the lake kind of serve. The
second is how we made Pinot point at it
and query it. The third one is proof
benchmarks against Trino and ClickHouse.
So, before that, just a quick primer on
Pinot. Pinot was born at LinkedIn to
serve who who viewed my profile use case
for hundreds to hundreds of millions of
users. And it is actually user-facing
analytics
from day one, not for internal BI. So,
the today today the open source
community is strong. 2,000 companies,
300 plus contributors, and 30% of
Fortune thousands are actually using it.
So, in fact, take any food delivery
company in the world today, all of them
are using Pinot underneath for their
restaurant business, for whatever you're
seeing on the app, and whatnot. So, I'm
not showing you a research prototype
here today, and this is just a engine
that already carries production
workloads at a very large scale. So,
what's new here today is where it reads
from.
Quick slide on StarTree. StarTree is
just a managed solution on top of Apache
Pinot. You get SAS, BYOC, BYOK kind of
deployments, no-code ingestion, absurd
query tuning, and whatnot, right? So, so
you don't have to even take care of
security, resilience, and everything is
covered. Right? So, let's talk about the
lake.
Iceberg.
So, I want to start with a very deep
respect with what it really does. With
Without Iceberg, you just have a pile of
files with
of storage with a bit of conventions
here and there. But with Iceberg, what
you get is something that behaves like a
database. You get asset semantics, you
get updates, multi-engine access, which
with which you don't have to get into a
vendor like right? So, the question is
once you have the data in Iceberg as
single source of truth, what happens
when you try to serve from it? And keep
that phrase in mind, single source of
source of truth, because that
Today, the whole talk is about not
breaking it.
So, let's let's see
Apart from storing data, what people are
actually using it for. In the previous
talk, Amit just mentioned about how he
is utilizing Iceberg to actually serve
the reporting use cases. But
companies like fintech and other other
sort of merchant analytical solutions
are actually providing real-time
dashboards on top of the data that is
actually the transactions and cash flows
and whatnot is happening on a real-time
basis. Like for logistic companies, they
are serving historical delivery
performances, route optimizations, cost
optimizations. And the newest one we
have got is the generative agents.
Everybody is talking about it today. So,
notice what these have in common. None
of these are analyst who are actually
waiting to run a query. They are product
surfaces. And product surfaces comes
with SLA, right? So, let me make it more
concrete with the real workload with
real numbers. This is one of the
one of the customers who are actually
using us.
They run live dashboard
for their business to
basically staff each stores correctly.
And this is we are talking about a quick
commerce company here
that hit
and they are actually staffing each
stores correctly and basically hitting
the delivery time promised at the lowest
cost possible. The other one is
basically keeping the fast moving
perishable so that they don't over order
and spoil them. The third one is move
goods from warehouse to stores to
customers on time and at minimum cost
possible.
So these are important decisions and
people are and there are people like
next door managers that actually using
this kind of dashboard on a minute to
minute basis. So let now look at on the
right hand side of this where you have
the numbers. They are running half a
million dashboard queries a day across
India and peaking at 80 QPS.
And you can see that the scan one of the
a few scans actually can touch terabyte
of data at once. I notice that there is
one petabyte of first copy data and 300
terabyte of a second copy that is
actually utilized to serve the those
kind of dashboard. So the data was
already there in the lake, but why does
it need another serving layer? Why does
it need duplicating a third
third of a petabyte to just answer that?
So let's look at how query query engines
actually came around data lake as of
today. There are two families that we
see often. One is processing engines
like Spark, Flink which is good for
transforming data and a job taking
minute is fine there. For exploratory
engines like Trino, Presto, Athena where
it is actually answering analyst where a
query taking 5, 10, 30 second or a
minute is fine, right? Both are valuable
and both are actually useful for very
throughput machines where they are like
actually really good good at, but
neither of them are actually built for a
promise like you run any kind of
workload on top of it, any number of uh
querying it and actually still responds
within a second every time.
So, we have got a third kind of
workloads, right? That's what I'm
talking about, the dashboard that a
quick commerce company is using for.
Where they really need is sub-second
latency SLAs, freshness measured in
minutes, right? And that's the part
actually breaks in the whole system. So,
the latency has to hold under a very
high concurrent queries because you can
understand that there are many, many
dark stores in India, and everybody's
actually querying the same database.
Right? So, so today's lake engines were
not cut for it, were not designed for
this. So, what do we do today?
The same thing. We actually copy this
data, we reverse ETL this data out of
the lake into Redis, Elasticsearch,
Druid, ClickHouse, into Postgres, and
whatnot, right? Whatever gets the
latency down. And suddenly your
beautiful data
lake has become a data backwater. You've
got five copies of the same data sitting
in five different systems, each with its
own problem, sync pipeline issues,
consistency lag issues, and ops burden
in itself. Right? So,
I would assume that everybody in the
room has actually built it or rather
inherited it. Right? So, it works. It's
but it it's expensive, it's fragile, and
it defeats the purpose of having a lake
in the first place. Remember that quick
commerce company again copying the data
300 terabyte of data again. So, what if
you didn't have to copy? What if there
was a query engine that actually sits on
the lake itself and serving with the
same SLAs, fast, fresh, and high
concurrent, and read directly from your
iceberg? No copy at all. So, that's what
Pinot does. That's what StarTree is
actually solving today. You keep your
Spark for processing, you keep Trino for
exploration, but the moment you get a
get into third workload SLA, you have to
put a serving engine on top of it. So,
so the rest of the talk here today is me
showing you how it really works and then
proving it with numbers.
So, let's take another quick primer on
the architecture of Pinot. You have four
um main components here, a controller
with Helix and Zookeeper that
coordinates the cluster, the broker that
takes in the query and fans it out to
the servers, and the servers that own
the data and execute, and the server
that server contains the segments, which
are the fundamental unit of of the data
in Pinot. And carry the term segment
because servers um
contain and own segments, and segments
actually carry the data and the indexes.
So, the ownership model here is exactly
what we actually exploited. So, this is
the new architecture from the
traditional one where you can see that
the segments in in this one, the
segments are in red, which are actually
filled in with the data. In the new
architecture, there is no data in the
segments.
The data actually sits in your lake
directly. We have introduced two two
different um
sort of new components to understand the
lake itself, the catalog itself, the
metadata that sits there. Right? So, one
is creator, one is watcher that takes in
all the incremental data for you. Notice
the data is still sitting in your
parquet files.
So, um again, remember the e-commerce
company whose files land every few
minutes, like every 5 minutes in this
case. So, the watcher actually takes up
all of that, registers with the
controller, say that create a pseudo
segment, and map it to the files that
you just got.
But, it raises obvious question. If the
data is still in plain parquet, how are
you getting it um under a second or a
kind of second latency, right? So, let's
talk about parquet, and be fair about
it. Parquet is a columnar storage
format. Inside parquet, you get um data
splitted into row groups and then column
chunks, and each column chunk has has
its own further splitted into pages.
Which gives you a lot, right? It gives
you I mean, max stats, row groups,
dictionary filtering, and whatnot.
As a storage format, it is really good,
and it is close to perfect. But, look at
what serving really needs, and parquet
lacks here. It does not have range or
timestamp indexes, no text or JSON
indexes for partial match filtering, no
skipping structures. It doesn't know
anything about the other file that is
present. Right? So, it's a perfect
storage format, but it's insufficient
for the kind of access
pattern that we are talking about today.
So, the gap is how the bytes are laid
and how it how we need to reach them.
Right? So, that is that is a very um
Yeah, so
>> [clears throat]
>> what we have done is we have filled this
gap that gap with indexes. And
uh I'll talk about the indexes in a bit,
but
what that was the Parquet and what
happens when you query a Parquet file or
from Iceberg. It reads the metadata in
the first step and prunes the file. In
the second, it scans the column blocks.
And when pruning fails, it actually gets
the whole file in itself
while you scan. And the third step is
you it moves the bytes to to compute and
then serve from it. Notice the second
step here is actually the killer one. It
is actually scanning your whole data
when you know don't require a lot of
data in it. Right? You actually are
ending up scanning files to
find a few rows in it.
So, and on object storage, every wasted
byte is both
latency and actually dollar. Right? So,
the whole game is about reading less. At
one QPS, this will work, but at a
thousand, it will just fail.
Right? So,
let's talk about this. Right? So,
every query engine today, what they're
doing is they're actually pruning at
first two levels only. They're not
getting into at page level, which
actually has the fundamental data that
you are actually
trying to query from. Right? So,
it has time partitioning, min/max
starts, bloom filters, all that. Right?
It gets you again through the rows, but
not to the not to the pages. And even
after pruning, engines today are left
with
say in this representational case, it
actually queried 25 megs, where whereas
it only required to query or fetch half
a meg of data. Right? So, here what has
happened is a range index gave you a
roaring bitmap with exact doc IDs, and
the inverted index narrowed it down to
only the matching doc IDs that is
present. And I'm talking about Pinot
here on the right hand side. So, same
data, same file format, um half a meg
read instead of 25. Understand what kind
of optimizations we are talking about.
That's a 50x difference in IO, and less
the IO per query is really more query
per node. And and that is all all of
that is powered by not one index, but a
family of indexes, where you have
inverted and sorted for exact matches,
range for um time windows, text and FST
for pattern searching, JSON for
semi-structured, Bloom for pruning,
vector for what? For similarity and AI
use cases, and start tree index again
for aggregations, right? So,
index Each index was built once when the
data was ingested and discovered, and it
was registered and stored on the server
with the segment. So, it can be utilized
by every query. So, now you can say
that, "Oh, now like the
why doesn't every other engine uh add
another index to it, right?" Because I
would say it's not a missing feature, uh
it's rather an architectural change. For
instance, here Pinot plans and schedules
every query from scratch. It is the
right design for your ad hoc queries,
but it is a tax that you have to pay per
query if you're running a dashboard um
kind of use cases. Right? So,
uh on pruning pruning side, pruning on
data lake is just not efficient with uh
Trino, where it only uses row group uh
min max or ClickHouse's to sort keys and
primary keys. Uh they're just not cut
for it. Uh what Pinot is actually doing
is index-driven pruning, which is down
to the page level. Right? On on the
other side, you have um indexes as well,
where they just don't have the
persistent index with them. Right? And
as we saw Pinot has family of indexes
serving different kind of use cases.
It is actually built on parquet itself.
So that allows reading only matching
pages. Right? So have this table in mind
the difference here. I'll show the
numbers that translates this.
Right? So
So that was the first technique where
it was more about being precise of what
we read and this is the technique of
being aggressive in what we read.
Right? So there these are the there are
differences there. So we all know that
any blob store is not known for latency,
but they are know what they're known for
is throughput. Right? So what do what do
we do? We actually
do smart parallel asynchronous fetching
and as soon as Pinot starts
query planning, it already knows from
the sparse index which blocks are likely
to be relevant and it immediately starts
pre-fetching those blocks in parallel
from S3. Right? So the
the kind of throughput that it gives you
is just amazing and that is what really
powers at a very high scale. So once you
once it gets in the filter stage, it
already has the matching doc IDs and it
uses the same metadata mapping as before
which
which actually maps the pages to the to
the doc IDs that it lives in. Right? So
So what we get from this is very for
very high selective queries,
you are
it can actually the time that it gets to
project the output, it has already done
all kind of aggregations and everything
and loaded that data into memory. So
it's not sequencing it. It actually
parallelizes it. Right? so that gives
that gives you the sub-second at a very
high QPS. The first one was get being
precise and basically fetching less less
less amount of data and this is a more
about doing the same thing at a very
high scale. Right? So the third
technique is essentially
having a cash hierarchy. Say you don't
require a very
basically low latency analytic kind of
use case,
no problem, right? So, what you get is a
kind of a configuration where you can
configure your indexes and your data and
and whatnot in in multiple hierarchies
where the indexes get pinned on your
local disk. So, index lookups doesn't
really need a network routing from S3.
And the parquet data itself is actually
in the in the S3 itself and which is
actually read at a block level. So, so
it effectively reaches the it
effectively works as a forward index in
itself, right? So, so what I was trying
to say is it's it's not a dial rather is
it's it's a dial not a not a knob that
you can switch on and off. You can pin
more index, pay more for disk and get
least amount of latency there. And you
pin less, you you actually run it
cheaper but again you pay for you
actually lean on pre-fetching rather and
and basically your latency takes a hit,
right? So, so what what with this
solution what we are giving is choose
your own Pinot on the kind of cost
latency curve so that and this is the
curve that I was talking about where
cheapest lowest everything is in S3,
right? No no sub-second latency there.
And then the second one is indexed
indexes cached in local and the data is
in S3. The third one is partial data and
indexes in local and the fourth is
traditional Pinot where everything is on
Pinot servers with in in Pinot
proprietary segments, right? So,
you configure whatever you want from
this based on your budget and latency
SLAs. So, let's look at benchmarking,
right? So, all of that was just theory.
So,
I want to start with the setup because
it actually gives you a base to the
benchmarks, right? So, data was about a
terabyte around 12 billion rows. I'll
also show a small demo of this. So, uh
uh and I'll show the comparison against
Pinot with on Trino and ClickHouse. And
each engine got its fair setup. Um Trino
actually got the partitioned um uh data
and file system cache, which is actually
I mean, people recommended. Uh for
ClickHouse, it was partitioned and
sorted again. Pinot got a lot of indexes
like inverted and composite and whatnot,
right? So, it was around uh 10 to 15% of
overhead on top of the data that you
have. So, basically, you have 1 TB of
data and around 130 gigs of indexes that
it requires.
Right? So, this was comparison with
Trino. Um
with this, I'm not here to say that
Trino is bad. It is actually excellent
for your ad hoc analytics
um use cases. But, the moment it goes to
um serving kind of use cases, um you can
see the tight latency, high concurrency,
and selective access, it just doesn't
work. The kind of gap you're seeing on
on the right-hand side here
are actually on different kind of
queries, right? And these are not
trivial queries, actually. These are
aggregations. Um these are group buys,
order buys on top of the map kind of
column, right? So, you are actually
accessing uh the data that sits in the
map and actually grouping by on top of
it. So, these are very heavy queries
that you are running. And you can see
that the pattern fits. Um this is also
from a um a real customer workload where
they did a POC on and then they saw
these numbers.
The other one is similar story with
ClickHouse, uh same hardware, cache
enabled, parquet V3 reader here, and
data again sorted, which actually
ClickHouse benefits from. And ClickHouse
is again fast for analytics, but the
moment it becomes um
um a complex query that you need for
serving, you can see the latency there
on the right-hand side. The the gap has
actually increased. And mind that the
scale is 10 seconds, 15 seconds, not
like a 1 second. Right? So, that's the
cost uh
direct cost saving I would say and the
throughput that you are getting.
So, let's turn that to more money,
right? So, how
So, you have seen the latency has
decreased with Pinot and basically I've
used around a 10 to 15% of cash. So, you
can see that the CPU utilization was low
in start three because the amount of um
amount of data it it was actually
scanning and loading into servers,
right? So, so basically
um so you the moment you scan you you
stop scanning wider data or rather
parquet, it actually reduces your
network cost, CPU cost, and what you get
is very high QPS in the same hardware,
right? So, and this chart shows that
relationship across engines, right? So,
And the other one, right? So, basically
you will say this was all historical
data, this was all data that sits in
say a lake, right? What about my
real-time data? That is actually getting
in the last minute of it I want to query
it. No worries, we have a solution of a
federated engine as well where you can
sit where you can keep your
real fresh data coming from Kafka into
into traditional Pinot segments and you
can say after a while, say an hour or 24
hour, whatever you can configure it,
move that data to a catalog, right? And
I'll use another engines like Spark or
any anything else to run on it. And you
can also power the same dashboard from
these two, right? So, basically you your
app did not change, your the API did not
change, underneath the data actually is
now sitting in the older data is now
sitting in a different low-cost storage,
right? So,
So, one table, one query with full
context, right? So,
Yeah.
Yeah, so
a bit of honesty here.
You have to use Pinot when you require a
very high QPS
workloads where it spans
uh
from hundreds to hundreds of thousands
of repeated queries,
right? And when P95 must stay under a
second and the and it you need fresh and
and basically historical queries in in
one query, right? So and you use
everything else for ad hoc exploration,
big joins, low QPS, internal analytics,
OLTP and and small data, right? So So,
yeah.
And
from all of this, the fourth thing that
I want you to take from this talk is
first, serving is a third kind of
workload where you do not
need to serve
do not need to copy to another layer.
That was just a workaround. This was the
actual solution, right? The second,
indexes, prefetching, pinning and all of
this is actually coming together to
serve
your queries directly from the pages,
not from the column chunks or the
parquet files. And the third one is
reduced data copy and data processed per
copy is actually getting your cost lower
and more throughput from the same node.
And the fourth one is
of course, yeah, it pays at a very high
QPS. You keep Trino and and
Spark for processing and kind of
exploration kind of queries. So that was
the talk. Let's talk about let's let's
show a quick demo here if it loads.
So, 1 second.
So this is just a Starburst Data
Platform tool. We have two tables here,
same parquet files sitting in S3 which
had like 12.2 billion records
rows which I'll show you here just now.
Yeah, this is 12.2. The two tables, one
is without index, one is with indexes
and
and basically it's observability data
that
and yeah, people use it
um these kind of queries and what you're
seeing right now is just a um
uh I'll just pause a minute. What it did
did it
in the in the table that did not have a
index or any kind of index, I actually
ran it for 6 minutes of slice only and
it returned around what 841 million 42
million records. And you can just see
that
um
yeah. You can see that it actually ran
in 3.4 milli 4 seconds and it scanned
You can see this value, right? That is
around what 29 billion uh sorry, 2.9
billion records. And what it means that
um the filters are not getting applied.
And I actually ran the same query on the
table which which is which has the
indexes and it ran in 26 milliseconds.
Right? And I'll just show you the
I'll just show you this, right? So, and
what you can also notice is the range
here is not six um minute, it is 10
hours. So, it is actually scanning more
and still performing better in 26
milliseconds. The data is not even in
Pinot, it's actually in S3. Right? And
you can see that the scan uh just a
moment, you'll see that the scan
data is actually zero under filters.
Right? Which which means that the
indexes are working.
Yeah. Yeah, you can see that the num
entries scanned in filter is just zero.
Right? So, um what happened here really
is give given you we had inverted index
and everything it built in.
It did not even require it to take a
round trip um from S3. It actually
served from the indexes itself.
Right? So, because that was a count star
query. Let's run a very um complex query
itself in just a moment.
Yeah, and you can see that there are
indexes applied on logical timestamp and
resolution with inverted index and this
one didn't does not have it. It just
have the forward index there
pointing to the file itself. And this is
the complex query I'm talking about. And
notice that it is actually grouping by
on
a value inside a map. It is actually
aggregating
again a data that sits in the map,
right? And there a bunch of filters
here. And it responded in 2.25
seconds, right? And you can see that it
has 10-hour window slice to actually
scan through.
Right, so that was it.
That was the talk.
Of course, you can scan this QR to get a
lot of blogs that we have wrote written
in the past few quarters.
So, yeah, I'll take questions if you
have any.