diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java index e138b32c58fe..ef0961e6d11d 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java +++ b/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 * 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 Globally globally( return globally(fn, displayDataForFn(fn)); } - private static DisplayData.ItemSpec> displayDataForFn(T fn) { + private static DisplayData.ItemSpec> displayDataForFn(Object fn) { return DisplayData.item("combineFn", fn.getClass()).withLabel("Combiner"); } @@ -533,7 +531,7 @@ public Holder createAccumulator() { @Override public Holder addInput(Holder 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 mergeAccumulators(Iterable> accumulators) { Holder 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 mergeAccumulators(Iterable> accumulators) { } @Override + @SuppressWarnings("nullness") // identity() is nullable; combining an empty set may yield null public V extractOutput(Holder accumulator) { if (accumulator.present) { - return accumulator.value; + return accumulator.get(); } else { return identity(); } @@ -593,7 +592,8 @@ public static class Holder { private Holder() {} private Holder(V value) { - set(value); + this.present = true; + this.value = value; } private void set(V value) { @@ -601,6 +601,15 @@ private void set(V value) { 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 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 insertDefaultValueIfEmpty(PCollection mayb PCollection 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 expand(PCollection input) { PCollectionView view = PCollectionViews.singletonView( combined, - (TypeDescriptorSupplier) - () -> outputCoder != null ? outputCoder.getEncodedTypeDescriptor() : null, + (TypeDescriptorSupplier) () -> outputCoder.getEncodedTypeDescriptor(), input.getWindowingStrategy(), insertDefault, insertDefault ? fn.defaultValue() : null, @@ -1756,7 +1766,7 @@ public AccumT createAccumulator() { @Override public AccumT addInput(AccumT accumulator, InputOrAccum 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 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); } } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java index 09bf20997f65..69bd00fe0650 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Max.java @@ -43,9 +43,6 @@ * .apply(Max.integersPerKey()); * } */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class Max { private Max() { @@ -228,7 +225,7 @@ private & Serializable> MaxFn( } @Override - public T identity() { + public @Nullable T identity() { return identity; } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java index a9daed951b61..873429e78f80 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Min.java @@ -43,9 +43,6 @@ * .apply(Min.integersPerKey()); * } */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class Min { private Min() { @@ -228,7 +225,7 @@ private & Serializable> MinFn( } @Override - public T identity() { + public @Nullable T identity() { return identity; }