Speed up compaction by skipping the segment download in the planning phase

Start with CompactionTask.runTask, ExistingSegmentAnalyzer.fetchAndProcessIfNeeded(), CompactionIntervalSpec, and the issuer paths in the Overlord supervisors and Coordinator CompactSegments duty. Trace how stored schema payloads are read, merged, and guarded by the segment ID hash; compare the merge results with the existing analysis behavior. Done means the opt-in flag skips downloads only for fully supported inputs, while all specified fallback cases retain current behavior.

Motivation

Before a compaction task starts its subtasks, it downloads every input segment just to work out the schema: dimensions, metrics, rollup and query granularity. Unless the compaction config spells all of these out, which few do, this happens on every run. So each auto-compaction task reads its whole input from deep storage twice: once to plan, once to compact.

When centralized datasource schema is enabled, the metadata store already has the schema of every published segment. This proposal has the compaction issuer read those stored schemas when it creates the task, put the merged schema into the task spec, and record the hash of the segment IDs it used. The task's existing segment-set check then confirms nothing changed before it relies on the embedded schema. Whenever the stored schema can't fully describe the input, the task downloads and analyzes as it does today.

Numbers from a production cluster ingesting from Kafka:

Segment granularity	hourly
Input per compaction interval	44–63 GB, 90–140 segments
Daily volume of the largest datasource	~550 GB
Planning phase per task	~16 min, all of it spent downloading segments
One subtask	~6 min (45% read/persist, 52% merge/push)
maxNumConcurrentSubTasks	4, so the task reserves 5 slots

What this costs:

• Every task starts with ~16 minutes of planning. Only the parent task runs during this time, while the 4 slots reserved for subtasks sit idle. That's about 64 idle slot-minutes per task.
• Deep storage is read twice. For one hourly datasource that is ~24 × 55 GB ≈ 1.3 TB of extra reads per day, plus the same amount of temporary peon disk.
• Compaction falls behind. With a fixed compaction slot budget, a long planning phase per task means fewer intervals compacted per hour. Fragmented intervals stay that way longer, which slows queries.

Today, skipping the planning download means writing a full dimensionsSpec, metricsSpec and granularitySpec into each compaction config by hand. That's tedious, and the hand-written copy drifts whenever the ingestion schema changes. Columns added later are silently dropped from compacted segments.

Proposed changes

Current behaviour

1. CompactionTask.runTask builds the input schemas (createInputDataSchemas).
2. ExistingSegmentAnalyzer.fetchAndProcessIfNeeded() skips the analysis only when rollup, query granularity, dimensionsSpec and metricsSpec are all already known. For MSQ, it must also not need to detect multi-value dimensions (identifyMultiValuedDimensions()).
3. Otherwise it downloads every segment (AcquireMode.FULL). From each one it reads the rollup, query granularity, dimensions, metrics, multi-value dimensions and projections (processRollup, processQueryGranularity, processDimensionsSpec, processMetricsSpec, processMultiValuedDimensions, processProjections).
4. Separately, the task already supports guarding its input. CompactionIntervalSpec has a sha256OfSortedSegmentIds field. After the task takes its lock, checkSegments compares that hash with the current used segments and fails the task if they differ. No issuer sets the hash today: CompactSegments passes null.

With centralized datasource schema enabled:

• Each row in druid_segments has a schema_fingerprint.
• druid_segmentSchemas stores a SchemaPayload (row signature and aggregator factories) for each fingerprint.
• Many segments share one fingerprint, so an interval with a hundred segments usually has only a few distinct schemas.

Issuer side

Applies to the Overlord's compaction supervisors and the Coordinator's CompactSegments duty, which both build tasks the same way.

When the feature is enabled and the compaction config leaves dimensionsSpec or metricsSpec unset:

1. Take the exact list of segments the job will compact.
2. Read their schema fingerprints and payloads from the metadata store in one query for the job's segments. The Coordinator can use its in-memory schema cache instead.
3. If any segment has no stored schema, embed nothing and leave the task as it is today.
4. Merge the schemas the same way the analysis does:
   • Dimension order: take segments newest first, then the union of their dimensions.
   • Types: map the column type to a dimension schema (string, long, float, double).
   • Metrics: combine aggregators with the same merge logic the analysis uses.
5. Put the result into dimensionsSpec and metricsSpec. Put SegmentUtils.hashIds(segments), computed from the same list, into inputSpec.sha256OfSortedSegmentIds.

Rollup and query granularity aren't in the stored schema. Phase 1 embeds only when the compaction config's granularitySpec sets them; otherwise the task falls back to the download (see Phasing).

Task side

No change is needed for correctness.

• The embedded specs make shouldDownloadSegments() return false, so the download is skipped.
• checkSegments runs after the lock is taken and before any data is read. If the segment set changed between planning and running, for example because late data added a segment with a new column, the task fails within seconds. The next supervisor run plans that interval again from the new segment set.
• Once the check passes, the time-chunk lock stops the input from being replaced or dropped. Segments appended concurrently under an APPEND lock aren't part of the input. They are upgraded to the new version as they are, so the embedded schema never needs to cover them.

When the issuer must not embed

In each of these cases the task falls back to today's download:

• Any input segment has no stored schema.
• A column's stored type can't be mapped to one dimension schema without guessing. This covers auto or nested JSON columns and complex dimensions, and columns whose type differs across segments.
• Any input segment has projections (DataSegment.getProjections() is not empty) and the compaction config doesn't specify them. The analysis is how existing projections are carried over today.
• The task will run on MSQ and needs to detect multi-value dimensions: rollup is possible, or the partitioning is range on string dimensions.
• The compaction config doesn't set rollup and query granularity (phase 1 only).

Feature flag: useStoredSegmentSchema

A new boolean field in the cluster compaction config (ClusterCompactionConfig):

{
  "useSupervisors": true,
  "useStoredSegmentSchema": true
}

• Default: false in phase 1. Change the default to true once real clusters show the output matches today's analysis.
• Dynamic: it's set through POST /druid/indexer/v1/compaction/config/cluster like the other cluster compaction settings, so turning it on or off needs no restart. It affects only tasks created after the change.
• Scope: read by every issuer, the Overlord's compaction supervisors and the Coordinator's CompactSegments duty, so behaviour is the same whichever one runs compaction.
• Needs centralized schema: with druid.centralizedDatasourceSchema.enabled=false there are no stored schemas. The flag then does nothing and the issuer logs that once.
• Per-datasource override: not in phase 1. A datasource whose stored schema can't be used already falls back on its own (see "When the issuer must not embed"). An override can be added later if a real case needs it.

Why a flag

When the flag is on, compaction tasks are built differently from today. If the merged schema differs in any detail from what the analysis would have produced (dimension order, a column type, how aggregators are merged), the compacted segments come out different, for example with a column stored as a different type. Nothing reports this, and once the input segments are replaced it can't be undone. A default-off flag keeps that risk opt-in while the merge logic is new, and gives operators a way to turn it off without a restart.

It doesn't cause intervals to be compacted again. The issuer embeds a schema only when the compaction config leaves dimensionsSpec and metricsSpec unset, and in that case CompactionStatus doesn't compare them with lastCompactionState.

Phasing

• Phase 1:
  • Native compaction, plus MSQ jobs that don't need multi-value detection.
  • Only when the compaction config sets rollup and query granularity.
  • Off by default, behind useStoredSegmentSchema.
• Phase 2:
  • Add optional fields to SchemaPayload: rollup, query granularity, a multi-value flag per column, and the original dimension schema.
  • These remove the remaining fallbacks, including the MSQ ones.
  • Rows without the new fields keep meaning "unknown, fall back", so the change is backward compatible.
• Phase 3 (optional): if the fallback is still common, use partial segment reads to make it cheaper.

Rationale

Have the task fetch the stored schema itself

The task would call a new Overlord API at run time instead of getting the schema in its spec.

• Pro: no segment-set race, because the task reads after taking its lock.
• Con: needs a new task action or endpoint, and a new fetch-and-fallback path inside the task. The schema used is also not visible in the task spec.

Embedding needs no new API. It reuses the existing segment-set check, and the schema can be inspected in the task payload.

Run a SegmentMetadataQuery against Historicals

• It needs the segments loaded. Compaction candidates may have been dropped by load rules, be on a cold tier, or not be loaded yet.
• It makes compaction depend on the Broker and Historicals, and adds metadata queries to the query tier. Today compaction only reads deep storage and the metadata store.
• With centralized schema on, the Coordinator already uses segment metadata queries to fill in missing schemas and stores the results. Reading the store gives the same information without running the query again.

It could serve as an optional middle fallback (stored schema, then segment metadata query, then download), but this proposal doesn't depend on it.

Partial segment reads (V10 segment format)

Reading only the metadata part of each segment would make the download cheaper, but it remains one remote read per segment. It needs the V10 format and a range reader for each deep-storage type. It works well as a speed-up for the fallback path, and doesn't conflict with this proposal.

Put the schema in the compaction config

This is already possible. It has to be maintained by hand, and it silently drops columns added to ingestion later. The stored schema follows ingestion automatically.

Per-segment schema cache on the Overlord

The Overlord doesn't bind a per-segment schema cache today. Reading the schema for each job is enough: one indexed query for a few distinct fingerprints, compared with downloading tens of GB. It adds no Overlord heap and no new sync loop. HeapMemorySegmentMetadataCache already has a place to sync schemas, if a cache is wanted later.

Operational impact

• Nothing is deprecated or removed. With useStoredSegmentSchema off (the default), or with centralized datasource schema disabled, compaction behaves exactly as it does today.
• No migration is needed. Operators who already run centralized datasource schema turn the feature on by setting the flag in the cluster compaction config.
• Rolling upgrade and downgrade are safe. The issuer only fills fields the task spec already has (dimensionsSpec, metricsSpec, sha256OfSortedSegmentIds), so older task versions run these tasks unchanged. Setting the flag back to false restores today's behaviour for every task created after the change.

Future work

• Phase 2 fields in SchemaPayload.
• Partial-read fallback (V10 segment format).
• If a mismatch failure is common, fall back to the download inside the task instead of failing. This needs a separate field so the schema guard is independent of the existing input check.