Skip to content

Commit 4ae92f5

Browse files
committed
Reapply "Refactor: separate SplittableTruncateSizedRestrictions" (#35700)
This reverts commit 800f78b.
1 parent 800f78b commit 4ae92f5

15 files changed

Lines changed: 2699 additions & 1976 deletions

File tree

sdks/java/core/src/main/java/org/apache/beam/sdk/fn/splittabledofn/RestrictionTrackers.java

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,4 +119,25 @@ public static <RestrictionT, PositionT> RestrictionTracker<RestrictionT, Positio
119119
return new RestrictionTrackerObserver<>(restrictionTracker, claimObserver);
120120
}
121121
}
122+
123+
public static <RestrictionT, PositionT> RestrictionTracker<RestrictionT, PositionT> synchronize(
124+
RestrictionTracker<RestrictionT, PositionT> restrictionTracker) {
125+
if (restrictionTracker instanceof RestrictionTracker.HasProgress) {
126+
return new RestrictionTrackerObserverWithProgress<>(
127+
restrictionTracker, (ClaimObserver<PositionT>) NOOP_CLAIM_OBSERVER);
128+
} else {
129+
return new RestrictionTrackerObserver<>(
130+
restrictionTracker, (ClaimObserver<PositionT>) NOOP_CLAIM_OBSERVER);
131+
}
132+
}
133+
134+
static class NoopClaimObserver<PositionT> implements ClaimObserver<PositionT> {
135+
@Override
136+
public void onClaimed(PositionT position) {}
137+
138+
@Override
139+
public void onClaimFailed(PositionT position) {}
140+
}
141+
142+
private static final NoopClaimObserver<Object> NOOP_CLAIM_OBSERVER = new NoopClaimObserver<>();
122143
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,6 +211,6 @@ public static <RestrictionT> TruncateResult of(RestrictionT restriction) {
211211
return new AutoValue_RestrictionTracker_TruncateResult(restriction);
212212
}
213213

214-
public abstract @Nullable RestrictionT getTruncatedRestriction();
214+
public abstract RestrictionT getTruncatedRestriction();
215215
}
216216
}
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.sdk.util;
19+
20+
import org.apache.beam.sdk.annotations.Internal;
21+
22+
/**
23+
* A trivial boxing of a value, used when nullability needs to be added to a generic type. (Optional
24+
* does not work for this)
25+
*
26+
* <p>Example: For a generic type `T` the actual parameter may be nullable or not. So you cannot
27+
* check values for null to determine presence/absence. Instead you can store a {@code @Nullable
28+
* Holder<T>}.
29+
*/
30+
@Internal
31+
public class Holder<T> {
32+
private T value;
33+
34+
private Holder(T value) {
35+
this.value = value;
36+
}
37+
38+
public static <ValueT> Holder<ValueT> of(ValueT value) {
39+
return new Holder<>(value);
40+
}
41+
42+
public T get() {
43+
return value;
44+
};
45+
}

sdks/java/core/src/main/java/org/apache/beam/sdk/values/KV.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
*/
4141
public class KV<K, V> implements Serializable {
4242
/** Returns a {@link KV} with the given key and value. */
43+
@Pure
4344
public static <K, V> KV<K, V> of(K key, V value) {
4445
return new KV<>(key, value);
4546
}

sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValue.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import java.util.Collection;
2121
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
2222
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
23+
import org.checkerframework.dataflow.qual.Pure;
2324
import org.joda.time.Instant;
2425

2526
/**
@@ -29,26 +30,32 @@
2930
*/
3031
public interface WindowedValue<T> {
3132
/** The primary data for this value. */
33+
@Pure
3234
T getValue();
3335

3436
/** The timestamp of this value in event time. */
37+
@Pure
3538
Instant getTimestamp();
3639

3740
/** Returns the windows of this {@code WindowedValue}. */
41+
@Pure
3842
Collection<? extends BoundedWindow> getWindows();
3943

4044
/** The {@link PaneInfo} associated with this WindowedValue. */
45+
@Pure
4146
PaneInfo getPaneInfo();
4247

4348
/**
4449
* A representation of each of the actual values represented by this compressed {@link
4550
* WindowedValue}, one per window.
4651
*/
52+
@Pure
4753
Iterable<WindowedValue<T>> explodeWindows();
4854

4955
/**
5056
* A {@link WindowedValue} with identical metadata to the current one, but with the provided
5157
* value.
5258
*/
59+
@Pure
5360
<OtherT> WindowedValue<OtherT> withValue(OtherT value);
5461
}

0 commit comments

Comments
 (0)