You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/05/15 00:53:05 UTC
[GitHub] [beam] boyuanzz opened a new pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
boyuanzz opened a new pull request #11715:
URL: https://github.com/apache/beam/pull/11715
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r425995961
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ long cur = (lastAttemptedOffset == null) ? range.getFrom() - 1 : lastAttemptedOffset;
+
+ // If current range has been done, there is no more space to split.
+ if (cur == Long.MAX_VALUE) {
+ return null;
+ }
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ long estimateLatestOffset = Long.max(poller.estimateRangeEnd(), cur + 1);
+ long splitPos =
+ cur
+ + Math.max(
+ 1L,
+ (Double.valueOf((estimateLatestOffset - cur) * fractionOfRemainder)).longValue());
+
+ if (splitPos > estimateLatestOffset) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(splitPos, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), splitPos);
+ return SplitResult.of(range, res);
Review comment:
btw we will also loss `lastAttemptedOffset` by creating a new tracker.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-632402839
Now that this is merged. Can I ask a question. What is the intrinsic limitation that did not allow old `OffsetRangeTracker` to be refactored for this use case? or why we want to have both?
Does this mean also that we might need `GrowableBytekeyRangeTracker` and basically 'dynamic' versions for every `RestrictionTracker` ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-631204566
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r428343752
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Suppliers;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as the end of the range to indicate infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * Provides the estimated end offset of the range.
+ *
+ * <p>{@link #estimate} is called to give the end offset when {@link #trySplit} or {@link
+ * #getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * required to monotonically increase as it will only be taken into consideration when the
+ * estimated end offset is larger than the current position. Returning {@code Long.MAX_VALUE} as
+ * the estimate implies the largest possible position for the range is {@code Long.MAX_VALUE - 1}.
+ * Return {@code Long.MIN_VALUE} if an estimate can not be provided.
+ *
+ * <p>Providing a good estimate is important for an accurate progress signal and will impact
+ * splitting decisions by the runner.
+ *
+ * <p>If {@link #estimate} is expensive to compute, consider wrapping the implementation with
+ * {@link Suppliers#memoizeWithExpiration} or equivalent as an optimization.
+ *
+ * <p>TODO(BEAM-10032): Also consider using {@link RangeEndEstimator} when the range is not ended
+ * with {@code Long.MAX_VALUE}.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ BigDecimal cur =
+ (lastAttemptedOffset == null)
+ ? BigDecimal.valueOf(range.getFrom()).subtract(BigDecimal.ONE, MathContext.DECIMAL128)
+ : BigDecimal.valueOf(lastAttemptedOffset);
+
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ BigDecimal estimateRangeEnd =
+ BigDecimal.valueOf(rangeEndEstimator.estimate())
+ .max(cur.add(BigDecimal.ONE, MathContext.DECIMAL128));
+
+ // Convert to BigDecimal in computation to prevent overflow, which may result in loss of
+ // precision.
+ // split = cur + max(1, (estimateRangeEnd - cur) * fractionOfRemainder)
+ BigDecimal splitPos =
+ cur.add(
+ estimateRangeEnd
+ .subtract(cur, MathContext.DECIMAL128)
+ .multiply(BigDecimal.valueOf(fractionOfRemainder), MathContext.DECIMAL128)
+ .max(BigDecimal.ONE),
+ MathContext.DECIMAL128);
+ long split = splitPos.longValue();
+ if (split > estimateRangeEnd.longValue()) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(split, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), split);
+ return SplitResult.of(range, res);
+ }
+
+ @Override
+ public Progress getProgress() {
+ // If current tracking range is no longer growable, get progress as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.getProgress();
+ }
+
+ // Convert to BigDecimal in computation to prevent overflow, which may result in lost of
+ // precision.
+ BigDecimal estimateRangeEnd = BigDecimal.valueOf(rangeEndEstimator.estimate());
+
+ if (lastAttemptedOffset == null) {
+ return Progress.from(
+ 0,
+ estimateRangeEnd
+ .subtract(BigDecimal.valueOf(range.getFrom()), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO)
+ .doubleValue());
+ }
+
+ BigDecimal workRemaining =
+ estimateRangeEnd
+ .subtract(BigDecimal.valueOf(lastAttemptedOffset), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO);
+ BigDecimal totalWork =
+ estimateRangeEnd
+ .max(BigDecimal.valueOf(lastAttemptedOffset))
+ .subtract(BigDecimal.valueOf(range.getFrom()));
Review comment:
It's needed as well.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-629344196
R: @lukecwik
CC: @robertwb
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r425553418
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
Review comment:
Currently I take `Long.MAX_VALUE` as a numeric end offset. But we may also need a notion to say that the `OffsetPoller` doesn't have a good estimate and still want to keep current range as infinite. `Long.MAX_VALUE` is not suitable because it's possible that the actual end is `Long.MAX_VALUE`. Do we want to provide a notion here? Like `null`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r427431140
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
Review comment:
We should only use doubles if there would be an issue with overflow. For really large long values we can have the math be incorrect which can lead to the splitPos being before `from` or after `to` even though with higher precision we would get something that works. For example:
```
long a = 123456789012345677L;
long b = 123456789012345679L;
long c = 123456789012345680L;
double ad = a;
double bd = b;
double cd = c;
System.out.println((long)ad + " " + (long)bd + " " + (long)cd);
```
prints
```
123456789012345680 123456789012345680 123456789012345680
```
Worthwhile to have a test which exercises a case where `from = 123456789012345677L` and `to = 123456789012345679L` and we can have a split in between. Also worthwhile to test `from = 123456789012345681L` and `to = 123456789012345683L`. All these values (and the values in between) when converted to double and back to long round to `123456789012345680L`.
We can solve this by having two ways to calculate this or using BigDecimal. The two ways to calculate method would use longs when (estimateRangeEnd - cur) won't overflow and doubles otherwise. Using BigDecimal with a MathContext with enough precision wouldn't run into this either and would only require one implemention (I believe `DECIMAL128` should have enough precision to not run into this issue).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r425885914
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
Review comment:
```suggestion
@FunctionalInterface
public interface RangeEndEstimator {
long estimate();
}
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
Review comment:
nit: Doesn't super.trySplit handle the to == from case already?
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
Review comment:
The user should return Long.MIN_VALUE and then we will use the current position and the estimate will not take into effect.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
Review comment:
We might want to suggest people wrap their estimator with https://guava.dev/releases/19.0/api/docs/com/google/common/base/Suppliers.html#memoizeWithExpiration(com.google.common.base.Supplier,%20long,%20java.util.concurrent.TimeUnit) if calling the estimator is expensive.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java
##########
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker.Progress;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GrowableOffsetRangeTracker}. */
+@RunWith(JUnit4.class)
+public class GrowableOffsetRangeTrackerTest {
+ private static class SimplePoller implements GrowableOffsetRangeTracker.OffsetPoller {
+ private long estimateRangeEnd = 0;
+
+ @Override
+ public long estimateRangeEnd() {
+ return estimateRangeEnd;
+ }
+
+ public void setEstimateRangeEnd(long offset) {
+ estimateRangeEnd = offset;
+ }
+ }
+
+ @Rule public final ExpectedException expected = ExpectedException.none();
+
+ @Test
+ public void testIllegalInitialization() throws Exception {
+ expected.expect(NullPointerException.class);
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, null);
+ }
+
+ @Test
+ public void testTryClaim() throws Exception {
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, new SimplePoller());
+ assertTrue(tracker.tryClaim(10L));
+ assertTrue(tracker.tryClaim(100L));
+ assertFalse(tracker.tryClaim(Long.MAX_VALUE));
+ tracker.checkDone();
+ }
+
+ @Test
+ public void testCheckpointBeforeStart() throws Exception {
+ SimplePoller poller = new SimplePoller();
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, poller);
+ poller.setEstimateRangeEnd(10);
+ SplitResult res = tracker.trySplit(0);
+ tracker.checkDone();
+ assertEquals(new OffsetRange(0, 0), res.getPrimary());
+ assertEquals(new OffsetRange(0, 0), tracker.currentRestriction());
+ assertEquals(new OffsetRange(0, Long.MAX_VALUE), res.getResidual());
+ }
+
+ @Test
+ public void testCheckpointJustStar() throws Exception {
Review comment:
```suggestion
public void testCheckpointJustStarted() throws Exception {
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
Review comment:
nit: poller -> estimator
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
Review comment:
It might be worthwhile to add this comment to OffsetRangeTracker that the largest position is Long.MAX_VALUE - 1.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
Review comment:
```suggestion
* <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
* getProgress} is invoked. The end offset is exclusive for the range. It is not necessary to
* increase monotonically as its only taken into consideration when it is larger than the current
* position. When returning {@code Long.MAX_VALUE} as estimate, it means the largest possible position for
* the range is {@code Long.MAX_VALUE - 1}. Having a good estimate is important for providing a good
* signal of progress and splitting at a proper position.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
Review comment:
```suggestion
* An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
* used as end range to indicate the possibility of infinity.
*
* <p>An offset range is considered growable when the end offset could grow (or change) during
* execution time (e.g., Kafka topic partition offset, appended file, ...).
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
Review comment:
We should mention in this class comment that the user marks the growable offset range tracker as done by claiming Long.MAX_VALUE when they detect "end of stream"
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ long cur = (lastAttemptedOffset == null) ? range.getFrom() - 1 : lastAttemptedOffset;
+
+ // If current range has been done, there is no more space to split.
+ if (cur == Long.MAX_VALUE) {
+ return null;
+ }
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ long estimateLatestOffset = Long.max(poller.estimateRangeEnd(), cur + 1);
+ long splitPos =
+ cur
+ + Math.max(
+ 1L,
+ (Double.valueOf((estimateLatestOffset - cur) * fractionOfRemainder)).longValue());
+
+ if (splitPos > estimateLatestOffset) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(splitPos, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), splitPos);
+ return SplitResult.of(range, res);
Review comment:
Why not construct a new OffsetRangeTracker over `[cur, estimateLatestOffset)` and call trySplit on it?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r428343813
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java
##########
@@ -120,13 +136,27 @@ public String toString() {
public Progress getProgress() {
// If we have never attempted an offset, we return the length of the entire range as work
// remaining.
+ // Convert to BigDecimal in computation to prevent overflow, which may result in loss of
+ // precision.
if (lastAttemptedOffset == null) {
- return Progress.from(0, range.getTo() - range.getFrom());
+ return Progress.from(
+ 0,
+ BigDecimal.valueOf(range.getTo())
+ .subtract(BigDecimal.valueOf(range.getFrom()), MathContext.DECIMAL128)
+ .doubleValue());
}
// Compute the amount of work remaining from where we are to where we are attempting to get to
// with a minimum of zero in case we have claimed beyond the end of the range.
- long workRemaining = Math.max(range.getTo() - lastAttemptedOffset, 0);
- return Progress.from(range.getTo() - range.getFrom() - workRemaining, workRemaining);
+ BigDecimal workRemaining =
+ BigDecimal.valueOf(range.getTo())
+ .subtract(BigDecimal.valueOf(lastAttemptedOffset), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO);
+ BigDecimal wholeWork =
Review comment:
Done. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-632434300
Thanks for answering and for the clear explaination @boyuanzz
I would have tended towards having the additional complexity in `OffsetRangeTracker` just because it is the de-facto reference, but I understand the different preference.
Nice to see this an the Kafka SDF happening, congrats!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-631794981
Thanks for your help!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r427654115
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
Review comment:
Using `BigDecimal` in the latest revision. Thanks for your help!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630952259
> Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
>
> Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
I can see the benefit of using `RangeEndEstimator` for the finite range here. But as long as we don't modify the range end to estimate end or use estimate ed in `tryClaim`, we still cannot say the residual is empty.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r428304723
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Suppliers;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as the end of the range to indicate infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * Provides the estimated end offset of the range.
+ *
+ * <p>{@link #estimate} is called to give the end offset when {@link #trySplit} or {@link
+ * #getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * required to monotonically increase as it will only be taken into consideration when the
+ * estimated end offset is larger than the current position. Returning {@code Long.MAX_VALUE} as
+ * the estimate implies the largest possible position for the range is {@code Long.MAX_VALUE - 1}.
+ * Return {@code Long.MIN_VALUE} if an estimate can not be provided.
+ *
+ * <p>Providing a good estimate is important for an accurate progress signal and will impact
+ * splitting decisions by the runner.
+ *
+ * <p>If {@link #estimate} is expensive to compute, consider wrapping the implementation with
+ * {@link Suppliers#memoizeWithExpiration} or equivalent as an optimization.
+ *
+ * <p>TODO(BEAM-10032): Also consider using {@link RangeEndEstimator} when the range is not ended
+ * with {@code Long.MAX_VALUE}.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ BigDecimal cur =
+ (lastAttemptedOffset == null)
+ ? BigDecimal.valueOf(range.getFrom()).subtract(BigDecimal.ONE, MathContext.DECIMAL128)
+ : BigDecimal.valueOf(lastAttemptedOffset);
+
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ BigDecimal estimateRangeEnd =
+ BigDecimal.valueOf(rangeEndEstimator.estimate())
+ .max(cur.add(BigDecimal.ONE, MathContext.DECIMAL128));
+
+ // Convert to BigDecimal in computation to prevent overflow, which may result in loss of
+ // precision.
+ // split = cur + max(1, (estimateRangeEnd - cur) * fractionOfRemainder)
+ BigDecimal splitPos =
+ cur.add(
+ estimateRangeEnd
+ .subtract(cur, MathContext.DECIMAL128)
+ .multiply(BigDecimal.valueOf(fractionOfRemainder), MathContext.DECIMAL128)
+ .max(BigDecimal.ONE),
+ MathContext.DECIMAL128);
+ long split = splitPos.longValue();
+ if (split > estimateRangeEnd.longValue()) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(split, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), split);
+ return SplitResult.of(range, res);
+ }
+
+ @Override
+ public Progress getProgress() {
+ // If current tracking range is no longer growable, get progress as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.getProgress();
+ }
+
+ // Convert to BigDecimal in computation to prevent overflow, which may result in lost of
+ // precision.
+ BigDecimal estimateRangeEnd = BigDecimal.valueOf(rangeEndEstimator.estimate());
+
+ if (lastAttemptedOffset == null) {
+ return Progress.from(
+ 0,
+ estimateRangeEnd
+ .subtract(BigDecimal.valueOf(range.getFrom()), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO)
+ .doubleValue());
+ }
+
+ BigDecimal workRemaining =
+ estimateRangeEnd
+ .subtract(BigDecimal.valueOf(lastAttemptedOffset), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO);
+ BigDecimal totalWork =
+ estimateRangeEnd
+ .max(BigDecimal.valueOf(lastAttemptedOffset))
+ .subtract(BigDecimal.valueOf(range.getFrom()));
Review comment:
Doesn't this subtract need the DECIMAL128 math context?
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java
##########
@@ -120,13 +136,27 @@ public String toString() {
public Progress getProgress() {
// If we have never attempted an offset, we return the length of the entire range as work
// remaining.
+ // Convert to BigDecimal in computation to prevent overflow, which may result in loss of
+ // precision.
if (lastAttemptedOffset == null) {
- return Progress.from(0, range.getTo() - range.getFrom());
+ return Progress.from(
+ 0,
+ BigDecimal.valueOf(range.getTo())
+ .subtract(BigDecimal.valueOf(range.getFrom()), MathContext.DECIMAL128)
+ .doubleValue());
}
// Compute the amount of work remaining from where we are to where we are attempting to get to
// with a minimum of zero in case we have claimed beyond the end of the range.
- long workRemaining = Math.max(range.getTo() - lastAttemptedOffset, 0);
- return Progress.from(range.getTo() - range.getFrom() - workRemaining, workRemaining);
+ BigDecimal workRemaining =
+ BigDecimal.valueOf(range.getTo())
+ .subtract(BigDecimal.valueOf(lastAttemptedOffset), MathContext.DECIMAL128)
+ .max(BigDecimal.ZERO);
+ BigDecimal wholeWork =
Review comment:
nit: `wholeWork` -> `totalWork`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-631722098
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-632408026
> What is the intrinsic limitation that did not allow old `OffsetRangeTracker` to be refactored for this use case? or why we want to have both?
>
`GrowableOffsetRangeTracker` and `OffsetRangeTracker` should be applied to different kind of `OffsetRange`. `GrowableOffsetRangeTracker` is for the `OffsetRange` that the end could be changed during execution time, which mostly happens in streaming case. `OffsetRangeTracker` is for the range that we know what the exact end is, which is perfect for batch. The reason that I didn't make them into one is because I don't want to introduce additional complexity to `OffsetRangeTracker`. It's doable to have the dynamic one as general case where the range with fixed end is a special case. But I want to make them specifically with less confusion.
> Does this mean also that we might need `GrowableBytekeyRangeTracker` and basically 'dynamic' versions for every `RestrictionTracker` ?
I think it will depend on the actual usage. If we have an application scenario that requires for a dynamic version, then I would say yes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630965824
> > > > Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
> > > > Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
> > >
> > >
> > > I can see the benefit of using `RangeEndEstimator` for the finite range here. But as long as we don't modify the range end to estimate end or use estimate ed in `tryClaim`, we still cannot say the residual is empty.
> >
> >
> > That is true but I was thinking it would make better splitting decisions instead of creating a bunch of empty splits trimming the range down. The advantage of not using the estimator is that we don't have to invoke since it could be expensive for the user and in many situations will produce a value greater than `to`.
> > We can leave it out for now unless some compelling use case comes up.
>
> Do we want to have a TODO here to track this?
Sure.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r427431140
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
Review comment:
We should only use doubles if there would be an issue with overflow. For really large long values we can have the math be incorrect which can lead to the splitPos being before from or after to even though with higher precision we would get something that works. For example:
```
long a = 123456789012345677L;
long b = 123456789012345679L;
long c = 123456789012345680L;
double ad = a;
double bd = b;
double cd = c;
System.out.println((long)ad + " " + (long)bd + " " + (long)cd);
```
prints
```
123456789012345680 123456789012345680 123456789012345680
```
Worthwhile to have a test which exercises a case where `from = 123456789012345677L` and `to = 123456789012345679L` and we can have a split in between. Also worthwhile to test `from = 123456789012345681L` and `to = 123456789012345683L`. All these values (and the values in between) when converted to double and back to long round to `123456789012345680L`.
We can solve this by having two ways to calculate this or using BigDecimal. The two ways to calculate method would use longs when (estimateRangeEnd - cur) won't overflow and doubles otherwise. Using BigDecimal with a MathContext with enough precision wouldn't run into this either and would only require one implemention (I believe `DECIMAL128` should have enough precision to not run into this issue).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r427400011
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker.Progress;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GrowableOffsetRangeTracker}. */
+@RunWith(JUnit4.class)
+public class GrowableOffsetRangeTrackerTest {
+ private static class SimplePoller implements GrowableOffsetRangeTracker.RangeEndEstimator {
+ private long estimateRangeEnd = 0;
+
+ @Override
+ public long estimate() {
+ return estimateRangeEnd;
+ }
+
+ public void setEstimateRangeEnd(long offset) {
+ estimateRangeEnd = offset;
+ }
+ }
+
+ @Rule public final ExpectedException expected = ExpectedException.none();
+
+ @Test
+ public void testIllegalInitialization() throws Exception {
+ expected.expect(NullPointerException.class);
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, null);
+ }
+
+ @Test
+ public void testTryClaim() throws Exception {
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, new SimplePoller());
+ assertTrue(tracker.tryClaim(10L));
+ assertTrue(tracker.tryClaim(100L));
+ assertFalse(tracker.tryClaim(Long.MAX_VALUE));
+ tracker.checkDone();
+ }
+
+ @Test
+ public void testCheckpointBeforeStart() throws Exception {
+ SimplePoller poller = new SimplePoller();
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, poller);
+ poller.setEstimateRangeEnd(10);
+ SplitResult res = tracker.trySplit(0);
+ tracker.checkDone();
+ assertEquals(new OffsetRange(0, 0), res.getPrimary());
+ assertEquals(new OffsetRange(0, 0), tracker.currentRestriction());
+ assertEquals(new OffsetRange(0, Long.MAX_VALUE), res.getResidual());
+ }
+
+ @Test
+ public void testCheckpointJustStarted() throws Exception {
+ SimplePoller poller = new SimplePoller();
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, poller);
+ assertTrue(tracker.tryClaim(5L));
+ poller.setEstimateRangeEnd(0L);
+ SplitResult res = tracker.trySplit(0);
+ tracker.checkDone();
+ assertEquals(new OffsetRange(0, 6), res.getPrimary());
+ assertEquals(new OffsetRange(0, 6), tracker.currentRestriction());
+ assertEquals(new OffsetRange(6, Long.MAX_VALUE), res.getResidual());
+
+ tracker = new GrowableOffsetRangeTracker(0L, poller);
+ assertTrue(tracker.tryClaim(5L));
+ poller.setEstimateRangeEnd(20L);
+ res = tracker.trySplit(0);
+ tracker.checkDone();
+ assertEquals(new OffsetRange(0, 6), res.getPrimary());
+ assertEquals(new OffsetRange(6, Long.MAX_VALUE), res.getResidual());
+ }
+
+ @Test
+ public void testCheckpointAfterAllProcessed() throws Exception {
+ SimplePoller poller = new SimplePoller();
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, poller);
+ assertFalse(tracker.tryClaim(Long.MAX_VALUE));
+ tracker.checkDone();
+ assertNull(tracker.trySplit(0));
+ }
+
+ @Test
+ public void testCheckpointAtEmptyRange() throws Exception {
+ GrowableOffsetRangeTracker tracker =
+ new GrowableOffsetRangeTracker(Long.MAX_VALUE, new SimplePoller());
+ tracker.checkDone();
+ assertNull(tracker.trySplit(0));
+ }
+
+ @Test
+ public void testSplit() throws Exception {
+ SimplePoller poller = new SimplePoller();
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, poller);
+ assertTrue(tracker.tryClaim(0L));
+
+ poller.setEstimateRangeEnd(16L);
+ // The split of infinite range results in one finite range and on infinite range.
Review comment:
```suggestion
// The split of infinite range results in one finite range and one infinite range.
```
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker.Progress;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GrowableOffsetRangeTracker}. */
+@RunWith(JUnit4.class)
+public class GrowableOffsetRangeTrackerTest {
+ private static class SimplePoller implements GrowableOffsetRangeTracker.RangeEndEstimator {
Review comment:
nit: SimplePoller -> SimpleEstimator
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.beam.sdk.io.range.OffsetRange;
+import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker.Progress;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link GrowableOffsetRangeTracker}. */
+@RunWith(JUnit4.class)
+public class GrowableOffsetRangeTrackerTest {
+ private static class SimplePoller implements GrowableOffsetRangeTracker.RangeEndEstimator {
+ private long estimateRangeEnd = 0;
+
+ @Override
+ public long estimate() {
+ return estimateRangeEnd;
+ }
+
+ public void setEstimateRangeEnd(long offset) {
+ estimateRangeEnd = offset;
+ }
+ }
+
+ @Rule public final ExpectedException expected = ExpectedException.none();
+
+ @Test
+ public void testIllegalInitialization() throws Exception {
+ expected.expect(NullPointerException.class);
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, null);
+ }
+
+ @Test
+ public void testTryClaim() throws Exception {
+ GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, new SimplePoller());
+ assertTrue(tracker.tryClaim(10L));
+ assertTrue(tracker.tryClaim(100L));
+ assertFalse(tracker.tryClaim(Long.MAX_VALUE));
+ tracker.checkDone();
+ }
+
+ @Test
+ public void testCheckpointBeforeStart() throws Exception {
+ SimplePoller poller = new SimplePoller();
Review comment:
nit: poller -> estimator
here and elsewhere in this test
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
Review comment:
```suggestion
* Provides the estimated end offset of the range.
*
* <p>{@link #estimate} is called to give the end offset when {@link #trySplit} or {@link
* #getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
* required to monotonically increase as it will only be taken into consideration when the estimated
* end offset is larger than the current position. Returning {@code Long.MAX_VALUE} as the estimate
* implies the largest possible position for the range is {@code Long.MAX_VALUE - 1}. Return {@code Long.MIN_VALUE} if an estimate can not be provided.
*
* <p>Providing a good estimate is important for an accurate progress signal and will impact splitting decisions by
* the runner.
*
* <p>If {@link #estimate} is expensive to compute, consider wrapping the implementation with
* {@link Suppliers.memoizeWithExpiration} or equivalent as an optimization.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
+ (lastAttemptedOffset == null) ? (double) range.getFrom() - 1 : (double) lastAttemptedOffset;
+
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ double estimateRangeEnd = Math.max(rangeEndEstimator.estimate(), cur + 1);
+
+ // Convert to double in computation to prevent overflow, which may result in lost of some
+ // accuracy.
+ double splitPos = cur + Math.max(1L, (estimateRangeEnd - cur) * fractionOfRemainder);
+ long split = Double.valueOf(splitPos).longValue();
+ if (split > Double.valueOf(estimateRangeEnd).longValue()) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(split, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), split);
+ return SplitResult.of(range, res);
+ }
+
+ @Override
+ public Progress getProgress() {
+ // If current tracking range is no longer growable, get progress as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.getProgress();
+ }
+
+ // Convert to double in computation to prevent overflow, which may result in lost of some
+ // accuracy.
Review comment:
```suggestion
// Convert to double in computation to prevent overflow, which may result in lost of
// precision.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
Review comment:
```suggestion
* used as the end of the range to indicate infinity.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
Review comment:
Sorry for the confusion, I misread the if statement earlier.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java
##########
@@ -50,16 +53,17 @@ public OffsetRange currentRestriction() {
@Override
public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
- long cur = (lastAttemptedOffset == null) ? range.getFrom() - 1 : lastAttemptedOffset;
- long splitPos =
- cur
- + Math.max(
- 1L, (Double.valueOf((range.getTo() - cur) * fractionOfRemainder)).longValue());
- if (splitPos >= range.getTo()) {
+ // Convert to double in computation to prevent overflow, which may result in lost of some
+ // accuracy.
Review comment:
```suggestion
// Convert to double in computation to prevent overflow, which may result in loss of
// precision.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
+ (lastAttemptedOffset == null) ? (double) range.getFrom() - 1 : (double) lastAttemptedOffset;
+
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ double estimateRangeEnd = Math.max(rangeEndEstimator.estimate(), cur + 1);
+
+ // Convert to double in computation to prevent overflow, which may result in lost of some
+ // accuracy.
Review comment:
```suggestion
// Convert to double in computation to prevent overflow, which may result in loss of
// precision.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * An {@link OffsetRangeTracker} for tracking a growable offset range. {@code Long.MAX_VALUE} is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>An offset range is considered growable when the end offset could grow (or change) during
+ * execution time (e.g., Kafka topic partition offset, appended file, ...).
+ *
+ * <p>The growable range is marked as done by claiming {@code Long.MAX_VALUE}.
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of the range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give the end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. The estimated end is not
+ * necessary to increase monotonically as it will only be taken into computation when the estimate
+ * end is larger than the current position. When returning {@code Long.MAX_VALUE} as estimate, it
+ * means the largest possible position for the range is {@code Long.MAX_VALUE - 1}. If there is
+ * not an estimate yet, {@code Long.MIN_VALUE} should be returned, where estimated end will not
+ * effect progress and split.
+ *
+ * <p>Having a good estimate is important for providing a good signal of progress and splitting at
+ * a proper position.
+ *
+ * <p>If {@code estimate()} is expensive to call, please consider wrapping the implementation with
+ * {@code Suppliers.memoizeWithExpiration} as an optimization.
+ */
+ @FunctionalInterface
+ public interface RangeEndEstimator {
+ long estimate();
+ }
+
+ private final RangeEndEstimator rangeEndEstimator;
+
+ public GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.rangeEndEstimator = checkNotNull(rangeEndEstimator);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ // If current range has been done, there is no more space to split.
+ if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) {
+ return null;
+ }
+ double cur =
Review comment:
We should only use doubles if there would be an issue with overflow. For really large long values we can have the math be incorrect which can lead to the splitPos being before from or after to even though with higher precision we would get something that works. For example:
```
long a = 123456789012345677L;
long b = 123456789012345679L;
long c = 123456789012345680L;
double ad = a;
double bd = b;
double cd = c;
System.out.println((long)ad + " " + (long)bd + " " + (long)cd);
```
prints
```
123456789012345680 123456789012345680 123456789012345680
```
Worthwhile to have a test which exercises a case where `from = 123456789012345677L` and `to = 123456789012345679L` and we can have a split in between. Also worthwhile to test `from = 123456789012345681L` and `to = 123456789012345683L`. All these values (and the values in between) when converted to double and back to long round to `123456789012345680L`.
We can solve this by having two ways to calculate this or using BigDecimal. The two ways to calculate method would use longs when (estimateRangeEnd - cur) won't overflow and doubles otherwise. Using BigDecimal with a MathContext with enough precision wouldn't run into this either and would only require one implemention (I believe DECIMAL128 should have enough precision to not run into this issue).
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/OffsetRangeTracker.java
##########
@@ -120,13 +124,16 @@ public String toString() {
public Progress getProgress() {
// If we have never attempted an offset, we return the length of the entire range as work
// remaining.
+ // Convert to double in computation to prevent overflow, which may result in lost of some
+ // accuracy.
Review comment:
```suggestion
// Convert to double in computation to prevent overflow, which may result in loss of
// precision.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630536199
Latest changes are for addressing comments and using double during computation. @lukecwik PTAL. Thanks for your help!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630934841
Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on a change in pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #11715:
URL: https://github.com/apache/beam/pull/11715#discussion_r425986350
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
+ return super.trySplit(fractionOfRemainder);
+ }
+ long cur = (lastAttemptedOffset == null) ? range.getFrom() - 1 : lastAttemptedOffset;
+
+ // If current range has been done, there is no more space to split.
+ if (cur == Long.MAX_VALUE) {
+ return null;
+ }
+ // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
+ // the next offset as end.
+ long estimateLatestOffset = Long.max(poller.estimateRangeEnd(), cur + 1);
+ long splitPos =
+ cur
+ + Math.max(
+ 1L,
+ (Double.valueOf((estimateLatestOffset - cur) * fractionOfRemainder)).longValue());
+
+ if (splitPos > estimateLatestOffset) {
+ return null;
+ }
+ OffsetRange res = new OffsetRange(splitPos, range.getTo());
+ this.range = new OffsetRange(range.getFrom(), splitPos);
+ return SplitResult.of(range, res);
Review comment:
There is a slight difference around comparison of `splitPos` and `end`. `GrowableOffsetTracker` allows split when `splitPos <= end` but `OffsetRangeTracker` allows split when `splitPos < end`. The purpose of `OffsetRangeTracker` is to reduce resuming from an empty range. The purpose of `GrowableOffsetRangeTracker` is to handle split at `cur + 1`.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTracker.java
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.transforms.splittabledofn;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+import org.apache.beam.sdk.io.range.OffsetRange;
+
+/**
+ * A special {@link OffsetRangeTracker} for tracking a growable offset range. The Long.MAX_VALUE is
+ * used as end range to indicate the possibility of infinity.
+ *
+ * <p>A offset range is considered as growable when the end offset could grow(or change) during
+ * execution time(e.g., Kafka backlog, appended file).
+ */
+@Experimental(Kind.SPLITTABLE_DO_FN)
+public class GrowableOffsetRangeTracker extends OffsetRangeTracker {
+ /**
+ * An interface that should be implemented to fetch estimated end offset of range.
+ *
+ * <p>{@code estimateRangeEnd} is called to give te end offset when {@code trySplit} or {@code
+ * getProgress} is invoked. The end offset is exclusive for the range. It's not necessary to
+ * increase monotonically but it's only taken into computation when it's larger than the current
+ * position. When returning Long.MAX_VALUE as estimate, it means the largest possible position for
+ * the range is Long.MAX_VALUE - 1. Having a good estimate is important for providing a good
+ * signal of progress and splitting at a proper position.
+ */
+ public interface OffsetPoller {
+ long estimateRangeEnd();
+ }
+
+ private final OffsetPoller poller;
+
+ public GrowableOffsetRangeTracker(long start, OffsetPoller offsetPoller) {
+ super(new OffsetRange(start, Long.MAX_VALUE));
+ this.poller = checkNotNull(offsetPoller);
+ }
+
+ @Override
+ public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
+ // If current tracking range is no longer growable, split it as a normal range.
+ if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
Review comment:
Yes `super.trySplit()` handles empty range. The special case here is to handle `range.getTo() == range.getFrom == Long.MAX_VALUE`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik edited a comment on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik edited a comment on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630955102
> > Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
> > Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
>
> I can see the benefit of using `RangeEndEstimator` for the finite range here. But as long as we don't modify the range end to estimate end or use estimate ed in `tryClaim`, we still cannot say the residual is empty.
That is true but I was thinking it would make better splitting decisions instead of creating a bunch of empty splits trimming the range down. The advantage of not using the estimator is that we don't have to invoke since it could be expensive for the user and in many situations will produce a value greater than `to`.
We can leave it out for now unless some compelling use case comes up.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630955102
> > Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
> > Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
>
> I can see the benefit of using `RangeEndEstimator` for the finite range here. But as long as we don't modify the range end to estimate end or use estimate ed in `tryClaim`, we still cannot say the residual is empty.
That is true but I was thinking it would make better splitting decisions instead of creating a bunch of empty splits trimming the range down. The advantage of not using the estimator is that we don't have to invoke since it could be expensive for the user and in many situations will produce a value greater than `to`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz merged pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz merged pull request #11715:
URL: https://github.com/apache/beam/pull/11715
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] boyuanzz commented on pull request #11715: [BEAM-9977] Implement GrowableOffsetRangeTracker
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #11715:
URL: https://github.com/apache/beam/pull/11715#issuecomment-630963389
> > > Should we be using the RangeEndEstimator when providing progress/splitting for ranges not ending at `Long.MAX_VALUE`?
> > > Lets say the range estimate is bad and is `MAX_VALUE - 3` but the real end is `5000`, then after a split we end up with `[0, (MAX_VALUE - 3) * 0.5)` and `[(MAX_VALUE - 3) * 0.5, MAX_VALUE)`. We may quickly learn that the residual is empty and then lose all effective progress on the primary.
> >
> >
> > I can see the benefit of using `RangeEndEstimator` for the finite range here. But as long as we don't modify the range end to estimate end or use estimate ed in `tryClaim`, we still cannot say the residual is empty.
>
> That is true but I was thinking it would make better splitting decisions instead of creating a bunch of empty splits trimming the range down. The advantage of not using the estimator is that we don't have to invoke since it could be expensive for the user and in many situations will produce a value greater than `to`.
>
> We can leave it out for now unless some compelling use case comes up.
Do we want to have a TODO here to track this?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org