Segment Merge Architecture

See Configure Segment Merge for how to enable and run merges. This page explains what a merge does, which segments it picks, and what it costs.

Diagram of segment merge: the top band shows how a merge runs — a scheduled run (Quartz cron schedule, window of the last day) or an on-demand run (POST /tasks/execute, window set per request) starts the Pinot Controller’s Select step, which groups consecutive sealed segments of each partition, oldest first, up to the table’s target size of 1 GiB or 2 GiB or 8 segments, and skips the latest sealed segment, segments that ended in the last hour, segments already large enough, segments that expire within 7 days, segments outside the window, segments still being merged, and groups whose merge failed, until cleanup; up to tableMaxNumTasks KfMergeTask tasks wait in a task queue for a free Pinot Minion, whose Merge step downloads the group, sorts rows by the table’s sort order, and rebuilds every index, and, for the rollup table with its resolution filter on, writes one segment per resolution and drops any resolutions listed to drop; the Replace step swaps the merged segment in for its inputs in one atomic step, with no missing or duplicate rows. The bottom band shows one stream partition before and after: five consecutive sealed segments form one merge group and become a single merged segment with the same rows, sorted, every index rebuilt, usually smaller, while a segment that ended in the last hour, the latest sealed segment, and the consuming segment are skipped and left as they are

Why merge segments

Each stream partition of a realtime table seals a segment when the segment reaches its size or time threshold. Over time a busy table accumulates many moderately sized segments, and every segment adds memory on the Pinot controller and servers, plus planning work on every query. Merging consecutive segments into fewer, larger ones reduces that overhead without changing the data.

How a merge works

The segment merge task, KfMergeTask, runs on the Pinot minions and rewrites sealed segments into fewer, larger ones:

  1. Select. The Pinot controller looks at the sealed segments of each partition and groups consecutive segments, oldest first, until a group’s input size reaches the table’s target size (1 GiB or 2 GiB, see Table groups) or the group holds 8 segments. A group never mixes stream partitions.

  2. Merge. A minion downloads the group, merges its rows into one segment sorted by the table’s sort order, and rebuilds every index of the table.

  3. Replace. The merged segment replaces its inputs in one atomic step. Queries never see missing or duplicated rows, before, during, or after the swap.

A merge keeps every row unchanged. The one exception is the metrics rollup table: since 4.4.0, its merges delete the rows of legacy resolutions by default (see Rollup table resolutions). The merged segment is usually smaller than the sum of its inputs, because sorted data compresses better.

Stopping or aborting a merge is always safe: a merge either replaces its inputs completely, or leaves them untouched.

Which segments are merged

Segment merge runs only on realtime tables, and skips tables with upsert or dedup enabled. Within a table, the controller skips segments that should not be merged:

  • the latest sealed segment of each partition;

  • segments that ended in the last hour;

  • segments that are already as large as the target size;

  • segments whose rows expire within 7 days. This applies only to rows that carry their own expiry time; segments whose rows have none never count as expiring;

  • segments outside the merge window. Scheduled runs only consider data from the last day (mergeNewerThan: "1d"). On-demand runs can set their own window to merge older data;

  • segments that another task is still working on: a running merge, a delete, or an object storage conversion.

A skipped segment ends the group in front of it: the controller never merges segments across a skipped one, so every merged segment covers a continuous stretch of the partition. A group that ends before it reaches the target size is merged only if it holds at least 2 segments and at least half the target size. As a result, a quiet partition, or one with many skipped segments, can stay unmerged.

Because of the 7-day rule, a table whose retention is 7 days or less merges nothing by default when its rows carry an expiry time: every one of its segments expires within the minimum remaining lifetime.

A merge that fails leaves its inputs untouched. Later runs skip the failed group until its task records are cleaned up, so a merge that can’t succeed doesn’t keep using minion time.

On every run, the controller logs how many segments it considered, how many were eligible, and why it skipped the others; see Read the controller log.

How scheduled runs execute

The Pinot controller runs scheduled merges with the Quartz scheduler. At each time the table group’s cron expression matches, the lead controller for the table runs one selection pass, as described in How a merge works. It submits up to tableMaxNumTasks merge tasks, which wait in a queue until a minion is free. Segments that are still being merged are skipped, so a run that starts before the previous one finishes never merges a segment twice. The schedule uses the controller’s time zone, which is normally UTC, and the controller picks up a new schedule without a restart.

An on-demand run uses the same selection pass, with settings that apply to that run only. The output of a merge can only be merged again once its task is done.

Rollup table resolutions

kf_metrics_rollup holds rows of several resolutions. Since Kloudfuse 4.4.0, each resolution is ingested into its own stream partitions, so every new segment holds a single resolution. Segments ingested before 4.4.0 hold rows of every resolution mixed together, and segments of both kinds can remain until retention deletes the older ones.

A merge never combines stream partitions, so segments ingested since 4.4.0 stay single-resolution when they are merged. Older, mixed segments are rewritten into one segment per resolution, the layout of segments ingested since 4.4.0, by a resolution filter that pinot.metrics.mergeTask.rollupSplitByResolution turns on. It is on by default. With the filter on, a merge filters rows on the resolution column:

  • It keeps the resolutions in global.metrics.rollupResolutions, plus global.metrics.legacyRollupIntervalSecs, each in its own segment. Rollup rows written before 3.5.3 carry no resolution and are read as the legacy interval.

  • It deletes the rows of the resolutions in filterDropValues. By default, that is the legacy resolutions 600, 3600, and 14400 that the merge doesn’t keep.

  • A row whose resolution is in neither list fails its task and leaves the data untouched.

With the filter off, a merge keeps every row and writes mixed segments, still mixed, into one segment. It also skips segments that an earlier filtered merge already split. See Merge the metrics rollup table for the settings.

Resource use

  • CPU: a merge sorts its inputs and builds its columns on several threads at once: one for every 8 CPUs of the minion, between 1 and 4. A minion with a CPU limit of 32 or more therefore uses up to 4 cores per merge. Merges running at the same time on one minion share the extra threads, so N concurrent merges use at most N + 3 cores. If that exceeds the minion’s CPU limit, the minion can lose its cluster session and drop its running merges.

  • Disk: temporary disk of about three times the target segment size per running merge.