How Apache Pinot real-time ingestion and pauseless consumption work
Follow Pinot server code from stream fetch through indexing, segment commit, failure recovery, and seven detailed stages on pauseless ingestion.
The whole story in 16 lines
Pinot keeps real-time rows queryable by indexing in memory, coordinating one immutable commit, and handing consumption to the next...
- One stream partition is consumed by several replicas whose mutable rows are immediately queryable.
- RealtimeSegmentDataManager fetches, decodes, transforms, indexes, and advances the next offset.
- A mutable row updates forward data, dictionaries, filter indexes, and value ranges together.
- Row, time, stream-end, force, or index-capacity limits can seal a consuming segment.
- Normal completion stops replica cursors while they report segmentConsumed to the lead controller.
- The controller selects the highest reported next offset and asks lagging replicas to catch up.
- RealtimeSegmentConverter turns mutable columns into a query-optimized immutable segment and tar file.
- Split commit brackets upload with commitStart and commitEnd before metadata becomes DONE.
- Retries, catch-up, keep, discard, peer download, and FSM restart contain blocking-protocol failures.
- Pauseless shortens the query freshness gap by moving the generation handoff ahead of build.
- The server calls segmentCommitStart before build and PauselessSegmentCommitter does not repeat it.
- The controller marks the old segment COMMITTING while opening the next one as CONSUMING.
- Old-generation build and new-generation ingestion proceed at the same time on the server fleet.
- Non-winners keep equal data, catch up when behind, or download the committed immutable copy.
- CommitEnd fills CRC, size, and download URL before COMMITTING becomes DONE.
- Validation repairs stale metadata, missing deep-store copies, and recoverable replica failures.
Real-Time Pinot Vocabulary
We will follow one Kafka partition through Pinot, so first we need the small vocabulary that names each moving part.
A consuming segment is the in-memory, still-growing slice that accepts stream records and answers queries immediately.
A commit seals one generation into immutable files while ZooKeeper metadata records the shared offset and location.
The roadmap now shows ingestion, indexing, commit, and pauseless recovery, so which value connects every chapter?
The shared offset connects every chapter because it ends one segment and becomes the exact start of the next. Now let us follow that partition into Pinot.
One partition, several queryable replicas
The source is one ordered partition and Pinot assigns several servers that each maintain a local consuming segment. Each copy uses the same offset coordinate, but it owns separate mutable index memory.
Records travel to every assigned server and each cursor advances independently after its local index accepts the row.
Queries can already read those mutable rows, so what protects the table when one replica briefly falls behind?
Replication preserves live availability because another cursor still exposes the newest indexed rows. The query lens remains connected to an up-to-date mutable copy.
Adjust Replicas between one and three. Compare whether the query lens keeps an alternate live path when one server falls behind, while each cursor still advances independently.
The stream supplies ordered records while replicas supply availability. One lagging server does not erase the other replicas' newest indexed rows. Next we will open one server and trace the exact Java call path.
Inside the Pinot real-time server loop
Now that one server owns a cursor, PartitionConsumer.run drives a state machine and calls consumeLoop while the state should consume. The loop retains its next offset as the checkpoint for recovery and later segment completion.
consumeLoop fetches a MessageBatch and passes it into processStreamEvents, which throttles the batch before touching individual records.
The decoder and TransformPipeline now inspect the record, so what happens before Pinot advances its next offset?
A clean record reaches MutableSegmentImpl.index and then the next offset advances. Decode and filter branches leave distinct downstream traces.
Switch Record through Clean, Decode error, and Filtered. Compare whether the source offset advances and whether a new document reaches the mutable index.
The hot loop separates source reading from row shaping and indexing. Advancing the source offset and adding a query document remain separate branch decisions. Next we will inspect the data structures changed by that final index call.
A row becomes several query paths
We are now inside MutableSegmentImpl.index, where one row feeds several query structures under a single document identifier. One accepted document receives a stable docId before configured index writers receive its column values.
The forward strip stores values by document id while the dictionary turns repeated regions into compact integer ids.
Filters need faster entry points than a full scan, so which structures receive this row at the same moment?
The inverted bitmap gains the document id while the latency range expands to include the value. Those structures make the same new row immediately useful to filters.
Switch Index set through Forward only, Filter indexes, and Full. Compare whether queries must scan values or can use bitmap and range pruning.
Queries are fast before commit because mutable indexes already exist. Their configured combination determines which predicates avoid scanning every forward value. Next we will see the conditions that tell this open segment to stop growing.
Five ways a segment stops consuming
Mutable indexes keep growing until endCriteriaReached finds a reason to stop the current generation and enter completion. The server checks these conditions around the same consume loop that advances records and offsets.
Rows and elapsed time are common limits while force commit, stream end, or index capacity can also win.
Several gauges can be close at once, so which cause becomes the recorded stop reason?
The first satisfied condition seals the generation, and every cause converges on the same completion entry point. Only the recorded reason changes.
Switch Trigger through Rows, Time, Force, and Capacity. Follow each active gauge into the same seal gate, then compare the recorded completion reason.
The recorded reason changes while the safe next protocol action stays constant. Pinot preserves the selected cause for diagnostics and later completion decisions. This invariant keeps commit logic independent from the rollover source. Next all replicas will stop their cursors and report completion.
Normal completion stops the cursors
The seal trigger fires on each replica and the normal server state changes from INITIAL_CONSUMING to HOLDING. Each server now protects a fixed local prefix instead of fetching more stream records.
Each server sends segmentConsumed with its next offset and stop reason while its stream cursor remains still.
The first controller might not be the partition leader, so how does the protocol avoid choosing two winners?
Only the lead controller owns the completion FSM. A NOT_LEADER response keeps every server holding its stable prefix until a retry reaches that owner.
Switch Controller between Leader ready and Failing over. Compare whether reports enter the lead FSM or retry while every stream cursor remains fixed.
The HOLDING state makes coordination safe while query freshness waits. Server retries clock controller progress without moving any protected cursor forward. One stable authority now receives every completion report. Next the controller will compare offsets and choose one committer.
The highest reported offset becomes the target
The controller now has replica reports and BlockingSegmentCompletionFSM records each next offset in its commit state map. Those comparable positions describe exactly how much of the ordered partition each server accepted.
After all available replicas report or the election window expires, the FSM scans for the maximum offset.
One replica is behind and another is almost tied, so can either safely commit a shorter prefix?
No, the highest offset wins and every smaller offset receives CATCH_UP. The ruler makes the missing suffix explicit before any immutable file is accepted.
Adjust Replica skew from zero through three offsets. Compare whether prefixes already agree or lagging servers must extend their catch-up path to the winner.
Every assigned replica must converge on the complete winning stream prefix. The winner defines content equivalence, not merely which machine uploads first. Smaller prefixes cannot become committed alternatives. Next it will convert that mutable prefix into immutable files.
The real-time converter seals the indexes
The winner owns the target prefix, so buildSegmentInternal creates a RealtimeSegmentConverter around its committed MutableSegmentImpl state. Conversion begins only after mutable writers have closed their generation-specific artifacts.
The converter commits mutable index artifacts and computes sorted document order when the table config names a sort column.
The source rows still live in mutable structures, so what physical result makes the generation portable and identical?
SegmentIndexCreationDriver writes immutable column files and a compressed tar archive. That durable result can replace every replica copy of the winning mutable prefix.
Switch Builder between Column major and Row reader. Compare whether committed columns feed the driver directly or values replay before producing the same portable archive.
The winner now holds a local immutable segment and its uploadable tar archive. Other replicas can install this deterministic result instead of inventing another prefix. The archive makes that result transferable. Next split commit will publish both data and metadata safely.
Commit gates surround the upload
The tar exists locally and SplitSegmentCommitter first asks segmentCommitStart whether this server and offset still own the lease. That validation prevents an obsolete winner from publishing after the controller has changed its decision.
COMMIT_CONTINUE opens the gate and the uploader moves the tar through the configured controller, deep-store, or peer path.
The bytes reached storage but ZooKeeper still describes an open segment, so what closes the publish transaction?
segmentCommitEnd moves any temporary file and writes DONE metadata plus the next CONSUMING segment. The permanent location now closes the publish transaction.
Switch Upload path through Via controller, Direct store, and Peer fallback. Follow the archive destination, then compare which permanent location commitEnd publishes.
Split commit keeps large bytes outside the lead controller when configured. The two gates still bind those transferred bytes to one accepted metadata transaction. Transport cannot bypass the winning lease. Next we will stress the protocol with realistic failures.
Each failure returns to a safe prefix
The happy path committed one prefix and failures matter because a server or controller can disappear between any two protocol calls. Recovery must never turn that interruption into a second accepted generation boundary.
A missing winner makes the FSM election time out and abort. Lagging replicas keep the mutable prefix available for another round.
Controller loss can erase an in-memory FSM and upload can leave temporary bytes, so how can a retry avoid publishing a second prefix?
Every retry revalidates the server and offset while committed replicas keep or download one immutable result. The safe prefix survives every interruption lane.
Switch Failure through Winner crash, Controller loss, Upload fails, and Healthy. Trace where each lane breaks and which recovery returns it to one safe prefix.
Blocking completion is safe because it waits and retries around one prefix. Interrupted work never authorizes a competing result. Next we will measure the freshness cost of that waiting.
Move the freshness handoff before build
The blocking protocol preserved one prefix while build and upload kept every replica cursor stopped, which creates a query freshness gap. Incoming stream records remain unavailable to Pinot queries throughout that stopped interval.
Pauseless performs the same expensive build and upload work while moving the next-generation handoff close to the start.
If conversion pressure doubles during this commit, must the newest queryable record fall farther behind the incoming stream?
Only the blocking gap grows with the full build path because pauseless keeps ingesting after its short handoff. Both lanes still perform the same conversion work.
Adjust Build load from one through four. Compare how the blocking freshness bracket grows while the pauseless handoff remains near the start.
Pauseless improves freshness by reordering work rather than removing work. Conversion and upload still consume the same resources after the early handoff. Only the queryable handoff moves earlier. Next we will inspect the exact server code change.
startSegmentCommit runs before conversion
The timeline showed an early handoff and RealtimeSegmentDataManager implements it when a COMMIT response arrives for a pauseless table. The server must preserve that response lease while it rearranges the later conversion calls.
The server checkpoints its winning offset and calls startSegmentCommit before buildSegmentForCommit begins the expensive immutable conversion work.
The later PauselessSegmentCommitter still uploads and calls commitEnd, so why must it omit another segmentCommitStart call?
The first call already opened the next generation, and repeating it would violate the completed FSM lease. The later committer only uploads and closes metadata.
Switch Protocol between Blocking and Pauseless. Compare whether commitStart appears after immutable conversion or opens the next generation before conversion begins.
Server code now separates the durable generation metadata handoff from later immutable file construction work on the elected server. PauselessSegmentCommitter therefore begins from an already opened controller transition. Next the controller will perform that handoff in ZooKeeper and IdealState.
COMMITTING opens the next CONSUMING segment
The early server call reaches PauselessSegmentCompletionFSM, which validates the elected winner and constructs one committing descriptor. This descriptor carries the fixed boundary before any immutable segment location exists.
The controller writes the old end offset and COMMITTING status before creating next-generation metadata with the same start offset.
Both metadata pages exist, so what tells Helix servers to serve the old segment and consume the next?
One IdealState update flips the old segment ONLINE and the new segment CONSUMING. The identical boundary offset now connects their metadata.
Switch Handoff between Healthy and ZK failure. Compare whether matching offsets activate both generations or leave the boundary unpublished for a safe retry.
The shared boundary is durable before build begins and offset continuity remains explicit. A failed metadata batch cannot expose only one side of that generation change. Both sides succeed together or retry safely. Next the two generations will perform work simultaneously.
Build generation N while consuming N+1
★ If you remember one thing · Pinot builds the sealed generation while the next generation keeps ingesting new records.
The metadata handoff fixed generation N at the winning offset and opened generation N+1 from that exact boundary. Each record therefore belongs to exactly one generation even while their physical work overlaps.
The old mutable strip freezes at its boundary while RealtimeSegmentConverter starts writing immutable column files below that fixed prefix.
Conversion and upload still need time, so must the new stream records wait outside Pinot until those files finish?
Generation N descends into immutable files while generation N plus one grows concurrently. Their offsets meet at one boundary without mixing the two generations.
Adjust New rows from one through four. Compare how the queryable suffix grows while the frozen old prefix keeps building from the same boundary.
Pauseless overlaps generations without mixing their offsets. Query freshness advances through the new suffix while immutable durability catches up behind it. Next we will see how non-winning replicas join the same boundary.
Non-winners keep, catch up, or download
Generation N+1 is already consuming and other replicas continue reporting their generation N offsets to the same controller FSM. Their replacement choice depends on prefix equality and the table completion policy.
An equal offset receives KEEP and may build locally. A smaller offset receives CATCH_UP before it can retain the prefix.
Local build can create memory pressure or conflict with dedup ordering, so when should the replica prefer an immutable download?
Completion mode and parallel policy choose local build or download while preserving the boundary. Every route ends with one equivalent immutable generation.
Switch Replica path through Equal offset, Lagging, and Download. Compare whether the non-winner keeps, catches up, or installs the committed immutable copy.
Every replica eventually serves one immutable generation N while consuming N+1 safely. Equivalence comes from the winning offset even when installation routes differ. Local path choice cannot change the prefix. Next commitEnd will finish the winner metadata.
CommitEnd fills the missing immutable facts
The old segment has been building while the new one consumes and its ZooKeeper page still says COMMITTING. That explicit state distinguishes a valid early handoff from a fully published immutable segment.
PauselessSegmentCommitter uploads the tar and sends commitEnd with generated metadata files and the resolved segment location.
The generation boundary is already live, so which facts must arrive before the old segment can honestly become DONE?
CRC, size, document count, and download URL fill the page before status becomes DONE. The completed file facts also clear the debugging breadcrumb.
Switch Upload result between Success and Missing URL. Compare whether immutable facts close metadata as DONE or leave COMMITTING recoverable.
CommitEnd closes the immutable file half of the handoff and removes the controller debugging breadcrumb. Operators can now distinguish completed generations from ones that still require repair. DONE means the permanent facts agree. Next we will recover cases where that closure never arrives.
Validation repairs an unfinished generation
A server or controller may fail after the early handoff, leaving COMMITTING metadata while generation N+1 continues to ingest. Recovery must repair the sealed old prefix without rolling back the active new generation.
RealtimeSegmentValidationManager first waits beyond the configurable completion window so it does not race a healthy commit.
The segment is now stale, so should every failure trigger the same replay from the stream?
Metadata is reconciled, peers repair missing files, and total replica loss may require reingestion. The failure location and ordering policy jointly select the safe action.
Switch Failure point, then toggle Disaster recovery. Compare the repair action with the replay policy.
Pauseless recovery preserves freshness while treating replay as a policy decision for dedup and partial-upsert tables. Peer evidence is preferred because replay can disturb their strict ordering guarantees. Automatic repair therefore stops before any unsafe reconstruction. Now let us reconnect the whole system.
The complete real-time ingestion map
We started with one stream partition and several independently advancing replicas whose mutable rows answered queries immediately.
Then RealtimeSegmentDataManager fetched, decoded, transformed, indexed, and checkpointed each record through one executable server loop.
Mutable forward data, dictionaries, filter bitmaps, and ranges created query paths before any immutable file existed.
Rollover gates converted row, time, force, stream-end, and capacity limits into one completion entry point.
Normal completion stopped every replica cursor while repeated segmentConsumed reports clocked the controller state machine.
The controller selected the highest reported offset and made lagging replicas catch up to one shared prefix.
RealtimeSegmentConverter froze that winning prefix into sorted immutable column files and one uploadable tar archive.
Split commit wrapped storage transfer between commitStart and commitEnd so bytes and metadata published coherently.
Failure handling retried leases, aborted stale FSMs, and replaced replicas without allowing two committed prefixes.
The first pauseless lesson measured the freshness gap and moved the next-generation handoff ahead of expensive build work.
Server code made that ordering change explicit by calling startSegmentCommit before RealtimeSegmentConverter begins its expensive conversion work.
The controller fixed the old end offset, marked COMMITTING, created the next metadata, and flipped IdealState.
Two generations then overlapped safely because the old prefix built while the new suffix kept growing.
Non-winning replicas used offset equality and policy to keep, catch up, build, or download one immutable copy.
commitEnd supplied CRC, size, and location before turning the old metadata from COMMITTING into DONE.
Validation waited for staleness and then repaired metadata, deep-store copies, replica errors, or policy-approved reingestion.
The unified lesson is that offsets preserve continuity, mutable indexes preserve freshness, immutable commit preserves equivalence, and pauseless overlaps their work safely.
The whole story in 16 lines
Pinot keeps real-time rows queryable by indexing in memory, coordinating one immutable commit, and handing consumption to the next...
- One stream partition is consumed by several replicas whose mutable rows are immediately queryable.
- RealtimeSegmentDataManager fetches, decodes, transforms, indexes, and advances the next offset.
- A mutable row updates forward data, dictionaries, filter indexes, and value ranges together.
- Row, time, stream-end, force, or index-capacity limits can seal a consuming segment.
- Normal completion stops replica cursors while they report segmentConsumed to the lead controller.
- The controller selects the highest reported next offset and asks lagging replicas to catch up.
- RealtimeSegmentConverter turns mutable columns into a query-optimized immutable segment and tar file.
- Split commit brackets upload with commitStart and commitEnd before metadata becomes DONE.
- Retries, catch-up, keep, discard, peer download, and FSM restart contain blocking-protocol failures.
- Pauseless shortens the query freshness gap by moving the generation handoff ahead of build.
- The server calls segmentCommitStart before build and PauselessSegmentCommitter does not repeat it.
- The controller marks the old segment COMMITTING while opening the next one as CONSUMING.
- Old-generation build and new-generation ingestion proceed at the same time on the server fleet.
- Non-winners keep equal data, catch up when behind, or download the committed immutable copy.
- CommitEnd fills CRC, size, and download URL before COMMITTING becomes DONE.
- Validation repairs stale metadata, missing deep-store copies, and recoverable replica failures.