This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new bb9924e59fd perf: speed-up coordinator segment metadata cache delta
syncs (#19672)
bb9924e59fd is described below
commit bb9924e59fd477cb1cd21712e9339a3b1029d95c
Author: jtuglu1 <[email protected]>
AuthorDate: Tue Jul 14 00:49:52 2026 -0700
perf: speed-up coordinator segment metadata cache delta syncs (#19672)
Changes:
- Update `DataSourcesSnapshot` for only the datasources that have been
updated
- Maintain a `hasNewWrites` flag in `HeapMemoryDatasourceSegmentCache` to
track
new writes since the last snapshot creation
- Add new index on `druid_segments` metadata table to speed up cache syncs
- Add docs
---
docs/operations/metrics.md | 11 ++
.../storage/sqlserver/SQLServerConnector.java | 7 ++
.../metadata/storage/mysql/MySQLConnector.java | 7 ++
.../apache/druid/client/DataSourcesSnapshot.java | 79 +++++++++++++
.../druid/metadata/SQLMetadataConnector.java | 79 ++++++++++++-
.../cache/HeapMemoryDatasourceSegmentCache.java | 50 +++++++++
.../cache/HeapMemorySegmentMetadataCache.java | 84 ++++++++++++--
.../druid/client/DataSourcesSnapshotTest.java | 125 +++++++++++++++++++++
.../druid/metadata/SQLMetadataConnectorTest.java | 24 ++--
.../cache/HeapMemorySegmentMetadataCacheTest.java | 30 +++++
10 files changed, 473 insertions(+), 23 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 81f8f21b917..ece838c8a93 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -383,11 +383,19 @@ The following metrics are emitted only when [segment
metadata caching](../config
|`segment/pending/count`|Number of pending segments currently present in the
metadata store.|`dataSource`|
|`segment/metadataCache/interval/count`|Total number of intervals present in
the cache for a single datasource.|`dataSource`|
|`segment/metadataCache/used/count`|Total number of used segments present in
the cache for a single datasource.|`dataSource`|
+|`segment/metadataCache/unused/count`|Total number of unused segments present
in the cache for a single datasource.|`dataSource`|
|`segment/metadataCache/pending/count`|Total number of pending segments
present in the cache for a single datasource.|`dataSource`|
|`segment/metadataCache/transactions/readOnly`|Number of read-only
transactions performed on the cache for a single datasource.|`dataSource`|
|`segment/metadataCache/transactions/readWrite`|Number of read-write
transactions performed on the cache for a single datasource.|`dataSource`|
|`segment/metadataCache/transactions/writeOnly`|Number of write-only
transactions performed on the cache for a single datasource. These transactions
happen only if the cache is operating in mode `ifSynced` and the first sync on
the leader Overlord is not complete yet.|`dataSource`|
|`segment/metadataCache/sync/time`|Number of milliseconds taken for the cache
to sync with the metadata store.||
+|`segment/metadataCache/fetchIds/time`|Time taken in a sync to fetch the IDs
of used segments from the metadata store.||
+|`segment/metadataCache/fetchPayloads/time`|Time taken in a sync to fetch the
payloads of used segments that have changed since the last sync.||
+|`segment/metadataCache/fetchPending/time`|Time taken in a sync to fetch
pending segments from the metadata store.||
+|`segment/metadataCache/fetchSchemas/time`|Time taken in a sync to fetch
segment schemas from the metadata store. Emitted only when centralized
datasource schema is enabled.||
+|`segment/metadataCache/fetchIndexingStates/time`|Time taken in a sync to
fetch segment indexing states from the metadata store.||
+|`segment/metadataCache/updateIds/time`|Time taken in a sync to reconcile the
cached segment IDs of each datasource with the metadata store.||
+|`segment/metadataCache/updateSnapshot/time`|Time taken in a sync to rebuild
the datasource snapshot from the cache.||
|`segment/metadataCache/dataSource/deleted`|Indicates that a datasource has no
used or pending segments anymore and has been removed from the
cache.|`dataSource`|
|`segment/metadataCache/deleted`|Total number of segments deleted from the
cache during the latest sync.||
|`segment/metadataCache/skipped`|Total number of unparseable segment records
that were skipped in the latest sync.||
@@ -396,6 +404,9 @@ The following metrics are emitted only when [segment
metadata caching](../config
|`segment/metadataCache/pending/deleted`|Number of pending segments deleted
from the cache during the latest sync.|`dataSource`|
|`segment/metadataCache/pending/updated`|Number of pending segments updated in
the cache during the latest sync.|`dataSource`|
|`segment/metadataCache/pending/skipped`|Number of unparseable pending segment
records that were skipped in the latest sync.|`dataSource`|
+|`segment/metadataCache/schema/skipped`|Number of unparseable segment schema
records that were skipped in the latest sync. Emitted only when centralized
datasource schema is enabled.||
+|`segment/metadataCache/indexingState/added`|Number of segment indexing states
added to the cache during the latest sync.||
+|`segment/metadataCache/indexingState/deleted`|Number of segment indexing
states deleted from the cache during the latest sync.||
### Auto-kill unused segments
diff --git
a/extensions-contrib/sqlserver-metadata-storage/src/main/java/org/apache/druid/metadata/storage/sqlserver/SQLServerConnector.java
b/extensions-contrib/sqlserver-metadata-storage/src/main/java/org/apache/druid/metadata/storage/sqlserver/SQLServerConnector.java
index be7147cc5d2..ad76cf8a0b5 100644
---
a/extensions-contrib/sqlserver-metadata-storage/src/main/java/org/apache/druid/metadata/storage/sqlserver/SQLServerConnector.java
+++
b/extensions-contrib/sqlserver-metadata-storage/src/main/java/org/apache/druid/metadata/storage/sqlserver/SQLServerConnector.java
@@ -224,6 +224,13 @@ public class SQLServerConnector extends
SQLMetadataConnector
return String.format(Locale.ENGLISH, "FETCH NEXT %d ROWS ONLY", limit);
}
+ @Override
+ protected String getDropIndexStatement(String indexName, String tableName)
+ {
+ // SQL Server requires the target table in a DROP INDEX statement.
+ return StringUtils.format("DROP INDEX %s ON %s", indexName, tableName);
+ }
+
/**
*
* {@inheritDoc}
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 8c4f928be76..c9599eb0fd7 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 class MySQLConnector extends SQLMetadataConnector
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 574e1ec76ac..7a6b2666caa 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 javax.annotation.Nullable;
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 class DataSourcesSnapshot
);
}
+ /**
+ * 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.
+ * <p>
+ * 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<String, Set<DataSegment>> datasourcesToRefresh,
+ Set<String> removedDataSources,
+ DateTime snapshotTime
+ )
+ {
+ final Map<String, String> properties = Map.of("created",
snapshotTime.toString());
+ final Map<String, ImmutableDruidDataSource> dataSources = new
HashMap<>(dataSourcesWithAllUsedSegments);
+ final Map<String, SegmentTimeline> timelines = new
HashMap<>(usedSegmentsTimelinesPerDataSource);
+
+ final Set<String> 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<DataSegment> 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<String, ImmutableDruidDataSource>
dataSourcesWithAllUsedSegments;
private final Map<String, SegmentTimeline>
usedSegmentsTimelinesPerDataSource;
@@ -110,6 +172,23 @@ public class 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<String, ImmutableDruidDataSource> dataSourcesWithAllUsedSegments,
+ Map<String, SegmentTimeline> usedSegmentsTimelinesPerDataSource,
+ Collection<DataSegment> 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 74829cddbfb..2be87f57b01 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 @@ public abstract class SQLMetadataConnector implements
MetadataStorageConnector
)
);
- createIndex(
- tableName,
- "IDX_%S_USED",
- List.of("used")
- );
createIndex(
tableName,
"IDX_%S_DATASOURCE_USED_END_START",
@@ -404,6 +399,17 @@ public abstract class SQLMetadataConnector implements
MetadataStorageConnector
"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 @@ public abstract class SQLMetadataConnector implements
MetadataStorageConnector
"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 abstract class SQLMetadataConnector implements
MetadataStorageConnector
}
}
+ /**
+ * 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<String> indexCols
+ )
+ {
+ final Set<String> 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<Void>) 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 <name>};
+ * MySQL requires the {@code ON <table>} 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 3a091a6c0b4..1b697d1f06f 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.Map;
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 @@ class HeapMemoryDatasourceSegmentCache extends
ReadWriteCache implements AutoClo
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<DataSegment> 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<DataSegment> 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 f2accfe5541..d9d2cfce159 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
@@ -336,6 +336,10 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
try (final HeapMemoryDatasourceSegmentCache datasourceCache =
getCacheWithReference(dataSource)) {
return datasourceCache.withWriteLock(
() -> {
+ // Flag the write-through before performing it, so that even a
write
+ // which throws after partially mutating the cache still causes
the next sync
+ // to refresh this datasource's snapshot.
+ datasourceCache.markHasNewWrites();
try {
return writeAction.perform(datasourceCache);
}
@@ -570,12 +574,16 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
final Map<String, DatasourceSegmentSummary> 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<String> 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 +599,41 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
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}.
+ * <p>
+ * 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<String>
datasourcesToRefresh)
{
final Stopwatch updateDuration = Stopwatch.createStarted();
+ final DataSourcesSnapshot previousSnapshot = datasourcesSnapshot.get();
+ // updateSegmentIdsInCache never returns null, so datasourcesToRefresh is
null
+ // only on the first sync which also has no previous snapshot to build on.
+ final boolean incremental = previousSnapshot != null;
+
final Set<String> 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<String, Set<DataSegment>> datasourceToUsedSegments = new
HashMap<>();
+ final Set<String> removedDatasources = new HashSet<>();
for (String dataSource : cachedDatasources) {
final HeapMemoryDatasourceSegmentCache cache =
datasourceToSegmentCache.getOrDefault(
@@ -627,19 +655,43 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
}
}
);
+ 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<DataSegment> 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 +794,21 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
* 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<String> updateSegmentIdsInCache(
Map<String, DatasourceSegmentSummary> 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<String> 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 +818,9 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
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 +828,14 @@ public class HeapMemorySegmentMetadataCache implements
SegmentMetadataCache
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 00000000000..7df7148a0d9
--- /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<DataSegment> 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<String> 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<String, Set<DataSegment>> 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<DataSegment> 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<DataSegment> baseDs2 = previous.getDataSource(DS2).getSegments()
+
.stream().collect(Collectors.toSet());
+
+ final Set<DataSegment> newDs1 = segmentsOf(DS1, "2024-02-01", 3);
+ final DataSourcesSnapshot incremental =
previous.updateSnapshotForDataSources(
+ Map.of(DS1, newDs1),
+ Set.of(),
+ SNAPSHOT_TIME
+ );
+
+ final Map<String, Set<DataSegment>> 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 3fef28a963a..51bd3dd8390 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 class SQLMetadataConnectorTest
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 class SQLMetadataConnectorTest
connector.alterSegmentTable();
final Set<String> expectedIndices = Sets.newHashSet(
- "IDX_DRUIDTEST_SEGMENTS_USED",
+ "IDX_DRUIDTEST_SEGMENTS_USED_USLU_DATASOURCE",
"IDX_D011BD6ED76268701273CE512704C5AFA060D672",
"IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3"
);
@@ -372,8 +375,11 @@ public class SQLMetadataConnectorTest
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 class SQLMetadataConnectorTest
connector.alterSegmentTable();
final Set<String> 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 class SQLMetadataConnectorTest
final String segmentsTable = tablesConfig.getSegmentsTable();
final Set<String> expectedIndices = Sets.newHashSet(
- "IDX_8FE3D20EC8C9CA932EA3FF6AC497D1A9E75ADDA0",
+ "IDX_93A18EE829B37C5F38FC6DAFB070261D88503835",
"IDX_D011BD6ED76268701273CE512704C5AFA060D672",
"IDX_6381EF2DB4824C35C0E72EF9E166626ADB2B21A3"
);
@@ -431,9 +437,9 @@ public class SQLMetadataConnectorTest
final String segmentsTable = tablesConfig.getSegmentsTable();
final Set<String> 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 e739744480e..95a8a344bc5 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 @@ public class HeapMemorySegmentMetadataCacheTest
);
}
+ @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<DataSegmentPlus> segments)
{
IndexerSqlMetadataStorageCoordinatorTestBase
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
