You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "jtuglu1 (via GitHub)" <gi...@apache.org> on 2026/05/30 17:03:04 UTC

[PR] perf: optimize coordinator duty runtime (druid)

jtuglu1 opened a new pull request, #19532:
URL: https://github.com/apache/druid/pull/19532

   <!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
   
   <!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
   
   Fixes #XXXX.
   
   <!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
   
   <!-- If you are a committer, follow the PR action item checklist for committers:
   https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
   
   ### Description
   
   <!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
   
   <!-- Describe your patch: what did you change in code? How did you fix the problem? -->
   
   <!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
   
   #### Fixed the bug ...
   #### Renamed the class ...
   #### Added a forbidden-apis entry ...
   
   <!--
   In each section, please describe design decisions made, including:
    - Choice of algorithms
    - Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
    - Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
    - Method organization and design (how the logic is split between methods, parameters and return types)
    - Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
   -->
   
   
   <!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
   
   <!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
   
   <!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
   
   #### Release note
   <!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users. 
   
   If your change doesn't have end user impact, you can skip this section.
   
   For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
   
   -->
   
   
   <hr>
   
   ##### Key changed/added classes in this PR
    * `MyFoo`
    * `OurBar`
    * `TheirBaz`
   
   <hr>
   
   <!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
   
   This PR has:
   
   - [ ] been self-reviewed.
      - [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
   - [ ] added documentation for new or modified features or behaviors.
   - [ ] a release note entry in the PR description.
   - [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
   - [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
   - [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
   - [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
   - [ ] added integration tests.
   - [ ] been tested in a test Druid cluster.


-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3595344321


##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -86,6 +87,7 @@ public StrategicSegmentAssigner(
     this.cluster = cluster;
     this.strategy = strategy;
     this.loadQueueManager = loadQueueManager;
+    this.allTiersInCluster = Set.copyOf(cluster.getTierNames());

Review Comment:
   [P2] Preserve null tier handling in the cached set
   
   `DruidCluster.Builder.addTier` and `DruidServerMetadata` both permit a null tier, and the previous `HashSet` copy preserved such a key. `Set.copyOf` rejects null elements, so one Historical with a null or omitted tier now throws here and aborts the HistoricalManagement duty before replica assignment. Use a null-tolerant copy, or normalize or reject null when constructing the cluster.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3579178396


##########
server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java:
##########
@@ -50,25 +51,49 @@ public BalanceSegments(Duration coordinatorPeriod)
   @Override
   public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
   {
+    final Stopwatch totalTime = Stopwatch.createStarted();
+
     if (params.getUsedSegmentCount() <= 0) {
+      log.info("BalanceSegments skipped: usedSegmentCount[%,d].", params.getUsedSegmentCount());

Review Comment:
   As already called out in the description, please ensure that all the logs are removed before merging. Otherwise, coordinator logs will become very noisy.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3597455328


##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }
+
+      final Map<String, SegmentReplicaCount> tierToReplicaCount = replicaCountsInTier.get(segmentId);
+      if (tierToReplicaCount != null) {
+        tierToReplicaCount.forEach((tier, counts) -> {
+          final int underReplicated = ignoreMissingServers ? counts.missing() : counts.missingAndLoadable();
+          if (underReplicated >= 0) {
+            Object2LongOpenHashMap<String> datasourceToUnderReplicated = (Object2LongOpenHashMap<String>)
+                tierToUnderReplicated.computeIfAbsent(tier, ds -> new Object2LongOpenHashMap<>());
+            datasourceToUnderReplicated.addTo(datasource, underReplicated);
+          }
+        });

Review Comment:
   I can run a few benchmarks and see if this has any material impact (the original goal was to avoid looping through 3x and instead fuse everything into a single pass).



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3593547411


##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -801,26 +803,46 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     {
       // Collect stats for unavailable and under-replicated segments
       final CoordinatorRunStats stats = params.getCoordinatorStats();
-      getDatasourceToUnavailableSegmentCount().forEach(
-          (dataSource, numUnavailable) -> stats.add(
-              Stats.Segments.UNAVAILABLE,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numUnavailable
-          )
-      );
-      getTierToDatasourceToUnderReplicatedCount(false).forEach(
-          (tier, countsPerDatasource) -> countsPerDatasource.forEach(
-              (dataSource, underReplicatedCount) ->
-                  stats.addToSegmentStat(Stats.Segments.UNDER_REPLICATED, tier, dataSource, underReplicatedCount)
-          )
-      );
-      getDatasourceToDeepStorageQueryOnlySegmentCount().forEach(
-          (dataSource, numDeepStorageOnly) -> stats.add(
-              Stats.Segments.DEEP_STORAGE_ONLY,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numDeepStorageOnly
-          )
-      );
+
+      final Object2IntMap<String> dsToUnavailable;
+      final Map<String, Object2LongMap<String>> tierToDsToUnderRepl;
+      final Object2IntMap<String> dsToDeepStorageOnly;
+
+      if (segmentReplicationStatus == null) {
+        dsToUnavailable = Object2IntMaps.emptyMap();
+        tierToDsToUnderRepl = Collections.emptyMap();
+        dsToDeepStorageOnly = Object2IntMaps.emptyMap();
+      } else {
+        // Single fused pass replaces three independent full iterations over
+        // metadataManager.iterateAllUsedSegments(). ignoreMissingServers=true replicates
+        // the !false inversion inside computeUnderReplicated(segments, false).

Review Comment:
   simplify the comment. details may be found in the javadoc of `computeSegmentStats` itself.
   ```suggestion
           // Compute unavailable, under-replicated, deep-storage only counts in a single pass
   ```



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator duty runtime (druid)

Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3473075645


##########
benchmarks/src/test/java/org/apache/druid/server/coordinator/ComputePlacementCostBenchmark.java:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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.server.coordinator;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.client.ImmutableDruidDataSource;
+import org.apache.druid.client.ImmutableDruidServer;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.server.coordination.DruidServerMetadata;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.balancer.CostBalancerStrategy;
+import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
+import org.apache.druid.timeline.DataSegment;
+import org.joda.time.Interval;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmarks {@link CostBalancerStrategy#computePlacementCost}, the per-server placement-cost computation the balancer
+ * invokes for every candidate server when placing a segment. The cost of a single call scales with the number of
+ * interval buckets the server's segments occupy, so {@code historyDays} (the span of daily intervals held by the
+ * server) is the primary parameter to vary.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(1)
+public class ComputePlacementCostBenchmark
+{
+  private static final long DAY_MILLIS = TimeUnit.DAYS.toMillis(1);
+  private static final long T0 = DateTimes.of("2026-01-01T00:00:00Z").getMillis();
+  private static final String DATASOURCE = "ds0";
+
+  /** Span of contiguous daily intervals held by the server. */
+  @Param({"180", "730", "3650"})
+  private int historyDays;
+
+  private ListeningExecutorService exec;
+  private ExposedCostBalancerStrategy strategy;
+  private ServerHolder server;
+  private DataSegment proposalSegment;
+
+  @Setup(Level.Trial)
+  public void setup()
+  {
+    exec = MoreExecutors.newDirectExecutorService();
+    strategy = new ExposedCostBalancerStrategy(exec);
+
+    final List<DataSegment> segments = new ArrayList<>(historyDays);
+    for (int day = 0; day < historyDays; day++) {
+      final long start = T0 - (long) (day + 1) * DAY_MILLIS;
+      segments.add(createSegment(Intervals.utc(start, start + DAY_MILLIS)));
+    }
+
+    final ImmutableDruidServer immutableServer = new ImmutableDruidServer(
+        new DruidServerMetadata("server", "host", null, 1L << 40, null, ServerType.HISTORICAL, "_default_tier", 0),
+        0L,
+        ImmutableMap.of(DATASOURCE, new ImmutableDruidDataSource(DATASOURCE, Collections.emptyMap(), segments)),
+        segments.size()
+    );
+    server = new ServerHolder(immutableServer, new TestLoadQueuePeon());
+
+    proposalSegment = createSegment(Intervals.utc(T0 - DAY_MILLIS, T0));
+  }
+
+  @TearDown(Level.Trial)
+  public void tearDown()
+  {
+    exec.shutdownNow();
+  }
+
+  @Benchmark
+  public double computePlacementCost()
+  {
+    return strategy.computePlacementCost(proposalSegment, server);
+  }
+
+  private static DataSegment createSegment(Interval interval)
+  {
+    return new DataSegment(
+        DATASOURCE,
+        interval,
+        "v1",
+        Collections.emptyMap(),
+        Collections.emptyList(),
+        Collections.emptyList(),
+        null,
+        0,
+        1
+    );

Review Comment:
   ## CodeQL / Deprecated method or constructor invocation
   
   Invoking [DataSegment.DataSegment](1) should be avoided because it has been deprecated.
   
   [Show more details](https://github.com/apache/druid/security/code-scanning/11322)



##########
benchmarks/src/test/java/org/apache/druid/server/coordinator/MarkOvershadowedSegmentsAsUnusedBenchmark.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.server.coordinator;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.client.DataSourcesSnapshot;
+import org.apache.druid.client.ImmutableDruidDataSource;
+import org.apache.druid.client.ImmutableDruidServer;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.server.coordination.DruidServerMetadata;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.balancer.RandomBalancerStrategy;
+import org.apache.druid.server.coordinator.duty.MarkOvershadowedSegmentsAsUnused;
+import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
+import org.apache.druid.timeline.DataSegment;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmarks the {@code MarkOvershadowedSegmentsAsUnused} coordinator duty's {@code run} method against a cluster where
+ * most datasources have no overshadowed segments (as in production). The duty builds segment timelines from the served
+ * segments only for the datasources that actually have overshadowed segments, so the cost is dominated by how many
+ * datasources are relevant relative to the total served.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(1)
+public class MarkOvershadowedSegmentsAsUnusedBenchmark
+{
+  private static final DateTime START = DateTimes.of("2024-01-01");
+
+  /** Total datasources served by the cluster. */
+  @Param({"1000"})
+  private int numDatasources;
+
+  /** Datasources that have overshadowed segments (an older version shadowed by a newer one). */
+  @Param({"10", "100"})
+  private int relevantDatasources;
+
+  @Param({"4"})
+  private int intervalsPerDatasource;
+
+  private MarkOvershadowedSegmentsAsUnused duty;
+  private DruidCoordinatorRuntimeParams params;
+
+  @Setup(Level.Trial)
+  public void setup()
+  {
+    final List<DataSegment> usedSegments = new ArrayList<>();
+    final ImmutableMap.Builder<String, ImmutableDruidDataSource> dataSources = ImmutableMap.builder();
+
+    for (int d = 0; d < numDatasources; d++) {
+      final String datasource = "ds_" + d;
+      final boolean hasOvershadowed = d < relevantDatasources;
+      final List<DataSegment> dsSegments = new ArrayList<>();
+      for (int i = 0; i < intervalsPerDatasource; i++) {
+        final Interval interval = new Interval(START.plusDays(i), START.plusDays(i + 1));
+        dsSegments.add(segment(datasource, interval, "v1"));
+        if (hasOvershadowed) {
+          // A newer version overshadows the v1 segment for the same interval.
+          dsSegments.add(segment(datasource, interval, "v2"));
+        }
+      }
+      usedSegments.addAll(dsSegments);
+      dataSources.put(datasource, new ImmutableDruidDataSource(datasource, ImmutableMap.of(), dsSegments));
+    }
+
+    final ImmutableDruidServer server = new ImmutableDruidServer(
+        new DruidServerMetadata("server", "host", null, 1L << 40, null, ServerType.HISTORICAL, "_default_tier", 0),
+        0L,
+        dataSources.build(),
+        usedSegments.size()
+    );
+    final DruidCluster cluster = DruidCluster
+        .builder()
+        .add(new ServerHolder(server, new TestLoadQueuePeon()))
+        .build();
+
+    params = DruidCoordinatorRuntimeParams
+        .builder()
+        .withDataSourcesSnapshot(DataSourcesSnapshot.fromUsedSegments(usedSegments))
+        .withDruidCluster(cluster)
+        .withDynamicConfigs(CoordinatorDynamicConfig.builder().withMarkSegmentAsUnusedDelayMillis(0).build())
+        .withBalancerStrategy(new RandomBalancerStrategy())
+        .withSegmentAssignerUsing(new SegmentLoadQueueManager(null, null))
+        .build();
+
+    // A no-op delete handler so the benchmark measures the timeline build + overshadow check, not the metadata write.
+    duty = new MarkOvershadowedSegmentsAsUnused((datasource, segmentIds) -> segmentIds.size());
+  }
+
+  @Benchmark
+  public DruidCoordinatorRuntimeParams run()
+  {
+    return duty.run(params);
+  }
+
+  private static DataSegment segment(String datasource, Interval interval, String version)
+  {
+    return DataSegment.builder()

Review Comment:
   ## CodeQL / Deprecated method or constructor invocation
   
   Invoking [DataSegment.builder](1) should be avoided because it has been deprecated.
   
   [Show more details](https://github.com/apache/druid/security/code-scanning/11323)



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3579171625


##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -1093,6 +1093,7 @@ public List<Interval> retrieveUnusedSegmentIntervals(String dataSource, int limi
     return intervals.stream().filter(Objects::nonNull).collect(Collectors.toList());
   }
 
+

Review Comment:
   nit: not required



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java:
##########
@@ -50,25 +51,49 @@ public BalanceSegments(Duration coordinatorPeriod)
   @Override
   public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
   {
+    final Stopwatch totalTime = Stopwatch.createStarted();
+
     if (params.getUsedSegmentCount() <= 0) {
+      log.info("BalanceSegments skipped: usedSegmentCount[%,d].", params.getUsedSegmentCount());

Review Comment:
   As already called out in the description, please ensure that all the logs are removed before merging. Otherwise, coordinator logs will become very noisy.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -64,6 +63,10 @@ public class StrategicSegmentAssigner implements SegmentActionHandler
   private final RoundRobinServerSelector serverSelector;
   private final BalancerStrategy strategy;
 
+  // Fixed for the lifetime of this assigner (a new instance is created each coordinator run),
+  // so it is computed once here rather than on every replicateSegment/replicateSegmentPartially call.

Review Comment:
   This comment can be omitted since we are already maintaining other similar fields which are valid for the current run such as `historicalTierAliases` and `tierToAliasName`.



##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -782,11 +783,32 @@ private class UpdateReplicationStatus implements CoordinatorDuty
     @Override
     public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     {
+      final Stopwatch totalTime = Stopwatch.createStarted();
+
+      final Stopwatch broadcastTime = Stopwatch.createStarted();

Review Comment:
   Please remove these stopwatches as well before merging this PR.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -86,6 +89,9 @@ public StrategicSegmentAssigner(
     this.cluster = cluster;
     this.strategy = strategy;
     this.loadQueueManager = loadQueueManager;
+    final Set<String> allTiersInCluster = new HashSet<>();
+    cluster.getTierNames().forEach(allTiersInCluster::add);
+    this.allTiersInCluster = allTiersInCluster;

Review Comment:
   to ensure immutability:
   
   ```suggestion
       this.allTiersInCluster = Set.copyOf(cluster.getTierNames());
   ```



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkOvershadowedSegmentsAsUnused.java:
##########
@@ -80,23 +80,33 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
       return params;
     }
 
+    // Identify the datasources that actually have overshadowed segments to check.
+    // Timelines only need to be built for these datasources.
+    final Set<String> relevantDatasources = new HashSet<>();
+    for (DataSegment s : allOvershadowedSegments) {
+      relevantDatasources.add(s.getDataSource());
+    }

Review Comment:
   Nit: rename and make immutable
   ```suggestion
       final Set<String> eligibleDatasources = allOvershadowedSegments
           .stream()
           .map(DataSegment::getDataSource)
           .collect(Collectors.toSet());
   ```



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkOvershadowedSegmentsAsUnused.java:
##########
@@ -80,23 +80,33 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
       return params;
     }
 
+    // Identify the datasources that actually have overshadowed segments to check.
+    // Timelines only need to be built for these datasources.
+    final Set<String> relevantDatasources = new HashSet<>();

Review Comment:
   ```suggestion
       final Set<String> eligibleDatasources = new HashSet<>();
   ```



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams par
   private Map<String, Set<SegmentId>> determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot dataSourcesSnapshot)
   {
     final Map<String, Set<SegmentId>> datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+    final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new HashMap<>();
 
-    dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+    for (final DataSegment overshadowedSegment : dataSourcesSnapshot.getOvershadowedSegments()) {
+      overshadowedSegmentsByDatasource
+          .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new HashSet<>())
+          .add(overshadowedSegment);
+    }
+
+    for (final ImmutableDruidDataSource dataSource : dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+      final String datasource = dataSource.getName();

Review Comment:
   To avoid confusion with the other variable `dataSource`.
   ```suggestion
         final String datasourceName = dataSource.getName();
   ```



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicaCount.java:
##########
@@ -36,6 +36,25 @@ public class SegmentReplicaCount
   private int movingTo;
   private int movingFrom;
 
+  SegmentReplicaCount()
+  {
+  }
+
+  /**
+   * Copies the counts out of {@code other} into a new, independent instance.
+   */
+  SegmentReplicaCount(SegmentReplicaCount other)

Review Comment:
   Instead of this, maybe add a `copy()` method.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicaCount.java:
##########
@@ -36,6 +36,25 @@ public class SegmentReplicaCount
   private int movingTo;
   private int movingFrom;
 
+  SegmentReplicaCount()

Review Comment:
   This shouldn't be needed.



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/PrepareBalancerAndLoadQueues.java:
##########
@@ -203,14 +203,14 @@ private void collectHistoricalStats(
 
   private void collectUsedSegmentStats(DruidCoordinatorRuntimeParams params, CoordinatorRunStats stats)
   {
-    params.getUsedSegmentsTimelinesPerDataSource().forEach((dataSource, timeline) -> {
-      long totalSizeOfUsedSegments = timeline.iterateAllObjects().stream()
-                                             .mapToLong(DataSegment::getSize).sum();
+    for (final ImmutableDruidDataSource dataSource : params.getDataSourcesSnapshot().getDataSourcesWithAllUsedSegments()) {
+      final long totalSizeOfUsedSegments = dataSource.getTotalSizeOfSegments();
+      final int numUsedSegments = dataSource.getSegments().size();

Review Comment:
   We should also add a method `DataSourcesSnapshot.getNumSegments()` to avoid materializing the collection inside `DataSourcesSnapshot.getSegments()`.



##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -801,26 +803,46 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     {
       // Collect stats for unavailable and under-replicated segments
       final CoordinatorRunStats stats = params.getCoordinatorStats();
-      getDatasourceToUnavailableSegmentCount().forEach(
-          (dataSource, numUnavailable) -> stats.add(
-              Stats.Segments.UNAVAILABLE,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numUnavailable
-          )
-      );
-      getTierToDatasourceToUnderReplicatedCount(false).forEach(
-          (tier, countsPerDatasource) -> countsPerDatasource.forEach(
-              (dataSource, underReplicatedCount) ->
-                  stats.addToSegmentStat(Stats.Segments.UNDER_REPLICATED, tier, dataSource, underReplicatedCount)
-          )
-      );
-      getDatasourceToDeepStorageQueryOnlySegmentCount().forEach(
-          (dataSource, numDeepStorageOnly) -> stats.add(
-              Stats.Segments.DEEP_STORAGE_ONLY,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numDeepStorageOnly
-          )
-      );
+
+      final Object2IntMap<String> dsToUnavailable;
+      final Map<String, Object2LongMap<String>> tierToDsToUnderRepl;
+      final Object2IntMap<String> dsToDeepStorageOnly;
+
+      if (segmentReplicationStatus == null) {
+        dsToUnavailable = Object2IntMaps.emptyMap();
+        tierToDsToUnderRepl = Collections.emptyMap();
+        dsToDeepStorageOnly = Object2IntMaps.emptyMap();
+      } else {
+        // Single fused pass replaces three independent full iterations over
+        // metadataManager.iterateAllUsedSegments(). ignoreMissingServers=true replicates
+        // the !false inversion inside computeUnderReplicated(segments, false).
+        final SegmentReplicationStatus.SegmentStatsSnapshot snapshot = segmentReplicationStatus.computeSegmentStats(
+            metadataManager.iterateAllUsedSegments(),
+            true
+        );
+        dsToUnavailable = snapshot.getDatasourceToUnavailableCount();
+        tierToDsToUnderRepl = snapshot.getTierToDatasourceToUnderReplicatedCount();
+        dsToDeepStorageOnly = snapshot.getDatasourceToDeepStorageOnlyCount();
+      }
+
+      for (final Object2IntMap.Entry<String> e : dsToUnavailable.object2IntEntrySet()) {
+        stats.add(Stats.Segments.UNAVAILABLE, RowKey.of(Dimension.DATASOURCE, e.getKey()), e.getIntValue());
+      }
+
+      for (final Map.Entry<String, Object2LongMap<String>> tierEntry : tierToDsToUnderRepl.entrySet()) {

Review Comment:
   Using `forEach` and `fastForEach` instead of traditional for loops may make this more readable.



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams par
   private Map<String, Set<SegmentId>> determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot dataSourcesSnapshot)
   {
     final Map<String, Set<SegmentId>> datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+    final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new HashMap<>();
 
-    dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+    for (final DataSegment overshadowedSegment : dataSourcesSnapshot.getOvershadowedSegments()) {
+      overshadowedSegmentsByDatasource
+          .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new HashSet<>())
+          .add(overshadowedSegment);
+    }
+
+    for (final ImmutableDruidDataSource dataSource : dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+      final String datasource = dataSource.getName();
       final SegmentTimeline usedSegmentsTimeline
           = dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+      if (usedSegmentsTimeline == null) {

Review Comment:
   Can this ever happen? We are already iterating over the datasources provided by the snapshot itself.



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams par
   private Map<String, Set<SegmentId>> determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot dataSourcesSnapshot)
   {
     final Map<String, Set<SegmentId>> datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+    final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new HashMap<>();
 
-    dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+    for (final DataSegment overshadowedSegment : dataSourcesSnapshot.getOvershadowedSegments()) {
+      overshadowedSegmentsByDatasource
+          .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new HashSet<>())
+          .add(overshadowedSegment);
+    }
+
+    for (final ImmutableDruidDataSource dataSource : dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+      final String datasource = dataSource.getName();
       final SegmentTimeline usedSegmentsTimeline
           = dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+      if (usedSegmentsTimeline == null) {
+        continue;
+      }
 
-      final Optional<Set<DataSegment>> usedNonOvershadowedSegments =
-          Optional.fromNullable(usedSegmentsTimeline)
-                  .transform(timeline -> timeline.findNonOvershadowedObjectsInInterval(
-                      Intervals.ETERNITY,
-                      Partitions.ONLY_COMPLETE
-                  ));
-
-      if (usedNonOvershadowedSegments.isPresent()) {
-        usedNonOvershadowedSegments.get().forEach(candidateSegment -> {
-          if (isNewGenerationEternityTombstone(candidateSegment)) {
-            boolean overlaps = dataSourcesSnapshot.getOvershadowedSegments().stream()
-                                                  .filter(overshadowedSegment ->
-                                                              candidateSegment.getDataSource()
-                                                                              .equals(overshadowedSegment.getDataSource()))
-                                                  .anyMatch(
-                                                      overshadowedSegment ->
-                                                          candidateSegment.getInterval()
-                                                                          .overlaps(overshadowedSegment.getInterval())
-                                                  );
-            if (!overlaps) {
-              datasourceToNonOvershadowedEternityTombstones
-                  .computeIfAbsent(datasource, ds -> new HashSet<>())
-                  .add(candidateSegment.getId());
-            }
-          }
-        });
+      for (final DataSegment candidateSegment : dataSource.getSegments()) {
+        if (isNewGenerationEternityTombstone(candidateSegment)
+            && !usedSegmentsTimeline.isOvershadowed(candidateSegment)
+            && !overlapsAnyOvershadowedSegment(
+                candidateSegment,
+                overshadowedSegmentsByDatasource.get(candidateSegment.getDataSource())
+            )) {

Review Comment:
   Let's collapse these 3 conditions into a single new method for readability of this for loop:
   
   ```
   shouldMarkAsUnused(candidateSegment, overshadowedSegmentsByDatasource, timeline)
   ```
   
   the new method can check the 3 required conditions by invoking respective methods.



##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -801,26 +803,46 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     {
       // Collect stats for unavailable and under-replicated segments
       final CoordinatorRunStats stats = params.getCoordinatorStats();
-      getDatasourceToUnavailableSegmentCount().forEach(
-          (dataSource, numUnavailable) -> stats.add(
-              Stats.Segments.UNAVAILABLE,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numUnavailable
-          )
-      );
-      getTierToDatasourceToUnderReplicatedCount(false).forEach(
-          (tier, countsPerDatasource) -> countsPerDatasource.forEach(
-              (dataSource, underReplicatedCount) ->
-                  stats.addToSegmentStat(Stats.Segments.UNDER_REPLICATED, tier, dataSource, underReplicatedCount)
-          )
-      );
-      getDatasourceToDeepStorageQueryOnlySegmentCount().forEach(
-          (dataSource, numDeepStorageOnly) -> stats.add(
-              Stats.Segments.DEEP_STORAGE_ONLY,
-              RowKey.of(Dimension.DATASOURCE, dataSource),
-              numDeepStorageOnly
-          )
-      );
+
+      final Object2IntMap<String> dsToUnavailable;
+      final Map<String, Object2LongMap<String>> tierToDsToUnderRepl;
+      final Object2IntMap<String> dsToDeepStorageOnly;

Review Comment:
   Please use full names e.g. `datasourceToUnavailableCount`, `datasourceToUnderReplicatedCount`.
   
   Do we need to keep 3 separate map references? Maybe just use an empty instance of `SegmentStatsSnapshot` and invoke getters on it where needed.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -40,15 +43,27 @@ public class SegmentReplicationStatus
 
   public SegmentReplicationStatus(Map<SegmentId, Map<String, SegmentReplicaCount>> replicaCountsInTier)
   {
-    this.replicaCountsInTier = ImmutableMap.copyOf(replicaCountsInTier);
-
-    final Map<SegmentId, SegmentReplicaCount> totalReplicaCounts = new HashMap<>();
-    replicaCountsInTier.forEach((segmentId, tierToReplicaCount) -> {
+    // replicaCountsInTier is the caller's live SegmentReplicaCountMap and is mutated further
+    // in the same coordinator cycle (e.g. by BalanceSegments), so we must snapshot both the
+    // structure and the per-tier SegmentReplicaCount values, not just alias the outer map.

Review Comment:
   Nit: simpler language
   ```suggestion
       // Make a deep copy of replicaCountsInTier as it may be mutated further
       // in the same coordinator cycle (e.g. by BalanceSegments)
   ```



##########
server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java:
##########
@@ -98,14 +99,15 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
 
     final DateTime now = DateTimes.nowUtc();
     final Object2IntOpenHashMap<String> datasourceToSegmentsWithNoRule = new Object2IntOpenHashMap<>();
+    final Map<String, List<Rule>> datasourceToRules = new HashMap<>();

Review Comment:
   Rules need not be cached as they are already served from the cache in `SQLMetadataRuleManager` itself. The changes to this class may be reverted.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }
+
+      final Map<String, SegmentReplicaCount> tierToReplicaCount = replicaCountsInTier.get(segmentId);
+      if (tierToReplicaCount != null) {
+        tierToReplicaCount.forEach((tier, counts) -> {
+          final int underReplicated = ignoreMissingServers ? counts.missing() : counts.missingAndLoadable();
+          if (underReplicated >= 0) {
+            Object2LongOpenHashMap<String> datasourceToUnderReplicated = (Object2LongOpenHashMap<String>)
+                tierToUnderReplicated.computeIfAbsent(tier, ds -> new Object2LongOpenHashMap<>());
+            datasourceToUnderReplicated.addTo(datasource, underReplicated);
+          }
+        });

Review Comment:
   Can we replace this by a call to `getTierToDatasourceToUnderReplicated()` since it is the same logic?



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }
+
+      final Map<String, SegmentReplicaCount> tierToReplicaCount = replicaCountsInTier.get(segmentId);
+      if (tierToReplicaCount != null) {
+        tierToReplicaCount.forEach((tier, counts) -> {
+          final int underReplicated = ignoreMissingServers ? counts.missing() : counts.missingAndLoadable();
+          if (underReplicated >= 0) {
+            Object2LongOpenHashMap<String> datasourceToUnderReplicated = (Object2LongOpenHashMap<String>)
+                tierToUnderReplicated.computeIfAbsent(tier, ds -> new Object2LongOpenHashMap<>());
+            datasourceToUnderReplicated.addTo(datasource, underReplicated);
+          }
+        });
+      }
+    }
+
+    return new SegmentStatsSnapshot(datasourceToUnavailable, tierToUnderReplicated, datasourceToDeepStorageOnly);
+  }
+
+  /**
+   * Holder for the three segment-stat views computed together by {@link #computeSegmentStats}.
+   */
+  public static class SegmentStatsSnapshot

Review Comment:
   Please move out this class into a separate file as it seems completely independent.



##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }

Review Comment:
   This logic is now duplicated with `DruidCoordinator`.
   Please move unavailable logic and deep storage logic into new methods in this class so that `DruidCoordinator` may reuse it too.
   
   ```java
   boolean isUnavailable(SegmentId segmentId);
   
   boolean isUnderReplicated(SegmentId segmentId);
   ```



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3579220402


##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -782,11 +783,32 @@ private class UpdateReplicationStatus implements CoordinatorDuty
     @Override
     public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     {
+      final Stopwatch totalTime = Stopwatch.createStarted();
+
+      final Stopwatch broadcastTime = Stopwatch.createStarted();

Review Comment:
   Please remove these stopwatches as well before merging this PR.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3593350519


##########
server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java:
##########
@@ -98,14 +99,15 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
 
     final DateTime now = DateTimes.nowUtc();
     final Object2IntOpenHashMap<String> datasourceToSegmentsWithNoRule = new Object2IntOpenHashMap<>();
+    final Map<String, List<Rule>> datasourceToRules = new HashMap<>();

Review Comment:
   Rules need not be cached here as they are already served from the cache in `SQLMetadataRuleManager` itself. The changes to this class may be reverted.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3595932502


##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -86,6 +87,7 @@ public StrategicSegmentAssigner(
     this.cluster = cluster;
     this.strategy = strategy;
     this.loadQueueManager = loadQueueManager;
+    this.allTiersInCluster = Set.copyOf(cluster.getTierNames());

Review Comment:
   We can reject nulls when constructing the `DruidCluster` in a follow up PR.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3592960643


##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams par
   private Map<String, Set<SegmentId>> determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot dataSourcesSnapshot)
   {
     final Map<String, Set<SegmentId>> datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+    final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new HashMap<>();
 
-    dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+    for (final DataSegment overshadowedSegment : dataSourcesSnapshot.getOvershadowedSegments()) {
+      overshadowedSegmentsByDatasource
+          .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new HashSet<>())
+          .add(overshadowedSegment);
+    }
+
+    for (final ImmutableDruidDataSource dataSource : dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+      final String datasource = dataSource.getName();
       final SegmentTimeline usedSegmentsTimeline
           = dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+      if (usedSegmentsTimeline == null) {

Review Comment:
   Can this ever happen? We are already iterating over the datasources provided by the snapshot itself, so the timeline must be non 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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3593018167


##########
server/src/main/java/org/apache/druid/server/coordinator/duty/PrepareBalancerAndLoadQueues.java:
##########
@@ -203,14 +203,14 @@ private void collectHistoricalStats(
 
   private void collectUsedSegmentStats(DruidCoordinatorRuntimeParams params, CoordinatorRunStats stats)
   {
-    params.getUsedSegmentsTimelinesPerDataSource().forEach((dataSource, timeline) -> {
-      long totalSizeOfUsedSegments = timeline.iterateAllObjects().stream()
-                                             .mapToLong(DataSegment::getSize).sum();
+    for (final ImmutableDruidDataSource dataSource : params.getDataSourcesSnapshot().getDataSourcesWithAllUsedSegments()) {
+      final long totalSizeOfUsedSegments = dataSource.getTotalSizeOfSegments();
+      final int numUsedSegments = dataSource.getSegments().size();

Review Comment:
   We should also add a method `ImmutableDataSource.getNumSegments()` to avoid materializing the collection inside `ImmutableDataSource.getSegments()`.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3595979463


##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }
+
+      final Map<String, SegmentReplicaCount> tierToReplicaCount = replicaCountsInTier.get(segmentId);
+      if (tierToReplicaCount != null) {
+        tierToReplicaCount.forEach((tier, counts) -> {
+          final int underReplicated = ignoreMissingServers ? counts.missing() : counts.missingAndLoadable();
+          if (underReplicated >= 0) {
+            Object2LongOpenHashMap<String> datasourceToUnderReplicated = (Object2LongOpenHashMap<String>)
+                tierToUnderReplicated.computeIfAbsent(tier, ds -> new Object2LongOpenHashMap<>());
+            datasourceToUnderReplicated.addTo(datasource, underReplicated);
+          }
+        });

Review Comment:
   @jtuglu1 , is this possible? If yes, then we can do it in a follow up.
   Otherwise, please feel free to resolve this comment.



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #19532:
URL: https://github.com/apache/druid/pull/19532


-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org


Re: [PR] perf: optimize coordinator HistoricalManagement duty runtime (druid)

Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3597452016


##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>> getTierToDatasourceToUnderReplicated(
 
     return tierToUnderReplicated;
   }
+
+  /**
+   * Computes unavailable, under-replicated and deep-storage-only segment counts in a single
+   * pass over {@code usedSegments}, instead of three independent full iterations. Produces
+   * results identical to calling {@link #getReplicaCountsInCluster} and
+   * {@link #getTierToDatasourceToUnderReplicated} independently for each segment.
+   *
+   * @param ignoreMissingServers same semantics as in {@link #getTierToDatasourceToUnderReplicated}.
+   */
+  public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment> usedSegments, boolean ignoreMissingServers)
+  {
+    final Object2IntOpenHashMap<String> datasourceToUnavailable = new Object2IntOpenHashMap<>();
+    final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new Object2IntOpenHashMap<>();
+    final Map<String, Object2LongMap<String>> tierToUnderReplicated = new HashMap<>();
+
+    for (DataSegment segment : usedSegments) {
+      final SegmentId segmentId = segment.getId();
+      final String datasource = segment.getDataSource();
+
+      final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+      if (totalCount != null && (totalCount.totalLoaded() > 0 || totalCount.required() == 0)) {
+        datasourceToUnavailable.addTo(datasource, 0);
+      } else {
+        datasourceToUnavailable.addTo(datasource, 1);
+      }
+      if (totalCount != null && totalCount.totalLoaded() == 0 && totalCount.required() == 0) {
+        datasourceToDeepStorageOnly.addTo(datasource, 1);
+      }
+
+      final Map<String, SegmentReplicaCount> tierToReplicaCount = replicaCountsInTier.get(segmentId);
+      if (tierToReplicaCount != null) {
+        tierToReplicaCount.forEach((tier, counts) -> {
+          final int underReplicated = ignoreMissingServers ? counts.missing() : counts.missingAndLoadable();
+          if (underReplicated >= 0) {
+            Object2LongOpenHashMap<String> datasourceToUnderReplicated = (Object2LongOpenHashMap<String>)
+                tierToUnderReplicated.computeIfAbsent(tier, ds -> new Object2LongOpenHashMap<>());
+            datasourceToUnderReplicated.addTo(datasource, underReplicated);
+          }
+        });

Review Comment:
   The goal for this change was to avoid another full pass over usedSegments. I suppose we can do something like:
   
   ```java
   final Map<...> tierToUnderReplicated =
       getTierToDatasourceToUnderReplicated(usedSegments, ignoreMissingServers);
   ```
   
   as well to avoid any n^2 behavior during computeSegmentStats().



-- 
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: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org