[GOBBLIN-2187] Fix NaN issue while generating size summary

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/GenerateWorkUnitsImpl.java

@@ -95,7 +95,11 @@ public WorkUnitsSizeSummary asSizeSummary(int numQuantiles) {
     private static List<Double> getQuantiles(TDigest digest, int numQuantiles) {
       List<Double> quantileMinSizes = Lists.newArrayList();
       for (int i = 1; i <= numQuantiles; i++) {
-        quantileMinSizes.add(digest.quantile((i * 1.0) / numQuantiles));
+        double currQuantileMinSize = digest.quantile((i * 1.0) / numQuantiles);
+        if (Double.isNaN(currQuantileMinSize)) {
+          currQuantileMinSize = 0.0;
+        }
+        quantileMinSizes.add(currQuantileMinSize);
       }
       return quantileMinSizes;
     }

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/work/WorkUnitsSizeSummary.java

@@ -78,12 +78,12 @@ public Distillation distill() {
 
   @JsonIgnore // (because no-arg method resembles 'java bean property')
   public double getTopLevelWorkUnitsMeanSize() {
-    return this.totalSize * 1.0 / this.topLevelWorkUnitsCount;
+    return this.topLevelWorkUnitsCount == 0 ? 0.0 : (this.totalSize * 1.0 / this.topLevelWorkUnitsCount);
   }
 
   @JsonIgnore // (because no-arg method resembles 'java bean property')
   public double getConstituentWorkUnitsMeanSize() {
-    return this.totalSize * 1.0 / this.constituentWorkUnitsCount;
+    return this.constituentWorkUnitsCount == 0 ? 0.0 : (this.totalSize * 1.0 / this.constituentWorkUnitsCount);
   }
 
   @JsonIgnore // (because no-arg method resembles 'java bean property')

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/activity/impl/GenerateWorkUnitsImplTest.java

@@ -174,6 +174,31 @@ public void testDigestWorkUnitsSize() {
             400.0 }); // with only one 20-quantile remaining, non-MWU [5] completes the "100-percentile" (all WUs)
   }
 
+  @Test
+  public void testDigestWorkUnitsSizeWithEmptyWorkUnits() {
+    List<WorkUnit> workUnits = new ArrayList<>();
+    GenerateWorkUnitsImpl.WorkUnitsSizeDigest wuSizeDigest = GenerateWorkUnitsImpl.digestWorkUnitsSize(workUnits);
+
+    Assert.assertEquals(wuSizeDigest.getTotalSize(), 0L);
+    Assert.assertEquals(wuSizeDigest.getTopLevelWorkUnitsSizeDigest().size(), 0);
+    Assert.assertEquals(wuSizeDigest.getConstituentWorkUnitsSizeDigest().size(), 0);
+
+    int numQuantilesDesired = 10;
+    WorkUnitsSizeSummary wuSizeInfo = wuSizeDigest.asSizeSummary(numQuantilesDesired);
+    Assert.assertEquals(wuSizeInfo.getTotalSize(), 0L);
+    Assert.assertEquals(wuSizeInfo.getTopLevelWorkUnitsCount(), 0);
+    Assert.assertEquals(wuSizeInfo.getConstituentWorkUnitsCount(), 0);
+    Assert.assertEquals(wuSizeInfo.getQuantilesCount(), numQuantilesDesired);
+    Assert.assertEquals(wuSizeInfo.getQuantilesWidth(), 1.0 / numQuantilesDesired);
+    Assert.assertEquals(wuSizeInfo.getTopLevelQuantilesMinSizes().size(), numQuantilesDesired); // same as `asSizeInfo` param
+    Assert.assertEquals(wuSizeInfo.getConstituentQuantilesMinSizes().size(), numQuantilesDesired); // same as `asSizeInfo` param
+    Assert.assertEquals(wuSizeInfo.getConstituentWorkUnitsMeanSize(), 0.0);
+    Assert.assertEquals(wuSizeInfo.getTopLevelWorkUnitsMeanSize(), 0.0);
+    Assert.assertEquals(wuSizeInfo.getConstituentWorkUnitsMeanSize(), 0.0);
+    Assert.assertEquals(wuSizeInfo.getTopLevelWorkUnitsMedianSize(), 0.0);
+    Assert.assertEquals(wuSizeInfo.getConstituentWorkUnitsMedianSize(), 0.0);
+  }
+
   public static WorkUnit createWorkUnitOfSize(long size) {
     WorkUnit workUnit = WorkUnit.createEmpty();
     workUnit.setProp(ServiceConfigKeys.WORK_UNIT_SIZE, size);