kfaraz commented on code in PR #19672:
URL: https://github.com/apache/druid/pull/19672#discussion_r3568112913
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -345,7 +337,12 @@ public <T> T writeCacheForDataSource(String dataSource,
Action<T> 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();
Review Comment:
The flag should be set before the write action is performed, so that cases
that throw exception and end up doing a partial write on the cache (which
should never happen in practice) are still considered.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -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<DataSegment> segmentsFromWrites =
cache.getUsedSegmentsIfHasNewWrites();
+ if (segmentsFromWrites != null) {
+ datasourceToUsedSegments.put(dataSource, segmentsFromWrites);
Review Comment:
I suppose we need not distinguish between these 2 cases since the snapshot
only cares about whether the cache has been updated or not, regardless of the
state of the metadata store. To avoid missing internal writes done outside the
method `writeCacheForDataSource`, we could maintain the `hasNewWrites` flag in
`ReadWriteCache`.
That would also allow us to avoid passing the `datasourcesToRefresh`
argument into this method.
But the current implementation is fine too.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -570,12 +575,16 @@ private long syncWithMetadataStore()
final Map<String, DatasourceSegmentSummary> datasourceToSummary = new
HashMap<>();
+ // Datasources whose used-segment set changed this sync. Null forces a full
Review Comment:
Does `updateSegmentIdsInCache` ever return null?
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -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}.
+ * <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();
+ final boolean incremental = datasourcesToRefresh != null &&
previousSnapshot != null;
Review Comment:
`updateSegmentIdsInCache` never seems to return null, which is okay.
We can just check if the previousSnapshot was null or not.
```suggestion
final boolean incremental = previousSnapshot != null;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
