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 574e1ec76ac7..7a6b2666caad 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,67 @@ public static DataSourcesSnapshot fromUsedSegments(Iterable segment ); } + /** + * 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. + *

+ * 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 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 DataSourcesSnapshot updateSnapshotForDataSources( + Map> datasourcesToRefresh, + Set removedDataSources, + DateTime snapshotTime + ) + { + final Map properties = Map.of("created", snapshotTime.toString()); + final Map dataSources = new HashMap<>(dataSourcesWithAllUsedSegments); + final Map timelines = new HashMap<>(usedSegmentsTimelinesPerDataSource); + + final Set changedDataSources = new HashSet<>(removedDataSources); + changedDataSources.addAll(datasourcesToRefresh.keySet()); + + removedDataSources.forEach(ds -> { + dataSources.remove(ds); + timelines.remove(ds); + }); + 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 : overshadowedSegments) { + if (!changedDataSources.contains(segment.getDataSource())) { + overshadowed.add(segment); + } + } + for (String ds : datasourcesToRefresh.keySet()) { + final ImmutableDruidDataSource dataSource = dataSources.get(ds); + final SegmentTimeline timeline = timelines.get(ds); + 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 +172,23 @@ private DataSourcesSnapshot( this.overshadowedSegments = ImmutableSet.copyOf(determineOvershadowedSegments()); } + /** + * Constructor used by {@link #updateSnapshotForDataSources} 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..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", @@ -404,6 +399,17 @@ 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). + // 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", "id") + ); } private void createUpgradeSegmentsTable(final String tableName) @@ -663,6 +669,16 @@ 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", "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 @@ -1256,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/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 f2accfe55415..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 @@ -337,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); @@ -570,12 +575,16 @@ private long syncWithMetadataStore() final Map datasourceToSummary = new HashMap<>(); + // Datasources whose used-segment set changed this sync. Null forces a full + // 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); - updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); + datasourcesToRefresh = updateSegmentIdsInCache(datasourceToSummary, syncStartTime.minus(SYNC_BUFFER_DURATION)); retrieveUsedSegmentPayloads(datasourceToSummary); } @@ -591,21 +600,39 @@ private long syncWithMetadataStore() retrieveAndResetUsedIndexingStates(); } - markCacheSynced(syncStartTime); + markCacheSynced(syncStartTime, datasourcesToRefresh); 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 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 datasourcesToRefresh Datasources changed by the metadata-store diff this + * sync, or null to force a full rebuild. */ - private void markCacheSynced(DateTime syncStartTime) + private void markCacheSynced(DateTime syncStartTime, @Nullable Set datasourcesToRefresh) { final Stopwatch updateDuration = Stopwatch.createStarted(); + final DataSourcesSnapshot previousSnapshot = datasourcesSnapshot.get(); + final boolean incremental = datasourcesToRefresh != null && previousSnapshot != null; + final Set cachedDatasources = Set.copyOf(datasourceToSegmentCache.keySet()); + // 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<>(); for (String dataSource : cachedDatasources) { final HeapMemoryDatasourceSegmentCache cache = datasourceToSegmentCache.getOrDefault( @@ -627,19 +654,43 @@ 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())); + 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); + } + } } } - datasourcesSnapshot.set( - DataSourcesSnapshot.fromUsedSegments(datasourceToUsedSegments, syncStartTime) - ); + if (incremental) { + datasourcesSnapshot.set( + previousSnapshot.updateSnapshotForDataSources( + datasourceToUsedSegments, + removedDatasources, + syncStartTime + ) + ); + } else { + datasourcesSnapshot.set( + DataSourcesSnapshot.fromUsedSegments(datasourceToUsedSegments, syncStartTime) + ); + } emitMetric(Metric.UPDATE_SNAPSHOT_DURATION_MILLIS, updateDuration.millisElapsed()); } @@ -742,14 +793,21 @@ 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 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 datasourcesToRefresh = new HashSet<>(); + // Sync segments for datasources which were retrieved in the latest poll datasourceToSummary.forEach((dataSource, summary) -> { final HeapMemoryDatasourceSegmentCache cache = getCacheForDatasource(dataSource); @@ -759,6 +817,9 @@ private void updateSegmentIdsInCache( emitNonZeroMetric(dataSource, Metric.DELETED_SEGMENTS, result.getDeleted()); summary.usedSegmentIdsToRefresh.addAll(result.getExpiredIds()); + if (result.getDeleted() > 0 || !result.getExpiredIds().isEmpty()) { + datasourcesToRefresh.add(dataSource); + } }); // Update cache for datasources which returned no segments in the latest poll @@ -766,10 +827,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) { + datasourcesToRefresh.add(dataSource); + } } }); emitMetric(Metric.UPDATE_IDS_DURATION_MILLIS, updateDuration.millisElapsed()); + 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 new file mode 100644 index 000000000000..7df7148a0d92 --- /dev/null +++ b/server/src/test/java/org/apache/druid/client/DataSourcesSnapshotTest.java @@ -0,0 +1,125 @@ +/* + * 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()); + } + + 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)); + 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 = previous.updateSnapshotForDataSources( + 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 testUpdateSnapshot_equalsFullRebuild() + { + 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 = previous.updateSnapshotForDataSources( + Map.of(DS1, newDs1), + Set.of(), + SNAPSHOT_TIME + ); + + final Map> finalState = new HashMap<>(); + finalState.put(DS1, newDs1); + finalState.put(DS2, baseDs2); + 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 testUpdateSnapshot_removesDatasource() + { + final DataSourcesSnapshot previous = baseSnapshot(); + + final DataSourcesSnapshot updated = previous.updateSnapshotForDataSources( + Map.of(), + Set.of(DS2), + SNAPSHOT_TIME + ); + + Assert.assertNull(updated.getDataSource(DS2)); + Assert.assertNotNull(updated.getDataSource(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..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,7 +349,7 @@ 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" ); @@ -372,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; }); @@ -381,7 +387,7 @@ 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" ); @@ -405,7 +411,7 @@ 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" ); @@ -431,9 +437,9 @@ 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_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/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