Enable segments read/published stats on Compaction task completion reports

docs/ingestion/tasks.md

@@ -155,6 +155,14 @@ For some task types, the indexing task can wait for the newly ingested segments
 |`segmentAvailabilityWaitTimeMs`|Milliseconds waited by the ingestion task for the newly ingested segments to be available for query after completing ingestion was completed.|
 |`recordsProcessed`| Partitions that were processed by an ingestion task and includes count of records processed from each partition.|
 
+
+#### Compaction Task Fields
+
+| Field               | Description                                           |
+|---------------------|-------------------------------------------------------|
+| `segmentsRead`      | # of segments read by a parallel compaction task      |
+| `segmentsPublished` | # of segments published by a parallel compaction task |
+
 ### Live report
 
 When a task is running, a live report containing ingestion state, unparseable events and moving average for number of events processed for 1 min, 5 min, 15 min time window can be retrieved at:

indexing-service/src/main/java/org/apache/druid/indexing/common/IngestionStatsAndErrorsTaskReportData.java

@@ -20,12 +20,19 @@
 package org.apache.druid.indexing.common;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import org.apache.druid.indexer.IngestionState;
 
 import javax.annotation.Nullable;
 import java.util.Map;
 import java.util.Objects;
 
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "kind", defaultImpl = IngestionStatsAndErrorsTaskReport.class)
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = "ingestion_stats", value = IngestionStatsAndErrorsTaskReport.class),
+    @JsonSubTypes.Type(name = "parallel_compaction_stats", value = ParallelCompactionTaskReportData.class)
+})
 public class IngestionStatsAndErrorsTaskReportData
 {
   @JsonProperty

indexing-service/src/main/java/org/apache/druid/indexing/common/ParallelCompactionTaskReportData.java

@@ -0,0 +1,109 @@
+/*
+ * 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.indexing.common;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.indexer.IngestionState;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+public class ParallelCompactionTaskReportData extends IngestionStatsAndErrorsTaskReportData
+{
+  @JsonProperty
+  private Long segmentsRead;
+  @JsonProperty
+  private Long segmentsPublished;
+
+  public ParallelCompactionTaskReportData(
+      @JsonProperty("ingestionState") IngestionState ingestionState,
+      @JsonProperty("unparseableEvents") Map<String, Object> unparseableEvents,
+      @JsonProperty("rowStats") Map<String, Object> rowStats,
+      @JsonProperty("errorMsg") @Nullable String errorMsg,
+      @JsonProperty("segmentAvailabilityConfirmed") boolean segmentAvailabilityConfirmed,
+      @JsonProperty("segmentAvailabilityWaitTimeMs") long segmentAvailabilityWaitTimeMs,
+      @JsonProperty("recordsProcessed") Map<String, Long> recordsProcessed,
+      @JsonProperty("segmentsRead") Long segmentsRead,
+      @JsonProperty("segmentsPublished") Long segmentsPublished
+  )
+  {
+    super(
+        ingestionState,
+        unparseableEvents,
+        rowStats,
+        errorMsg,
+        segmentAvailabilityConfirmed,
+        segmentAvailabilityWaitTimeMs,
+        recordsProcessed
+    );
+    this.segmentsRead = segmentsRead;
+    this.segmentsPublished = segmentsPublished;
+  }
+
+  @JsonProperty
+  @Nullable
+  public Long getSegmentsRead()
+  {
+    return segmentsRead;
+  }
+
+  @JsonProperty
+  @Nullable
+  public Long getSegmentsPublished()
+  {
+    return segmentsPublished;
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ParallelCompactionTaskReportData that = (ParallelCompactionTaskReportData) o;
+    return super.equals(o) &&
+           Objects.equals(getSegmentsRead(), that.getSegmentsRead()) &&
+           Objects.equals(getSegmentsPublished(), that.getSegmentsPublished());
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(
+        super.hashCode(),
+        getSegmentsRead(),
+        getSegmentsPublished()
+    );
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ParallelCompactionTaskReportData {" +
+           "IngestionStatsAndErrorsTaskReportData=" + super.toString() +
+           ", segmentsRead=" + segmentsRead +
+           ", segmentsPublished=" + segmentsPublished +
+           '}';
+  }
+}

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/GeneratedPartitionsMetadataReport.java

@@ -38,9 +38,10 @@ class GeneratedPartitionsMetadataReport extends GeneratedPartitionsReport
   GeneratedPartitionsMetadataReport(
       @JsonProperty("taskId") String taskId,
       @JsonProperty("partitionStats") List<PartitionStat> partitionStats,
-      @JsonProperty("taskReport") Map<String, TaskReport> taskReport
+      @JsonProperty("taskReport") Map<String, TaskReport> taskReport,
+      @JsonProperty("segmentsRead") Long segmentsRead
   )
   {
-    super(taskId, partitionStats, taskReport);
+    super(taskId, partitionStats, taskReport, segmentsRead);
   }
 }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/GeneratedPartitionsReport.java

@@ -22,6 +22,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.druid.indexing.common.TaskReport;
 
+import javax.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -37,11 +38,19 @@ public class GeneratedPartitionsReport implements SubTaskReport
   private final List<PartitionStat> partitionStats;
   private final Map<String, TaskReport> taskReport;
 
-  GeneratedPartitionsReport(String taskId, List<PartitionStat> partitionStats, Map<String, TaskReport> taskReport)
+  private final Long segmentsRead;
+
+  GeneratedPartitionsReport(
+      String taskId,
+      List<PartitionStat> partitionStats,
+      Map<String, TaskReport> taskReport,
+      Long segmentsRead
+  )
   {
     this.taskId = taskId;
     this.partitionStats = partitionStats;
     this.taskReport = taskReport;
+    this.segmentsRead = segmentsRead;
   }
 
   @Override
@@ -63,6 +72,13 @@ public List<PartitionStat> getPartitionStats()
     return partitionStats;
   }
 
+  @JsonProperty
+  @Nullable
+  public Long getSegmentsRead()
+  {
+    return segmentsRead;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -75,13 +91,14 @@ public boolean equals(Object o)
     GeneratedPartitionsReport that = (GeneratedPartitionsReport) o;
     return Objects.equals(taskId, that.taskId) &&
            Objects.equals(partitionStats, that.partitionStats) &&
-           Objects.equals(taskReport, that.taskReport);
+           Objects.equals(taskReport, that.taskReport) &&
+           Objects.equals(segmentsRead, that.segmentsRead);
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(taskId, partitionStats, taskReport);
+    return Objects.hash(taskId, partitionStats, taskReport, segmentsRead);
   }
 
   @Override
@@ -91,6 +108,7 @@ public String toString()
         "taskId='" + taskId + '\'' +
         ", partitionStats=" + partitionStats +
         ", taskReport=" + taskReport +
+        ", segmentsRead=" + segmentsRead +
         '}';
   }
 }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java

@@ -42,6 +42,7 @@
 import org.apache.druid.indexer.partitions.PartitionsSpec;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReport;
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
+import org.apache.druid.indexing.common.ParallelCompactionTaskReportData;
 import org.apache.druid.indexing.common.TaskLockType;
 import org.apache.druid.indexing.common.TaskReport;
 import org.apache.druid.indexing.common.TaskToolbox;
@@ -204,6 +205,8 @@ public class ParallelIndexSupervisorTask extends AbstractBatchIndexTask implemen
   private IngestionState ingestionState;
   private Map<String, TaskReport> completionReports;
   private final Boolean isCompactionTask;
+  private Long segmentsRead;
+  private Long segmentsPublished;
 
 
   @JsonCreator
@@ -651,6 +654,14 @@ private TaskStatus runSinglePhaseParallel(TaskToolbox toolbox) throws Exception
     if (state.isSuccess()) {
       //noinspection ConstantConditions
       publishSegments(toolbox, parallelSinglePhaseRunner.getReports());
+      segmentsRead = parallelSinglePhaseRunner.getReports()
+                                              .values()
+                                              .stream()
+                                              .mapToLong(report -> report.getOldSegments().size()).sum();
+      segmentsPublished = parallelSinglePhaseRunner.getReports()
+                                              .values()
+                                              .stream()
+                                                   .mapToLong(report -> report.getNewSegments().size()).sum();
       if (awaitSegmentAvailabilityTimeoutMillis > 0) {
         waitForSegmentAvailability(parallelSinglePhaseRunner.getReports());
       }
@@ -822,7 +833,7 @@ TaskStatus runHashPartitionMultiPhaseParallel(TaskToolbox toolbox) throws Except
     TaskStatus taskStatus;
     if (state.isSuccess()) {
       //noinspection ConstantConditions
-      publishSegments(toolbox, mergeRunner.getReports());
+      segmentsPublished = publishSegments(toolbox, mergeRunner.getReports());
       if (awaitSegmentAvailabilityTimeoutMillis > 0) {
         waitForSegmentAvailability(mergeRunner.getReports());
       }
@@ -920,7 +931,7 @@ TaskStatus runRangePartitionMultiPhaseParallel(TaskToolbox toolbox) throws Excep
     TaskState mergeState = runNextPhase(mergeRunner);
     TaskStatus taskStatus;
     if (mergeState.isSuccess()) {
-      publishSegments(toolbox, mergeRunner.getReports());
+      segmentsPublished = publishSegments(toolbox, mergeRunner.getReports());
       if (awaitSegmentAvailabilityTimeoutMillis > 0) {
         waitForSegmentAvailability(mergeRunner.getReports());
       }
@@ -1132,7 +1143,7 @@ private static Pair<Integer, Integer> getPartitionBoundaries(int index, int tota
     return Pair.of(start, stop);
   }
 
-  private void publishSegments(
+  private long publishSegments(
       TaskToolbox toolbox,
       Map<String, PushedSegmentsReport> reportsMap
   )
@@ -1200,6 +1211,8 @@ private void publishSegments(
     } else {
       throw new ISE("Failed to publish segments");
     }
+
+    return newSegments.size();
   }
 
   private TaskStatus runSequential(TaskToolbox toolbox) throws Exception
@@ -1248,6 +1261,18 @@ private Map<String, TaskReport> getTaskCompletionReports(TaskStatus taskStatus,
     return TaskReport.buildTaskReports(
         new IngestionStatsAndErrorsTaskReport(
             getId(),
+            isCompactionTask ?
+            new ParallelCompactionTaskReportData(
+                IngestionState.COMPLETED,
+                rowStatsAndUnparseableEvents.rhs,
+                rowStatsAndUnparseableEvents.lhs,
+                taskStatus.getErrorMsg(),
+                segmentAvailabilityConfirmed,
+                segmentAvailabilityWaitTimeMs,
+                Collections.emptyMap(),
+                segmentsRead,
+                segmentsPublished
+            ) :
             new IngestionStatsAndErrorsTaskReportData(
                 IngestionState.COMPLETED,
                 rowStatsAndUnparseableEvents.rhs,
@@ -1633,6 +1658,7 @@ private Pair<Map<String, Object>, Map<String, Object>> doGetRowStatsAndUnparseab
 
       final SimpleRowIngestionMeters buildSegmentsRowStats = new SimpleRowIngestionMeters();
       final List<ParseExceptionReport> unparseableEvents = new ArrayList<>();
+      long totalSegmentsRead = 0L;
       for (GeneratedPartitionsReport generatedPartitionsReport : completedSubtaskReports.values()) {
         Map<String, TaskReport> taskReport = generatedPartitionsReport.getTaskReport();
         if (taskReport == null || taskReport.isEmpty()) {
@@ -1643,6 +1669,9 @@ private Pair<Map<String, Object>, Map<String, Object>> doGetRowStatsAndUnparseab
             getBuildSegmentsStatsFromTaskReport(taskReport, true, unparseableEvents);
 
         buildSegmentsRowStats.addRowIngestionMetersTotals(rowStatsForCompletedTask);
+        if (generatedPartitionsReport.getSegmentsRead() != null) {
+          totalSegmentsRead += generatedPartitionsReport.getSegmentsRead();
+        }
       }
 
       RowIngestionMetersTotals rowStatsForRunningTasks = getRowStatsAndUnparseableEventsForRunningTasks(
@@ -1651,6 +1680,9 @@ private Pair<Map<String, Object>, Map<String, Object>> doGetRowStatsAndUnparseab
           includeUnparseable
       );
       buildSegmentsRowStats.addRowIngestionMetersTotals(rowStatsForRunningTasks);
+      if (totalSegmentsRead > 0) {
+        segmentsRead = totalSegmentsRead;
+      }
 
       return createStatsAndErrorsReport(buildSegmentsRowStats.getTotals(), unparseableEvents);
     }

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialHashSegmentGenerateTask.java

@@ -177,12 +177,17 @@ SegmentAllocatorForBatch createSegmentAllocator(TaskToolbox toolbox, ParallelInd
   }
 
   @Override
-  GeneratedPartitionsMetadataReport createGeneratedPartitionsReport(TaskToolbox toolbox, List<DataSegment> segments, Map<String, TaskReport> taskReport)
+  GeneratedPartitionsMetadataReport createGeneratedPartitionsReport(
+      TaskToolbox toolbox,
+      List<DataSegment> segments,
+      Map<String, TaskReport> taskReport,
+      Long segmentsRead
+  )
   {
     List<PartitionStat> partitionStats = segments.stream()
                                                         .map(segment -> toolbox.getIntermediaryDataManager().generatePartitionStat(toolbox, segment))
                                                         .collect(Collectors.toList());
-    return new GeneratedPartitionsMetadataReport(getId(), partitionStats, taskReport);
+    return new GeneratedPartitionsMetadataReport(getId(), partitionStats, taskReport, segmentsRead);
   }
 
   /**

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialRangeSegmentGenerateTask.java

@@ -192,11 +192,16 @@ SegmentAllocatorForBatch createSegmentAllocator(TaskToolbox toolbox, ParallelInd
   }
 
   @Override
-  GeneratedPartitionsMetadataReport createGeneratedPartitionsReport(TaskToolbox toolbox, List<DataSegment> segments, Map<String, TaskReport> taskReport)
+  GeneratedPartitionsMetadataReport createGeneratedPartitionsReport(
+      TaskToolbox toolbox,
+      List<DataSegment> segments,
+      Map<String, TaskReport> taskReport,
+      Long segmentsRead
+  )
   {
     List<PartitionStat> partitionStats = segments.stream()
                                                         .map(segment -> toolbox.getIntermediaryDataManager().generatePartitionStat(toolbox, segment))
                                                         .collect(Collectors.toList());
-    return new GeneratedPartitionsMetadataReport(getId(), partitionStats, taskReport);
+    return new GeneratedPartitionsMetadataReport(getId(), partitionStats, taskReport, segmentsRead);
   }
 }