Nullness improvements for Combine, Min, Max

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.transforms;
 
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import java.io.IOException;
@@ -82,9 +83,6 @@
  * href="https://beam.apache.org/documentation/programming-guide/#transforms-combine">documentation</a>
  * for how to use the operations in this class.
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class Combine {
   private Combine() {
     // do not instantiate
@@ -142,7 +140,7 @@ public static <InputT, OutputT> Globally<InputT, OutputT> globally(
     return globally(fn, displayDataForFn(fn));
   }
 
-  private static <T> DisplayData.ItemSpec<? extends Class<?>> displayDataForFn(T fn) {
+  private static DisplayData.ItemSpec<? extends Class<?>> displayDataForFn(Object fn) {
     return DisplayData.item("combineFn", fn.getClass()).withLabel("Combiner");
   }
 
@@ -533,7 +531,7 @@ public Holder<V> createAccumulator() {
     @Override
     public Holder<V> addInput(Holder<V> accumulator, V input) {
       if (accumulator.present) {
-        accumulator.set(apply(accumulator.value, input));
+        accumulator.set(apply(accumulator.get(), input));
       } else {
         accumulator.set(input);
       }
@@ -551,9 +549,9 @@ public Holder<V> mergeAccumulators(Iterable<Holder<V>> accumulators) {
           Holder<V> accum = iter.next();
           if (accum.present) {
             if (running.present) {
-              running.set(apply(running.value, accum.value));
+              running.set(apply(running.get(), accum.get()));
             } else {
-              running.set(accum.value);
+              running.set(accum.get());
             }
           }
         }
@@ -562,9 +560,10 @@ public Holder<V> mergeAccumulators(Iterable<Holder<V>> accumulators) {
     }
 
     @Override
+    @SuppressWarnings("nullness") // identity() is nullable; combining an empty set may yield null
     public V extractOutput(Holder<V> accumulator) {
       if (accumulator.present) {
-        return accumulator.value;
+        return accumulator.get();
       } else {
         return identity();
       }
@@ -593,14 +592,24 @@ public static class Holder<V> {
     private Holder() {}
 
     private Holder(V value) {
-      set(value);
+      this.present = true;
+      this.value = value;
     }
 
     private void set(V value) {
       this.present = true;
       this.value = value;
     }
 
+    /**
+     * Returns the held value, which is meaningful only when {@link #present}. The result may be
+     * null, because {@link BinaryCombineFn} supports null values.
+     */
+    @SuppressWarnings("nullness")
+    private V get() {
+      return value;
+    }
+
     @Override
     public String toString() {
       return "Combine.Holder(value=" + value + ", present=" + present + ")";
@@ -627,7 +636,7 @@ public void encode(Holder<V> accumulator, OutputStream outStream, Coder.Context
         throws CoderException, IOException {
       if (accumulator.present) {
         outStream.write(1);
-        valueCoder.encode(accumulator.value, outStream, context);
+        valueCoder.encode(accumulator.get(), outStream, context);
       } else {
         outStream.write(0);
       }
@@ -1224,7 +1233,9 @@ private PCollection<OutputT> insertDefaultValueIfEmpty(PCollection<OutputT> mayb
       PCollection<OutputT> defaultIfEmpty =
           maybeEmpty
               .getPipeline()
-              .apply("CreateVoid", Create.of((Void) null).withCoder(VoidCoder.of()))
+              .apply(
+                  "CreateVoid",
+                  Create.<@Nullable Void>of((@Nullable Void) null).withCoder(VoidCoder.of()))
               .apply(
                   "ProduceDefault",
                   ParDo.of(
@@ -1324,8 +1335,7 @@ public PCollectionView<OutputT> expand(PCollection<InputT> input) {
       PCollectionView<OutputT> view =
           PCollectionViews.singletonView(
               combined,
-              (TypeDescriptorSupplier<OutputT>)
-                  () -> outputCoder != null ? outputCoder.getEncodedTypeDescriptor() : null,
+              (TypeDescriptorSupplier<OutputT>) () -> outputCoder.getEncodedTypeDescriptor(),
               input.getWindowingStrategy(),
               insertDefault,
               insertDefault ? fn.defaultValue() : null,
@@ -1756,7 +1766,7 @@ public AccumT createAccumulator() {
               @Override
               public AccumT addInput(AccumT accumulator, InputOrAccum<InputT, AccumT> value) {
                 if (value.accum == null) {
-                  return fn.addInput(accumulator, value.input);
+                  return fn.addInput(accumulator, checkStateNotNull(value.input));
                 } else {
                   return fn.mergeAccumulators(ImmutableList.of(accumulator, value.accum));
                 }
@@ -1854,7 +1864,7 @@ public AccumT addInput(
                   InputOrAccum<InputT, AccumT> value,
                   CombineWithContext.Context c) {
                 if (value.accum == null) {
-                  return fnWithContext.addInput(accumulator, value.input, c);
+                  return fnWithContext.addInput(accumulator, checkStateNotNull(value.input), c);
                 } else {
                   return fnWithContext.mergeAccumulators(
                       ImmutableList.of(accumulator, value.accum), c);
@@ -2075,7 +2085,7 @@ public void encode(
             inputCoder.encode(value.input, outStream, context);
           } else {
             outStream.write(1);
-            accumCoder.encode(value.accum, outStream, context);
+            accumCoder.encode(checkStateNotNull(value.accum), outStream, context);
           }
         }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java

@@ -43,9 +43,6 @@
  *     .apply(Max.<String>integersPerKey());
  * }</pre>
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class Max {
 
   private Max() {
@@ -228,7 +225,7 @@ private <ComparatorT extends Comparator<? super T> & Serializable> MaxFn(
     }
 
     @Override
-    public T identity() {
+    public @Nullable T identity() {
       return identity;
     }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java

@@ -43,9 +43,6 @@
  *     .apply(Min.<String>integersPerKey());
  * }</pre>
  */
-@SuppressWarnings({
-  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
 public class Min {
 
   private Min() {
@@ -228,7 +225,7 @@ private <ComparatorT extends Comparator<? super T> & Serializable> MinFn(
     }
 
     @Override
-    public T identity() {
+    public @Nullable T identity() {
       return identity;
     }