kfaraz commented on code in PR #19672:
URL: https://github.com/apache/druid/pull/19672#discussion_r3558982244
##########
server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java:
##########
@@ -91,6 +92,78 @@ public static DataSourcesSnapshot
fromUsedSegments(Iterable<DataSegment> 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.
+ * <p>
+ * Overshadowing is computed per-datasource, so reusing the prior
overshadowed
+ * segments for unchanged datasources is correct. This turns the
O(all-segments)
+ * snapshot rebuild into an O(changed-segments) operation, while producing a
+ * 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,
Review Comment:
For a cleaner API, let's not use a static method which allows passing a null
`previous` snapshot.
Instead, add a non-static method `updateSnapshotForDatasources()` which will
be called on the existing snapshot by the caller itself.
```java
newSnapshot = previousSnapshot.updateSnapshotForDatasources(changed,
removed, snapshotTime);
```
##########
server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java:
##########
@@ -91,6 +92,78 @@ public static DataSourcesSnapshot
fromUsedSegments(Iterable<DataSegment> 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.
+ * <p>
+ * Overshadowing is computed per-datasource, so reusing the prior
overshadowed
+ * segments for unchanged datasources is correct. This turns the
O(all-segments)
+ * snapshot rebuild into an O(changed-segments) operation, while producing a
+ * 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<String, Set<DataSegment>> changedDataSources,
+ Set<String> removedDataSources,
+ DateTime snapshotTime
+ )
+ {
+ if (previous == null) {
+ return fromUsedSegments(changedDataSources, snapshotTime);
+ }
+
+ final Map<String, String> properties = Map.of("created",
snapshotTime.toString());
+ final Map<String, ImmutableDruidDataSource> dataSources = new
HashMap<>(previous.dataSourcesWithAllUsedSegments);
+ final Map<String, SegmentTimeline> timelines = new
HashMap<>(previous.usedSegmentsTimelinesPerDataSource);
+
+ final Set<String> dirtyDataSources = new HashSet<>(removedDataSources);
+ dirtyDataSources.addAll(changedDataSources.keySet());
+
+ removedDataSources.forEach(ds -> {
+ dataSources.remove(ds);
+ timelines.remove(ds);
+ });
+ changedDataSources.forEach((ds, segments) -> {
+ if (segments.isEmpty()) {
Review Comment:
Wouldn't this case already be present in `removedDatasources`? Is this just
a safe-side measure?
##########
server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java:
##########
@@ -91,6 +92,78 @@ public static DataSourcesSnapshot
fromUsedSegments(Iterable<DataSegment> 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.
+ * <p>
+ * Overshadowing is computed per-datasource, so reusing the prior
overshadowed
+ * segments for unchanged datasources is correct. This turns the
O(all-segments)
+ * snapshot rebuild into an O(changed-segments) operation, while producing a
+ * 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<String, Set<DataSegment>> changedDataSources,
+ Set<String> removedDataSources,
+ DateTime snapshotTime
+ )
+ {
+ if (previous == null) {
+ return fromUsedSegments(changedDataSources, snapshotTime);
+ }
+
+ final Map<String, String> properties = Map.of("created",
snapshotTime.toString());
+ final Map<String, ImmutableDruidDataSource> dataSources = new
HashMap<>(previous.dataSourcesWithAllUsedSegments);
+ final Map<String, SegmentTimeline> timelines = new
HashMap<>(previous.usedSegmentsTimelinesPerDataSource);
+
+ final Set<String> dirtyDataSources = new HashSet<>(removedDataSources);
Review Comment:
Please use some other term here like `outdated` or `modified` or
`datasourcesToUpdate`, etc.
##########
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
Review Comment:
We can skip this config and have this feature always on.
##########
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")
Review Comment:
Yeah, I have my doubts about this too.
But it is okay since after this, we can get rid of the older `IDX_USED`
index.
@jtuglu1 , should we do it in this PR itself? It will affect only new
clusters anyway.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -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<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> dirtyDatasources = new HashSet<>();
Review Comment:
```suggestion
final Set<String> datasourcesToRefresh = new HashSet<>();
```
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -167,6 +167,13 @@ private enum CacheState
private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
private final AtomicReference<DataSourcesSnapshot> 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;
Review Comment:
I don't think this needs to be optional since it is a perf improvement that
all clusters can benefit from.
Are there any drawbacks to the approach?
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -570,12 +578,16 @@ private long syncWithMetadataStore()
final Map<String, DatasourceSegmentSummary> 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<String> dirtyDatasources = null;
Review Comment:
```suggestion
Set<String> modifiedDatasources = null;
```
##########
server/src/main/java/org/apache/druid/client/DataSourcesSnapshot.java:
##########
@@ -91,6 +92,78 @@ public static DataSourcesSnapshot
fromUsedSegments(Iterable<DataSegment> 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.
+ * <p>
+ * Overshadowing is computed per-datasource, so reusing the prior
overshadowed
+ * segments for unchanged datasources is correct. This turns the
O(all-segments)
+ * snapshot rebuild into an O(changed-segments) operation, while producing a
Review Comment:
This is not exactly true since we would still be rebuilding the entire
timeline of a datasource only if a single segment changed in that datasource.
That said, it would still be better than what we have today.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -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)) {
Review Comment:
Nice catch! This does seem to be an issue that will cause the snapshot and
the cache for a datasource to diverge. Normally, the snapshot reconciles itself
with what's in the cache in the next sync.
But we are removing that capability in this PR by updating snapshots for
modified datasources only.
> Record used-segment mutations from writeCacheForDataSource or otherwise
compare against the prior snapshot before reusing it.
Comparing against prior snapshot might be too costly, I think we could do
the following instead.
- `ReadWriteCache` keeps a volatile boolean `hasNewWrites`.
- Set this flag to true in `withWriteLock()`.
- Also, add method `ReadWriteCache.markAllWritesAsSynced()`
- Add new methods to `HeapMemoryDatasourceSegmentCache`:
`getAllUsedSegmentsIfCacheHasNewWrites()`.
- Invoke this method in incremental mode only
- This method should acquire a write lock, return all the used segments only
if there are writes and finally set `hasNewWrites` to false.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -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.
Review Comment:
```suggestion
* 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.
*/
```
--
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]
