From 984d09e37b5d1d45d70d3b6d183cf59227736728 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jesse=20Tu=C4=9Flu?= Date: Thu, 9 Jul 2026 17:18:53 -0700 Subject: [PATCH 1/2] perf: speed-up coordinator segment metadata cache delta syncs --- .../druid/client/DataSourcesSnapshot.java | 90 +++++++++++ .../druid/metadata/SQLMetadataConnector.java | 16 ++ .../SegmentsMetadataManagerConfig.java | 29 +++- .../cache/HeapMemorySegmentMetadataCache.java | 74 +++++++-- .../druid/client/DataSourcesSnapshotTest.java | 143 ++++++++++++++++++ .../metadata/SQLMetadataConnectorTest.java | 12 +- .../SegmentsMetadataManagerConfigTest.java | 50 ++++++ 7 files changed, 400 insertions(+), 14 deletions(-) create mode 100644 server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java create mode 100644 server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java diff --git a/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java b/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java index 574e1ec76ac7..9a0fd0808886 100644 --- a/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java +++ b/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java @@ -35,6 +35,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -91,6 +92,78 @@ public static DataSourcesSnapshot fromUsedSegments(Iterable segment ); } + /** + * Builds a new snapshot from a previous one by recomputing only the datasources + * that changed, reusing the previous snapshot's {@link ImmutableDruidDataSource}, + * timeline, and overshadowed-segment computation for all unchanged datasources. + *

+ * 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 + * snapshot identical to a full {@link #fromUsedSegments} rebuild of the same + * final state. + * + * @param previous Previous snapshot to build upon (null → full build). + * @param changedDataSources Changed datasource → its complete current set of + * used segments. An empty set removes the datasource. + * @param removedDataSources Datasources whose caches are now empty/gone. + * @param snapshotTime Time of this snapshot (poll start time). + */ + public static DataSourcesSnapshot withUpdatedDataSources( + @Nullable DataSourcesSnapshot previous, + Map> changedDataSources, + Set removedDataSources, + DateTime snapshotTime + ) + { + if (previous == null) { + return fromUsedSegments(changedDataSources, snapshotTime); + } + + final Map properties = Map.of("created", snapshotTime.toString()); + final Map dataSources = new HashMap<>(previous.dataSourcesWithAllUsedSegments); + final Map timelines = new HashMap<>(previous.usedSegmentsTimelinesPerDataSource); + + final Set dirtyDataSources = new HashSet<>(removedDataSources); + dirtyDataSources.addAll(changedDataSources.keySet()); + + removedDataSources.forEach(ds -> { + dataSources.remove(ds); + timelines.remove(ds); + }); + changedDataSources.forEach((ds, segments) -> { + if (segments.isEmpty()) { + dataSources.remove(ds); + timelines.remove(ds); + } else { + dataSources.put(ds, new ImmutableDruidDataSource(ds, properties, segments)); + timelines.put(ds, SegmentTimeline.forSegments(segments)); + } + }); + + // Reuse prior overshadowed segments for unchanged datasources; recompute only the changed ones. + final List overshadowed = new ArrayList<>(); + for (DataSegment segment : previous.overshadowedSegments) { + if (!dirtyDataSources.contains(segment.getDataSource())) { + overshadowed.add(segment); + } + } + for (String ds : changedDataSources.keySet()) { + final ImmutableDruidDataSource dataSource = dataSources.get(ds); + final SegmentTimeline timeline = timelines.get(ds); + if (dataSource == null || timeline == null) { + continue; + } + for (DataSegment segment : dataSource.getSegments()) { + if (timeline.isOvershadowed(segment)) { + overshadowed.add(segment); + } + } + } + + return new DataSourcesSnapshot(snapshotTime, dataSources, timelines, overshadowed); + } + private final DateTime snapshotTime; private final Map dataSourcesWithAllUsedSegments; private final Map usedSegmentsTimelinesPerDataSource; @@ -110,6 +183,23 @@ private DataSourcesSnapshot( this.overshadowedSegments = ImmutableSet.copyOf(determineOvershadowedSegments()); } + /** + * Constructor used by {@link #withUpdatedDataSources} where timelines and + * overshadowed segments have already been computed incrementally. + */ + private DataSourcesSnapshot( + DateTime snapshotTime, + Map dataSourcesWithAllUsedSegments, + Map usedSegmentsTimelinesPerDataSource, + Collection overshadowedSegments + ) + { + this.snapshotTime = snapshotTime; + this.dataSourcesWithAllUsedSegments = dataSourcesWithAllUsedSegments; + this.usedSegmentsTimelinesPerDataSource = usedSegmentsTimelinesPerDataSource; + this.overshadowedSegments = ImmutableSet.copyOf(overshadowedSegments); + } + /** * Time when this snapshot was taken. Since polling segments from the database * may be a slow operation, this represents the poll start time. diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java index 74829cddbfbf..befcd43a3006 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java @@ -404,6 +404,15 @@ tableName, getPayloadType(), getQuoteString(), getCollation() "start" ) ); + // Covering index for the used-segment ID scan performed on every metadata + // cache sync (SELECT id, dataSource, used_status_last_updated WHERE used=true). + // id rides along as the implicit primary key, so this makes the scan + // index-only and avoids reading the payload-bearing clustered index. + createIndex( + tableName, + "IDX_%S_USED_USLU_DATASOURCE", + List.of("used", "used_status_last_updated", "dataSource") + ); } private void createUpgradeSegmentsTable(final String tableName) @@ -663,6 +672,13 @@ protected void alterSegmentTable() "IDX_%S_DATASOURCE_UPGRADED_FROM_SEGMENT_ID", List.of("dataSource", "upgraded_from_segment_id") ); + // Migration for existing tables: covering index backing the used-segment ID + // scan on every cache sync (see createSegmentTable). + createIndex( + tableName, + "IDX_%S_USED_USLU_DATASOURCE", + List.of("used", "used_status_last_updated", "dataSource") + ); } @Override diff --git a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java index 8fd23da5e15e..c5b5bcaa71c5 100644 --- a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java +++ b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java @@ -43,16 +43,38 @@ public class SegmentsMetadataManagerConfig @JsonProperty private final UnusedSegmentKillerConfig killUnused; + /** + * When enabled, the incremental cache applies each drift-free full sync + * incrementally: it parses/allocates only the segment rows that changed since + * the last sync and rebuilds only the affected part of the datasource snapshot, + * instead of re-parsing and rebuilding everything. This does not change what is + * read from the metadata store (still the complete used-segment set every poll), + * so it introduces no staleness/drift. Opt-in; default false. + */ + @JsonProperty + private final boolean useIncrementalSync; + + public SegmentsMetadataManagerConfig( + Period pollDuration, + SegmentMetadataCache.UsageMode useIncrementalCache, + UnusedSegmentKillerConfig killUnused + ) + { + this(pollDuration, useIncrementalCache, killUnused, null); + } + @JsonCreator public SegmentsMetadataManagerConfig( @JsonProperty("pollDuration") Period pollDuration, @JsonProperty("useIncrementalCache") SegmentMetadataCache.UsageMode useIncrementalCache, - @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused + @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused, + @JsonProperty("useIncrementalSync") Boolean useIncrementalSync ) { this.pollDuration = Configs.valueOrDefault(pollDuration, Period.minutes(1)); this.useIncrementalCache = Configs.valueOrDefault(useIncrementalCache, SegmentMetadataCache.UsageMode.IF_SYNCED); this.killUnused = Configs.valueOrDefault(killUnused, new UnusedSegmentKillerConfig(null, null, null)); + this.useIncrementalSync = Configs.valueOrDefault(useIncrementalSync, false); if (this.killUnused.isEnabled() && this.useIncrementalCache == SegmentMetadataCache.UsageMode.NEVER) { throw DruidException .forPersona(DruidException.Persona.OPERATOR) @@ -82,4 +104,9 @@ public UnusedSegmentKillerConfig getKillUnused() { return killUnused; } + + public boolean isUseIncrementalSync() + { + return useIncrementalSync; + } } diff --git a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java index f2accfe55415..c8113bb13995 100644 --- a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java +++ b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java @@ -167,6 +167,13 @@ private enum CacheState private final AtomicReference syncFinishTime = new AtomicReference<>(); private final AtomicReference datasourcesSnapshot = new AtomicReference<>(null); + /** + * When true, each sync rebuilds only the changed datasources' portion of the + * {@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; + @Inject public HeapMemorySegmentMetadataCache( ObjectMapper jsonMapper, @@ -182,6 +189,7 @@ public HeapMemorySegmentMetadataCache( this.jsonMapper = jsonMapper; this.cacheMode = config.get().getCacheUsageMode(); this.pollDuration = config.get().getPollDuration().toStandardDuration(); + this.useIncrementalSync = config.get().isUseIncrementalSync(); this.tablesConfig = tablesConfig.get(); this.useSchemaCache = segmentSchemaCache.isEnabled(); this.segmentSchemaCache = segmentSchemaCache; @@ -570,12 +578,16 @@ private long syncWithMetadataStore() final Map datasourceToSummary = new HashMap<>(); + // Datasources whose used-segment set changed this sync. Null forces a full + // snapshot rebuild (first sync, or when incremental sync is disabled). + Set dirtyDatasources = null; + // Fetch all used segments if this is the first sync if (syncFinishTime.get() == null) { retrieveAllUsedSegments(datasourceToSummary); } else { retrieveUsedSegmentIds(datasourceToSummary); - updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); + dirtyDatasources = updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); retrieveUsedSegmentPayloads(datasourceToSummary); } @@ -591,21 +603,38 @@ private long syncWithMetadataStore() retrieveAndResetUsedIndexingStates(); } - markCacheSynced(syncStartTime); + markCacheSynced(syncStartTime, dirtyDatasources); syncFinishTime.set(DateTimes.nowUtc()); return totalSyncDuration.millisElapsed(); } /** - * Marks the cache for all datasources as synced and emit total stats. + * Marks the cache for all datasources as synced and emits stats, then rebuilds + * the {@link DataSourcesSnapshot}. + *

+ * When {@code dirtyDatasources} is non-null, incremental sync is enabled, and a + * previous snapshot exists, only the changed datasources' portion of the + * snapshot is rebuilt (reusing the prior snapshot's timelines/overshadow for the + * rest); this is identical to a full rebuild but O(changed) instead of + * O(all-segments). Otherwise the snapshot is fully rebuilt. + * + * @param dirtyDatasources Datasources whose used set changed this sync, or null + * to force a full rebuild (e.g. first sync). */ - private void markCacheSynced(DateTime syncStartTime) + private void markCacheSynced(DateTime syncStartTime, @Nullable Set dirtyDatasources) { final Stopwatch updateDuration = Stopwatch.createStarted(); + final DataSourcesSnapshot previousSnapshot = datasourcesSnapshot.get(); + final boolean incremental = + useIncrementalSync && dirtyDatasources != null && previousSnapshot != null; + final Set cachedDatasources = Set.copyOf(datasourceToSegmentCache.keySet()); + // In incremental mode this holds only the changed (non-empty) datasources; + // in full mode it holds every non-empty datasource. final Map> datasourceToUsedSegments = new HashMap<>(); + final Set removedDatasources = new HashSet<>(); for (String dataSource : cachedDatasources) { final HeapMemoryDatasourceSegmentCache cache = datasourceToSegmentCache.getOrDefault( @@ -627,19 +656,35 @@ private void markCacheSynced(DateTime syncStartTime) } } ); + removedDatasources.add(dataSource); } else { emitMetric(dataSource, Metric.CACHED_INTERVALS, stats.getNumIntervals()); emitMetric(dataSource, Metric.CACHED_USED_SEGMENTS, stats.getNumUsedSegments()); emitMetric(dataSource, Metric.CACHED_UNUSED_SEGMENTS, stats.getNumUnusedSegments()); emitMetric(dataSource, Metric.CACHED_PENDING_SEGMENTS, stats.getNumPendingSegments()); - 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)) { + datasourceToUsedSegments.put(dataSource, cache.findUsedSegmentsOverlappingAnyOf(List.of())); + } } } - datasourcesSnapshot.set( - DataSourcesSnapshot.fromUsedSegments(datasourceToUsedSegments, syncStartTime) - ); + if (incremental) { + datasourcesSnapshot.set( + DataSourcesSnapshot.withUpdatedDataSources( + previousSnapshot, + datasourceToUsedSegments, + removedDatasources, + syncStartTime + ) + ); + } else { + datasourcesSnapshot.set( + DataSourcesSnapshot.fromUsedSegments(datasourceToUsedSegments, syncStartTime) + ); + } emitMetric(Metric.UPDATE_SNAPSHOT_DURATION_MILLIS, updateDuration.millisElapsed()); } @@ -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. */ - private void updateSegmentIdsInCache( + private Set updateSegmentIdsInCache( Map datasourceToSummary, DateTime syncStartTime ) { final Stopwatch updateDuration = Stopwatch.createStarted(); + // Datasources whose used-segment set actually changed this sync (used to + // rebuild only the affected part of the snapshot). + final Set dirtyDatasources = new HashSet<>(); + // Sync segments for datasources which were retrieved in the latest poll datasourceToSummary.forEach((dataSource, summary) -> { final HeapMemoryDatasourceSegmentCache cache = getCacheForDatasource(dataSource); @@ -759,6 +808,9 @@ private void updateSegmentIdsInCache( emitNonZeroMetric(dataSource, Metric.DELETED_SEGMENTS, result.getDeleted()); summary.usedSegmentIdsToRefresh.addAll(result.getExpiredIds()); + if (result.getDeleted() > 0 || !result.getExpiredIds().isEmpty()) { + dirtyDatasources.add(dataSource); + } }); // Update cache for datasources which returned no segments in the latest poll @@ -766,10 +818,14 @@ private void updateSegmentIdsInCache( if (!datasourceToSummary.containsKey(dataSource)) { final SegmentSyncResult result = cache.syncSegmentIds(List.of(), syncStartTime); emitNonZeroMetric(dataSource, Metric.DELETED_SEGMENTS, result.getDeleted()); + if (result.getDeleted() > 0) { + dirtyDatasources.add(dataSource); + } } }); emitMetric(Metric.UPDATE_IDS_DURATION_MILLIS, updateDuration.millisElapsed()); + return dirtyDatasources; } /** diff --git a/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java b/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java new file mode 100644 index 000000000000..097c1c92ef0f --- /dev/null +++ b/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.client; + +import org.apache.druid.java.util.common.DateTimes; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.server.coordinator.CreateDataSegments; +import org.apache.druid.timeline.DataSegment; +import org.joda.time.DateTime; +import org.junit.Assert; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class DataSourcesSnapshotTest +{ + private static final String DS1 = "ds1"; + private static final String DS2 = "ds2"; + private static final DateTime SNAPSHOT_TIME = DateTimes.of("2024-01-01"); + + private static Set segmentsOf(String dataSource, String start, int numIntervals) + { + return CreateDataSegments.ofDatasource(dataSource) + .forIntervals(numIntervals, Granularities.DAY) + .startingAt(start) + .withNumPartitions(1) + .eachOfSizeInMb(500) + .stream() + .collect(Collectors.toSet()); + } + + private static Set idsOf(DataSourcesSnapshot snapshot, String dataSource) + { + return snapshot.getDataSource(dataSource).getSegments().stream() + .map(s -> s.getId().toString()) + .collect(Collectors.toSet()); + } + + @Test + public void testWithUpdatedDatasources_reusesUnchangedDatasourceByReference() + { + final Map> base = new HashMap<>(); + base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); + base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); + final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + + final Set newDs1 = segmentsOf(DS1, "2024-02-01", 3); + final DataSourcesSnapshot updated = DataSourcesSnapshot.withUpdatedDataSources( + previous, + Map.of(DS1, newDs1), + Set.of(), + SNAPSHOT_TIME + ); + + Assert.assertSame(previous.getDataSource(DS2), updated.getDataSource(DS2)); + Assert.assertNotSame(previous.getDataSource(DS1), updated.getDataSource(DS1)); + Assert.assertEquals( + newDs1.stream().map(s -> s.getId().toString()).collect(Collectors.toSet()), + idsOf(updated, DS1) + ); + } + + @Test + public void testWithUpdatedDatasources_equalsFullRebuild() + { + final Map> base = new HashMap<>(); + base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); + base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); + final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + + final Set newDs1 = segmentsOf(DS1, "2024-02-01", 3); + final DataSourcesSnapshot incremental = DataSourcesSnapshot.withUpdatedDataSources( + previous, + Map.of(DS1, newDs1), + Set.of(), + SNAPSHOT_TIME + ); + + final Map> finalState = new HashMap<>(); + finalState.put(DS1, newDs1); + finalState.put(DS2, base.get(DS2)); + final DataSourcesSnapshot fullRebuild = DataSourcesSnapshot.fromUsedSegments(finalState, SNAPSHOT_TIME); + + Assert.assertEquals(idsOf(fullRebuild, DS1), idsOf(incremental, DS1)); + Assert.assertEquals(idsOf(fullRebuild, DS2), idsOf(incremental, DS2)); + Assert.assertEquals(fullRebuild.getOvershadowedSegments(), incremental.getOvershadowedSegments()); + } + + @Test + public void testWithUpdatedDatasources_removesDatasource() + { + final Map> base = new HashMap<>(); + base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); + base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); + final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + + final DataSourcesSnapshot updated = DataSourcesSnapshot.withUpdatedDataSources( + previous, + Map.of(), + Set.of(DS2), + SNAPSHOT_TIME + ); + + Assert.assertNull(updated.getDataSource(DS2)); + Assert.assertNotNull(updated.getDataSource(DS1)); + } + + @Test + public void testWithUpdatedDatasources_nullPreviousFallsBackToFullBuild() + { + final Set ds1 = segmentsOf(DS1, "2024-01-01", 2); + final DataSourcesSnapshot snapshot = DataSourcesSnapshot.withUpdatedDataSources( + null, + Map.of(DS1, ds1), + Set.of(), + SNAPSHOT_TIME + ); + Assert.assertEquals( + ds1.stream().map(s -> s.getId().toString()).collect(Collectors.toSet()), + idsOf(snapshot, DS1) + ); + } +} diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java index 3fef28a963a6..b6b8f27863b1 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java @@ -348,7 +348,8 @@ public void test_useShortIndexNames_true_tableIndices_areNotAdded_ifExist() final Set expectedIndices = Sets.newHashSet( "IDX_DRUIDTEST_SEGMENTS_USED", "IDX_D011BD6ED76268701273CE512704C5AFA060D672", - "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3" + "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3", + "IDX_139C146E7C6A2C3B15E0EF04B4A3896519466F60" ); assertIndicesPresentOnTable(segmentsTable, expectedIndices); @@ -383,7 +384,8 @@ public void test_useShortIndexNames_false_tableIndices_areNotAdded_ifExist() final Set expectedIndices = Sets.newHashSet( "IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0", "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_USED_END_START", - "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID" + "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID", + "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE" ); assertIndicesPresentOnTable(segmentsTable, expectedIndices); @@ -407,7 +409,8 @@ public void test_useShortIndexNames_true_tableIndices_areAdded_IfNotExist() final Set expectedIndices = Sets.newHashSet( "IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0", "IDX_D011BD6ED76268701273CE512704C5AFA060D672", - "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3" + "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3", + "IDX_139C146E7C6A2C3B15E0EF04B4A3896519466F60" ); connector.createSegmentTable(segmentsTable); @@ -433,7 +436,8 @@ public void test_useShortIndexNames_false_tableIndices_areAdded_IfNotExist() final Set expectedIndices = Sets.newHashSet( "IDX_DRUIDTEST_SEGMENTS_USED", "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_USED_END_START", - "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID" + "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID", + "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE" ); connector.createSegmentTable(segmentsTable); diff --git a/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java b/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java new file mode 100644 index 000000000000..161eab0f7a04 --- /dev/null +++ b/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.metadata; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.druid.metadata.segment.cache.SegmentMetadataCache; +import org.apache.druid.segment.TestHelper; +import org.junit.Assert; +import org.junit.Test; + +public class SegmentsMetadataManagerConfigTest +{ + private final ObjectMapper mapper = TestHelper.makeJsonMapper(); + + @Test + public void testDefaults() throws Exception + { + final SegmentsMetadataManagerConfig config = + mapper.readValue("{}", SegmentsMetadataManagerConfig.class); + + Assert.assertEquals(SegmentMetadataCache.UsageMode.IF_SYNCED, config.getCacheUsageMode()); + // Incremental sync is opt-in and off by default + Assert.assertFalse(config.isUseIncrementalSync()); + } + + @Test + public void testDeserializeUseIncrementalSync() throws Exception + { + final SegmentsMetadataManagerConfig config = + mapper.readValue("{\"useIncrementalSync\": true}", SegmentsMetadataManagerConfig.class); + Assert.assertTrue(config.isUseIncrementalSync()); + } +} From 25353611bdf8f674b25267c225c65875ecf73949 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jesse=20Tu=C4=9Flu?= Date: Fri, 10 Jul 2026 10:42:00 -0700 Subject: [PATCH 2/2] more changes --- .../storage/mysql/MySQLConnector.java | 7 ++ .../druid/client/DataSourcesSnapshot.java | 67 +++++++--------- .../druid/metadata/SQLMetadataConnector.java | 71 ++++++++++++++--- .../SegmentsMetadataManagerConfig.java | 29 +------ .../HeapMemoryDatasourceSegmentCache.java | 50 ++++++++++++ .../cache/HeapMemorySegmentMetadataCache.java | 77 +++++++++++-------- .../druid/client/DataSourcesSnapshotTest.java | 54 +++++-------- .../metadata/SQLMetadataConnectorTest.java | 30 ++++---- .../SegmentsMetadataManagerConfigTest.java | 50 ------------ .../HeapMemorySegmentMetadataCacheTest.java | 30 ++++++++ 10 files changed, 255 insertions(+), 210 deletions(-) delete mode 100644 server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java diff --git a/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/mysql/MySQLConnector.java b/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/mysql/MySQLConnector.java index 8c4f928be76b..c9599eb0fd7a 100644 --- a/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/mysql/MySQLConnector.java +++ b/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/metadata/storage/mysql/MySQLConnector.java @@ -191,6 +191,13 @@ public String limitClause(int limit) return String.format(Locale.ENGLISH, "LIMIT %d", limit); } + @Override + protected String getDropIndexStatement(String indexName, String tableName) + { + // MySQL requires the target table in a DROP INDEX statement. + return StringUtils.format("DROP INDEX %s ON %s", indexName, tableName); + } + @Override public boolean tableExists(Handle handle, String tableName) { diff --git a/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java b/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java index 9a0fd0808886..7a6b2666caad 100644 --- a/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java +++ b/server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java @@ -93,67 +93,56 @@ public static DataSourcesSnapshot fromUsedSegments(Iterable segment } /** - * Builds a new snapshot from a previous one by recomputing only the datasources - * that changed, reusing the previous snapshot's {@link ImmutableDruidDataSource}, - * timeline, and overshadowed-segment computation for all unchanged datasources. + * Builds a new snapshot from this one by recomputing only the datasources that + * changed, reusing this snapshot's {@link ImmutableDruidDataSource}, timeline, + * and overshadowed-segment computation for every unchanged datasource. *

- * 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 - * snapshot identical to a full {@link #fromUsedSegments} rebuild of the same - * final state. + * Note that a changed datasource has its entire timeline rebuilt even if only a + * single one of its segments changed; the saving is that untouched datasources + * are reused as is. Overshadowing is computed per-datasource, so reusing the + * prior overshadowed segments for unchanged datasources is correct. The result + * is identical to a full {@link #fromUsedSegments} rebuild of the same final + * state, at a cost proportional to the changed datasources rather than all of them. * - * @param previous Previous snapshot to build upon (null → full build). - * @param changedDataSources Changed datasource → its complete current set of - * used segments. An empty set removes the datasource. - * @param removedDataSources Datasources whose caches are now empty/gone. - * @param snapshotTime Time of this snapshot (poll start time). + * @param datasourcesToRefresh Datasource → its complete current set of used + * segments, for each datasource that changed. Sets + * must be non-empty; removals are conveyed via + * {@code removedDataSources}. + * @param removedDataSources Datasources whose caches are now empty/gone. + * @param snapshotTime Time of this snapshot (poll start time). */ - public static DataSourcesSnapshot withUpdatedDataSources( - @Nullable DataSourcesSnapshot previous, - Map> changedDataSources, + public DataSourcesSnapshot updateSnapshotForDataSources( + Map> datasourcesToRefresh, Set removedDataSources, DateTime snapshotTime ) { - if (previous == null) { - return fromUsedSegments(changedDataSources, snapshotTime); - } - final Map properties = Map.of("created", snapshotTime.toString()); - final Map dataSources = new HashMap<>(previous.dataSourcesWithAllUsedSegments); - final Map timelines = new HashMap<>(previous.usedSegmentsTimelinesPerDataSource); + final Map dataSources = new HashMap<>(dataSourcesWithAllUsedSegments); + final Map timelines = new HashMap<>(usedSegmentsTimelinesPerDataSource); - final Set dirtyDataSources = new HashSet<>(removedDataSources); - dirtyDataSources.addAll(changedDataSources.keySet()); + final Set changedDataSources = new HashSet<>(removedDataSources); + changedDataSources.addAll(datasourcesToRefresh.keySet()); removedDataSources.forEach(ds -> { dataSources.remove(ds); timelines.remove(ds); }); - changedDataSources.forEach((ds, segments) -> { - if (segments.isEmpty()) { - dataSources.remove(ds); - timelines.remove(ds); - } else { - dataSources.put(ds, new ImmutableDruidDataSource(ds, properties, segments)); - timelines.put(ds, SegmentTimeline.forSegments(segments)); - } + datasourcesToRefresh.forEach((ds, segments) -> { + dataSources.put(ds, new ImmutableDruidDataSource(ds, properties, segments)); + timelines.put(ds, SegmentTimeline.forSegments(segments)); }); // Reuse prior overshadowed segments for unchanged datasources; recompute only the changed ones. final List overshadowed = new ArrayList<>(); - for (DataSegment segment : previous.overshadowedSegments) { - if (!dirtyDataSources.contains(segment.getDataSource())) { + for (DataSegment segment : overshadowedSegments) { + if (!changedDataSources.contains(segment.getDataSource())) { overshadowed.add(segment); } } - for (String ds : changedDataSources.keySet()) { + for (String ds : datasourcesToRefresh.keySet()) { final ImmutableDruidDataSource dataSource = dataSources.get(ds); final SegmentTimeline timeline = timelines.get(ds); - if (dataSource == null || timeline == null) { - continue; - } for (DataSegment segment : dataSource.getSegments()) { if (timeline.isOvershadowed(segment)) { overshadowed.add(segment); @@ -184,7 +173,7 @@ private DataSourcesSnapshot( } /** - * Constructor used by {@link #withUpdatedDataSources} where timelines and + * Constructor used by {@link #updateSnapshotForDataSources} where timelines and * overshadowed segments have already been computed incrementally. */ private DataSourcesSnapshot( diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java index befcd43a3006..2be87f57b013 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataConnector.java @@ -389,11 +389,6 @@ tableName, getPayloadType(), getQuoteString(), getCollation() ) ); - createIndex( - tableName, - "IDX_%S_USED", - List.of("used") - ); createIndex( tableName, "IDX_%S_DATASOURCE_USED_END_START", @@ -406,12 +401,14 @@ tableName, getPayloadType(), getQuoteString(), getCollation() ); // Covering index for the used-segment ID scan performed on every metadata // cache sync (SELECT id, dataSource, used_status_last_updated WHERE used=true). - // id rides along as the implicit primary key, so this makes the scan - // index-only and avoids reading the payload-bearing clustered index. + // Includes id explicitly so the scan is index-only on all backends (not only + // engines like InnoDB that append the primary key to secondary indexes). + // Its leading 'used' column also serves plain 'WHERE used = ?' lookups, so a + // separate IDX_%S_USED index is not created. createIndex( tableName, "IDX_%S_USED_USLU_DATASOURCE", - List.of("used", "used_status_last_updated", "dataSource") + List.of("used", "used_status_last_updated", "dataSource", "id") ); } @@ -677,8 +674,11 @@ protected void alterSegmentTable() createIndex( tableName, "IDX_%S_USED_USLU_DATASOURCE", - List.of("used", "used_status_last_updated", "dataSource") + List.of("used", "used_status_last_updated", "dataSource", "id") ); + // The covering index above leads with 'used', so it supersedes the single-column + // IDX_%S_USED. Drop the now-redundant index on existing tables. + dropIndex(tableName, "IDX_%S_USED", List.of("used")); } @Override @@ -1272,6 +1272,59 @@ public void createIndex( } } + /** + * Drops an index on {@code tableName} if it exists, under either the full or + * short naming convention (see {@link #createIndex}). No-op if absent, so it is + * safe to call on every startup. + * + * @param tableName Name of the table the index is on + * @param fullIndexNameFormat Same format string that was passed to {@link #createIndex} + * @param indexCols Same columns that were passed to {@link #createIndex} + */ + public void dropIndex( + final String tableName, + final String fullIndexNameFormat, + final List indexCols + ) + { + final Set createdIndexSet = getIndexOnTable(tableName); + final String shortIndexName = generateShortIndexName(tableName, indexCols); + final String fullIndexName = StringUtils.toUpperCase(StringUtils.format(fullIndexNameFormat, tableName)); + + final String indexName; + if (createdIndexSet.contains(fullIndexName)) { + indexName = fullIndexName; + } else if (createdIndexSet.contains(shortIndexName)) { + indexName = shortIndexName; + } else { + log.info("Index[%s] on table[%s] does not exist, skipping drop.", fullIndexName, tableName); + return; + } + + try { + retryWithHandle( + (HandleCallback) handle -> { + final String dropSQL = getDropIndexStatement(indexName, tableName); + log.info("Dropping index[%s] on table[%s] using SQL[%s].", indexName, tableName, dropSQL); + handle.execute(dropSQL); + return null; + } + ); + } + catch (Exception e) { + log.warn(e, "Could not drop index[%s] on table[%s]", indexName, tableName); + } + } + + /** + * SQL to drop an index. Standard SQL / Derby / PostgreSQL use {@code DROP INDEX }; + * MySQL requires the {@code ON } clause (see {@code MySQLConnector}). + */ + protected String getDropIndexStatement(String indexName, String tableName) + { + return StringUtils.format("DROP INDEX %s", indexName); + } + /** * Checks table metadata to determine if the given column exists in the table. * diff --git a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java index c5b5bcaa71c5..8fd23da5e15e 100644 --- a/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java +++ b/server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java @@ -43,38 +43,16 @@ public class SegmentsMetadataManagerConfig @JsonProperty private final UnusedSegmentKillerConfig killUnused; - /** - * When enabled, the incremental cache applies each drift-free full sync - * incrementally: it parses/allocates only the segment rows that changed since - * the last sync and rebuilds only the affected part of the datasource snapshot, - * instead of re-parsing and rebuilding everything. This does not change what is - * read from the metadata store (still the complete used-segment set every poll), - * so it introduces no staleness/drift. Opt-in; default false. - */ - @JsonProperty - private final boolean useIncrementalSync; - - public SegmentsMetadataManagerConfig( - Period pollDuration, - SegmentMetadataCache.UsageMode useIncrementalCache, - UnusedSegmentKillerConfig killUnused - ) - { - this(pollDuration, useIncrementalCache, killUnused, null); - } - @JsonCreator public SegmentsMetadataManagerConfig( @JsonProperty("pollDuration") Period pollDuration, @JsonProperty("useIncrementalCache") SegmentMetadataCache.UsageMode useIncrementalCache, - @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused, - @JsonProperty("useIncrementalSync") Boolean useIncrementalSync + @JsonProperty("killUnused") UnusedSegmentKillerConfig killUnused ) { this.pollDuration = Configs.valueOrDefault(pollDuration, Period.minutes(1)); this.useIncrementalCache = Configs.valueOrDefault(useIncrementalCache, SegmentMetadataCache.UsageMode.IF_SYNCED); this.killUnused = Configs.valueOrDefault(killUnused, new UnusedSegmentKillerConfig(null, null, null)); - this.useIncrementalSync = Configs.valueOrDefault(useIncrementalSync, false); if (this.killUnused.isEnabled() && this.useIncrementalCache == SegmentMetadataCache.UsageMode.NEVER) { throw DruidException .forPersona(DruidException.Persona.OPERATOR) @@ -104,9 +82,4 @@ public UnusedSegmentKillerConfig getKillUnused() { return killUnused; } - - public boolean isUseIncrementalSync() - { - return useIncrementalSync; - } } diff --git a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemoryDatasourceSegmentCache.java b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemoryDatasourceSegmentCache.java index 3a091a6c0b47..1b697d1f06fc 100644 --- a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemoryDatasourceSegmentCache.java +++ b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemoryDatasourceSegmentCache.java @@ -38,6 +38,7 @@ import java.util.Objects; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Predicate; import java.util.stream.Collectors; @@ -68,6 +69,15 @@ class HeapMemoryDatasourceSegmentCache extends ReadWriteCache implements AutoClo */ private final AtomicInteger references = new AtomicInteger(0); + /** + * Set when a write-through transaction mutates this cache directly (via + * {@link HeapMemorySegmentMetadataCache#writeCacheForDataSource}), so that the + * incremental snapshot rebuild in the next sync refreshes this datasource even + * though the metadata store and cache already agree (the DB diff would report + * no change). Cleared while the snapshot is being rebuilt. + */ + private final AtomicBoolean hasNewWrites = new AtomicBoolean(false); + HeapMemoryDatasourceSegmentCache(String dataSource) { super(true); @@ -116,6 +126,46 @@ boolean isBeingUsedByTransaction() return references.get() > 0; } + /** + * Marks that a write-through transaction has mutated this cache directly, so + * the next snapshot rebuild must refresh this datasource. Called under the + * write lock held by {@link HeapMemorySegmentMetadataCache#writeCacheForDataSource}. + */ + void markHasNewWrites() + { + hasNewWrites.set(true); + } + + /** + * If a write-through mutation occurred since the last snapshot rebuild, returns + * all used segments and clears the flag (atomically under the write lock); + * otherwise returns null. Used to catch datasources whose cache diverged from + * the last published snapshot without a corresponding metadata-store diff. + */ + @Nullable + Set getUsedSegmentsIfHasNewWrites() + { + return withWriteLock(() -> { + if (!hasNewWrites.getAndSet(false)) { + return null; + } + return findUsedSegmentsOverlappingAnyOf(List.of()); + }); + } + + /** + * Returns all used segments and clears the new-writes flag, atomically under + * the write lock. Used for datasources already being refreshed from the + * metadata-store diff, so a concurrent write-through is not lost. + */ + Set getUsedSegmentsAndClearNewWrites() + { + return withWriteLock(() -> { + hasNewWrites.set(false); + return findUsedSegmentsOverlappingAnyOf(List.of()); + }); + } + /** * Checks if a record in the cache needs to be updated. * diff --git a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java index c8113bb13995..6a341c45c96e 100644 --- a/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java +++ b/server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java @@ -167,13 +167,6 @@ private enum CacheState private final AtomicReference syncFinishTime = new AtomicReference<>(); private final AtomicReference datasourcesSnapshot = new AtomicReference<>(null); - /** - * When true, each sync rebuilds only the changed datasources' portion of the - * {@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; - @Inject public HeapMemorySegmentMetadataCache( ObjectMapper jsonMapper, @@ -189,7 +182,6 @@ public HeapMemorySegmentMetadataCache( this.jsonMapper = jsonMapper; this.cacheMode = config.get().getCacheUsageMode(); this.pollDuration = config.get().getPollDuration().toStandardDuration(); - this.useIncrementalSync = config.get().isUseIncrementalSync(); this.tablesConfig = tablesConfig.get(); this.useSchemaCache = segmentSchemaCache.isEnabled(); this.segmentSchemaCache = segmentSchemaCache; @@ -345,7 +337,12 @@ public T writeCacheForDataSource(String dataSource, Action writeAction) return datasourceCache.withWriteLock( () -> { try { - return writeAction.perform(datasourceCache); + final T result = writeAction.perform(datasourceCache); + // A write-through transaction may have mutated the cache directly. + // Flag it so the next sync refreshes this datasource's snapshot even + // if the metadata-store diff shows no change. + datasourceCache.markHasNewWrites(); + return result; } catch (Exception e) { Throwables.throwIfUnchecked(e); @@ -579,15 +576,15 @@ private long syncWithMetadataStore() final Map datasourceToSummary = new HashMap<>(); // Datasources whose used-segment set changed this sync. Null forces a full - // snapshot rebuild (first sync, or when incremental sync is disabled). - Set dirtyDatasources = null; + // snapshot rebuild (the first sync, which has no previous snapshot to reuse). + Set datasourcesToRefresh = null; // Fetch all used segments if this is the first sync if (syncFinishTime.get() == null) { retrieveAllUsedSegments(datasourceToSummary); } else { retrieveUsedSegmentIds(datasourceToSummary); - dirtyDatasources = updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); + datasourcesToRefresh = updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); retrieveUsedSegmentPayloads(datasourceToSummary); } @@ -603,7 +600,7 @@ private long syncWithMetadataStore() retrieveAndResetUsedIndexingStates(); } - markCacheSynced(syncStartTime, dirtyDatasources); + markCacheSynced(syncStartTime, datasourcesToRefresh); syncFinishTime.set(DateTimes.nowUtc()); return totalSyncDuration.millisElapsed(); @@ -613,26 +610,27 @@ private long syncWithMetadataStore() * Marks the cache for all datasources as synced and emits stats, then rebuilds * the {@link DataSourcesSnapshot}. *

- * When {@code dirtyDatasources} is non-null, incremental sync is enabled, and a - * previous snapshot exists, only the changed datasources' portion of the - * snapshot is rebuilt (reusing the prior snapshot's timelines/overshadow for the - * rest); this is identical to a full rebuild but O(changed) instead of - * O(all-segments). Otherwise the snapshot is fully rebuilt. + * When {@code datasourcesToRefresh} is non-null and a previous snapshot exists, + * only the changed datasources are rebuilt into the snapshot (reusing the prior + * snapshot's per-datasource timelines/overshadow for the rest); the result is + * identical to a full rebuild. A datasource is rebuilt if the metadata-store + * diff changed it ({@code datasourcesToRefresh}) or a write-through transaction + * mutated its cache since the last sync ({@code hasNewWrites}). Otherwise the + * whole snapshot is rebuilt (e.g. first sync, when there is no previous snapshot). * - * @param dirtyDatasources Datasources whose used set changed this sync, or null - * to force a full rebuild (e.g. first sync). + * @param datasourcesToRefresh Datasources changed by the metadata-store diff this + * sync, or null to force a full rebuild. */ - private void markCacheSynced(DateTime syncStartTime, @Nullable Set dirtyDatasources) + private void markCacheSynced(DateTime syncStartTime, @Nullable Set datasourcesToRefresh) { final Stopwatch updateDuration = Stopwatch.createStarted(); final DataSourcesSnapshot previousSnapshot = datasourcesSnapshot.get(); - final boolean incremental = - useIncrementalSync && dirtyDatasources != null && previousSnapshot != null; + final boolean incremental = datasourcesToRefresh != null && previousSnapshot != null; final Set cachedDatasources = Set.copyOf(datasourceToSegmentCache.keySet()); - // In incremental mode this holds only the changed (non-empty) datasources; - // in full mode it holds every non-empty datasource. + // In incremental mode this holds only the datasources being rebuilt; in full + // mode it holds every non-empty datasource. final Map> datasourceToUsedSegments = new HashMap<>(); final Set removedDatasources = new HashSet<>(); @@ -663,18 +661,26 @@ private void markCacheSynced(DateTime syncStartTime, @Nullable Set dirty emitMetric(dataSource, Metric.CACHED_UNUSED_SEGMENTS, stats.getNumUnusedSegments()); emitMetric(dataSource, Metric.CACHED_PENDING_SEGMENTS, stats.getNumPendingSegments()); - // Only materialize the (potentially large) used-segment set for datasources - // that must be rebuilt into the snapshot. - if (!incremental || dirtyDatasources.contains(dataSource)) { + if (!incremental) { datasourceToUsedSegments.put(dataSource, cache.findUsedSegmentsOverlappingAnyOf(List.of())); + } else if (datasourcesToRefresh.contains(dataSource)) { + // Changed by the metadata-store diff; materialize and clear the + // write-through flag atomically so a concurrent write is not lost. + datasourceToUsedSegments.put(dataSource, cache.getUsedSegmentsAndClearNewWrites()); + } else { + // Not changed in the store, but a write-through transaction may have + // mutated the cache directly since the last snapshot; catch that here. + final Set segmentsFromWrites = cache.getUsedSegmentsIfHasNewWrites(); + if (segmentsFromWrites != null) { + datasourceToUsedSegments.put(dataSource, segmentsFromWrites); + } } } } if (incremental) { datasourcesSnapshot.set( - DataSourcesSnapshot.withUpdatedDataSources( - previousSnapshot, + previousSnapshot.updateSnapshotForDataSources( datasourceToUsedSegments, removedDatasources, syncStartTime @@ -787,6 +793,9 @@ private void retrieveRequiredUsedSegments( * metadata store in {@link #retrieveUsedSegmentIds}. The update done on each * 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. + * + * @return Set of datasource names whose used segment set has changed in this + * sync and need to be refreshed in the snapshot. */ private Set updateSegmentIdsInCache( Map datasourceToSummary, @@ -797,7 +806,7 @@ private Set updateSegmentIdsInCache( // Datasources whose used-segment set actually changed this sync (used to // rebuild only the affected part of the snapshot). - final Set dirtyDatasources = new HashSet<>(); + final Set datasourcesToRefresh = new HashSet<>(); // Sync segments for datasources which were retrieved in the latest poll datasourceToSummary.forEach((dataSource, summary) -> { @@ -809,7 +818,7 @@ private Set updateSegmentIdsInCache( summary.usedSegmentIdsToRefresh.addAll(result.getExpiredIds()); if (result.getDeleted() > 0 || !result.getExpiredIds().isEmpty()) { - dirtyDatasources.add(dataSource); + datasourcesToRefresh.add(dataSource); } }); @@ -819,13 +828,13 @@ private Set updateSegmentIdsInCache( final SegmentSyncResult result = cache.syncSegmentIds(List.of(), syncStartTime); emitNonZeroMetric(dataSource, Metric.DELETED_SEGMENTS, result.getDeleted()); if (result.getDeleted() > 0) { - dirtyDatasources.add(dataSource); + datasourcesToRefresh.add(dataSource); } } }); emitMetric(Metric.UPDATE_IDS_DURATION_MILLIS, updateDuration.millisElapsed()); - return dirtyDatasources; + return datasourcesToRefresh; } /** diff --git a/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java b/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java index 097c1c92ef0f..7df7148a0d92 100644 --- a/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java +++ b/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java @@ -56,17 +56,21 @@ private static Set idsOf(DataSourcesSnapshot snapshot, String dataSource .collect(Collectors.toSet()); } - @Test - public void testWithUpdatedDatasources_reusesUnchangedDatasourceByReference() + private static DataSourcesSnapshot baseSnapshot() { final Map> base = new HashMap<>(); base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); - final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + return DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + } + + @Test + public void testUpdateSnapshot_reusesUnchangedDatasourceByReference() + { + final DataSourcesSnapshot previous = baseSnapshot(); final Set newDs1 = segmentsOf(DS1, "2024-02-01", 3); - final DataSourcesSnapshot updated = DataSourcesSnapshot.withUpdatedDataSources( - previous, + final DataSourcesSnapshot updated = previous.updateSnapshotForDataSources( Map.of(DS1, newDs1), Set.of(), SNAPSHOT_TIME @@ -81,16 +85,14 @@ public void testWithUpdatedDatasources_reusesUnchangedDatasourceByReference() } @Test - public void testWithUpdatedDatasources_equalsFullRebuild() + public void testUpdateSnapshot_equalsFullRebuild() { - final Map> base = new HashMap<>(); - base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); - base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); - final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + final DataSourcesSnapshot previous = baseSnapshot(); + final Set baseDs2 = previous.getDataSource(DS2).getSegments() + .stream().collect(Collectors.toSet()); final Set newDs1 = segmentsOf(DS1, "2024-02-01", 3); - final DataSourcesSnapshot incremental = DataSourcesSnapshot.withUpdatedDataSources( - previous, + final DataSourcesSnapshot incremental = previous.updateSnapshotForDataSources( Map.of(DS1, newDs1), Set.of(), SNAPSHOT_TIME @@ -98,7 +100,7 @@ public void testWithUpdatedDatasources_equalsFullRebuild() final Map> finalState = new HashMap<>(); finalState.put(DS1, newDs1); - finalState.put(DS2, base.get(DS2)); + finalState.put(DS2, baseDs2); final DataSourcesSnapshot fullRebuild = DataSourcesSnapshot.fromUsedSegments(finalState, SNAPSHOT_TIME); Assert.assertEquals(idsOf(fullRebuild, DS1), idsOf(incremental, DS1)); @@ -107,15 +109,11 @@ public void testWithUpdatedDatasources_equalsFullRebuild() } @Test - public void testWithUpdatedDatasources_removesDatasource() + public void testUpdateSnapshot_removesDatasource() { - final Map> base = new HashMap<>(); - base.put(DS1, segmentsOf(DS1, "2024-01-01", 2)); - base.put(DS2, segmentsOf(DS2, "2024-01-01", 2)); - final DataSourcesSnapshot previous = DataSourcesSnapshot.fromUsedSegments(base, SNAPSHOT_TIME); + final DataSourcesSnapshot previous = baseSnapshot(); - final DataSourcesSnapshot updated = DataSourcesSnapshot.withUpdatedDataSources( - previous, + final DataSourcesSnapshot updated = previous.updateSnapshotForDataSources( Map.of(), Set.of(DS2), SNAPSHOT_TIME @@ -124,20 +122,4 @@ public void testWithUpdatedDatasources_removesDatasource() Assert.assertNull(updated.getDataSource(DS2)); Assert.assertNotNull(updated.getDataSource(DS1)); } - - @Test - public void testWithUpdatedDatasources_nullPreviousFallsBackToFullBuild() - { - final Set ds1 = segmentsOf(DS1, "2024-01-01", 2); - final DataSourcesSnapshot snapshot = DataSourcesSnapshot.withUpdatedDataSources( - null, - Map.of(DS1, ds1), - Set.of(), - SNAPSHOT_TIME - ); - Assert.assertEquals( - ds1.stream().map(s -> s.getId().toString()).collect(Collectors.toSet()), - idsOf(snapshot, DS1) - ); - } } diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java index b6b8f27863b1..51bd3dd8390b 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataConnectorTest.java @@ -337,8 +337,11 @@ public void test_useShortIndexNames_true_tableIndices_areNotAdded_ifExist() connector.createSegmentTable(segmentsTable); connector.alterSegmentTable(); connector.getDBI().withHandle(handle -> { - handle.execute("DROP INDEX IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0"); - handle.execute("CREATE INDEX IDX_DRUIDTEST_SEGMENTS_USED ON druidTest_segments(used)"); + handle.execute("DROP INDEX IDX_93A18EE829B37C5F38FC6DAFB070261D88503835"); + handle.execute( + "CREATE INDEX IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE" + + " ON druidTest_segments(used,used_status_last_updated,dataSource,id)" + ); return null; }); @@ -346,10 +349,9 @@ public void test_useShortIndexNames_true_tableIndices_areNotAdded_ifExist() connector.alterSegmentTable(); final Set expectedIndices = Sets.newHashSet( - "IDX_DRUIDTEST_SEGMENTS_USED", + "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE", "IDX_D011BD6ED76268701273CE512704C5AFA060D672", - "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3", - "IDX_139C146E7C6A2C3B15E0EF04B4A3896519466F60" + "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3" ); assertIndicesPresentOnTable(segmentsTable, expectedIndices); @@ -373,8 +375,11 @@ public void test_useShortIndexNames_false_tableIndices_areNotAdded_ifExist() connector.createSegmentTable(segmentsTable); connector.alterSegmentTable(); connector.getDBI().withHandle(handle -> { - handle.execute("DROP INDEX IDX_DRUIDTEST_SEGMENTS_USED"); - handle.execute("CREATE INDEX IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0 ON druidTest_segments(used)"); + handle.execute("DROP INDEX IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE"); + handle.execute( + "CREATE INDEX IDX_93A18EE829B37C5F38FC6DAFB070261D88503835" + + " ON druidTest_segments(used,used_status_last_updated,dataSource,id)" + ); return null; }); @@ -382,10 +387,9 @@ public void test_useShortIndexNames_false_tableIndices_areNotAdded_ifExist() connector.alterSegmentTable(); final Set expectedIndices = Sets.newHashSet( - "IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0", + "IDX_93A18EE829B37C5F38FC6DAFB070261D88503835", "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_USED_END_START", - "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID", - "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE" + "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID" ); assertIndicesPresentOnTable(segmentsTable, expectedIndices); @@ -407,10 +411,9 @@ public void test_useShortIndexNames_true_tableIndices_areAdded_IfNotExist() final String segmentsTable = tablesConfig.getSegmentsTable(); final Set expectedIndices = Sets.newHashSet( - "IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0", + "IDX_93A18EE829B37C5F38FC6DAFB070261D88503835", "IDX_D011BD6ED76268701273CE512704C5AFA060D672", - "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3", - "IDX_139C146E7C6A2C3B15E0EF04B4A3896519466F60" + "IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3" ); connector.createSegmentTable(segmentsTable); @@ -434,7 +437,6 @@ public void test_useShortIndexNames_false_tableIndices_areAdded_IfNotExist() final String segmentsTable = tablesConfig.getSegmentsTable(); final Set expectedIndices = Sets.newHashSet( - "IDX_DRUIDTEST_SEGMENTS_USED", "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_USED_END_START", "IDX_DRUIDTEST_SEGMENTS_DATASOURCE_UPGRADED_FROM_SEGMENT_ID", "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE" diff --git a/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java b/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java deleted file mode 100644 index 161eab0f7a04..000000000000 --- a/server/src/test/java/org/apache/druid/metadata/SegmentsMetadataManagerConfigTest.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.druid.metadata; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.druid.metadata.segment.cache.SegmentMetadataCache; -import org.apache.druid.segment.TestHelper; -import org.junit.Assert; -import org.junit.Test; - -public class SegmentsMetadataManagerConfigTest -{ - private final ObjectMapper mapper = TestHelper.makeJsonMapper(); - - @Test - public void testDefaults() throws Exception - { - final SegmentsMetadataManagerConfig config = - mapper.readValue("{}", SegmentsMetadataManagerConfig.class); - - Assert.assertEquals(SegmentMetadataCache.UsageMode.IF_SYNCED, config.getCacheUsageMode()); - // Incremental sync is opt-in and off by default - Assert.assertFalse(config.isUseIncrementalSync()); - } - - @Test - public void testDeserializeUseIncrementalSync() throws Exception - { - final SegmentsMetadataManagerConfig config = - mapper.readValue("{\"useIncrementalSync\": true}", SegmentsMetadataManagerConfig.class); - Assert.assertTrue(config.isUseIncrementalSync()); - } -} diff --git a/server/src/test/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCacheTest.java b/server/src/test/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCacheTest.java index e739744480eb..95a8a344bc52 100644 --- a/server/src/test/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCacheTest.java +++ b/server/src/test/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCacheTest.java @@ -851,6 +851,36 @@ private static String getSchemaFingerprint(SchemaPayload payload) ); } + @Test + public void testSync_reflectsWriteThroughInsertInSnapshot() + { + setupAndSyncCache(); + + final DataSegmentPlus segment = + CreateDataSegments.ofDatasource(TestDataSource.WIKI).updatedNow().markUsed().asPlus(); + + // Simulate a write-through transaction: the segment is written to BOTH the + // metadata store and the datasource cache, so the next sync's store diff finds + // nothing to refresh for this datasource (DB and cache already agree). + insertSegmentsInMetadataStore(Set.of(segment)); + cache.writeCacheForDataSource( + TestDataSource.WIKI, + wikiCache -> wikiCache.insertSegments(Set.of(segment)) + ); + + syncCache(); + + // The published snapshot must still include the write-through segment, even + // though the incremental rebuild saw no metadata-store change for WIKI. + Assert.assertNotNull(cache.getDataSourcesSnapshot().getDataSource(TestDataSource.WIKI)); + Assert.assertTrue( + cache.getDataSourcesSnapshot() + .getDataSource(TestDataSource.WIKI) + .getSegments() + .contains(segment.getDataSegment()) + ); + } + private void insertSegmentsInMetadataStore(Set segments) { IndexerSqlMetadataStorageCoordinatorTestBase