Druid segments, partitioning, and the hot key that broke my assumptions

At work we run Apache Druid as the storage and query engine behind our observability platform for logs ; and logs on the order of five billion a week flow through it. For a long time everything behaved. Then one tenant’s queries started getting slow, and the reason turned out to be something Druid never told us about: some of our segments were many times larger than the size we had explicitly configured, and every ingestion task had reported success anyway.

This post is the writeup of that investigation. I’ll build up the concepts first: what a segment is, what partitioning means, primary vs secondary partitioning, and how hash and range partitioning actually work, and then walk through the specific failure mode I hit, why it’s silent, and how we fixed it. If you already know Druid well, you can skip to the problem.

What is a segment?

Druid doesn’t store a table as one big file. It stores it as a large number of segments, self-contained, immutable columnar files that are the fundamental unit of storage, distribution, and query in the system. A segment typically holds a few million rows and is designed to be scanned in the low hundreds of milliseconds. Each segment carries its own dictionary-encoded columns, bitmap indexes, and compressed value arrays, so it can be queried entirely on its own.

Every segment is identified by four things:

• Datasource: the table it belongs to.
• Time interval: the chunk of time it covers (e.g. one day).
• Version: a timestamp used to resolve overwrites; newer versions shadow older ones.
• Partition number (shard spec): which slice of that time interval this segment is.

That partition number is where this whole story lives, so let’s pull it apart.

Primary partitioning: time

Druid’s primary partitioning is always by time. When you ingest, you choose a segmentGranularity: HOUR, DAY, MONTH, and so on. Each time chunk (Druid calls it an interval or “time chunk”) becomes a bucket, and all the rows whose timestamp falls in that chunk go into segments for that chunk.

We run daily segment granularity. So conceptually the data is first split into one bucket per day. This is what makes time-range queries cheap: a query for “last Tuesday” only has to touch the segments for that day and can prune everything else.

Secondary partitioning: splitting within a time chunk

One day of data is usually far too large to be a single segment. So within each time chunk, Druid applies secondary partitioning to break the day into multiple segments. This is the knob that decides how many shards a day becomes and which row goes into which shard.

How big should each of those segments be? You tell Druid, with a single number in the ingestion config called maxRowsPerSegment, the rough number of rows you want per segment. We set it to 5,000,000. So the mental model everyone carries is simple: “a day gets sliced into ~5M-row segments.” Hold onto that sentence. It is exactly the assumption that turned out to be wrong for one tenant.

Druid’s native batch ingestion offers three secondary partitioning strategies. They differ entirely in how they assign a row to a shard.

1. Dynamic partitioning

The simplest one. Rows are appended to the current segment until it hits maxRowsPerSegment, then a new segment is started. It’s essentially “fill a bucket, start another bucket.” It’s cheap (single pass, no shuffle) but it gives you no control over which rows land together, so it doesn’t help with query pruning beyond time.

2. Hash partitioning

You pick one or more partitionDimensions. Druid hashes those dimension values for each row and assigns the row to a shard based on the hash. All rows with the same partition-key value land in the same shard. This is great for pruning: if you query WHERE orgId = X, Druid can compute the hash of X up front and only read the shard(s) that value could be in.

We hash by orgId. The core of the assignment is a Murmur3 hash, modulo the number of buckets:

// HashPartitionFunction.MURMUR3_32_ABS
return Math.abs(
    Hashing.murmur3_32().hashBytes(serializedRow).asInt() % numBuckets
);

Two things to notice here, because they matter later. First, this is deterministic: a given orgId always hashes to the same bucket. Second, the shard is chosen purely from the hash of the key: nothing in this function looks at how many rows share that key.

3. Range partitioning

You pick one or more dimensions, and Druid partitions by ranges of their values rather than by a hash. It samples the data, finds boundaries so that each range holds roughly the target number of rows, and assigns each row to the range it falls into. Segments end up sorted and non-overlapping on the range dimension, which enables both pruning and better compression (similar values sit together).

Range partitioning is more aware of distribution than hashing. It’s literally trying to size ranges by row count. But it has its own sharp edge when a single value dominates, which I’ll come back to.

The problem: one hot key, one giant segment

Here’s our setup, concretely:

• Daily segment granularity.
• Hash partitioning by orgId.
• maxRowsPerSegment = 5,000,000.

For the vast majority of tenants this is perfect. Their daily volume is modest, hashing spreads orgs across shards, and segments come out near the 5M target. Then we onboarded one very large tenant, roughly 5 billion logs a week, all under a single orgId.

Now trace one row through the hash function above. Every row for that org shares the same orgId, so every row hashes to the same bucket. Hashing simply cannot split one key across shards. That’s the whole point of it, it’s what lets a query for orgId = X read just one shard. So a full day of that org’s data piles into a single shard, and that segment blows straight past the 5M target, not by a little, by orders of magnitude.

Here’s the deeper reason Druid doesn’t catch this. Before ingesting, Druid decides how many shards to make for the day, and it does that with simple division: roughly total rows for the day ÷ maxRowsPerSegment. If a day has 50 million rows and the target is 5 million, it plans for about 10 shards.

But that division only tells Druid how many shards to create. It silently assumes the rows will land in them evenly, about 5 million each. It never checks how rows are actually distributed across orgIds. So when one orgId owns most of the day’s rows, Druid still plans those ~10 shards as if the load were even, then hashing dumps that entire org into one of them. That one shard ends up enormous while the other nine sit nearly empty. The plan and the reality quietly disagree, and nothing reconciles them.

It’s worse than uniformly big. It’s spiky

What made this genuinely hard to spot: this org’s volume isn’t even steady through the day. In one hour it might send ~20 million rows, and in the next ~300 million. So its segments aren’t uniformly oversized: a few hours are monstrous and the rest look completely normal. There’s no single number you can glance at to catch it; you have to inspect segments one by one to even see the pattern.

And Druid says nothing

This is the part that actually bothered me. When a segment comes out at 10× or 50× the configured target, there is:

• no WARN log,
• no metric,
• no signal in the task report,
• nothing in the Overlord UI.

The ingestion task simply reports success. From the operator’s seat, the first symptom is “queries for this tenant are slow,” which sends you down the usual query-tuning rabbit hole. It took manual per-segment size inspection to find that the real cause was a handful of monster segments, oversized because of a partitioning assumption, not because of anything in the query path.

What I tried

Once I understood the mechanism, the interesting question was: can I fix it by changing the partitioning strategy? I spent a good while experimenting with segment layout across the different algorithms.

Lowering maxRowsPerSegment doesn’t help. It’s only a target, and it assumes rows are spread evenly; a single hot key ignores it no matter what value you pick, because hashing still can’t split that one key across shards.

Switching to range partitioning is more interesting. Range partitioning sizes ranges by actual row count, so in principle it reacts to distribution. But when one single value dominates, a range boundary can’t split within that value: a range is [value, value] at best, and one value’s rows can’t be broken across two ranges without breaking the “non-overlapping ranges” guarantee that makes pruning work. Druid’s own code even acknowledges this: there’s a // Future improvement: Handle skewed partitions better TODO sitting in PartitionBoundaries that points right at this gap.

Adding a secondary partition dimension is the standard advice: partition by (orgId, someOtherDim) so the hot org gets broken up by a second dimension. The catch is that it only works if that second dimension actually varies within the hot org. If you pick something that’s roughly constant for that tenant, you’ve gained nothing. So the real work was finding a dimension with enough spread inside the hot key.

What actually fixed it

Two changes together did it for our case:

• Hourly segment granularity instead of daily. This directly attacks the spikiness: each hour becomes its own set of segments, so a catastrophic hour is contained to that hour rather than dragging a whole day’s segment up with it. It also shrinks the base unit of work, so the sizing math has a smaller, more even amount of data to slice per time chunk.
• Adding the data-plane nodeId as a secondary partition dimension. Unlike the natural candidates we’d considered, nodeId actually varies within the hot org. That tenant’s traffic is spread across many data-plane nodes. Partitioning by (orgId, nodeId) means the hot key is no longer a single hash bucket; it fans out across nodes, so no one shard has to hold the entire org’s hour.

The reason this combination works is that each change fixes a different half of the problem. Hourly granularity handles the temporal skew (the 20M-vs-300M-rows-per-hour spikiness), and the nodeId secondary dimension handles the key skew (one orgId hashing to one bucket). Either one alone left a gap; both together brought the tenant’s segments back down toward the target and the query regression went away.

Takeaways

• A Druid segment’s size is a best-effort target, not a guarantee. maxRowsPerSegment assumes your partition key is well distributed.
• Hash partitioning cannot split a single key. One hot value = one (huge) shard, by design.
• Shard count is derived from total rows ÷ maxRowsPerSegment, assuming rows spread evenly across buckets, so per-key skew is invisible to the sizing math.
• Skew that’s also spiky in time is especially nasty: some segments are fine, some are monsters, and averages hide it.
• The fix was two-sided: hourly granularity for the temporal spikiness, and a secondary partition dimension that actually varies within the hot key (nodeId) for the key skew.

The thing I’d underline is how much of this was diagnosis rather than a clever fix. The mechanism (total-rows sizing, deterministic hashing, one key to one bucket) is all knowable up front, but nothing surfaces it at ingestion time. Once you can actually see which segments blew past the target, picking the right combination of granularity and partition dimensions is the comparatively easy part.