diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java index 641412772bf3..852e1fb31c90 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Distinct.java @@ -61,9 +61,6 @@ * * @param the type of the elements of the input and output {@code PCollection}s */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class Distinct extends PTransform, PCollection> { /** @@ -148,10 +145,10 @@ public void processElement( public static class WithRepresentativeValues extends PTransform, PCollection> { private final SerializableFunction fn; - private final TypeDescriptor representativeType; + private final @Nullable TypeDescriptor representativeType; private WithRepresentativeValues( - SerializableFunction fn, TypeDescriptor representativeType) { + SerializableFunction fn, @Nullable TypeDescriptor representativeType) { this.fn = fn; this.representativeType = representativeType; } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Latest.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Latest.java index cd52f05cad5e..dc75ea9d3b0b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Latest.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Latest.java @@ -17,8 +17,8 @@ */ package org.apache.beam.sdk.transforms; +import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; -import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; import java.util.Iterator; @@ -31,6 +31,7 @@ import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.TimestampedValue; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; +import org.checkerframework.checker.nullness.qual.Nullable; /** * {@link PTransform} and {@link Combine.CombineFn} for computing the latest element in a {@link @@ -50,9 +51,6 @@ * *

For elements with the same timestamp, the element chosen for output is arbitrary. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class Latest { // Do not instantiate private Latest() {} @@ -88,25 +86,37 @@ public static PTransform>, PCollection>> pe /** * A {@link Combine.CombineFn} that computes the latest element from a set of inputs. * + *

The accumulator holds a {@literal null} value until the first input arrives, so combining an + * empty input yields {@literal null}. {@code T} must therefore be instantiated at a nullable type + * for the output to be sound, which the declaration cannot say without changing the public + * signature of {@link Latest#combineFn()}. + * * @param Type of input element. * @see Latest */ @VisibleForTesting - static class LatestFn extends Combine.CombineFn, TimestampedValue, T> { + static class LatestFn + extends Combine.CombineFn, TimestampedValue<@Nullable T>, T> { /** Construct a new {@link LatestFn} instance. */ public LatestFn() {} @Override - public TimestampedValue createAccumulator() { + public TimestampedValue<@Nullable T> createAccumulator() { return TimestampedValue.atMinimumTimestamp(null); } @Override - public TimestampedValue addInput( - TimestampedValue accumulator, TimestampedValue input) { - checkNotNull(accumulator, "accumulator must be non-null"); - checkNotNull(input, "input must be non-null"); + public TimestampedValue<@Nullable T> addInput( + TimestampedValue<@Nullable T> accumulator, TimestampedValue input) { + checkArgumentNotNull(accumulator, "accumulator must be non-null"); + checkArgumentNotNull(input, "input must be non-null"); + + return latest(accumulator, input); + } + /** Returns whichever argument is later, preferring a non-null value when they are equal. */ + private static TimestampedValue<@Nullable T> latest( + TimestampedValue<@Nullable T> accumulator, TimestampedValue<@Nullable T> input) { if (input.getTimestamp().isBefore(accumulator.getTimestamp())) { return accumulator; } else if (input.getTimestamp().isAfter(accumulator.getTimestamp())) { @@ -117,13 +127,15 @@ public TimestampedValue addInput( } @Override - public Coder> getAccumulatorCoder( + @SuppressWarnings("nullness") // accumulated values may be null + public Coder> getAccumulatorCoder( CoderRegistry registry, Coder> inputCoder) throws CannotProvideCoderException { return NullableCoder.of(inputCoder); } @Override + @SuppressWarnings("nullness") // the output is null when the input is empty public Coder getDefaultOutputCoder( CoderRegistry registry, Coder> inputCoder) throws CannotProvideCoderException { @@ -138,24 +150,28 @@ public Coder getDefaultOutputCoder( } @Override - public TimestampedValue mergeAccumulators(Iterable> accumulators) { - checkNotNull(accumulators, "accumulators must be non-null"); + public TimestampedValue<@Nullable T> mergeAccumulators( + Iterable> accumulators) { + checkArgumentNotNull(accumulators, "accumulators must be non-null"); - Iterator> iter = accumulators.iterator(); + Iterator> iter = accumulators.iterator(); if (!iter.hasNext()) { return createAccumulator(); } - TimestampedValue merged = iter.next(); + TimestampedValue<@Nullable T> merged = iter.next(); while (iter.hasNext()) { - merged = addInput(merged, iter.next()); + TimestampedValue<@Nullable T> next = iter.next(); + checkArgumentNotNull(next, "input must be non-null"); + merged = latest(merged, next); } return merged; } @Override - public T extractOutput(TimestampedValue accumulator) { + @SuppressWarnings("nullness") // the output is null until the first input arrives + public T extractOutput(TimestampedValue<@Nullable T> accumulator) { return accumulator.getValue(); } } @@ -178,7 +194,7 @@ private static class PerKey extends PTransform>, PCollection>> { @Override public PCollection> expand(PCollection> input) { - checkNotNull(input); + checkArgumentNotNull(input); checkArgument( input.getCoder() instanceof KvCoder, "Input specifiedCoder must be an instance of KvCoder, but was %s", diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reify.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reify.java index da6feef92d65..2c4f16f1227b 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reify.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reify.java @@ -32,6 +32,7 @@ import org.apache.beam.sdk.values.TimestampedValue.TimestampedValueCoder; import org.apache.beam.sdk.values.ValueInSingleWindow; import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; import org.joda.time.Duration; import org.joda.time.Instant; @@ -39,9 +40,6 @@ * {@link PTransform PTransforms} for converting between explicit and implicit form of various Beam * values. */ -@SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) -}) public class Reify { private static class ReifyView extends PTransform, PCollection>> { private final PCollectionView view; @@ -80,7 +78,7 @@ private ReifyViewInGlobalWindow(PCollectionView view, Coder coder) { @Override public PCollection expand(PBegin input) { return input - .apply(Create.of((Void) null).withCoder(VoidCoder.of())) + .apply(Create.<@Nullable Void>of((@Nullable Void) null).withCoder(VoidCoder.of())) .apply(Reify.viewAsValues(view, coder)) .apply(Values.create()); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java index d0017c88f6a4..a838106ff376 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Sample.java @@ -43,10 +43,7 @@ *

{@link #combineFn} can also be used manually, in combination with state and with the {@link * Combine} transform. */ -@SuppressWarnings({ - "nullness", // TODO(https://github.com/apache/beam/issues/20497) - "rawtypes" -}) +@SuppressWarnings({"rawtypes"}) public class Sample { /** Returns a {@link CombineFn} that computes a fixed-sized uniform sample of its inputs. */ @@ -282,6 +279,7 @@ public List mergeAccumulators(Iterable> accumulators) { } @Override + @SuppressWarnings("nullness") // the output is null when the input is empty public T extractOutput(List accumulator) { Iterator it = internal.extractOutput(accumulator).iterator(); return it.hasNext() ? it.next() : null; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TimestampedValue.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TimestampedValue.java index 753ee4c2b9f8..8bcf9ffb0f28 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TimestampedValue.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TimestampedValue.java @@ -31,6 +31,7 @@ import org.apache.beam.sdk.coders.StructuredCoder; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.checkerframework.checker.nullness.qual.Nullable; +import org.checkerframework.framework.qual.Covariant; import org.joda.time.Instant; /** @@ -41,6 +42,8 @@ * * @param the type of the value */ +// Immutable, so the value type may be widened. +@Covariant(0) public class TimestampedValue { /** * Returns a new {@link TimestampedValue} with the {@link BoundedWindow#TIMESTAMP_MIN_VALUE diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java index 3eafbaa7c862..b37f0588438f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/LatestFnTest.java @@ -100,14 +100,14 @@ public void testAddInputSameTimestamp() { @Test public void testAddInputNullAccumulator() { - thrown.expect(NullPointerException.class); + thrown.expect(IllegalArgumentException.class); thrown.expectMessage("accumulator"); fn.addInput(null, TV); } @Test public void testAddInputNullInput() { - thrown.expect(NullPointerException.class); + thrown.expect(IllegalArgumentException.class); thrown.expectMessage("input"); fn.addInput(TV, null); } @@ -151,7 +151,7 @@ public void testMergeAccumulatorsAllDefaultAccumulators() { @Test public void testMergeAccumulatorsNullIterable() { - thrown.expect(NullPointerException.class); + thrown.expect(IllegalArgumentException.class); thrown.expectMessage("accumulators"); fn.mergeAccumulators(null); }