perf: speed-up coordinator segment metadata cache delta syncs by jtuglu1 · Pull Request #19672 · apache/druid

jtuglu1 · 2026-07-10T00:21:29Z

Description

Currently, coordinator segment metadata cache syncs (even deltas) are incredibly slow for large used segment sets. This increases the ingest-to-query latency of data as well as hindering coordinator recovery in the event of failover + leadership re-election.

Speed up coordinator delta syncs by rebuilding a delta snapshot of the changed datasources rather than the full thing. Currently, every poll it throws away the old snapshot and builds a brand new one for every datasource, walking all used segments to rebuild every timeline; it does this even when almost nothing changed since the last poll. For a cluster with 25M+ segments, this can take ~2mins of time.

This also adds a better-covering index. The segments table has an index on used, so the db can quickly find which rows are used. But that index does not contain dataSource or used_status_last_updated, so this scan is very slow for large #s of segments (4mins+). We should then delete the used-only index since it is fully covered by this one.

Release note

Speed-up coordinator metadata cache delta syncs

This PR has:

FrankChen021

Severity	Findings
P0	0
P1	1
P2	1
P3	0
Total	2

Severity	Findings
P0	0
P1	1
P2	1
P3	0
Total	2

Reviewed 7 of 7 changed files.

This is an automated review by Codex GPT-5.6-Sol

FrankChen021 · 2026-07-10T12:03:11Z

-        datasourceToUsedSegments.put(dataSource, cache.findUsedSegmentsOverlappingAnyOf(List.of()));
+        // Only materialize the (potentially large) used-segment set for datasources
+        // that must be rebuilt into the snapshot.
+        if (!incremental || dirtyDatasources.contains(dataSource)) {


[P1] Track write-through cache mutations as dirty

Incremental rebuilding only materializes datasources returned by updateSegmentIdsInCache. However, CachedSegmentMetadataTransaction applies successful inserts and deletes directly to the datasource cache. On the next poll, the database and cache already match, so syncSegmentIds reports no expired or deleted IDs and the datasource is skipped here. Consequently, newly inserted segments can remain absent from the published snapshot, and partial deletions can remain present, indefinitely until an unrelated update dirties that datasource. Record used-segment mutations from writeCacheForDataSource or otherwise compare against the prior snapshot before reusing it.

Nice catch! This does seem to be an issue that will cause the snapshot and the cache for a datasource to diverge. Normally, the snapshot reconciles itself with what's in the cache in the next sync.
But we are removing that capability in this PR by updating snapshots for modified datasources only.

Record used-segment mutations from writeCacheForDataSource or otherwise compare against the prior snapshot before reusing it.

Comparing against prior snapshot might be too costly, I think we could do the following instead.

ReadWriteCache keeps a volatile boolean hasNewWrites.

Set this flag to true in withWriteLock().

Also, add method ReadWriteCache.markAllWritesAsSynced()

Add new methods to HeapMemoryDatasourceSegmentCache: getAllUsedSegmentsIfCacheHasNewWrites().

Invoke this method in incremental mode only

This method should acquire a write lock, return all the used segments only if there are writes and finally set hasNewWrites to false.

FrankChen021 · 2026-07-10T12:03:11Z

+    createIndex(
+        tableName,
+        "IDX_%S_USED_USLU_DATASOURCE",
+        List.of("used", "used_status_last_updated", "dataSource")


[P2] Include id in the portable covering index

The assumption that a secondary index implicitly carries the primary key is specific to engines such as InnoDB. PostgreSQL stores the primary key in a separate index, so this index still requires a heap lookup for every selected id, defeating the intended index-only scan on a supported metadata backend. Include id explicitly or provide backend-specific index definitions.

Yeah, I have my doubts about this too.

But it is okay since after this, we can get rid of the older IDX_USED index.
@jtuglu1 , should we do it in this PR itself? It will affect only new clusters anyway.

kfaraz

Thanks for the changes, @jtuglu1 !
I have left some suggestions.

kfaraz · 2026-07-10T12:46:37Z

+  public static DataSourcesSnapshot withUpdatedDataSources(
+      @Nullable DataSourcesSnapshot previous,


For a cleaner API, let's not use a static method which allows passing a null previous snapshot.

Instead, add a non-static method updateSnapshotForDatasources() which will be called on the existing snapshot by the caller itself.

newSnapshot = previousSnapshot.updateSnapshotForDatasources(changed, removed, snapshotTime);

kfaraz · 2026-07-10T12:48:53Z

+    final Map<String, ImmutableDruidDataSource> dataSources = new HashMap<>(previous.dataSourcesWithAllUsedSegments);
+    final Map<String, SegmentTimeline> timelines = new HashMap<>(previous.usedSegmentsTimelinesPerDataSource);
+
+    final Set<String> dirtyDataSources = new HashSet<>(removedDataSources);


Please use some other term here like modified or datasourcesToRefresh, etc.

kfaraz · 2026-07-10T12:49:34Z

+      timelines.remove(ds);
+    });
+    changedDataSources.forEach((ds, segments) -> {
+      if (segments.isEmpty()) {


Wouldn't this case already be present in removedDatasources? Is this just a safe-side measure?

kfaraz · 2026-07-10T12:50:45Z

+   * {@link DataSourcesSnapshot} instead of the whole snapshot. Drift-free: the
+   * set read from the metadata store is unchanged (still the full used set).
+   */
+  private final boolean useIncrementalSync;


I don't think this needs to be optional since it is a perf improvement that all clusters can benefit from.
Are there any drawbacks to the approach?

There shouldn't be, I'll remove.

kfaraz · 2026-07-10T12:51:10Z

      @JsonProperty("useIncrementalCache") SegmentMetadataCache.UsageMode useIncrementalCache,
-      @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused
+      @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused,
+      @JsonProperty("useIncrementalSync") Boolean useIncrementalSync


We can skip this config and have this feature always on.

kfaraz · 2026-07-10T12:56:09Z


+    // Datasources whose used-segment set changed this sync. Null forces a full
+    // snapshot rebuild (first sync, or when incremental sync is disabled).
+    Set<String> dirtyDatasources = null;


Suggested change

Set<String> dirtyDatasources = null;

Set<String> modifiedDatasources = null;

kfaraz · 2026-07-10T12:57:33Z

+   * <p>
+   * Overshadowing is computed per-datasource, so reusing the prior overshadowed
+   * segments for unchanged datasources is correct. This turns the O(all-segments)
+   * snapshot rebuild into an O(changed-segments) operation, while producing a


This is not exactly true since we would still be rebuilding the entire timeline of a datasource only if a single segment changed in that datasource. That said, it would still be better than what we have today.

kfaraz · 2026-07-10T13:00:45Z

@@ -743,13 +788,17 @@ private void retrieveRequiredUsedSegments(
   * datasource cache is atomic. Also identifies the segment IDs which have been
   * updated in the metadata store and need to be refreshed in the cache.


Suggested change

* updated in the metadata store and need to be refreshed in the cache.

* updated in the metadata store and need to be refreshed in the cache.

*

* @return Set of datasource names whose used segment set has changed in this sync and need to be refreshed in the snapshot.

*/

kfaraz · 2026-07-10T13:01:14Z


+    // Datasources whose used-segment set actually changed this sync (used to
+    // rebuild only the affected part of the snapshot).
+    final Set<String> dirtyDatasources = new HashSet<>();


Suggested change

final Set<String> dirtyDatasources = new HashSet<>();

final Set<String> datasourcesToRefresh = new HashSet<>();

kfaraz · 2026-07-10T13:52:52Z

-        datasourceToUsedSegments.put(dataSource, cache.findUsedSegmentsOverlappingAnyOf(List.of()));
+        // Only materialize the (potentially large) used-segment set for datasources
+        // that must be rebuilt into the snapshot.
+        if (!incremental || dirtyDatasources.contains(dataSource)) {


Nice catch! This does seem to be an issue that will cause the snapshot and the cache for a datasource to diverge. Normally, the snapshot reconciles itself with what's in the cache in the next sync.
But we are removing that capability in this PR by updating snapshots for modified datasources only.

Record used-segment mutations from writeCacheForDataSource or otherwise compare against the prior snapshot before reusing it.

Comparing against prior snapshot might be too costly, I think we could do the following instead.

ReadWriteCache keeps a volatile boolean hasNewWrites.

Set this flag to true in withWriteLock().

Also, add method ReadWriteCache.markAllWritesAsSynced()

Add new methods to HeapMemoryDatasourceSegmentCache: getAllUsedSegmentsIfCacheHasNewWrites().

Invoke this method in incremental mode only

This method should acquire a write lock, return all the used segments only if there are writes and finally set hasNewWrites to false.

jtuglu1 added this to the 38.0.0 milestone Jul 10, 2026

jtuglu1 added the Performance label Jul 10, 2026

jtuglu1 requested a review from kfaraz July 10, 2026 00:38

perf: speed-up coordinator segment metadata cache delta syncs

984d09e

jtuglu1 force-pushed the speed-up-coordinator-delta-sync branch from 26405b1 to 984d09e Compare July 10, 2026 00:44

FrankChen021 reviewed Jul 10, 2026

View reviewed changes

kfaraz reviewed Jul 10, 2026

View reviewed changes

more changes

2535361

jtuglu1 requested review from FrankChen021 and kfaraz July 10, 2026 17:42

		public static DataSourcesSnapshot withUpdatedDataSources(
		@Nullable DataSourcesSnapshot previous,

	Set<String> dirtyDatasources = null;
	Set<String> modifiedDatasources = null;

		@@ -743,13 +788,17 @@ private void retrieveRequiredUsedSegments(
		* datasource cache is atomic. Also identifies the segment IDs which have been
		* updated in the metadata store and need to be refreshed in the cache.

	final Set<String> dirtyDatasources = new HashSet<>();
	final Set<String> datasourcesToRefresh = new HashSet<>();

Uh oh!

Conversation

jtuglu1 commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Release note

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kfaraz left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kfaraz Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

jtuglu1 commented Jul 10, 2026 •

edited

Loading

kfaraz left a comment •

edited

Loading

kfaraz Jul 10, 2026 •

edited

Loading