From 1ff88c3ab25de1ceec237b44302751615e113dc9 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 4 Sep 2026 19:07:38 +0000 Subject: [PATCH 1/5] Mark TimestampedValue as covariant in its value type TimestampedValue is immutable, so a TimestampedValue is safe to use wherever a TimestampedValue of a supertype of V is expected. Saying so lets the nullness checker accept passing a TimestampedValue where a TimestampedValue<@Nullable T> is wanted, which combiners need when their accumulator starts out holding a null value. This only loosens checking, so it cannot invalidate existing callers. --- .../main/java/org/apache/beam/sdk/values/TimestampedValue.java | 3 +++ 1 file changed, 3 insertions(+) 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 From f8508f18ad6c3d982ab116d923de0ef24d0f7734 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 4 Sep 2026 19:09:35 +0000 Subject: [PATCH 2/5] Nullness improvements for Latest Replace the blanket class-level nullness suppression with narrow, documented suppressions on the methods that genuinely produce a null. LatestFn's accumulator holds a null value until the first input arrives, so it is now typed TimestampedValue<@Nullable T>. That is invisible in the public API, since Latest.combineFn() erases AccumT to a wildcard. It relies on TimestampedValue being covariant in its value type, so that an input can still be passed where an accumulator is expected. No runtime changes beyond hoisting the timestamp comparison into a private helper so that mergeAccumulators no longer routes through addInput. The null-element check it relied on is retained. --- .../apache/beam/sdk/transforms/Latest.java | 52 ++++++++++++------- .../beam/sdk/transforms/LatestFnTest.java | 6 +-- 2 files changed, 37 insertions(+), 21 deletions(-) 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/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); } From 1c9f945fbdaaa64dd57ccc17c03f9aa256794430 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 4 Sep 2026 19:11:31 +0000 Subject: [PATCH 3/5] Nullness improvements for Distinct Remove the blanket class-level nullness suppression. The representative type of WithRepresentativeValues is genuinely optional -- expand() already branches on it -- so annotate it rather than suppressing. No runtime changes. --- .../main/java/org/apache/beam/sdk/transforms/Distinct.java | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) 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; } From 388687bc6e275e82ae371cab6200145a4d1ec708 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 4 Sep 2026 19:13:30 +0000 Subject: [PATCH 4/5] Nullness improvements for Reify Remove the blanket class-level nullness suppression. Reifying a view in the global window creates a single placeholder element, which is a null Void; spell that type out rather than casting. No runtime changes. --- .../src/main/java/org/apache/beam/sdk/transforms/Reify.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) 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()); } From c0ecc803afb8a3223d21bc8ead7e9069142e7ad7 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 4 Sep 2026 19:15:20 +0000 Subject: [PATCH 5/5] Nullness improvements for Sample Remove the blanket class-level nullness suppression, keeping the unrelated rawtypes one. AnyValueCombineFn.extractOutput returns null when the input is empty, and OutputT is T in the public signature of anyValueCombineFn(), so that keeps a narrow suppression. No runtime changes. --- .../main/java/org/apache/beam/sdk/transforms/Sample.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) 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;