You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/04/14 21:41:09 UTC
[1/4] beam git commit: This closes #2512
Repository: beam
Updated Branches:
refs/heads/master 89ff0b145 -> f30d5b9e9
This closes #2512
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/f30d5b9e
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/f30d5b9e
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/f30d5b9e
Branch: refs/heads/master
Commit: f30d5b9e9db80e35c9fae4c05d2a76655922db1c
Parents: 89ff0b1 113471d
Author: Thomas Groh <tg...@google.com>
Authored: Fri Apr 14 14:40:57 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Apr 14 14:40:57 2017 -0700
----------------------------------------------------------------------
.../apex/translation/ParDoTranslator.java | 6 +-
.../operators/ApexGroupByKeyOperator.java | 19 +-
.../operators/ApexParDoOperator.java | 48 +--
.../apex/translation/utils/NoOpStepContext.java | 2 +-
.../apex/translation/ParDoTranslatorTest.java | 18 +-
.../beam/runners/core/BaseExecutionContext.java | 13 +-
.../apache/beam/runners/core/DoFnAdapters.java | 16 +-
.../apache/beam/runners/core/DoFnRunners.java | 8 +-
.../beam/runners/core/ExecutionContext.java | 13 +-
.../GroupAlsoByWindowViaWindowSetNewDoFn.java | 6 +-
.../org/apache/beam/runners/core/OldDoFn.java | 38 +--
...eBoundedSplittableProcessElementInvoker.java | 8 +-
.../beam/runners/core/OutputWindowedValue.java | 10 +-
.../beam/runners/core/SimpleDoFnRunner.java | 54 ++--
.../beam/runners/core/SimpleOldDoFnRunner.java | 63 ++--
.../beam/runners/core/SplittableParDo.java | 21 +-
.../beam/runners/core/WindowingInternals.java | 8 +-
.../core/WindowingInternalsAdapters.java | 8 +-
.../core/GroupAlsoByWindowsProperties.java | 10 +-
.../apache/beam/runners/core/NoOpOldDoFn.java | 4 +-
.../apache/beam/runners/core/OldDoFnTest.java | 4 +-
...ndedSplittableProcessElementInvokerTest.java | 6 +-
.../beam/runners/core/ReduceFnTester.java | 8 +-
.../runners/core/SimpleOldDoFnRunnerTest.java | 4 +-
.../beam/runners/core/SplittableParDoTest.java | 8 +-
.../GroupAlsoByWindowEvaluatorFactory.java | 8 +-
.../beam/runners/direct/ParDoEvaluator.java | 4 +-
.../runners/direct/ParDoEvaluatorFactory.java | 10 +-
.../direct/ParDoMultiOverrideFactory.java | 2 +-
...littableProcessElementsEvaluatorFactory.java | 8 +-
.../direct/StatefulParDoEvaluatorFactory.java | 2 +-
.../beam/runners/direct/ParDoEvaluatorTest.java | 6 +-
.../FlinkStreamingTransformTranslators.java | 20 +-
.../functions/FlinkDoFnFunction.java | 4 +-
.../functions/FlinkNoOpStepContext.java | 2 +-
.../functions/FlinkStatefulDoFnFunction.java | 4 +-
.../wrappers/streaming/DoFnOperator.java | 14 +-
.../streaming/SplittableDoFnOperator.java | 10 +-
.../wrappers/streaming/WindowDoFnOperator.java | 4 +-
.../flink/streaming/DoFnOperatorTest.java | 34 +--
.../dataflow/BatchStatefulParDoOverrides.java | 2 +-
.../runners/dataflow/BatchViewOverrides.java | 6 +-
.../dataflow/BatchViewOverridesTest.java | 4 +-
.../SparkGroupAlsoByWindowViaWindowSet.java | 10 +-
.../spark/translation/MultiDoFnFunction.java | 4 +-
...SparkGroupAlsoByWindowViaOutputBufferFn.java | 8 +-
.../spark/translation/SparkProcessContext.java | 2 +-
.../streaming/StreamingTransformTranslator.java | 2 +-
.../org/apache/beam/sdk/transforms/Combine.java | 2 +-
.../org/apache/beam/sdk/transforms/DoFn.java | 14 +-
.../apache/beam/sdk/transforms/DoFnTester.java | 41 ++-
.../org/apache/beam/sdk/transforms/ParDo.java | 77 +++--
.../apache/beam/sdk/transforms/Partition.java | 2 +-
.../beam/sdk/values/PCollectionTuple.java | 3 +-
.../org/apache/beam/sdk/values/TupleTag.java | 26 +-
.../apache/beam/sdk/values/TupleTagList.java | 2 +-
.../org/apache/beam/sdk/values/TypedPValue.java | 4 +-
.../apache/beam/sdk/metrics/MetricsTest.java | 2 +-
.../apache/beam/sdk/transforms/ParDoTest.java | 293 ++++++++++---------
.../beam/sdk/transforms/SplittableDoFnTest.java | 21 +-
.../beam/sdk/values/PCollectionTupleTest.java | 8 +-
.../apache/beam/sdk/values/TypedPValueTest.java | 46 +--
.../beam/fn/harness/fake/FakeStepContext.java | 2 +-
.../control/ProcessBundleHandlerTest.java | 30 +-
.../sdk/io/gcp/bigquery/WritePartition.java | 6 +-
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 4 +-
66 files changed, 578 insertions(+), 578 deletions(-)
----------------------------------------------------------------------
[3/4] beam git commit: Rename DoFn.Context#sideOutput to output
Posted by tg...@apache.org.
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
index 9a66a2f..5496f71 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/DoFnOperator.java
@@ -97,7 +97,7 @@ import org.joda.time.Instant;
* @param <InputT> the input type of the {@link DoFn}
* @param <FnOutputT> the output type of the {@link DoFn}
* @param <OutputT> the output type of the operator, this can be different from the fn output
- * type when we have side outputs
+ * type when we have additional tagged outputs
*/
public class DoFnOperator<InputT, FnOutputT, OutputT>
extends AbstractStreamOperator<OutputT>
@@ -110,7 +110,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
protected final SerializedPipelineOptions serializedOptions;
protected final TupleTag<FnOutputT> mainOutputTag;
- protected final List<TupleTag<?>> sideOutputTags;
+ protected final List<TupleTag<?>> additionalOutputTags;
protected final Collection<PCollectionView<?>> sideInputs;
protected final Map<Integer, PCollectionView<?>> sideInputTagMapping;
@@ -155,7 +155,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
DoFn<InputT, FnOutputT> doFn,
Coder<WindowedValue<InputT>> inputCoder,
TupleTag<FnOutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
OutputManagerFactory<OutputT> outputManagerFactory,
WindowingStrategy<?, ?> windowingStrategy,
Map<Integer, PCollectionView<?>> sideInputTagMapping,
@@ -165,7 +165,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
this.doFn = doFn;
this.inputCoder = inputCoder;
this.mainOutputTag = mainOutputTag;
- this.sideOutputTags = sideOutputTags;
+ this.additionalOutputTags = additionalOutputTags;
this.sideInputTagMapping = sideInputTagMapping;
this.sideInputs = sideInputs;
this.serializedOptions = new SerializedPipelineOptions(options);
@@ -275,7 +275,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorFactory,
windowingStrategy);
@@ -619,7 +619,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
return new DoFnRunners.OutputManager() {
@Override
public <T> void output(TupleTag<T> tag, WindowedValue<T> value) {
- // with side outputs we can't get around this because we don't
+ // with tagged outputs we can't get around this because we don't
// know our own output type...
@SuppressWarnings("unchecked")
OutputT castValue = (OutputT) value;
@@ -675,7 +675,7 @@ public class DoFnOperator<InputT, FnOutputT, OutputT>
public void noteOutput(WindowedValue<?> output) {}
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {}
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {}
@Override
public <T, W extends BoundedWindow> void writePCollectionViewData(
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java
index 0724ac2..1a636c9 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/SplittableDoFnOperator.java
@@ -63,7 +63,7 @@ public class SplittableDoFnOperator<
WindowedValue<
KeyedWorkItem<String, ElementAndRestriction<InputT, RestrictionT>>>> inputCoder,
TupleTag<FnOutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
OutputManagerFactory<OutputT> outputManagerFactory,
WindowingStrategy<?, ?> windowingStrategy,
Map<Integer, PCollectionView<?>> sideInputTagMapping,
@@ -74,7 +74,7 @@ public class SplittableDoFnOperator<
doFn,
inputCoder,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
outputManagerFactory,
windowingStrategy,
sideInputTagMapping,
@@ -125,9 +125,9 @@ public class SplittableDoFnOperator<
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
index b015f66..8bbc6ef 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/WindowDoFnOperator.java
@@ -57,7 +57,7 @@ public class WindowDoFnOperator<K, InputT, OutputT>
SystemReduceFn<K, InputT, ?, OutputT, BoundedWindow> systemReduceFn,
Coder<WindowedValue<KeyedWorkItem<K, InputT>>> inputCoder,
TupleTag<KV<K, OutputT>> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
OutputManagerFactory<WindowedValue<KV<K, OutputT>>> outputManagerFactory,
WindowingStrategy<?, ?> windowingStrategy,
Map<Integer, PCollectionView<?>> sideInputTagMapping,
@@ -68,7 +68,7 @@ public class WindowDoFnOperator<K, InputT, OutputT>
null,
inputCoder,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
outputManagerFactory,
windowingStrategy,
sideInputTagMapping,
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java b/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java
index c1fdea3..4c826d1 100644
--- a/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java
+++ b/runners/flink/runner/src/test/java/org/apache/beam/runners/flink/streaming/DoFnOperatorTest.java
@@ -144,19 +144,19 @@ public class DoFnOperatorTest {
WindowedValue.getValueOnlyCoder(StringUtf8Coder.of());
TupleTag<String> mainOutput = new TupleTag<>("main-output");
- TupleTag<String> sideOutput1 = new TupleTag<>("side-output-1");
- TupleTag<String> sideOutput2 = new TupleTag<>("side-output-2");
+ TupleTag<String> additionalOutput1 = new TupleTag<>("output-1");
+ TupleTag<String> additionalOutput2 = new TupleTag<>("output-2");
ImmutableMap<TupleTag<?>, Integer> outputMapping = ImmutableMap.<TupleTag<?>, Integer>builder()
.put(mainOutput, 1)
- .put(sideOutput1, 2)
- .put(sideOutput2, 3)
+ .put(additionalOutput1, 2)
+ .put(additionalOutput2, 3)
.build();
DoFnOperator<String, String, RawUnionValue> doFnOperator = new DoFnOperator<>(
- new MultiOutputDoFn(sideOutput1, sideOutput2),
+ new MultiOutputDoFn(additionalOutput1, additionalOutput2),
windowedValueCoder,
mainOutput,
- ImmutableList.<TupleTag<?>>of(sideOutput1, sideOutput2),
+ ImmutableList.<TupleTag<?>>of(additionalOutput1, additionalOutput2),
new DoFnOperator.MultiOutputOutputManagerFactory(outputMapping),
WindowingStrategy.globalDefault(),
new HashMap<Integer, PCollectionView<?>>(), /* side-input mapping */
@@ -176,8 +176,8 @@ public class DoFnOperatorTest {
assertThat(
this.stripStreamRecordFromRawUnion(testHarness.getOutput()),
contains(
- new RawUnionValue(2, WindowedValue.valueInGlobalWindow("side: one")),
- new RawUnionValue(3, WindowedValue.valueInGlobalWindow("side: two")),
+ new RawUnionValue(2, WindowedValue.valueInGlobalWindow("extra: one")),
+ new RawUnionValue(3, WindowedValue.valueInGlobalWindow("extra: two")),
new RawUnionValue(1, WindowedValue.valueInGlobalWindow("got: hello")),
new RawUnionValue(2, WindowedValue.valueInGlobalWindow("got: hello")),
new RawUnionValue(3, WindowedValue.valueInGlobalWindow("got: hello"))));
@@ -542,24 +542,24 @@ public class DoFnOperatorTest {
}
private static class MultiOutputDoFn extends DoFn<String, String> {
- private TupleTag<String> sideOutput1;
- private TupleTag<String> sideOutput2;
+ private TupleTag<String> additionalOutput1;
+ private TupleTag<String> additionalOutput2;
- public MultiOutputDoFn(TupleTag<String> sideOutput1, TupleTag<String> sideOutput2) {
- this.sideOutput1 = sideOutput1;
- this.sideOutput2 = sideOutput2;
+ public MultiOutputDoFn(TupleTag<String> additionalOutput1, TupleTag<String> additionalOutput2) {
+ this.additionalOutput1 = additionalOutput1;
+ this.additionalOutput2 = additionalOutput2;
}
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
if (c.element().equals("one")) {
- c.sideOutput(sideOutput1, "side: one");
+ c.output(additionalOutput1, "extra: one");
} else if (c.element().equals("two")) {
- c.sideOutput(sideOutput2, "side: two");
+ c.output(additionalOutput2, "extra: two");
} else {
c.output("got: " + c.element());
- c.sideOutput(sideOutput1, "got: " + c.element());
- c.sideOutput(sideOutput2, "got: " + c.element());
+ c.output(additionalOutput1, "got: " + c.element());
+ c.output(additionalOutput2, "got: " + c.element());
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
index 3ded079..73f3728 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java
@@ -175,7 +175,7 @@ public class BatchStatefulParDoOverrides {
ParDo.of(new BatchStatefulDoFn<K, InputT, OutputT>(fn))
.withSideInputs(originalParDo.getSideInputs())
.withOutputTags(
- originalParDo.getMainOutputTag(), originalParDo.getSideOutputTags());
+ originalParDo.getMainOutputTag(), originalParDo.getAdditionalOutputTags());
return input.apply(new GbkBeforeStatefulParDo<K, InputT>()).apply(statefulParDo);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchViewOverrides.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchViewOverrides.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchViewOverrides.java
index 86bfeb6..ead2712 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchViewOverrides.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchViewOverrides.java
@@ -494,7 +494,7 @@ class BatchViewOverrides {
*/
private void outputMetadataRecordForSize(
ProcessContext c, KV<KV<K, W>, WindowedValue<V>> value, long uniqueKeyCount) {
- c.sideOutput(outputForSize,
+ c.output(outputForSize,
KV.of(ismCoder.hash(ImmutableList.of(IsmFormat.getMetadataKey(),
value.getKey().getValue())),
KV.of(value.getKey().getValue(), uniqueKeyCount)));
@@ -503,7 +503,7 @@ class BatchViewOverrides {
/** This outputs records which will be used to construct the entry set. */
private void outputMetadataRecordForEntrySet(
ProcessContext c, KV<KV<K, W>, WindowedValue<V>> value) {
- c.sideOutput(outputForEntrySet,
+ c.output(outputForEntrySet,
KV.of(ismCoder.hash(ImmutableList.of(IsmFormat.getMetadataKey(),
value.getKey().getValue())),
KV.of(value.getKey().getValue(), value.getKey().getKey())));
@@ -773,7 +773,7 @@ class BatchViewOverrides {
coderForMapLike(windowCoder, inputCoder.getKeyCoder(), inputCoder.getValueCoder());
// Create the various output tags representing the main output containing the data stream
- // and the side outputs containing the metadata about the size and entry set.
+ // and the additional outputs containing the metadata about the size and entry set.
TupleTag<IsmRecord<WindowedValue<V>>> mainOutputTag = new TupleTag<>();
TupleTag<KV<Integer, KV<W, Long>>> outputForSizeTag = new TupleTag<>();
TupleTag<KV<Integer, KV<W, K>>> outputForEntrySetTag = new TupleTag<>();
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchViewOverridesTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchViewOverridesTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchViewOverridesTest.java
index cd12c92..87395e6 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchViewOverridesTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/BatchViewOverridesTest.java
@@ -280,7 +280,7 @@ public class BatchViewOverridesTest {
// Verify the number of unique keys per window.
assertThat(
- doFnTester.takeSideOutputElements(outputForSizeTag),
+ doFnTester.takeOutputElements(outputForSizeTag),
contains(
KV.of(
ismCoder.hash(ImmutableList.of(IsmFormat.getMetadataKey(), windowA)),
@@ -294,7 +294,7 @@ public class BatchViewOverridesTest {
// Verify the output for the unique keys.
assertThat(
- doFnTester.takeSideOutputElements(outputForEntrySetTag),
+ doFnTester.takeOutputElements(outputForEntrySetTag),
contains(
KV.of(
ismCoder.hash(ImmutableList.of(IsmFormat.getMetadataKey(), windowA)),
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java
index 0e74fa2..029c28a 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java
@@ -412,12 +412,14 @@ public class SparkGroupAlsoByWindowViaWindowSet {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output, Instant timestamp,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
+ Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- throw new UnsupportedOperationException("Side outputs are not allowed in GroupAlsoByWindow.");
+ throw new UnsupportedOperationException(
+ "Tagged outputs are not allowed in GroupAlsoByWindow.");
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java
index a761954..4cd1683 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/MultiDoFnFunction.java
@@ -43,8 +43,8 @@ import scala.Tuple2;
/**
- * DoFunctions ignore side outputs. MultiDoFunctions deal with side outputs by enriching the
- * underlying data with multiple TupleTags.
+ * DoFunctions ignore outputs that are not the main output. MultiDoFunctions deal with additional
+ * outputs by enriching the underlying data with multiple TupleTags.
*
* @param <InputT> Input type for DoFunction.
* @param <OutputT> Output type for DoFunction.
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowViaOutputBufferFn.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowViaOutputBufferFn.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowViaOutputBufferFn.java
index d19c4a9..ccc0fa3 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowViaOutputBufferFn.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowViaOutputBufferFn.java
@@ -164,12 +164,12 @@ public class SparkGroupAlsoByWindowViaOutputBufferFn<K, InputT, W extends Bounde
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows, PaneInfo pane) {
- throw new UnsupportedOperationException("GroupAlsoByWindow should not use side outputs.");
+ throw new UnsupportedOperationException("GroupAlsoByWindow should not use tagged outputs.");
}
Iterable<WindowedValue<KV<K, Iterable<V>>>> getOutputs() {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java
index 4f8a1a5..3e8dde5 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java
@@ -113,7 +113,7 @@ class SparkProcessContext<FnInputT, FnOutputT, OutputT> {
public void noteOutput(WindowedValue<?> output) { }
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) { }
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) { }
@Override
public <T, W extends BoundedWindow> void writePCollectionViewData(
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
index 65892d2..000eada 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/StreamingTransformTranslator.java
@@ -385,7 +385,7 @@ public final class StreamingTransformTranslator {
JavaDStream<WindowedValue<InputT>> dStream = unboundedDataset.getDStream();
final String stepName = context.getCurrentTransform().getFullName();
- if (transform.getSideOutputTags().size() == 0) {
+ if (transform.getAdditionalOutputTags().size() == 0) {
JavaPairDStream<TupleTag<?>, WindowedValue<?>> all =
dStream.transformToPair(
new Function<
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
index 8fe4831..58d65d0 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
@@ -2191,7 +2191,7 @@ public class Combine {
c.output(kv);
} else {
int nonce = counter++ % spread;
- c.sideOutput(hot, KV.of(KV.of(kv.getKey(), nonce), kv.getValue()));
+ c.output(hot, KV.of(KV.of(kv.getKey(), nonce), kv.getValue()));
}
}
})
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
index 74a1348..d3da251 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
@@ -153,14 +153,14 @@ public abstract class DoFn<InputT, OutputT> implements Serializable, HasDisplayD
public abstract void outputWithTimestamp(OutputT output, Instant timestamp);
/**
- * Adds the given element to the side output {@code PCollection} with the
+ * Adds the given element to the output {@code PCollection} with the
* given tag.
*
- * <p>Once passed to {@code sideOutput} the element should not be modified
+ * <p>Once passed to {@code output} the element should not be modified
* in any way.
*
* <p>The caller of {@code ParDo} uses {@link ParDo.SingleOutput#withOutputTags} to
- * specify the tags of side outputs that it consumes. Non-consumed side
+ * specify the tags of outputs that it consumes. Non-consumed
* outputs, e.g., outputs for monitoring purposes only, don't necessarily
* need to be specified.
*
@@ -180,13 +180,13 @@ public abstract class DoFn<InputT, OutputT> implements Serializable, HasDisplayD
*
* @see ParDo.SingleOutput#withOutputTags
*/
- public abstract <T> void sideOutput(TupleTag<T> tag, T output);
+ public abstract <T> void output(TupleTag<T> tag, T output);
/**
- * Adds the given element to the specified side output {@code PCollection},
+ * Adds the given element to the specified output {@code PCollection},
* with the given timestamp.
*
- * <p>Once passed to {@code sideOutputWithTimestamp} the element should not be
+ * <p>Once passed to {@code outputWithTimestamp} the element should not be
* modified in any way.
*
* <p>If invoked from {@link ProcessElement}), the timestamp
@@ -207,7 +207,7 @@ public abstract class DoFn<InputT, OutputT> implements Serializable, HasDisplayD
*
* @see ParDo.SingleOutput#withOutputTags
*/
- public abstract <T> void sideOutputWithTimestamp(
+ public abstract <T> void outputWithTimestamp(
TupleTag<T> tag, T output, Instant timestamp);
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
index 88f4035..5446431 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
@@ -65,7 +65,7 @@ import org.joda.time.Instant;
*
* // Set arguments shared across all bundles:
* fnTester.setSideInputs(...); // If fn takes side inputs.
- * fnTester.setSideOutputTags(...); // If fn writes to side outputs.
+ * fnTester.setOutputTags(...); // If fn writes to more than one output.
*
* // Process a bundle containing a single input element:
* Input testInput = ...;
@@ -464,14 +464,14 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
}
/**
- * Returns the elements output so far to the side output with the
+ * Returns the elements output so far to the output with the
* given tag. Does not clear them, so subsequent calls will
* continue to include these elements.
*
- * @see #takeSideOutputElements
- * @see #clearSideOutputElements
+ * @see #takeOutputElements
+ * @see #clearOutputElements
*/
- public <T> List<T> peekSideOutputElements(TupleTag<T> tag) {
+ public <T> List<T> peekOutputElements(TupleTag<T> tag) {
// TODO: Should we return an unmodifiable list?
return Lists.transform(getImmutableOutput(tag),
new Function<ValueInSingleWindow<T>, T>() {
@@ -483,24 +483,23 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
}
/**
- * Clears the record of the elements output so far to the side
- * output with the given tag.
+ * Clears the record of the elements output so far to the output with the given tag.
*
- * @see #peekSideOutputElements
+ * @see #peekOutputElements
*/
- public <T> void clearSideOutputElements(TupleTag<T> tag) {
+ public <T> void clearOutputElements(TupleTag<T> tag) {
getMutableOutput(tag).clear();
}
/**
- * Returns the elements output so far to the side output with the given tag.
+ * Returns the elements output so far to the output with the given tag.
* Clears the list so these elements don't appear in future calls.
*
- * @see #peekSideOutputElements
+ * @see #peekOutputElements
*/
- public <T> List<T> takeSideOutputElements(TupleTag<T> tag) {
- List<T> resultElems = new ArrayList<>(peekSideOutputElements(tag));
- clearSideOutputElements(tag);
+ public <T> List<T> takeOutputElements(TupleTag<T> tag) {
+ List<T> resultElems = new ArrayList<>(peekOutputElements(tag));
+ clearOutputElements(tag);
return resultElems;
}
@@ -563,12 +562,12 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
throwUnsupportedOutputFromBundleMethods();
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
+ public <T> void output(TupleTag<T> tag, T output) {
throwUnsupportedOutputFromBundleMethods();
}
@@ -683,21 +682,21 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
@Override
public void output(OutputT output) {
- sideOutput(mainOutputTag, output);
+ output(mainOutputTag, output);
}
@Override
public void outputWithTimestamp(OutputT output, Instant timestamp) {
- sideOutputWithTimestamp(mainOutputTag, output, timestamp);
+ outputWithTimestamp(mainOutputTag, output, timestamp);
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- sideOutputWithTimestamp(tag, output, element.getTimestamp());
+ public <T> void output(TupleTag<T> tag, T output) {
+ outputWithTimestamp(tag, output, element.getTimestamp());
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
getMutableOutput(tag)
.add(ValueInSingleWindow.of(output, timestamp, element.getWindow(), element.getPane()));
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
index 3de845b..e3777ac 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
@@ -35,6 +35,7 @@ import org.apache.beam.sdk.coders.CoderRegistry;
import org.apache.beam.sdk.runners.PipelineRunner;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.transforms.display.DisplayData.Builder;
+import org.apache.beam.sdk.transforms.display.DisplayData.ItemSpec;
import org.apache.beam.sdk.transforms.display.HasDisplayData;
import org.apache.beam.sdk.transforms.reflect.DoFnSignature;
import org.apache.beam.sdk.transforms.reflect.DoFnSignature.MethodWithExtraParameters;
@@ -103,7 +104,7 @@ import org.apache.beam.sdk.values.TypedPValue;
* <p>Each of the calls to any of the {@link DoFn DoFn's} processing
* methods can produce zero or more output elements. All of the
* of output elements from all of the {@link DoFn} instances
- * are included in the output {@link PCollection}.
+ * are included in an output {@link PCollection}.
*
* <p>For example:
*
@@ -180,20 +181,20 @@ import org.apache.beam.sdk.values.TypedPValue;
* }}));
* }</pre>
*
- * <h2>Side Outputs</h2>
+ * <h2>Additional Outputs</h2>
*
* <p>Optionally, a {@link ParDo} transform can produce multiple
* output {@link PCollection PCollections}, both a "main output"
- * {@code PCollection<OutputT>} plus any number of "side output"
+ * {@code PCollection<OutputT>} plus any number of additional output
* {@link PCollection PCollections}, each keyed by a distinct {@link TupleTag},
* and bundled in a {@link PCollectionTuple}. The {@link TupleTag TupleTags}
* to be used for the output {@link PCollectionTuple} are specified by
- * invoking {@link SingleOutput#withOutputTags}. Unconsumed side outputs do not
+ * invoking {@link SingleOutput#withOutputTags}. Unconsumed outputs do not
* necessarily need to be explicitly specified, even if the {@link DoFn}
* generates them. Within the {@link DoFn}, an element is added to the
* main output {@link PCollection} as normal, using
- * {@link DoFn.Context#output}, while an element is added to a side output
- * {@link PCollection} using {@link DoFn.Context#sideOutput}. For example:
+ * {@link DoFn.Context#output(Object)}, while an element is added to any additional output
+ * {@link PCollection} using {@link DoFn.Context#output(TupleTag, Object)}. For example:
*
* <pre>{@code
* PCollection<String> words = ...;
@@ -201,7 +202,7 @@ import org.apache.beam.sdk.values.TypedPValue;
* // plus the lengths of words that are above the cut off.
* // Also select words starting with "MARKER".
* final int wordLengthCutOff = 10;
- * // Create tags to use for the main and side outputs.
+ * // Create tags to use for the main and additional outputs.
* final TupleTag<String> wordsBelowCutOffTag =
* new TupleTag<String>(){};
* final TupleTag<Integer> wordLengthsAboveCutOffTag =
@@ -212,7 +213,7 @@ import org.apache.beam.sdk.values.TypedPValue;
* words.apply(
* ParDo
* .of(new DoFn<String, String>() {
- * // Create a tag for the unconsumed side output.
+ * // Create a tag for the unconsumed output.
* final TupleTag<String> specialWordsTag =
* new TupleTag<String>(){};
* {@literal @}ProcessElement
@@ -222,19 +223,19 @@ import org.apache.beam.sdk.values.TypedPValue;
* // Emit this short word to the main output.
* c.output(word);
* } else {
- * // Emit this long word's length to a side output.
- * c.sideOutput(wordLengthsAboveCutOffTag, word.length());
+ * // Emit this long word's length to a specified output.
+ * c.output(wordLengthsAboveCutOffTag, word.length());
* }
* if (word.startsWith("MARKER")) {
- * // Emit this word to a different side output.
- * c.sideOutput(markedWordsTag, word);
+ * // Emit this word to a different specified output.
+ * c.output(markedWordsTag, word);
* }
* if (word.startsWith("SPECIAL")) {
- * // Emit this word to the unconsumed side output.
- * c.sideOutput(specialWordsTag, word);
+ * // Emit this word to the unconsumed output.
+ * c.output(specialWordsTag, word);
* }
* }})
- * // Specify the main and consumed side output tags of the
+ * // Specify the main and consumed output tags of the
* // PCollectionTuple result:
* .withOutputTags(wordsBelowCutOffTag,
* TupleTagList.of(wordLengthsAboveCutOffTag)
@@ -254,9 +255,9 @@ import org.apache.beam.sdk.values.TypedPValue;
* elements of the main output {@link PCollection PCollection<OutputT>} is
* inferred from the concrete type of the {@link DoFn DoFn<InputT, OutputT>}.
*
- * <p>By default, the {@link Coder Coder<SideOutputT>} for the elements of
- * a side output {@link PCollection PCollection<SideOutputT>} is inferred
- * from the concrete type of the corresponding {@link TupleTag TupleTag<SideOutputT>}.
+ * <p>By default, the {@link Coder Coder<AdditionalOutputT>} for the elements of
+ * an output {@link PCollection PCollection<AdditionalOutputT>} is inferred
+ * from the concrete type of the corresponding {@link TupleTag TupleTag<AdditionalOutputT>}.
* To be successful, the {@link TupleTag} should be created as an instance
* of a trivial anonymous subclass, with {@code {}} suffixed to the
* constructor call. Such uses block Java's generic type parameter
@@ -265,12 +266,12 @@ import org.apache.beam.sdk.values.TypedPValue;
* <pre> {@code
* // A TupleTag to use for a side input can be written concisely:
* final TupleTag<Integer> sideInputag = new TupleTag<>();
- * // A TupleTag to use for a side output should be written with "{}",
+ * // A TupleTag to use for an output should be written with "{}",
* // and explicit generic parameter type:
- * final TupleTag<String> sideOutputTag = new TupleTag<String>(){};
+ * final TupleTag<String> additionalOutputTag = new TupleTag<String>(){};
* } </pre>
* This style of {@code TupleTag} instantiation is used in the example of
- * multiple side outputs, above.
+ * {@link ParDo ParDos} that produce multiple outputs, above.
*
* <h2>Serializability of {@link DoFn DoFns}</h2>
*
@@ -358,7 +359,7 @@ import org.apache.beam.sdk.values.TypedPValue;
* that state across Java processes. All information should be
* communicated to {@link DoFn} instances via main and side inputs and
* serialized state, and all output should be communicated from a
- * {@link DoFn} instance via main and side outputs, in the absence of
+ * {@link DoFn} instance via output {@link PCollection PCollections}, in the absence of
* external communication mechanisms written by user code.
*
* <h2>Fault Tolerance</h2>
@@ -602,14 +603,14 @@ public class ParDo {
/**
* Returns a new multi-output {@link ParDo} {@link PTransform} that's like this {@link
- * PTransform} but with the specified main and side output tags. Does not modify this {@link
+ * PTransform} but with the specified output tags. Does not modify this {@link
* PTransform}.
*
- * <p>See the discussion of Side Outputs above for more explanation.
+ * <p>See the discussion of Additional Outputs above for more explanation.
*/
public MultiOutput<InputT, OutputT> withOutputTags(
- TupleTag<OutputT> mainOutputTag, TupleTagList sideOutputTags) {
- return new MultiOutput<>(fn, sideInputs, mainOutputTag, sideOutputTags, fnDisplayData);
+ TupleTag<OutputT> mainOutputTag, TupleTagList additionalOutputTags) {
+ return new MultiOutput<>(fn, sideInputs, mainOutputTag, additionalOutputTags, fnDisplayData);
}
@Override
@@ -671,11 +672,9 @@ public class ParDo {
}
/**
- * A {@link PTransform} that, when applied to a
- * {@code PCollection<InputT>}, invokes a user-specified
- * {@code DoFn<InputT, OutputT>} on all its elements, which can emit elements
- * to any of the {@link PTransform}'s main and side output
- * {@code PCollection}s, which are bundled into a result
+ * A {@link PTransform} that, when applied to a {@code PCollection<InputT>}, invokes a
+ * user-specified {@code DoFn<InputT, OutputT>} on all its elements, which can emit elements to
+ * any of the {@link PTransform}'s output {@code PCollection}s, which are bundled into a result
* {@code PCollectionTuple}.
*
* @param <InputT> the type of the (main) input {@code PCollection} elements
@@ -685,7 +684,7 @@ public class ParDo {
extends PTransform<PCollection<? extends InputT>, PCollectionTuple> {
private final List<PCollectionView<?>> sideInputs;
private final TupleTag<OutputT> mainOutputTag;
- private final TupleTagList sideOutputTags;
+ private final TupleTagList additionalOutputTags;
private final DisplayData.ItemSpec<? extends Class<?>> fnDisplayData;
private final DoFn<InputT, OutputT> fn;
@@ -693,11 +692,11 @@ public class ParDo {
DoFn<InputT, OutputT> fn,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- TupleTagList sideOutputTags,
- DisplayData.ItemSpec<? extends Class<?>> fnDisplayData) {
+ TupleTagList additionalOutputTags,
+ ItemSpec<? extends Class<?>> fnDisplayData) {
this.sideInputs = sideInputs;
this.mainOutputTag = mainOutputTag;
- this.sideOutputTags = sideOutputTags;
+ this.additionalOutputTags = additionalOutputTags;
this.fn = SerializableUtils.clone(fn);
this.fnDisplayData = fnDisplayData;
}
@@ -730,7 +729,7 @@ public class ParDo {
.addAll(sideInputs)
.build(),
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
fnDisplayData);
}
@@ -745,7 +744,7 @@ public class ParDo {
PCollectionTuple outputs = PCollectionTuple.ofPrimitiveOutputsInternal(
input.getPipeline(),
- TupleTagList.of(mainOutputTag).and(sideOutputTags.getAll()),
+ TupleTagList.of(mainOutputTag).and(additionalOutputTags.getAll()),
input.getWindowingStrategy(),
input.isBounded());
@@ -794,8 +793,8 @@ public class ParDo {
return mainOutputTag;
}
- public TupleTagList getSideOutputTags() {
- return sideOutputTags;
+ public TupleTagList getAdditionalOutputTags() {
+ return additionalOutputTags;
}
public List<PCollectionView<?>> getSideInputs() {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java
index 2031bc9..595d18c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Partition.java
@@ -169,7 +169,7 @@ public class Partition<T> extends PTransform<PCollection<T>, PCollectionList<T>>
if (0 <= partition && partition < numPartitions) {
@SuppressWarnings("unchecked")
TupleTag<X> typedTag = (TupleTag<X>) outputTags.get(partition);
- c.sideOutput(typedTag, input);
+ c.output(typedTag, input);
} else {
throw new IndexOutOfBoundsException(
"Partition function returned out of bounds index: "
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java
index 0ab26ca..ce67e94 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionTuple.java
@@ -37,8 +37,7 @@ import org.apache.beam.sdk.values.PCollection.IsBounded;
* {@link PTransform} taking
* or producing multiple PCollection inputs or outputs that can be of
* different types, for instance a
- * {@link ParDo} with side
- * outputs.
+ * {@link ParDo} with multiple outputs.
*
* <p>A {@link PCollectionTuple} can be created and accessed like follows:
* <pre> {@code
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTag.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTag.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTag.java
index a6b63ab..37d41f7 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTag.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTag.java
@@ -31,25 +31,23 @@ import org.apache.beam.sdk.util.CloudObject;
import org.apache.beam.sdk.util.PropertyNames;
/**
- * A {@link TupleTag} is a typed tag to use as the key of a
- * heterogeneously typed tuple, like {@link PCollectionTuple}.
- * Its generic type parameter allows tracking
- * the static type of things stored in tuples.
+ * A {@link TupleTag} is a typed tag to use as the key of a heterogeneously typed tuple, like {@link
+ * PCollectionTuple}. Its generic type parameter allows tracking the static type of things stored in
+ * tuples.
*
- * <p>To aid in assigning default {@link org.apache.beam.sdk.coders.Coder Coders} for results of
- * side outputs of {@link ParDo}, an output
- * {@link TupleTag} should be instantiated with an extra {@code {}} so
- * it is an instance of an anonymous subclass without generic type
- * parameters. Input {@link TupleTag TupleTags} require no such extra
- * instantiation (although it doesn't hurt). For example:
+ * <p>To aid in assigning default {@link org.apache.beam.sdk.coders.Coder Coders} for results of a
+ * {@link ParDo}, an output {@link TupleTag} should be instantiated with an extra {@code {}} so it
+ * is an instance of an anonymous subclass without generic type parameters. Input {@link TupleTag
+ * TupleTags} require no such extra instantiation (although it doesn't hurt). For example:
*
- * <pre> {@code
+ * <pre>{@code
* TupleTag<SomeType> inputTag = new TupleTag<>();
* TupleTag<SomeOtherType> outputTag = new TupleTag<SomeOtherType>(){};
- * } </pre>
+ * }
+ * </pre>
*
- * @param <V> the type of the elements or values of the tagged thing,
- * e.g., a {@code PCollection<V>}.
+ * @param <V> the type of the elements or values of the tagged thing, e.g., a {@code
+ * PCollection<V>}.
*/
public class TupleTag<V> implements Serializable {
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTagList.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTagList.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTagList.java
index b4ce941..5aeff5e 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTagList.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TupleTagList.java
@@ -28,7 +28,7 @@ import org.apache.beam.sdk.transforms.ParDo;
/**
* A {@link TupleTagList} is an immutable list of heterogeneously
* typed {@link TupleTag TupleTags}. A {@link TupleTagList} is used, for instance, to
- * specify the tags of the side outputs of a
+ * specify the tags of the additional outputs of a
* {@link ParDo}.
*
* <p>A {@link TupleTagList} can be created and accessed like follows:
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
index d353835..54af747 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypedPValue.java
@@ -148,14 +148,14 @@ public abstract class TypedPValue<T> extends PValueBase implements PValue {
return new CoderOrFailure<>(registry.getDefaultCoder(token), null);
} catch (CannotProvideCoderException exc) {
inferFromTokenException = exc;
- // Attempt to detect when the token came from a TupleTag used for a ParDo side output,
+ // Attempt to detect when the token came from a TupleTag used for a ParDo output,
// and provide a better error message if so. Unfortunately, this information is not
// directly available from the TypeDescriptor, so infer based on the type of the PTransform
// and the error message itself.
if (transform instanceof ParDo.MultiOutput
&& exc.getReason() == ReasonCode.TYPE_ERASURE) {
inferFromTokenException = new CannotProvideCoderException(exc.getMessage()
- + " If this error occurs for a side output of the producing ParDo, verify that the "
+ + " If this error occurs for an output of the producing ParDo, verify that the "
+ "TupleTag for this output is constructed with proper type information (see "
+ "TupleTag Javadoc) or explicitly set the Coder to use if this is not possible.");
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
index 3555db3..afe384d 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/metrics/MetricsTest.java
@@ -221,7 +221,7 @@ public class MetricsTest implements Serializable {
values.update(element);
gauge.set(12L);
c.output(element);
- c.sideOutput(output2, element);
+ c.output(output2, element);
}
})
.withOutputTags(output1, TupleTagList.of(output2)));
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
index b429eab..589c744 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
@@ -153,15 +153,15 @@ public class ParDoTest implements Serializable {
State state = State.NOT_SET_UP;
final List<PCollectionView<Integer>> sideInputViews = new ArrayList<>();
- final List<TupleTag<String>> sideOutputTupleTags = new ArrayList<>();
+ final List<TupleTag<String>> additionalOutputTupleTags = new ArrayList<>();
public TestDoFn() {
}
public TestDoFn(List<PCollectionView<Integer>> sideInputViews,
- List<TupleTag<String>> sideOutputTupleTags) {
+ List<TupleTag<String>> additionalOutputTupleTags) {
this.sideInputViews.addAll(sideInputViews);
- this.sideOutputTupleTags.addAll(sideOutputTupleTags);
+ this.additionalOutputTupleTags.addAll(additionalOutputTupleTags);
}
@Setup
@@ -197,9 +197,9 @@ public class ParDoTest implements Serializable {
private void outputToAll(Context c, String value) {
c.output(value);
- for (TupleTag<String> sideOutputTupleTag : sideOutputTupleTags) {
- c.sideOutput(sideOutputTupleTag,
- sideOutputTupleTag.getId() + ": " + value);
+ for (TupleTag<String> additionalOutputTupleTag : additionalOutputTupleTags) {
+ c.output(additionalOutputTupleTag,
+ additionalOutputTupleTag.getId() + ": " + value);
}
}
@@ -212,9 +212,9 @@ public class ParDoTest implements Serializable {
value += ": " + sideInputValues;
}
c.output(value);
- for (TupleTag<String> sideOutputTupleTag : sideOutputTupleTags) {
- c.sideOutput(sideOutputTupleTag,
- sideOutputTupleTag.getId() + ": " + value);
+ for (TupleTag<String> additionalOutputTupleTag : additionalOutputTupleTags) {
+ c.output(additionalOutputTupleTag,
+ additionalOutputTupleTag.getId() + ": " + value);
}
}
}
@@ -389,90 +389,90 @@ public class ParDoTest implements Serializable {
@Test
@Category(ValidatesRunner.class)
- public void testParDoWithSideOutputs() {
+ public void testParDoWithTaggedOutput() {
List<Integer> inputs = Arrays.asList(3, -42, 666);
TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- TupleTag<String> sideOutputTag1 = new TupleTag<String>("side1"){};
- TupleTag<String> sideOutputTag2 = new TupleTag<String>("side2"){};
- TupleTag<String> sideOutputTag3 = new TupleTag<String>("side3"){};
- TupleTag<String> sideOutputTagUnwritten = new TupleTag<String>("sideUnwritten"){};
+ TupleTag<String> additionalOutputTag1 = new TupleTag<String>("additional1"){};
+ TupleTag<String> additionalOutputTag2 = new TupleTag<String>("additional2"){};
+ TupleTag<String> additionalOutputTag3 = new TupleTag<String>("additional3"){};
+ TupleTag<String> additionalOutputTagUnwritten = new TupleTag<String>("unwrittenOutput"){};
PCollectionTuple outputs = pipeline
.apply(Create.of(inputs))
.apply(ParDo
.of(new TestDoFn(
Arrays.<PCollectionView<Integer>>asList(),
- Arrays.asList(sideOutputTag1, sideOutputTag2, sideOutputTag3)))
+ Arrays.asList(additionalOutputTag1, additionalOutputTag2, additionalOutputTag3)))
.withOutputTags(
mainOutputTag,
- TupleTagList.of(sideOutputTag3)
- .and(sideOutputTag1)
- .and(sideOutputTagUnwritten)
- .and(sideOutputTag2)));
+ TupleTagList.of(additionalOutputTag3)
+ .and(additionalOutputTag1)
+ .and(additionalOutputTagUnwritten)
+ .and(additionalOutputTag2)));
PAssert.that(outputs.get(mainOutputTag))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs));
- PAssert.that(outputs.get(sideOutputTag1))
+ PAssert.that(outputs.get(additionalOutputTag1))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag1));
- PAssert.that(outputs.get(sideOutputTag2))
+ .fromOutput(additionalOutputTag1));
+ PAssert.that(outputs.get(additionalOutputTag2))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag2));
- PAssert.that(outputs.get(sideOutputTag3))
+ .fromOutput(additionalOutputTag2));
+ PAssert.that(outputs.get(additionalOutputTag3))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag3));
- PAssert.that(outputs.get(sideOutputTagUnwritten)).empty();
+ .fromOutput(additionalOutputTag3));
+ PAssert.that(outputs.get(additionalOutputTagUnwritten)).empty();
pipeline.run();
}
@Test
@Category(ValidatesRunner.class)
- public void testParDoEmptyWithSideOutputs() {
+ public void testParDoEmptyWithTaggedOutput() {
TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- TupleTag<String> sideOutputTag1 = new TupleTag<String>("side1"){};
- TupleTag<String> sideOutputTag2 = new TupleTag<String>("side2"){};
- TupleTag<String> sideOutputTag3 = new TupleTag<String>("side3"){};
- TupleTag<String> sideOutputTagUnwritten = new TupleTag<String>("sideUnwritten"){};
+ TupleTag<String> additionalOutputTag1 = new TupleTag<String>("additional1"){};
+ TupleTag<String> additionalOutputTag2 = new TupleTag<String>("additional2"){};
+ TupleTag<String> additionalOutputTag3 = new TupleTag<String>("additional3"){};
+ TupleTag<String> additionalOutputTagUnwritten = new TupleTag<String>("unwrittenOutput"){};
PCollectionTuple outputs = pipeline
.apply(Create.empty(VarIntCoder.of()))
.apply(ParDo
.of(new TestDoFn(
Arrays.<PCollectionView<Integer>>asList(),
- Arrays.asList(sideOutputTag1, sideOutputTag2, sideOutputTag3)))
+ Arrays.asList(additionalOutputTag1, additionalOutputTag2, additionalOutputTag3)))
.withOutputTags(
mainOutputTag,
- TupleTagList.of(sideOutputTag3).and(sideOutputTag1)
- .and(sideOutputTagUnwritten).and(sideOutputTag2)));
+ TupleTagList.of(additionalOutputTag3).and(additionalOutputTag1)
+ .and(additionalOutputTagUnwritten).and(additionalOutputTag2)));
List<Integer> inputs = Collections.emptyList();
PAssert.that(outputs.get(mainOutputTag))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs));
- PAssert.that(outputs.get(sideOutputTag1))
+ PAssert.that(outputs.get(additionalOutputTag1))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag1));
- PAssert.that(outputs.get(sideOutputTag2))
+ .fromOutput(additionalOutputTag1));
+ PAssert.that(outputs.get(additionalOutputTag2))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag2));
- PAssert.that(outputs.get(sideOutputTag3))
+ .fromOutput(additionalOutputTag2));
+ PAssert.that(outputs.get(additionalOutputTag3))
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs)
- .fromSideOutput(sideOutputTag3));
- PAssert.that(outputs.get(sideOutputTagUnwritten)).empty();
+ .fromOutput(additionalOutputTag3));
+ PAssert.that(outputs.get(additionalOutputTagUnwritten)).empty();
pipeline.run();
}
@Test
@Category(ValidatesRunner.class)
- public void testParDoWithEmptySideOutputs() {
+ public void testParDoWithEmptyTaggedOutput() {
TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- TupleTag<String> sideOutputTag1 = new TupleTag<String>("side1"){};
- TupleTag<String> sideOutputTag2 = new TupleTag<String>("side2"){};
+ TupleTag<String> additionalOutputTag1 = new TupleTag<String>("additional1"){};
+ TupleTag<String> additionalOutputTag2 = new TupleTag<String>("additional2"){};
PCollectionTuple outputs = pipeline
.apply(Create.empty(VarIntCoder.of()))
@@ -480,12 +480,12 @@ public class ParDoTest implements Serializable {
.of(new TestNoOutputDoFn())
.withOutputTags(
mainOutputTag,
- TupleTagList.of(sideOutputTag1).and(sideOutputTag2)));
+ TupleTagList.of(additionalOutputTag1).and(additionalOutputTag2)));
PAssert.that(outputs.get(mainOutputTag)).empty();
- PAssert.that(outputs.get(sideOutputTag1)).empty();
- PAssert.that(outputs.get(sideOutputTag2)).empty();
+ PAssert.that(outputs.get(additionalOutputTag1)).empty();
+ PAssert.that(outputs.get(additionalOutputTag2)).empty();
pipeline.run();
}
@@ -493,12 +493,12 @@ public class ParDoTest implements Serializable {
@Test
@Category(ValidatesRunner.class)
- public void testParDoWithOnlySideOutputs() {
+ public void testParDoWithOnlyTaggedOutput() {
List<Integer> inputs = Arrays.asList(3, -42, 666);
final TupleTag<Void> mainOutputTag = new TupleTag<Void>("main"){};
- final TupleTag<Integer> sideOutputTag = new TupleTag<Integer>("side"){};
+ final TupleTag<Integer> additionalOutputTag = new TupleTag<Integer>("additional"){};
PCollectionTuple outputs = pipeline
.apply(Create.of(inputs))
@@ -506,29 +506,29 @@ public class ParDoTest implements Serializable {
.of(new DoFn<Integer, Void>(){
@ProcessElement
public void processElement(ProcessContext c) {
- c.sideOutput(sideOutputTag, c.element());
+ c.output(additionalOutputTag, c.element());
}})
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
PAssert.that(outputs.get(mainOutputTag)).empty();
- PAssert.that(outputs.get(sideOutputTag)).containsInAnyOrder(inputs);
+ PAssert.that(outputs.get(additionalOutputTag)).containsInAnyOrder(inputs);
pipeline.run();
}
@Test
@Category(NeedsRunner.class)
- public void testParDoWritingToUndeclaredSideOutput() {
+ public void testParDoWritingToUndeclaredTag() {
List<Integer> inputs = Arrays.asList(3, -42, 666);
- TupleTag<String> sideTag = new TupleTag<String>("side"){};
+ TupleTag<String> notOutputTag = new TupleTag<String>("additional"){};
PCollection<String> output = pipeline
.apply(Create.of(inputs))
.apply(ParDo.of(new TestDoFn(
Arrays.<PCollectionView<Integer>>asList(),
- Arrays.asList(sideTag))));
+ Arrays.asList(notOutputTag))));
PAssert.that(output)
.satisfies(ParDoTest.HasExpectedOutput.forInput(inputs));
@@ -539,7 +539,7 @@ public class ParDoTest implements Serializable {
@Test
// TODO: The exception thrown is runner-specific, even if the behavior is general
@Category(NeedsRunner.class)
- public void testParDoUndeclaredSideOutputLimit() {
+ public void testParDoUndeclaredTagLimit() {
PCollection<Integer> input = pipeline.apply(Create.of(Arrays.asList(3)));
@@ -548,13 +548,13 @@ public class ParDoTest implements Serializable {
.apply("Success1000", ParDo.of(new DoFn<Integer, String>() {
@ProcessElement
public void processElement(ProcessContext c) {
- TupleTag<String> specialSideTag = new TupleTag<String>(){};
- c.sideOutput(specialSideTag, "side");
- c.sideOutput(specialSideTag, "side");
- c.sideOutput(specialSideTag, "side");
+ TupleTag<String> specialOutputTag = new TupleTag<String>(){};
+ c.output(specialOutputTag, "special");
+ c.output(specialOutputTag, "special");
+ c.output(specialOutputTag, "special");
for (int i = 0; i < 998; i++) {
- c.sideOutput(new TupleTag<String>(){}, "side");
+ c.output(new TupleTag<String>(){}, "tag" + i);
}
}}));
pipeline.run();
@@ -565,12 +565,12 @@ public class ParDoTest implements Serializable {
@ProcessElement
public void processElement(ProcessContext c) {
for (int i = 0; i < 1000; i++) {
- c.sideOutput(new TupleTag<String>(){}, "side");
+ c.output(new TupleTag<String>(){}, "output" + i);
}
}}));
thrown.expect(RuntimeException.class);
- thrown.expectMessage("the number of side outputs has exceeded a limit");
+ thrown.expectMessage("the number of outputs has exceeded a limit");
pipeline.run();
}
@@ -647,7 +647,7 @@ public class ParDoTest implements Serializable {
List<Integer> inputs = Arrays.asList(3, -42, 666);
final TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- final TupleTag<Void> sideOutputTag = new TupleTag<Void>("sideOutput"){};
+ final TupleTag<Void> additionalOutputTag = new TupleTag<Void>("output"){};
PCollectionView<Integer> sideInput1 = pipeline
.apply("CreateSideInput1", Create.of(11))
@@ -668,7 +668,7 @@ public class ParDoTest implements Serializable {
.withSideInputs(sideInput1)
.withSideInputs(sideInputUnread)
.withSideInputs(sideInput2)
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
PAssert.that(outputs.get(mainOutputTag))
.satisfies(ParDoTest.HasExpectedOutput
@@ -685,7 +685,7 @@ public class ParDoTest implements Serializable {
List<Integer> inputs = Arrays.asList(3, -42, 666);
final TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- final TupleTag<Void> sideOutputTag = new TupleTag<Void>("sideOutput"){};
+ final TupleTag<Void> additionalOutputTag = new TupleTag<Void>("output"){};
PCollectionView<Integer> sideInput1 = pipeline
.apply("CreateSideInput1", Create.of(11))
@@ -706,7 +706,7 @@ public class ParDoTest implements Serializable {
.withSideInputs(sideInput1)
.withSideInputs(sideInputUnread)
.withSideInputs(sideInput2)
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
PAssert.that(outputs.get(mainOutputTag))
.satisfies(ParDoTest.HasExpectedOutput
@@ -853,37 +853,37 @@ public class ParDoTest implements Serializable {
@Test
public void testParDoMultiNameBasedDoFnWithTrimmerSuffix() {
assertThat(
- ParDo.of(new SideOutputDummyFn(null)).withOutputTags(null, null).getName(),
- containsString("ParMultiDo(SideOutputDummy)"));
+ ParDo.of(new TaggedOutputDummyFn(null)).withOutputTags(null, null).getName(),
+ containsString("ParMultiDo(TaggedOutputDummy)"));
}
@Test
- public void testParDoWithSideOutputsName() {
+ public void testParDoWithTaggedOutputName() {
pipeline.enableAbandonedNodeEnforcement(false);
TupleTag<String> mainOutputTag = new TupleTag<String>("main"){};
- TupleTag<String> sideOutputTag1 = new TupleTag<String>("side1"){};
- TupleTag<String> sideOutputTag2 = new TupleTag<String>("side2"){};
- TupleTag<String> sideOutputTag3 = new TupleTag<String>("side3"){};
- TupleTag<String> sideOutputTagUnwritten = new TupleTag<String>("sideUnwritten"){};
+ TupleTag<String> additionalOutputTag1 = new TupleTag<String>("output1"){};
+ TupleTag<String> additionalOutputTag2 = new TupleTag<String>("output2"){};
+ TupleTag<String> additionalOutputTag3 = new TupleTag<String>("output3"){};
+ TupleTag<String> additionalOutputTagUnwritten = new TupleTag<String>("unwrittenOutput"){};
PCollectionTuple outputs = pipeline
.apply(Create.of(Arrays.asList(3, -42, 666))).setName("MyInput")
.apply("MyParDo", ParDo
.of(new TestDoFn(
Arrays.<PCollectionView<Integer>>asList(),
- Arrays.asList(sideOutputTag1, sideOutputTag2, sideOutputTag3)))
+ Arrays.asList(additionalOutputTag1, additionalOutputTag2, additionalOutputTag3)))
.withOutputTags(
mainOutputTag,
- TupleTagList.of(sideOutputTag3).and(sideOutputTag1)
- .and(sideOutputTagUnwritten).and(sideOutputTag2)));
+ TupleTagList.of(additionalOutputTag3).and(additionalOutputTag1)
+ .and(additionalOutputTagUnwritten).and(additionalOutputTag2)));
assertEquals("MyParDo.main", outputs.get(mainOutputTag).getName());
- assertEquals("MyParDo.side1", outputs.get(sideOutputTag1).getName());
- assertEquals("MyParDo.side2", outputs.get(sideOutputTag2).getName());
- assertEquals("MyParDo.side3", outputs.get(sideOutputTag3).getName());
- assertEquals("MyParDo.sideUnwritten",
- outputs.get(sideOutputTagUnwritten).getName());
+ assertEquals("MyParDo.output1", outputs.get(additionalOutputTag1).getName());
+ assertEquals("MyParDo.output2", outputs.get(additionalOutputTag2).getName());
+ assertEquals("MyParDo.output3", outputs.get(additionalOutputTag3).getName());
+ assertEquals("MyParDo.unwrittenOutput",
+ outputs.get(additionalOutputTagUnwritten).getName());
}
@Test
@@ -892,29 +892,29 @@ public class ParDoTest implements Serializable {
PCollection<Long> longs = pipeline.apply(CountingInput.unbounded());
TupleTag<Long> mainOut = new TupleTag<>();
- final TupleTag<String> sideOutOne = new TupleTag<>();
- final TupleTag<Integer> sideOutTwo = new TupleTag<>();
+ final TupleTag<String> valueAsString = new TupleTag<>();
+ final TupleTag<Integer> valueAsInt = new TupleTag<>();
DoFn<Long, Long> fn =
new DoFn<Long, Long>() {
@ProcessElement
public void processElement(ProcessContext cxt) {
cxt.output(cxt.element());
- cxt.sideOutput(sideOutOne, Long.toString(cxt.element()));
- cxt.sideOutput(sideOutTwo, Long.valueOf(cxt.element()).intValue());
+ cxt.output(valueAsString, Long.toString(cxt.element()));
+ cxt.output(valueAsInt, Long.valueOf(cxt.element()).intValue());
}
};
ParDo.MultiOutput<Long, Long> parDo =
- ParDo.of(fn).withOutputTags(mainOut, TupleTagList.of(sideOutOne).and(sideOutTwo));
+ ParDo.of(fn).withOutputTags(mainOut, TupleTagList.of(valueAsString).and(valueAsInt));
PCollectionTuple firstApplication = longs.apply("first", parDo);
PCollectionTuple secondApplication = longs.apply("second", parDo);
assertThat(firstApplication, not(equalTo(secondApplication)));
assertThat(
firstApplication.getAll().keySet(),
- Matchers.<TupleTag<?>>containsInAnyOrder(mainOut, sideOutOne, sideOutTwo));
+ Matchers.<TupleTag<?>>containsInAnyOrder(mainOut, valueAsString, valueAsInt));
assertThat(
secondApplication.getAll().keySet(),
- Matchers.<TupleTag<?>>containsInAnyOrder(mainOut, sideOutOne, sideOutTwo));
+ Matchers.<TupleTag<?>>containsInAnyOrder(mainOut, valueAsString, valueAsInt));
}
@Test
@@ -1017,28 +1017,28 @@ public class ParDoTest implements Serializable {
}
}
- private static class SideOutputDummyFn extends DoFn<Integer, Integer> {
- private TupleTag<TestDummy> sideTag;
- public SideOutputDummyFn(TupleTag<TestDummy> sideTag) {
- this.sideTag = sideTag;
+ private static class TaggedOutputDummyFn extends DoFn<Integer, Integer> {
+ private TupleTag<TestDummy> dummyOutputTag;
+ public TaggedOutputDummyFn(TupleTag<TestDummy> dummyOutputTag) {
+ this.dummyOutputTag = dummyOutputTag;
}
@ProcessElement
public void processElement(ProcessContext c) {
c.output(1);
- c.sideOutput(sideTag, new TestDummy());
+ c.output(dummyOutputTag, new TestDummy());
}
}
private static class MainOutputDummyFn extends DoFn<Integer, TestDummy> {
- private TupleTag<Integer> sideTag;
- public MainOutputDummyFn(TupleTag<Integer> sideTag) {
- this.sideTag = sideTag;
+ private TupleTag<Integer> intOutputTag;
+ public MainOutputDummyFn(TupleTag<Integer> intOutputTag) {
+ this.intOutputTag = intOutputTag;
}
@ProcessElement
public void processElement(ProcessContext c) {
c.output(new TestDummy());
- c.sideOutput(sideTag, 1);
+ c.output(intOutputTag, 1);
}
}
@@ -1112,7 +1112,7 @@ public class ParDoTest implements Serializable {
implements SerializableFunction<Iterable<String>, Void>, Serializable {
private final List<Integer> inputs;
private final List<Integer> sideInputs;
- private final String sideOutput;
+ private final String additionalOutput;
private final boolean ordered;
public static HasExpectedOutput forInput(List<Integer> inputs) {
@@ -1125,11 +1125,11 @@ public class ParDoTest implements Serializable {
private HasExpectedOutput(List<Integer> inputs,
List<Integer> sideInputs,
- String sideOutput,
+ String additionalOutput,
boolean ordered) {
this.inputs = inputs;
this.sideInputs = sideInputs;
- this.sideOutput = sideOutput;
+ this.additionalOutput = additionalOutput;
this.ordered = ordered;
}
@@ -1138,18 +1138,18 @@ public class ParDoTest implements Serializable {
for (Integer sideInputValue : sideInputValues) {
sideInputs.add(sideInputValue);
}
- return new HasExpectedOutput(inputs, sideInputs, sideOutput, ordered);
+ return new HasExpectedOutput(inputs, sideInputs, additionalOutput, ordered);
}
- public HasExpectedOutput fromSideOutput(TupleTag<String> sideOutputTag) {
- return fromSideOutput(sideOutputTag.getId());
+ public HasExpectedOutput fromOutput(TupleTag<String> outputTag) {
+ return fromOutput(outputTag.getId());
}
- public HasExpectedOutput fromSideOutput(String sideOutput) {
- return new HasExpectedOutput(inputs, sideInputs, sideOutput, ordered);
+ public HasExpectedOutput fromOutput(String outputId) {
+ return new HasExpectedOutput(inputs, sideInputs, outputId, ordered);
}
public HasExpectedOutput inOrder() {
- return new HasExpectedOutput(inputs, sideInputs, sideOutput, true);
+ return new HasExpectedOutput(inputs, sideInputs, additionalOutput, true);
}
@Override
@@ -1174,17 +1174,17 @@ public class ParDoTest implements Serializable {
sideInputsSuffix = ": " + sideInputs;
}
- String sideOutputPrefix;
- if (sideOutput.isEmpty()) {
- sideOutputPrefix = "";
+ String additionalOutputPrefix;
+ if (additionalOutput.isEmpty()) {
+ additionalOutputPrefix = "";
} else {
- sideOutputPrefix = sideOutput + ": ";
+ additionalOutputPrefix = additionalOutput + ": ";
}
List<String> expectedProcesseds = new ArrayList<>();
for (Integer input : inputs) {
expectedProcesseds.add(
- sideOutputPrefix + "processing: " + input + sideInputsSuffix);
+ additionalOutputPrefix + "processing: " + input + sideInputsSuffix);
}
String[] expectedProcessedsArray =
expectedProcesseds.toArray(new String[expectedProcesseds.size()]);
@@ -1196,10 +1196,10 @@ public class ParDoTest implements Serializable {
assertEquals(starteds.size(), finisheds.size());
for (String started : starteds) {
- assertEquals(sideOutputPrefix + "started", started);
+ assertEquals(additionalOutputPrefix + "started", started);
}
for (String finished : finisheds) {
- assertEquals(sideOutputPrefix + "finished", finished);
+ assertEquals(additionalOutputPrefix + "finished", finished);
}
return null;
@@ -1208,15 +1208,15 @@ public class ParDoTest implements Serializable {
@Test
@Category(NeedsRunner.class)
- public void testSideOutputUnknownCoder() throws Exception {
+ public void testTaggedOutputUnknownCoder() throws Exception {
PCollection<Integer> input = pipeline
.apply(Create.of(Arrays.asList(1, 2, 3)));
final TupleTag<Integer> mainOutputTag = new TupleTag<Integer>("main");
- final TupleTag<TestDummy> sideOutputTag = new TupleTag<TestDummy>("unknownSide");
- input.apply(ParDo.of(new SideOutputDummyFn(sideOutputTag))
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ final TupleTag<TestDummy> additionalOutputTag = new TupleTag<TestDummy>("unknownSide");
+ input.apply(ParDo.of(new TaggedOutputDummyFn(additionalOutputTag))
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
thrown.expect(IllegalStateException.class);
thrown.expectMessage("Unable to return a default Coder");
@@ -1224,26 +1224,27 @@ public class ParDoTest implements Serializable {
}
@Test
- public void testSideOutputUnregisteredExplicitCoder() throws Exception {
+ public void testTaggedOutputUnregisteredExplicitCoder() throws Exception {
pipeline.enableAbandonedNodeEnforcement(false);
PCollection<Integer> input = pipeline
.apply(Create.of(Arrays.asList(1, 2, 3)));
final TupleTag<Integer> mainOutputTag = new TupleTag<Integer>("main");
- final TupleTag<TestDummy> sideOutputTag = new TupleTag<TestDummy>("unregisteredSide");
- ParDo.MultiOutput<Integer, Integer> pardo = ParDo.of(new SideOutputDummyFn(sideOutputTag))
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag));
+ final TupleTag<TestDummy> additionalOutputTag = new TupleTag<TestDummy>("unregisteredSide");
+ ParDo.MultiOutput<Integer, Integer> pardo =
+ ParDo.of(new TaggedOutputDummyFn(additionalOutputTag))
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag));
PCollectionTuple outputTuple = input.apply(pardo);
- outputTuple.get(sideOutputTag).setCoder(new TestDummyCoder());
+ outputTuple.get(additionalOutputTag).setCoder(new TestDummyCoder());
- outputTuple.get(sideOutputTag).apply(View.<TestDummy>asSingleton());
+ outputTuple.get(additionalOutputTag).apply(View.<TestDummy>asSingleton());
- assertEquals(new TestDummyCoder(), outputTuple.get(sideOutputTag).getCoder());
- outputTuple.get(sideOutputTag).finishSpecifyingOutput(input, pardo); // Check for crashes
+ assertEquals(new TestDummyCoder(), outputTuple.get(additionalOutputTag).getCoder());
+ outputTuple.get(additionalOutputTag).finishSpecifyingOutput(input, pardo); // Check for crashes
assertEquals(new TestDummyCoder(),
- outputTuple.get(sideOutputTag).getCoder()); // Check for corruption
+ outputTuple.get(additionalOutputTag).getCoder()); // Check for corruption
}
@Test
@@ -1254,9 +1255,11 @@ public class ParDoTest implements Serializable {
.apply(Create.of(Arrays.asList(1, 2, 3)));
final TupleTag<TestDummy> mainOutputTag = new TupleTag<TestDummy>("unregisteredMain");
- final TupleTag<Integer> sideOutputTag = new TupleTag<Integer>("side") {};
- PCollectionTuple outputTuple = input.apply(ParDo.of(new MainOutputDummyFn(sideOutputTag))
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ final TupleTag<Integer> additionalOutputTag = new TupleTag<Integer>("additionalOutput") {};
+ PCollectionTuple outputTuple =
+ input.apply(
+ ParDo.of(new MainOutputDummyFn(additionalOutputTag))
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
outputTuple.get(mainOutputTag).setCoder(new TestDummyCoder());
@@ -1265,13 +1268,13 @@ public class ParDoTest implements Serializable {
@Test
@Category(NeedsRunner.class)
- public void testMainOutputApplySideOutputNoCoder() {
+ public void testMainOutputApplyTaggedOutputNoCoder() {
// Regression test: applying a transform to the main output
// should not cause a crash based on lack of a coder for the
- // side output.
+ // additional output.
final TupleTag<TestDummy> mainOutputTag = new TupleTag<TestDummy>("main");
- final TupleTag<TestDummy> sideOutputTag = new TupleTag<TestDummy>("side");
+ final TupleTag<TestDummy> additionalOutputTag = new TupleTag<TestDummy>("additionalOutput");
PCollectionTuple tuple = pipeline
.apply(Create.of(new TestDummy())
.withCoder(TestDummyCoder.of()))
@@ -1282,14 +1285,14 @@ public class ParDoTest implements Serializable {
public void processElement(ProcessContext context) {
TestDummy element = context.element();
context.output(element);
- context.sideOutput(sideOutputTag, element);
+ context.output(additionalOutputTag, element);
}
})
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag))
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag))
);
// Before fix, tuple.get(mainOutputTag).apply(...) would indirectly trigger
- // tuple.get(sideOutputTag).finishSpecifyingOutput(), which would crash
+ // tuple.get(additionalOutputTag).finishSpecifyingOutput(), which would crash
// on a missing coder.
tuple.get(mainOutputTag)
.setCoder(TestDummyCoder.of())
@@ -1300,7 +1303,7 @@ public class ParDoTest implements Serializable {
}
}));
- tuple.get(sideOutputTag).setCoder(TestDummyCoder.of());
+ tuple.get(additionalOutputTag).setCoder(TestDummyCoder.of());
pipeline.run();
}
@@ -1328,13 +1331,13 @@ public class ParDoTest implements Serializable {
@Test
@Category(NeedsRunner.class)
- public void testParDoSideOutputWithTimestamp() {
+ public void testParDoTaggedOutputWithTimestamp() {
PCollection<Integer> input =
pipeline.apply(Create.of(Arrays.asList(3, 42, 6)));
final TupleTag<Integer> mainOutputTag = new TupleTag<Integer>("main"){};
- final TupleTag<Integer> sideOutputTag = new TupleTag<Integer>("side"){};
+ final TupleTag<Integer> additionalOutputTag = new TupleTag<Integer>("additional"){};
PCollection<String> output =
input
@@ -1342,11 +1345,11 @@ public class ParDoTest implements Serializable {
new DoFn<Integer, Integer>() {
@ProcessElement
public void processElement(ProcessContext c) {
- c.sideOutputWithTimestamp(
- sideOutputTag, c.element(), new Instant(c.element().longValue()));
+ c.outputWithTimestamp(
+ additionalOutputTag, c.element(), new Instant(c.element().longValue()));
}
- }).withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)))
- .get(sideOutputTag)
+ }).withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)))
+ .get(additionalOutputTag)
.apply(ParDo.of(new TestShiftTimestampDoFn<Integer>(Duration.ZERO, Duration.ZERO)))
.apply(ParDo.of(new TestFormatTimestampDoFn<Integer>()));
@@ -1914,7 +1917,7 @@ public class ParDoTest implements Serializable {
@Test
@Category({ValidatesRunner.class, UsesStatefulParDo.class})
- public void testValueStateSideOutput() {
+ public void testValueStateTaggedOutput() {
final String stateId = "foo";
final TupleTag<Integer> evenTag = new TupleTag<Integer>() {};
@@ -1934,7 +1937,7 @@ public class ParDoTest implements Serializable {
if (currentValue % 2 == 0) {
c.output(currentValue);
} else {
- c.sideOutput(oddTag, currentValue);
+ c.output(oddTag, currentValue);
}
state.write(currentValue + 1);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
index a122f67..9e8c12e 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/SplittableDoFnTest.java
@@ -218,12 +218,12 @@ public class SplittableDoFnTest {
private static class SDFWithSideInputsAndOutputs extends DoFn<Integer, String> {
private final PCollectionView<String> sideInput;
- private final TupleTag<String> sideOutput;
+ private final TupleTag<String> additionalOutput;
private SDFWithSideInputsAndOutputs(
- PCollectionView<String> sideInput, TupleTag<String> sideOutput) {
+ PCollectionView<String> sideInput, TupleTag<String> additionalOutput) {
this.sideInput = sideInput;
- this.sideOutput = sideOutput;
+ this.additionalOutput = additionalOutput;
}
@ProcessElement
@@ -231,7 +231,7 @@ public class SplittableDoFnTest {
checkState(tracker.tryClaim(tracker.currentRestriction().getFrom()));
String side = c.sideInput(sideInput);
c.output("main:" + side + ":" + c.element());
- c.sideOutput(sideOutput, "side:" + side + ":" + c.element());
+ c.output(additionalOutput, "additional:" + side + ":" + c.element());
}
@GetInitialRestriction
@@ -247,21 +247,22 @@ public class SplittableDoFnTest {
PCollectionView<String> sideInput =
p.apply("side input", Create.of("foo")).apply(View.<String>asSingleton());
TupleTag<String> mainOutputTag = new TupleTag<>("main");
- TupleTag<String> sideOutputTag = new TupleTag<>("side");
+ TupleTag<String> additionalOutputTag = new TupleTag<>("additional");
PCollectionTuple res =
p.apply("input", Create.of(0, 1, 2))
.apply(
- ParDo.of(new SDFWithSideInputsAndOutputs(sideInput, sideOutputTag))
+ ParDo.of(new SDFWithSideInputsAndOutputs(sideInput, additionalOutputTag))
.withSideInputs(sideInput)
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
res.get(mainOutputTag).setCoder(StringUtf8Coder.of());
- res.get(sideOutputTag).setCoder(StringUtf8Coder.of());
+ res.get(additionalOutputTag).setCoder(StringUtf8Coder.of());
PAssert.that(res.get(mainOutputTag))
.containsInAnyOrder(Arrays.asList("main:foo:0", "main:foo:1", "main:foo:2"));
- PAssert.that(res.get(sideOutputTag))
- .containsInAnyOrder(Arrays.asList("side:foo:0", "side:foo:1", "side:foo:2"));
+ PAssert.that(res.get(additionalOutputTag))
+ .containsInAnyOrder(
+ Arrays.asList("additional:foo:0", "additional:foo:1", "additional:foo:2"));
p.run();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
index 0a0abd6..9df0512 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/PCollectionTupleTest.java
@@ -82,7 +82,7 @@ public final class PCollectionTupleTest implements Serializable {
TupleTag<Integer> mainOutputTag = new TupleTag<Integer>("main") {};
TupleTag<Integer> emptyOutputTag = new TupleTag<Integer>("empty") {};
- final TupleTag<Integer> sideOutputTag = new TupleTag<Integer>("side") {};
+ final TupleTag<Integer> additionalOutputTag = new TupleTag<Integer>("extra") {};
PCollection<Integer> mainInput = pipeline
.apply(Create.of(inputs));
@@ -91,14 +91,14 @@ public final class PCollectionTupleTest implements Serializable {
.of(new DoFn<Integer, Integer>() {
@ProcessElement
public void processElement(ProcessContext c) {
- c.sideOutput(sideOutputTag, c.element());
+ c.output(additionalOutputTag, c.element());
}})
- .withOutputTags(emptyOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(emptyOutputTag, TupleTagList.of(additionalOutputTag)));
assertNotNull("outputs.getPipeline()", outputs.getPipeline());
outputs = outputs.and(mainOutputTag, mainInput);
PAssert.that(outputs.get(mainOutputTag)).containsInAnyOrder(inputs);
- PAssert.that(outputs.get(sideOutputTag)).containsInAnyOrder(inputs);
+ PAssert.that(outputs.get(additionalOutputTag)).containsInAnyOrder(inputs);
PAssert.that(outputs.get(emptyOutputTag)).empty();
pipeline.run();
[4/4] beam git commit: Rename DoFn.Context#sideOutput to output
Posted by tg...@apache.org.
Rename DoFn.Context#sideOutput to output
Having two methods, both named output, one which takes the "main output
type" and one that takes a tag to specify the type more clearly
communicates the actual behavior - sideOutput isn't a "special" way to
output, it's the same as output(T), just to a specified PCollection.
This will help pipeline authors understand the actual behavior of
outputting to a tag, and detangle it from "sideInput", which is a
special way to receive input. Giving them the same name means that it's
not even strange to call output and provide the main output type, which
is what we want - it's a more specific way to output, but does not have
different restrictions or capabilities.
Rename internal references to SideOutput, SideOutputT, etc to (largely)
AdditionalOutput(T).
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/113471d6
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/113471d6
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/113471d6
Branch: refs/heads/master
Commit: 113471d6457b4afa2523afc74b40be09935292d0
Parents: 89ff0b1
Author: Thomas Groh <tg...@google.com>
Authored: Mon Apr 10 17:14:15 2017 -0700
Committer: Thomas Groh <tg...@google.com>
Committed: Fri Apr 14 14:40:57 2017 -0700
----------------------------------------------------------------------
.../apex/translation/ParDoTranslator.java | 6 +-
.../operators/ApexGroupByKeyOperator.java | 19 +-
.../operators/ApexParDoOperator.java | 48 +--
.../apex/translation/utils/NoOpStepContext.java | 2 +-
.../apex/translation/ParDoTranslatorTest.java | 18 +-
.../beam/runners/core/BaseExecutionContext.java | 13 +-
.../apache/beam/runners/core/DoFnAdapters.java | 16 +-
.../apache/beam/runners/core/DoFnRunners.java | 8 +-
.../beam/runners/core/ExecutionContext.java | 13 +-
.../GroupAlsoByWindowViaWindowSetNewDoFn.java | 6 +-
.../org/apache/beam/runners/core/OldDoFn.java | 38 +--
...eBoundedSplittableProcessElementInvoker.java | 8 +-
.../beam/runners/core/OutputWindowedValue.java | 10 +-
.../beam/runners/core/SimpleDoFnRunner.java | 54 ++--
.../beam/runners/core/SimpleOldDoFnRunner.java | 63 ++--
.../beam/runners/core/SplittableParDo.java | 21 +-
.../beam/runners/core/WindowingInternals.java | 8 +-
.../core/WindowingInternalsAdapters.java | 8 +-
.../core/GroupAlsoByWindowsProperties.java | 10 +-
.../apache/beam/runners/core/NoOpOldDoFn.java | 4 +-
.../apache/beam/runners/core/OldDoFnTest.java | 4 +-
...ndedSplittableProcessElementInvokerTest.java | 6 +-
.../beam/runners/core/ReduceFnTester.java | 8 +-
.../runners/core/SimpleOldDoFnRunnerTest.java | 4 +-
.../beam/runners/core/SplittableParDoTest.java | 8 +-
.../GroupAlsoByWindowEvaluatorFactory.java | 8 +-
.../beam/runners/direct/ParDoEvaluator.java | 4 +-
.../runners/direct/ParDoEvaluatorFactory.java | 10 +-
.../direct/ParDoMultiOverrideFactory.java | 2 +-
...littableProcessElementsEvaluatorFactory.java | 8 +-
.../direct/StatefulParDoEvaluatorFactory.java | 2 +-
.../beam/runners/direct/ParDoEvaluatorTest.java | 6 +-
.../FlinkStreamingTransformTranslators.java | 20 +-
.../functions/FlinkDoFnFunction.java | 4 +-
.../functions/FlinkNoOpStepContext.java | 2 +-
.../functions/FlinkStatefulDoFnFunction.java | 4 +-
.../wrappers/streaming/DoFnOperator.java | 14 +-
.../streaming/SplittableDoFnOperator.java | 10 +-
.../wrappers/streaming/WindowDoFnOperator.java | 4 +-
.../flink/streaming/DoFnOperatorTest.java | 34 +--
.../dataflow/BatchStatefulParDoOverrides.java | 2 +-
.../runners/dataflow/BatchViewOverrides.java | 6 +-
.../dataflow/BatchViewOverridesTest.java | 4 +-
.../SparkGroupAlsoByWindowViaWindowSet.java | 10 +-
.../spark/translation/MultiDoFnFunction.java | 4 +-
...SparkGroupAlsoByWindowViaOutputBufferFn.java | 8 +-
.../spark/translation/SparkProcessContext.java | 2 +-
.../streaming/StreamingTransformTranslator.java | 2 +-
.../org/apache/beam/sdk/transforms/Combine.java | 2 +-
.../org/apache/beam/sdk/transforms/DoFn.java | 14 +-
.../apache/beam/sdk/transforms/DoFnTester.java | 41 ++-
.../org/apache/beam/sdk/transforms/ParDo.java | 77 +++--
.../apache/beam/sdk/transforms/Partition.java | 2 +-
.../beam/sdk/values/PCollectionTuple.java | 3 +-
.../org/apache/beam/sdk/values/TupleTag.java | 26 +-
.../apache/beam/sdk/values/TupleTagList.java | 2 +-
.../org/apache/beam/sdk/values/TypedPValue.java | 4 +-
.../apache/beam/sdk/metrics/MetricsTest.java | 2 +-
.../apache/beam/sdk/transforms/ParDoTest.java | 293 ++++++++++---------
.../beam/sdk/transforms/SplittableDoFnTest.java | 21 +-
.../beam/sdk/values/PCollectionTupleTest.java | 8 +-
.../apache/beam/sdk/values/TypedPValueTest.java | 46 +--
.../beam/fn/harness/fake/FakeStepContext.java | 2 +-
.../control/ProcessBundleHandlerTest.java | 30 +-
.../sdk/io/gcp/bigquery/WritePartition.java | 6 +-
.../sdk/io/gcp/bigquery/BigQueryIOTest.java | 4 +-
66 files changed, 578 insertions(+), 578 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java
index 9213c1f..2e3d902 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoTranslator.java
@@ -94,7 +94,7 @@ class ParDoTranslator<InputT, OutputT>
context.getPipelineOptions(),
doFn,
transform.getMainOutputTag(),
- transform.getSideOutputTags().getAll(),
+ transform.getAdditionalOutputTags().getAll(),
input.getWindowingStrategy(),
sideInputs,
wvInputCoder,
@@ -114,9 +114,9 @@ class ParDoTranslator<InputT, OutputT>
ports.put(pc, operator.output);
} else {
int portIndex = 0;
- for (TupleTag<?> tag : transform.getSideOutputTags().getAll()) {
+ for (TupleTag<?> tag : transform.getAdditionalOutputTags().getAll()) {
if (tag.equals(output.getKey())) {
- ports.put(pc, operator.sideOutputPorts[portIndex]);
+ ports.put(pc, operator.additionalOutputPorts[portIndex]);
break;
}
portIndex++;
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
index 230082e..1697921 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
@@ -353,13 +353,14 @@ public class ApexGroupByKeyOperator<K, V> implements Operator {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- throw new UnsupportedOperationException("GroupAlsoByWindow should not use side outputs");
+ throw new UnsupportedOperationException(
+ "GroupAlsoByWindow should not use tagged outputs");
}
@Override
@@ -390,15 +391,13 @@ public class ApexGroupByKeyOperator<K, V> implements Operator {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- // ignore the side output, this can happen when a user does not register
- // side outputs but then outputs using a freshly created TupleTag.
- throw new RuntimeException("sideOutput() is not available when grouping by window.");
+ public <T> void output(TupleTag<T> tag, T output) {
+ throw new RuntimeException("output() is not available when grouping by window.");
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- sideOutput(tag, output);
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ output(tag, output);
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java
index 1fc91c8..bad5be2 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java
@@ -88,7 +88,7 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
@Bind(JavaSerializer.class)
private final TupleTag<OutputT> mainOutputTag;
@Bind(JavaSerializer.class)
- private final List<TupleTag<?>> sideOutputTags;
+ private final List<TupleTag<?>> additionalOutputTags;
@Bind(JavaSerializer.class)
private final WindowingStrategy<?, ?> windowingStrategy;
@Bind(JavaSerializer.class)
@@ -108,15 +108,15 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
private transient PushbackSideInputDoFnRunner<InputT, OutputT> pushbackDoFnRunner;
private transient SideInputHandler sideInputHandler;
- private transient Map<TupleTag<?>, DefaultOutputPort<ApexStreamTuple<?>>> sideOutputPortMapping =
- Maps.newHashMapWithExpectedSize(5);
+ private transient Map<TupleTag<?>, DefaultOutputPort<ApexStreamTuple<?>>>
+ additionalOutputPortMapping = Maps.newHashMapWithExpectedSize(5);
private transient DoFnInvoker<InputT, OutputT> doFnInvoker;
public ApexParDoOperator(
ApexPipelineOptions pipelineOptions,
DoFn<InputT, OutputT> doFn,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
WindowingStrategy<?, ?> windowingStrategy,
List<PCollectionView<?>> sideInputs,
Coder<WindowedValue<InputT>> inputCoder,
@@ -125,15 +125,15 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
this.pipelineOptions = new SerializablePipelineOptions(pipelineOptions);
this.doFn = doFn;
this.mainOutputTag = mainOutputTag;
- this.sideOutputTags = sideOutputTags;
+ this.additionalOutputTags = additionalOutputTags;
this.windowingStrategy = windowingStrategy;
this.sideInputs = sideInputs;
this.sideInputStateInternals = new StateInternalsProxy<>(
stateBackend.newStateInternalsFactory(VoidCoder.of()));
- if (sideOutputTags.size() > sideOutputPorts.length) {
- String msg = String.format("Too many side outputs (currently only supporting %s).",
- sideOutputPorts.length);
+ if (additionalOutputTags.size() > additionalOutputPorts.length) {
+ String msg = String.format("Too many additional outputs (currently only supporting %s).",
+ additionalOutputPorts.length);
throw new UnsupportedOperationException(msg);
}
@@ -148,7 +148,7 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
this.pipelineOptions = null;
this.doFn = null;
this.mainOutputTag = null;
- this.sideOutputTags = null;
+ this.additionalOutputTags = null;
this.windowingStrategy = null;
this.sideInputs = null;
this.pushedBack = null;
@@ -218,29 +218,31 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
public final transient DefaultOutputPort<ApexStreamTuple<?>> output = new DefaultOutputPort<>();
@OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<ApexStreamTuple<?>> sideOutput1 =
+ public final transient DefaultOutputPort<ApexStreamTuple<?>> additionalOutput1 =
new DefaultOutputPort<>();
@OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<ApexStreamTuple<?>> sideOutput2 =
+ public final transient DefaultOutputPort<ApexStreamTuple<?>> additionalOutput2 =
new DefaultOutputPort<>();
@OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<ApexStreamTuple<?>> sideOutput3 =
+ public final transient DefaultOutputPort<ApexStreamTuple<?>> additionalOutput3 =
new DefaultOutputPort<>();
@OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<ApexStreamTuple<?>> sideOutput4 =
+ public final transient DefaultOutputPort<ApexStreamTuple<?>> additionalOutput4 =
new DefaultOutputPort<>();
@OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<ApexStreamTuple<?>> sideOutput5 =
+ public final transient DefaultOutputPort<ApexStreamTuple<?>> additionalOutput5 =
new DefaultOutputPort<>();
- public final transient DefaultOutputPort<?>[] sideOutputPorts = {sideOutput1, sideOutput2,
- sideOutput3, sideOutput4, sideOutput5};
+ public final transient DefaultOutputPort<?>[] additionalOutputPorts = {
+ additionalOutput1, additionalOutput2, additionalOutput3, additionalOutput4, additionalOutput5
+ };
@Override
public <T> void output(TupleTag<T> tag, WindowedValue<T> tuple) {
- DefaultOutputPort<ApexStreamTuple<?>> sideOutputPort = sideOutputPortMapping.get(tag);
- if (sideOutputPort != null) {
- sideOutputPort.emit(ApexStreamTuple.DataTuple.of(tuple));
+ DefaultOutputPort<ApexStreamTuple<?>> additionalOutputPort =
+ additionalOutputPortMapping.get(tag);
+ if (additionalOutputPort != null) {
+ additionalOutputPort.emit(ApexStreamTuple.DataTuple.of(tuple));
} else {
output.emit(ApexStreamTuple.DataTuple.of(tuple));
}
@@ -306,11 +308,11 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
sideInputReader = sideInputHandler;
}
- for (int i = 0; i < sideOutputTags.size(); i++) {
+ for (int i = 0; i < additionalOutputTags.size(); i++) {
@SuppressWarnings("unchecked")
DefaultOutputPort<ApexStreamTuple<?>> port = (DefaultOutputPort<ApexStreamTuple<?>>)
- sideOutputPorts[i];
- sideOutputPortMapping.put(sideOutputTags.get(i), port);
+ additionalOutputPorts[i];
+ additionalOutputPortMapping.put(additionalOutputTags.get(i), port);
}
NoOpStepContext stepContext = new NoOpStepContext() {
@@ -332,7 +334,7 @@ public class ApexParDoOperator<InputT, OutputT> extends BaseOperator implements
sideInputReader,
this,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
new NoOpAggregatorFactory(),
windowingStrategy
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java
index ad4de97..cc64c7c 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/utils/NoOpStepContext.java
@@ -48,7 +48,7 @@ public class NoOpStepContext implements ExecutionContext.StepContext, Serializab
}
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoTranslatorTest.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoTranslatorTest.java b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoTranslatorTest.java
index 2760d06..1a5c8be 100644
--- a/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoTranslatorTest.java
+++ b/runners/apex/src/test/java/org/apache/beam/runners/apex/translation/ParDoTranslatorTest.java
@@ -267,7 +267,7 @@ public class ParDoTranslatorTest {
List<Integer> inputs = Arrays.asList(3, -42, 666);
final TupleTag<String> mainOutputTag = new TupleTag<>("main");
- final TupleTag<Void> sideOutputTag = new TupleTag<>("sideOutput");
+ final TupleTag<Void> additionalOutputTag = new TupleTag<>("output");
PCollectionView<Integer> sideInput1 = pipeline
.apply("CreateSideInput1", Create.of(11))
@@ -288,10 +288,10 @@ public class ParDoTranslatorTest {
.withSideInputs(sideInput1)
.withSideInputs(sideInputUnread)
.withSideInputs(sideInput2)
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
outputs.get(mainOutputTag).apply(ParDo.of(new EmbeddedCollector()));
- outputs.get(sideOutputTag).setCoder(VoidCoder.of());
+ outputs.get(additionalOutputTag).setCoder(VoidCoder.of());
ApexRunnerResult result = (ApexRunnerResult) pipeline.run();
HashSet<String> expected = Sets.newHashSet("processing: 3: [11, 222]",
@@ -312,12 +312,12 @@ public class ParDoTranslatorTest {
private static final long serialVersionUID = 1L;
final List<PCollectionView<Integer>> sideInputViews = new ArrayList<>();
- final List<TupleTag<String>> sideOutputTupleTags = new ArrayList<>();
+ final List<TupleTag<String>> additionalOutputTupleTags = new ArrayList<>();
public TestMultiOutputWithSideInputsFn(List<PCollectionView<Integer>> sideInputViews,
- List<TupleTag<String>> sideOutputTupleTags) {
+ List<TupleTag<String>> additionalOutputTupleTags) {
this.sideInputViews.addAll(sideInputViews);
- this.sideOutputTupleTags.addAll(sideOutputTupleTags);
+ this.additionalOutputTupleTags.addAll(additionalOutputTupleTags);
}
@ProcessElement
@@ -334,9 +334,9 @@ public class ParDoTranslatorTest {
value += ": " + sideInputValues;
}
c.output(value);
- for (TupleTag<String> sideOutputTupleTag : sideOutputTupleTags) {
- c.sideOutput(sideOutputTupleTag,
- sideOutputTupleTag.getId() + ": " + value);
+ for (TupleTag<String> additionalOutputTupleTag : additionalOutputTupleTags) {
+ c.output(additionalOutputTupleTag,
+ additionalOutputTupleTag.getId() + ": " + value);
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java
index 0f23fea..cc7b574 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/BaseExecutionContext.java
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.DoFn.Context;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.TupleTag;
@@ -106,19 +107,17 @@ public abstract class BaseExecutionContext<T extends ExecutionContext.StepContex
/**
* Hook for subclasses to implement that will be called whenever
- * {@code DoFn.Context#output}
- * is called.
+ * {@link Context#output(Object)} is called.
*/
@Override
public void noteOutput(WindowedValue<?> output) {}
/**
* Hook for subclasses to implement that will be called whenever
- * {@code DoFn.Context#sideOutput}
- * is called.
+ * {@link Context#output(TupleTag, Object)} is called.
*/
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {}
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {}
/**
* Base class for implementations of {@link ExecutionContext.StepContext}.
@@ -153,8 +152,8 @@ public abstract class BaseExecutionContext<T extends ExecutionContext.StepContex
}
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {
- executionContext.noteSideOutput(tag, output);
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {
+ executionContext.noteOutput(tag, output);
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
index deb3b7e..66ad736 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnAdapters.java
@@ -162,13 +162,13 @@ public class DoFnAdapters {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- context.sideOutput(tag, output);
+ public <T> void output(TupleTag<T> tag, T output) {
+ context.output(tag, output);
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- context.sideOutputWithTimestamp(tag, output, timestamp);
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ context.outputWithTimestamp(tag, output, timestamp);
}
@Override
@@ -255,13 +255,13 @@ public class DoFnAdapters {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- context.sideOutput(tag, output);
+ public <T> void output(TupleTag<T> tag, T output) {
+ context.output(tag, output);
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- context.sideOutputWithTimestamp(tag, output, timestamp);
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ context.outputWithTimestamp(tag, output, timestamp);
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
index a1b7c8b..b09ee08 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunners.java
@@ -59,7 +59,7 @@ public class DoFnRunners {
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
StepContext stepContext,
AggregatorFactory aggregatorFactory,
WindowingStrategy<?, ?> windowingStrategy) {
@@ -69,7 +69,7 @@ public class DoFnRunners {
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorFactory,
windowingStrategy);
@@ -86,7 +86,7 @@ public class DoFnRunners {
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
StepContext stepContext,
AggregatorFactory aggregatorFactory,
WindowingStrategy<?, ?> windowingStrategy) {
@@ -96,7 +96,7 @@ public class DoFnRunners {
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorFactory,
windowingStrategy);
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java
index 40c0798..ecd30c0 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ExecutionContext.java
@@ -20,6 +20,7 @@ package org.apache.beam.runners.core;
import java.io.IOException;
import java.util.Collection;
import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.DoFn.Context;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.TupleTag;
@@ -41,17 +42,15 @@ public interface ExecutionContext {
/**
* Hook for subclasses to implement that will be called whenever
- * {@link org.apache.beam.sdk.transforms.DoFn.Context#output}
- * is called.
+ * {@link Context#output(TupleTag, Object)} is called.
*/
void noteOutput(WindowedValue<?> output);
/**
* Hook for subclasses to implement that will be called whenever
- * {@link org.apache.beam.sdk.transforms.DoFn.Context#sideOutput}
- * is called.
+ * {@link Context#output(TupleTag, Object)} is called.
*/
- void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output);
+ void noteOutput(TupleTag<?> tag, WindowedValue<?> output);
/**
* Per-step, per-key context used for retrieving state.
@@ -77,10 +76,10 @@ public interface ExecutionContext {
/**
* Hook for subclasses to implement that will be called whenever
- * {@link org.apache.beam.sdk.transforms.DoFn.Context#sideOutput}
+ * {@link org.apache.beam.sdk.transforms.DoFn.Context#output}
* is called.
*/
- void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output);
+ void noteOutput(TupleTag<?> tag, WindowedValue<?> output);
/**
* Writes the given {@code PCollectionView} data to a globally accessible location.
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
index 8fff0e4..0cf6e2d 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
@@ -104,9 +104,9 @@ public class GroupAlsoByWindowViaWindowSetNewDoFn<
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java
index e9d4740..507ee50 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java
@@ -135,16 +135,15 @@ public abstract class OldDoFn<InputT, OutputT> implements Serializable, HasDispl
public abstract void outputWithTimestamp(OutputT output, Instant timestamp);
/**
- * Adds the given element to the side output {@code PCollection} with the
+ * Adds the given element to the output {@code PCollection} with the
* given tag.
*
- * <p>Once passed to {@code sideOutput} the element should not be modified
+ * <p>Once passed to {@code output} the element should not be modified
* in any way.
*
* <p>The caller of {@code ParDo} uses {@link ParDo.SingleOutput#withOutputTags withOutputTags}
- * to specify the tags of side outputs that it consumes. Non-consumed side
- * outputs, e.g., outputs for monitoring purposes only, don't necessarily
- * need to be specified.
+ * to specify the tags of outputs that it consumes. Outputs that are not consumed, e.g., outputs
+ * for monitoring purposes only, don't necessarily need to be specified.
*
* <p>The output element will have the same timestamp and be in the same
* windows as the input element passed to {@link OldDoFn#processElement processElement}.
@@ -159,32 +158,27 @@ public abstract class OldDoFn<InputT, OutputT> implements Serializable, HasDispl
*
* @see ParDo.SingleOutput#withOutputTags
*/
- public abstract <T> void sideOutput(TupleTag<T> tag, T output);
+ public abstract <T> void output(TupleTag<T> tag, T output);
/**
- * Adds the given element to the specified side output {@code PCollection},
- * with the given timestamp.
+ * Adds the given element to the specified output {@code PCollection}, with the given timestamp.
*
- * <p>Once passed to {@code sideOutputWithTimestamp} the element should not be
- * modified in any way.
+ * <p>Once passed to {@code outputWithTimestamp} the element should not be modified in any way.
*
- * <p>If invoked from {@link OldDoFn#processElement processElement}, the timestamp
- * must not be older than the input element's timestamp minus
- * {@link OldDoFn#getAllowedTimestampSkew getAllowedTimestampSkew}. The output element will
- * be in the same windows as the input element.
+ * <p>If invoked from {@link OldDoFn#processElement processElement}, the timestamp must not be
+ * older than the input element's timestamp minus {@link OldDoFn#getAllowedTimestampSkew
+ * getAllowedTimestampSkew}. The output element will be in the same windows as the input
+ * element.
*
* <p>If invoked from {@link #startBundle startBundle} or {@link #finishBundle finishBundle},
- * this will attempt to use the
- * {@link org.apache.beam.sdk.transforms.windowing.WindowFn}
- * of the input {@code PCollection} to determine what windows the element
- * should be in, throwing an exception if the {@code WindowFn} attempts
- * to access any information about the input element except for the
- * timestamp.
+ * this will attempt to use the {@link org.apache.beam.sdk.transforms.windowing.WindowFn} of the
+ * input {@code PCollection} to determine what windows the element should be in, throwing an
+ * exception if the {@code WindowFn} attempts to access any information about the input element
+ * except for the timestamp.
*
* @see ParDo.SingleOutput#withOutputTags
*/
- public abstract <T> void sideOutputWithTimestamp(
- TupleTag<T> tag, T output, Instant timestamp);
+ public abstract <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp);
/**
* Creates an {@link Aggregator} in the {@link OldDoFn} context with the
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
index 27fd0a3..d132af6 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java
@@ -254,13 +254,13 @@ public class OutputAndTimeBoundedSplittableProcessElementInvoker<
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T value) {
- sideOutputWithTimestamp(tag, value, element.getTimestamp());
+ public <T> void output(TupleTag<T> tag, T value) {
+ outputWithTimestamp(tag, value, element.getTimestamp());
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T value, Instant timestamp) {
- output.sideOutputWindowedValue(
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T value, Instant timestamp) {
+ output.outputWindowedValue(
tag, value, timestamp, element.getWindows(), element.getPane());
noteOutput();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputWindowedValue.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputWindowedValue.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputWindowedValue.java
index 86eeb33..35d6737 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputWindowedValue.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OutputWindowedValue.java
@@ -25,7 +25,7 @@ import org.joda.time.Instant;
/**
* An object that can output a value with all of its windowing information to the main output or
- * a side output.
+ * any tagged output.
*/
public interface OutputWindowedValue<OutputT> {
/** Outputs a value with windowing information to the main output. */
@@ -35,10 +35,10 @@ public interface OutputWindowedValue<OutputT> {
Collection<? extends BoundedWindow> windows,
PaneInfo pane);
- /** Outputs a value with windowing information to a side output. */
- <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ /** Outputs a value with windowing information to a tagged output. */
+ <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane);
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
index 98d88b6..141bf20 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
@@ -106,7 +106,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
StepContext stepContext,
AggregatorFactory aggregatorFactory,
WindowingStrategy<?, ?> windowingStrategy) {
@@ -133,7 +133,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorFactory,
windowingStrategy.getWindowFn());
@@ -257,7 +257,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
StepContext stepContext,
AggregatorFactory aggregatorFactory,
WindowFn<?, ?> windowFn) {
@@ -270,8 +270,8 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
this.outputTags = Sets.newHashSet();
outputTags.add(mainOutputTag);
- for (TupleTag<?> sideOutputTag : sideOutputTags) {
- outputTags.add(sideOutputTag);
+ for (TupleTag<?> additionalOutputTag : additionalOutputTags) {
+ outputTags.add(additionalOutputTag);
}
this.stepContext = stepContext;
@@ -355,16 +355,16 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
}
}
- private <T> void sideOutputWindowedValue(
+ private <T> void outputWindowedValue(
TupleTag<T> tag,
T output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- sideOutputWindowedValue(tag, makeWindowedValue(output, timestamp, windows, pane));
+ outputWindowedValue(tag, makeWindowedValue(output, timestamp, windows, pane));
}
- private <T> void sideOutputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedElem) {
+ private <T> void outputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedElem) {
if (!outputTags.contains(tag)) {
// This tag wasn't declared nor was it seen before during this execution.
// Thus, this must be a new, undeclared and unconsumed output.
@@ -372,18 +372,18 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
// outputs.
if (outputTags.size() >= MAX_SIDE_OUTPUTS) {
throw new IllegalArgumentException(
- "the number of side outputs has exceeded a limit of " + MAX_SIDE_OUTPUTS);
+ "the number of outputs has exceeded a limit of " + MAX_SIDE_OUTPUTS);
}
outputTags.add(tag);
}
outputManager.output(tag, windowedElem);
if (stepContext != null) {
- stepContext.noteSideOutput(tag, windowedElem);
+ stepContext.noteOutput(tag, windowedElem);
}
}
- // Following implementations of output, outputWithTimestamp, and sideOutput
+ // Following implementations of output, outputWithTimestamp, and output
// are only accessible in DoFn.startBundle and DoFn.finishBundle, and will be shadowed by
// ProcessContext's versions in DoFn.processElement.
@Override
@@ -397,15 +397,15 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- checkNotNull(tag, "TupleTag passed to sideOutput cannot be null");
- sideOutputWindowedValue(tag, output, null, null, PaneInfo.NO_FIRING);
+ public <T> void output(TupleTag<T> tag, T output) {
+ checkNotNull(tag, "TupleTag passed to output cannot be null");
+ outputWindowedValue(tag, output, null, null, PaneInfo.NO_FIRING);
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- checkNotNull(tag, "TupleTag passed to sideOutputWithTimestamp cannot be null");
- sideOutputWindowedValue(tag, output, timestamp, null, PaneInfo.NO_FIRING);
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ checkNotNull(tag, "TupleTag passed to outputWithTimestamp cannot be null");
+ outputWindowedValue(tag, output, timestamp, null, PaneInfo.NO_FIRING);
}
@Override
@@ -559,16 +559,16 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- checkNotNull(tag, "Tag passed to sideOutput cannot be null");
- context.sideOutputWindowedValue(tag, windowedValue.withValue(output));
+ public <T> void output(TupleTag<T> tag, T output) {
+ checkNotNull(tag, "Tag passed to output cannot be null");
+ context.outputWindowedValue(tag, windowedValue.withValue(output));
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- checkNotNull(tag, "Tag passed to sideOutputWithTimestamp cannot be null");
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ checkNotNull(tag, "Tag passed to outputWithTimestamp cannot be null");
checkTimestamp(timestamp);
- context.sideOutputWindowedValue(
+ context.outputWindowedValue(
tag, output, timestamp, windowedValue.getWindows(), windowedValue.getPane());
}
@@ -787,14 +787,14 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- context.sideOutputWindowedValue(
+ public <T> void output(TupleTag<T> tag, T output) {
+ context.outputWindowedValue(
tag, output, timestamp, Collections.singleton(window()), PaneInfo.NO_FIRING);
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- context.sideOutputWindowedValue(
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ context.outputWindowedValue(
tag, output, timestamp, Collections.singleton(window()), PaneInfo.NO_FIRING);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleOldDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleOldDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleOldDoFnRunner.java
index c88f1c9..6320a3a 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleOldDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleOldDoFnRunner.java
@@ -60,11 +60,16 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
/** The context used for running the {@link OldDoFn}. */
private final DoFnContext<InputT, OutputT> context;
- public SimpleOldDoFnRunner(PipelineOptions options, OldDoFn<InputT, OutputT> fn,
+ public SimpleOldDoFnRunner(
+ PipelineOptions options,
+ OldDoFn<InputT, OutputT> fn,
SideInputReader sideInputReader,
OutputManager outputManager,
- TupleTag<OutputT> mainOutputTag, List<TupleTag<?>> sideOutputTags, StepContext stepContext,
- AggregatorFactory aggregatorFactory, WindowingStrategy<?, ?> windowingStrategy) {
+ TupleTag<OutputT> mainOutputTag,
+ List<TupleTag<?>> additionalOutputTags,
+ StepContext stepContext,
+ AggregatorFactory aggregatorFactory,
+ WindowingStrategy<?, ?> windowingStrategy) {
this.fn = fn;
this.context = new DoFnContext<>(
options,
@@ -72,7 +77,7 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorFactory,
windowingStrategy == null ? null : windowingStrategy.getWindowFn());
@@ -177,7 +182,7 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
SideInputReader sideInputReader,
OutputManager outputManager,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
StepContext stepContext,
AggregatorFactory aggregatorFactory,
WindowFn<?, ?> windowFn) {
@@ -190,8 +195,8 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
this.outputTags = Sets.newHashSet();
outputTags.add(mainOutputTag);
- for (TupleTag<?> sideOutputTag : sideOutputTags) {
- outputTags.add(sideOutputTag);
+ for (TupleTag<?> additionalOutputTag : additionalOutputTags) {
+ outputTags.add(additionalOutputTag);
}
this.stepContext = stepContext;
@@ -273,15 +278,15 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
}
}
- private <T> void sideOutputWindowedValue(TupleTag<T> tag,
+ private <T> void outputWindowedValue(TupleTag<T> tag,
T output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- sideOutputWindowedValue(tag, makeWindowedValue(output, timestamp, windows, pane));
+ outputWindowedValue(tag, makeWindowedValue(output, timestamp, windows, pane));
}
- private <T> void sideOutputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedElem) {
+ private <T> void outputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedElem) {
if (!outputTags.contains(tag)) {
// This tag wasn't declared nor was it seen before during this execution.
// Thus, this must be a new, undeclared and unconsumed output.
@@ -289,18 +294,18 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
// outputs.
if (outputTags.size() >= MAX_SIDE_OUTPUTS) {
throw new IllegalArgumentException(
- "the number of side outputs has exceeded a limit of " + MAX_SIDE_OUTPUTS);
+ "the number of outputs has exceeded a limit of " + MAX_SIDE_OUTPUTS);
}
outputTags.add(tag);
}
outputManager.output(tag, windowedElem);
if (stepContext != null) {
- stepContext.noteSideOutput(tag, windowedElem);
+ stepContext.noteOutput(tag, windowedElem);
}
}
- // Following implementations of output, outputWithTimestamp, and sideOutput
+ // Following implementations of output, outputWithTimestamp, and output
// are only accessible in OldDoFn.startBundle and OldDoFn.finishBundle, and will be shadowed by
// ProcessContext's versions in OldDoFn.processElement.
@Override
@@ -314,15 +319,15 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- checkNotNull(tag, "TupleTag passed to sideOutput cannot be null");
- sideOutputWindowedValue(tag, output, null, null, PaneInfo.NO_FIRING);
+ public <T> void output(TupleTag<T> tag, T output) {
+ checkNotNull(tag, "TupleTag passed to output cannot be null");
+ outputWindowedValue(tag, output, null, null, PaneInfo.NO_FIRING);
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- checkNotNull(tag, "TupleTag passed to sideOutputWithTimestamp cannot be null");
- sideOutputWindowedValue(tag, output, timestamp, null, PaneInfo.NO_FIRING);
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ checkNotNull(tag, "TupleTag passed to outputWithTimestamp cannot be null");
+ outputWindowedValue(tag, output, timestamp, null, PaneInfo.NO_FIRING);
}
@Override
@@ -428,16 +433,16 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
- checkNotNull(tag, "Tag passed to sideOutput cannot be null");
- context.sideOutputWindowedValue(tag, windowedValue.withValue(output));
+ public <T> void output(TupleTag<T> tag, T output) {
+ checkNotNull(tag, "Tag passed to output cannot be null");
+ context.outputWindowedValue(tag, windowedValue.withValue(output));
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
- checkNotNull(tag, "Tag passed to sideOutputWithTimestamp cannot be null");
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ checkNotNull(tag, "Tag passed to outputWithTimestamp cannot be null");
checkTimestamp(timestamp);
- context.sideOutputWindowedValue(
+ context.outputWindowedValue(
tag, output, timestamp, windowedValue.getWindows(), windowedValue.getPane());
}
@@ -471,13 +476,13 @@ class SimpleOldDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, OutputT
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- context.sideOutputWindowedValue(tag, output, timestamp, windows, pane);
+ context.outputWindowedValue(tag, output, timestamp, windows, pane);
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
index c16bf44..9cc965a 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDo.java
@@ -118,7 +118,7 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
input.getWindowingStrategy(),
parDo.getSideInputs(),
parDo.getMainOutputTag(),
- parDo.getSideOutputTags()));
+ parDo.getAdditionalOutputTags()));
}
private static <InputT, OutputT, RestrictionT>
@@ -188,14 +188,15 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
private final WindowingStrategy<?, ?> windowingStrategy;
private final List<PCollectionView<?>> sideInputs;
private final TupleTag<OutputT> mainOutputTag;
- private final TupleTagList sideOutputTags;
+ private final TupleTagList additionalOutputTags;
/**
* @param fn the splittable {@link DoFn}.
* @param windowingStrategy the {@link WindowingStrategy} of the input collection.
* @param sideInputs list of side inputs that should be available to the {@link DoFn}.
* @param mainOutputTag {@link TupleTag Tag} of the {@link DoFn DoFn's} main output.
- * @param sideOutputTags {@link TupleTagList Tags} of the {@link DoFn DoFn's} side outputs.
+ * @param additionalOutputTags {@link TupleTagList Tags} of the {@link DoFn DoFn's} additional
+ * outputs.
*/
public ProcessElements(
DoFn<InputT, OutputT> fn,
@@ -204,14 +205,14 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
WindowingStrategy<?, ?> windowingStrategy,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- TupleTagList sideOutputTags) {
+ TupleTagList additionalOutputTags) {
this.fn = fn;
this.elementCoder = elementCoder;
this.restrictionCoder = restrictionCoder;
this.windowingStrategy = windowingStrategy;
this.sideInputs = sideInputs;
this.mainOutputTag = mainOutputTag;
- this.sideOutputTags = sideOutputTags;
+ this.additionalOutputTags = additionalOutputTags;
}
public DoFn<InputT, OutputT> getFn() {
@@ -226,8 +227,8 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
return mainOutputTag;
}
- public TupleTagList getSideOutputTags() {
- return sideOutputTags;
+ public TupleTagList getAdditionalOutputTags() {
+ return additionalOutputTags;
}
public ProcessFn<InputT, OutputT, RestrictionT, TrackerT> newProcessFn(
@@ -244,7 +245,7 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
PCollectionTuple outputs =
PCollectionTuple.ofPrimitiveOutputsInternal(
input.getPipeline(),
- TupleTagList.of(mainOutputTag).and(sideOutputTags.getAll()),
+ TupleTagList.of(mainOutputTag).and(additionalOutputTags.getAll()),
windowingStrategy,
input.isBounded().and(signature.isBoundedPerElement()));
@@ -522,12 +523,12 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
+ public <T> void output(TupleTag<T> tag, T output) {
throwUnsupportedOutput();
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
throwUnsupportedOutput();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternals.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternals.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternals.java
index 8dc0bfc..5005065 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternals.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternals.java
@@ -49,11 +49,11 @@ public interface WindowingInternals<InputT, OutputT> {
Collection<? extends BoundedWindow> windows, PaneInfo pane);
/**
- * Output the value to a side output at the specified timestamp in the listed windows.
+ * Output the value to a tagged output at the specified timestamp in the listed windows.
*/
- <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane);
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternalsAdapters.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternalsAdapters.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternalsAdapters.java
index 48a39d6..1b36bf9 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternalsAdapters.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WindowingInternalsAdapters.java
@@ -62,13 +62,13 @@ public class WindowingInternalsAdapters {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- windowingInternals.sideOutputWindowedValue(tag, output, timestamp, windows, pane);
+ windowingInternals.outputWindowedValue(tag, output, timestamp, windows, pane);
}
};
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/GroupAlsoByWindowsProperties.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/GroupAlsoByWindowsProperties.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/GroupAlsoByWindowsProperties.java
index 6c7c4e0..d0a8923 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/GroupAlsoByWindowsProperties.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/GroupAlsoByWindowsProperties.java
@@ -677,9 +677,9 @@ public class GroupAlsoByWindowsProperties {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
@@ -729,12 +729,12 @@ public class GroupAlsoByWindowsProperties {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
+ public <T> void output(TupleTag<T> tag, T output) {
throw new UnsupportedOperationException();
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/NoOpOldDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/NoOpOldDoFn.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/NoOpOldDoFn.java
index 5cbea8c..2e5cd6d 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/NoOpOldDoFn.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/NoOpOldDoFn.java
@@ -57,10 +57,10 @@ class NoOpOldDoFn<InputT, OutputT> extends OldDoFn<InputT, OutputT> {
public void outputWithTimestamp(OutputT output, Instant timestamp) {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
+ public <T> void output(TupleTag<T> tag, T output) {
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output,
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output,
Instant timestamp) {
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/OldDoFnTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/OldDoFnTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/OldDoFnTest.java
index 651bc72..425de07 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/OldDoFnTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/OldDoFnTest.java
@@ -160,12 +160,12 @@ public class OldDoFnTest implements Serializable {
}
@Override
- public <T> void sideOutput(TupleTag<T> tag, T output) {
+ public <T> void output(TupleTag<T> tag, T output) {
throw new UnsupportedOperationException();
}
@Override
- public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+ public <T> void outputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
throw new UnsupportedOperationException();
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java
index 965380b..541e238 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvokerTest.java
@@ -80,9 +80,9 @@ public class OutputAndTimeBoundedSplittableProcessElementInvokerTest {
PaneInfo pane) {}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index 512420f..914550e 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -574,13 +574,13 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- throw new UnsupportedOperationException("GroupAlsoByWindow should not use side outputs");
+ throw new UnsupportedOperationException("GroupAlsoByWindow should not use tagged outputs");
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleOldDoFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleOldDoFnRunnerTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleOldDoFnRunnerTest.java
index 28698ca..8ded2dc 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleOldDoFnRunnerTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleOldDoFnRunnerTest.java
@@ -64,10 +64,10 @@ public class SimpleOldDoFnRunnerTest {
private DoFnRunner<String, String> createRunner(OldDoFn<String, String> fn) {
// Pass in only necessary parameters for the test
- List<TupleTag<?>> sideOutputTags = Arrays.asList();
+ List<TupleTag<?>> additionalOutputTags = Arrays.asList();
StepContext context = mock(StepContext.class);
return new SimpleOldDoFnRunner<>(
- null, fn, null, null, null, sideOutputTags, context, null, null);
+ null, fn, null, null, null, additionalOutputTags, context, null, null);
}
static class ThrowingDoFn extends OldDoFn<String, String> {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
index d301113..2c89543 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/SplittableParDoTest.java
@@ -356,13 +356,13 @@ public class SplittableParDoTest {
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- sideOutputWindowedValue(tester.getMainOutputTag(), output, timestamp, windows, pane);
+ outputWindowedValue(tester.getMainOutputTag(), output, timestamp, windows, pane);
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
index b4ca998..ce7b12a 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
@@ -264,13 +264,13 @@ class GroupAlsoByWindowEvaluatorFactory implements TransformEvaluatorFactory {
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
- throw new UnsupportedOperationException("GroupAlsoByWindow should not use side outputs");
+ throw new UnsupportedOperationException("GroupAlsoByWindow should not use tagged outputs");
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
index 328d139..49d0723 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluator.java
@@ -51,7 +51,7 @@ class ParDoEvaluator<InputT, OutputT> implements TransformEvaluator<InputT> {
StructuralKey<?> key,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
Map<TupleTag<?>, PCollection<?>> outputs) {
AggregatorContainer.Mutator aggregatorChanges = evaluationContext.getAggregatorMutator();
@@ -80,7 +80,7 @@ class ParDoEvaluator<InputT, OutputT> implements TransformEvaluator<InputT> {
sideInputReader,
outputManager,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
aggregatorChanges,
windowingStrategy);
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
index b8a13e2..0372295 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoEvaluatorFactory.java
@@ -80,7 +80,7 @@ final class ParDoEvaluatorFactory<InputT, OutputT> implements TransformEvaluator
doFn,
transform.getSideInputs(),
transform.getMainOutputTag(),
- transform.getSideOutputTags().getAll());
+ transform.getAdditionalOutputTags().getAll());
return evaluator;
}
@@ -103,7 +103,7 @@ final class ParDoEvaluatorFactory<InputT, OutputT> implements TransformEvaluator
DoFn<InputT, OutputT> doFn,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags)
+ List<TupleTag<?>> additionalOutputTags)
throws Exception {
String stepName = evaluationContext.getStepName(application);
DirectStepContext stepContext =
@@ -119,7 +119,7 @@ final class ParDoEvaluatorFactory<InputT, OutputT> implements TransformEvaluator
inputBundleKey,
sideInputs,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
stepContext,
fnManager.<InputT, OutputT>get(),
fnManager),
@@ -131,7 +131,7 @@ final class ParDoEvaluatorFactory<InputT, OutputT> implements TransformEvaluator
StructuralKey<?> key,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
DirectStepContext stepContext,
DoFn<InputT, OutputT> fn,
DoFnLifecycleManager fnManager)
@@ -147,7 +147,7 @@ final class ParDoEvaluatorFactory<InputT, OutputT> implements TransformEvaluator
key,
sideInputs,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
pcollections(application.getOutputs()));
} catch (Exception e) {
try {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
index 00c0d6a..366777b 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
@@ -192,7 +192,7 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
PCollectionTuple.ofPrimitiveOutputsInternal(
input.getPipeline(),
TupleTagList.of(underlyingParDo.getMainOutputTag())
- .and(underlyingParDo.getSideOutputTags().getAll()),
+ .and(underlyingParDo.getAdditionalOutputTags().getAll()),
input.getWindowingStrategy(),
input.isBounded());
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java
index 07affd8..64cef35 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SplittableProcessElementsEvaluatorFactory.java
@@ -105,7 +105,7 @@ class SplittableProcessElementsEvaluatorFactory<
inputBundle.getKey(),
transform.getSideInputs(),
transform.getMainOutputTag(),
- transform.getSideOutputTags().getAll(),
+ transform.getAdditionalOutputTags().getAll(),
stepContext,
processFn,
fnManager);
@@ -146,9 +146,9 @@ class SplittableProcessElementsEvaluatorFactory<
}
@Override
- public <SideOutputT> void sideOutputWindowedValue(
- TupleTag<SideOutputT> tag,
- SideOutputT output,
+ public <AdditionalOutputT> void outputWindowedValue(
+ TupleTag<AdditionalOutputT> tag,
+ AdditionalOutputT output,
Instant timestamp,
Collection<? extends BoundedWindow> windows,
PaneInfo pane) {
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
index f8fe3d6..be77ea1 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
@@ -120,7 +120,7 @@ final class StatefulParDoEvaluatorFactory<K, InputT, OutputT> implements Transfo
doFn,
application.getTransform().getUnderlyingParDo().getSideInputs(),
application.getTransform().getUnderlyingParDo().getMainOutputTag(),
- application.getTransform().getUnderlyingParDo().getSideOutputTags().getAll());
+ application.getTransform().getUnderlyingParDo().getAdditionalOutputTags().getAll());
return new StatefulParDoEvaluator<>(delegateEvaluator);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
index 2a94d48..65a1248 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ParDoEvaluatorTest.java
@@ -70,7 +70,7 @@ public class ParDoEvaluatorTest {
@Mock private EvaluationContext evaluationContext;
private PCollection<Integer> inputPc;
private TupleTag<Integer> mainOutputTag;
- private List<TupleTag<?>> sideOutputTags;
+ private List<TupleTag<?>> additionalOutputTags;
private BundleFactory bundleFactory;
@Rule
@@ -81,7 +81,7 @@ public class ParDoEvaluatorTest {
MockitoAnnotations.initMocks(this);
inputPc = p.apply(Create.of(1, 2, 3));
mainOutputTag = new TupleTag<Integer>() {};
- sideOutputTags = TupleTagList.empty().getAll();
+ additionalOutputTags = TupleTagList.empty().getAll();
bundleFactory = ImmutableListBundleFactory.create();
}
@@ -168,7 +168,7 @@ public class ParDoEvaluatorTest {
null /* key */,
ImmutableList.<PCollectionView<?>>of(singletonView),
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
ImmutableMap.<TupleTag<?>, PCollection<?>>of(mainOutputTag, output));
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
index af157f0..fbd7620 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkStreamingTransformTranslators.java
@@ -406,7 +406,7 @@ class FlinkStreamingTransformTranslators {
DoFn<InputT, OutputT> doFn,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
FlinkStreamingTranslationContext context,
WindowingStrategy<?, ?> windowingStrategy,
Map<TupleTag<?>, Integer> tagsToLabels,
@@ -422,7 +422,7 @@ class FlinkStreamingTransformTranslators {
List<PCollectionView<?>> sideInputs,
Map<TupleTag<?>, PValue> outputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
FlinkStreamingTranslationContext context,
DoFnOperatorFactory<InputT, OutputT> doFnOperatorFactory) {
@@ -460,7 +460,7 @@ class FlinkStreamingTransformTranslators {
doFn,
sideInputs,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
context,
windowingStrategy,
tagsToLabels,
@@ -485,7 +485,7 @@ class FlinkStreamingTransformTranslators {
doFn,
sideInputs,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
context,
windowingStrategy,
tagsToLabels,
@@ -605,7 +605,7 @@ class FlinkStreamingTransformTranslators {
transform.getSideInputs(),
context.getOutputs(transform),
transform.getMainOutputTag(),
- transform.getSideOutputTags().getAll(),
+ transform.getAdditionalOutputTags().getAll(),
context,
new ParDoTranslationHelper.DoFnOperatorFactory<InputT, OutputT>() {
@Override
@@ -613,7 +613,7 @@ class FlinkStreamingTransformTranslators {
DoFn<InputT, OutputT> doFn,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
FlinkStreamingTranslationContext context,
WindowingStrategy<?, ?> windowingStrategy,
Map<TupleTag<?>, Integer> tagsToLabels,
@@ -624,7 +624,7 @@ class FlinkStreamingTransformTranslators {
doFn,
inputCoder,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
new DoFnOperator.MultiOutputOutputManagerFactory(tagsToLabels),
windowingStrategy,
transformedSideInputs,
@@ -654,7 +654,7 @@ class FlinkStreamingTransformTranslators {
transform.getSideInputs(),
context.getOutputs(transform),
transform.getMainOutputTag(),
- transform.getSideOutputTags().getAll(),
+ transform.getAdditionalOutputTags().getAll(),
context,
new ParDoTranslationHelper.DoFnOperatorFactory<
KeyedWorkItem<String, ElementAndRestriction<InputT, RestrictionT>>, OutputT>() {
@@ -668,7 +668,7 @@ class FlinkStreamingTransformTranslators {
OutputT> doFn,
List<PCollectionView<?>> sideInputs,
TupleTag<OutputT> mainOutputTag,
- List<TupleTag<?>> sideOutputTags,
+ List<TupleTag<?>> additionalOutputTags,
FlinkStreamingTranslationContext context,
WindowingStrategy<?, ?> windowingStrategy,
Map<TupleTag<?>, Integer> tagsToLabels,
@@ -683,7 +683,7 @@ class FlinkStreamingTransformTranslators {
doFn,
inputCoder,
mainOutputTag,
- sideOutputTags,
+ additionalOutputTags,
new DoFnOperator.MultiOutputOutputManagerFactory(tagsToLabels),
windowingStrategy,
transformedSideInputs,
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
index 9687478..51582af 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
@@ -87,7 +87,7 @@ public class FlinkDoFnFunction<InputT, OutputT>
if (outputMap == null) {
outputManager = new FlinkDoFnFunction.DoFnOutputManager(out);
} else {
- // it has some sideOutputs
+ // it has some additional outputs
outputManager =
new FlinkDoFnFunction.MultiDoFnOutputManager((Collector) out, outputMap);
}
@@ -97,7 +97,7 @@ public class FlinkDoFnFunction<InputT, OutputT>
new FlinkSideInputReader(sideInputs, runtimeContext),
outputManager,
mainOutputTag,
- // see SimpleDoFnRunner, just use it to limit number of side outputs
+ // see SimpleDoFnRunner, just use it to limit number of additional outputs
Collections.<TupleTag<?>>emptyList(),
new FlinkNoOpStepContext(),
new FlinkAggregatorFactory(runtimeContext),
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java
index d901d8e..847a00a 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkNoOpStepContext.java
@@ -47,7 +47,7 @@ public class FlinkNoOpStepContext implements StepContext {
}
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java
index 0d8399e..c8193d2 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkStatefulDoFnFunction.java
@@ -88,7 +88,7 @@ public class FlinkStatefulDoFnFunction<K, V, OutputT>
if (outputMap == null) {
outputManager = new FlinkDoFnFunction.DoFnOutputManager(out);
} else {
- // it has some sideOutputs
+ // it has some additional Outputs
outputManager =
new FlinkDoFnFunction.MultiDoFnOutputManager((Collector) out, outputMap);
}
@@ -114,7 +114,7 @@ public class FlinkStatefulDoFnFunction<K, V, OutputT>
new FlinkSideInputReader(sideInputs, runtimeContext),
outputManager,
mainOutputTag,
- // see SimpleDoFnRunner, just use it to limit number of side outputs
+ // see SimpleDoFnRunner, just use it to limit number of additional outputs
Collections.<TupleTag<?>>emptyList(),
new FlinkNoOpStepContext() {
@Override
[2/4] beam git commit: Rename DoFn.Context#sideOutput to output
Posted by tg...@apache.org.
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/core/src/test/java/org/apache/beam/sdk/values/TypedPValueTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/TypedPValueTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/TypedPValueTest.java
index 211dfd9..f752b1c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/values/TypedPValueTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/values/TypedPValueTest.java
@@ -55,12 +55,12 @@ public class TypedPValueTest {
}
private PCollectionTuple buildPCollectionTupleWithTags(
- TupleTag<Integer> mainOutputTag, TupleTag<Integer> sideOutputTag) {
+ TupleTag<Integer> mainOutputTag, TupleTag<Integer> additionalOutputTag) {
PCollection<Integer> input = p.apply(Create.of(1, 2, 3));
PCollectionTuple tuple = input.apply(
ParDo
.of(new IdentityDoFn())
- .withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag)));
+ .withOutputTags(mainOutputTag, TupleTagList.of(additionalOutputTag)));
return tuple;
}
@@ -69,11 +69,11 @@ public class TypedPValueTest {
}
@Test
- public void testUntypedSideOutputTupleTagGivesActionableMessage() {
+ public void testUntypedOutputTupleTagGivesActionableMessage() {
TupleTag<Integer> mainOutputTag = new TupleTag<Integer>() {};
- // untypedSideOutputTag did not use anonymous subclass.
- TupleTag<Integer> untypedSideOutputTag = new TupleTag<Integer>();
- PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, untypedSideOutputTag);
+ // untypedOutputTag did not use anonymous subclass.
+ TupleTag<Integer> untypedOutputTag = new TupleTag<Integer>();
+ PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, untypedOutputTag);
thrown.expect(IllegalStateException.class);
thrown.expectMessage("No Coder has been manually specified");
@@ -84,15 +84,15 @@ public class TypedPValueTest {
thrown.expectMessage(
containsString("Building a Coder from the fallback CoderProvider failed"));
- tuple.get(untypedSideOutputTag).getCoder();
+ tuple.get(untypedOutputTag).getCoder();
}
@Test
- public void testStaticFactorySideOutputTupleTagGivesActionableMessage() {
+ public void testStaticFactoryOutputTupleTagGivesActionableMessage() {
TupleTag<Integer> mainOutputTag = new TupleTag<Integer>() {};
- // untypedSideOutputTag constructed from a static factory method.
- TupleTag<Integer> untypedSideOutputTag = makeTagStatically();
- PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, untypedSideOutputTag);
+ // untypedOutputTag constructed from a static factory method.
+ TupleTag<Integer> untypedOutputTag = makeTagStatically();
+ PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, untypedOutputTag);
thrown.expect(IllegalStateException.class);
thrown.expectMessage("No Coder has been manually specified");
@@ -103,27 +103,27 @@ public class TypedPValueTest {
thrown.expectMessage(
containsString("Building a Coder from the fallback CoderProvider failed"));
- tuple.get(untypedSideOutputTag).getCoder();
+ tuple.get(untypedOutputTag).getCoder();
}
@Test
- public void testTypedSideOutputTupleTag() {
+ public void testTypedOutputTupleTag() {
TupleTag<Integer> mainOutputTag = new TupleTag<Integer>() {};
- // typedSideOutputTag was constructed with compile-time type information.
- TupleTag<Integer> typedSideOutputTag = new TupleTag<Integer>() {};
- PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, typedSideOutputTag);
+ // typedOutputTag was constructed with compile-time type information.
+ TupleTag<Integer> typedOutputTag = new TupleTag<Integer>() {};
+ PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, typedOutputTag);
- assertThat(tuple.get(typedSideOutputTag).getCoder(), instanceOf(VarIntCoder.class));
+ assertThat(tuple.get(typedOutputTag).getCoder(), instanceOf(VarIntCoder.class));
}
@Test
- public void testUntypedMainOutputTagTypedSideOutputTupleTag() {
+ public void testUntypedMainOutputTagTypedOutputTupleTag() {
// mainOutputTag is allowed to be untyped because Coder can be inferred other ways.
TupleTag<Integer> mainOutputTag = new TupleTag<>();
- TupleTag<Integer> typedSideOutputTag = new TupleTag<Integer>() {};
- PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, typedSideOutputTag);
+ TupleTag<Integer> typedOutputTag = new TupleTag<Integer>() {};
+ PCollectionTuple tuple = buildPCollectionTupleWithTags(mainOutputTag, typedOutputTag);
- assertThat(tuple.get(typedSideOutputTag).getCoder(), instanceOf(VarIntCoder.class));
+ assertThat(tuple.get(typedOutputTag).getCoder(), instanceOf(VarIntCoder.class));
}
// A simple class for which there should be no obvious Coder.
@@ -139,13 +139,13 @@ public class TypedPValueTest {
}
@Test
- public void testParDoWithNoSideOutputsErrorDoesNotMentionTupleTag() {
+ public void testParDoWithNoOutputsErrorDoesNotMentionTupleTag() {
PCollection<EmptyClass> input =
p.apply(Create.of(1, 2, 3)).apply(ParDo.of(new EmptyClassDoFn()));
thrown.expect(IllegalStateException.class);
- // Output specific to ParDo TupleTag side outputs should not be present.
+ // Output specific to ParDo additional TupleTag outputs should not be present.
thrown.expectMessage(not(containsString("erasure")));
thrown.expectMessage(not(containsString("see TupleTag Javadoc")));
// Instead, expect output suggesting other possible fixes.
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java
----------------------------------------------------------------------
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java
index 6403e96..9714d72 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/fake/FakeStepContext.java
@@ -46,7 +46,7 @@ public class FakeStepContext implements StepContext {
}
@Override
- public void noteSideOutput(TupleTag<?> tag, WindowedValue<?> output) {
+ public void noteOutput(TupleTag<?> tag, WindowedValue<?> output) {
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
index de105d7..bd2fba9 100644
--- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
+++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/control/ProcessBundleHandlerTest.java
@@ -300,7 +300,7 @@ public class ProcessBundleHandlerTest {
private static class TestDoFn extends DoFn<String, String> {
private static final TupleTag<String> mainOutput = new TupleTag<>("mainOutput");
- private static final TupleTag<String> sideOutput = new TupleTag<>("sideOutput");
+ private static final TupleTag<String> additionalOutput = new TupleTag<>("output");
@StartBundle
public void startBundle(Context context) {
@@ -310,7 +310,7 @@ public class ProcessBundleHandlerTest {
@ProcessElement
public void processElement(ProcessContext context) {
context.output("MainOutput" + context.element());
- context.sideOutput(sideOutput, "SideOutput" + context.element());
+ context.output(additionalOutput, "AdditionalOutput" + context.element());
}
@FinishBundle
@@ -321,7 +321,7 @@ public class ProcessBundleHandlerTest {
/**
* Create a DoFn that has 3 inputs (inputATarget1, inputATarget2, inputBTarget) and 2 outputs
- * (mainOutput, sideOutput). Validate that inputs are fed to the {@link DoFn} and that outputs
+ * (mainOutput, output). Validate that inputs are fed to the {@link DoFn} and that outputs
* are directed to the correct consumers.
*/
@Test
@@ -329,7 +329,7 @@ public class ProcessBundleHandlerTest {
Map<String, Message> fnApiRegistry = ImmutableMap.of(STRING_CODER_SPEC_ID, STRING_CODER_SPEC);
String primitiveTransformId = "100L";
long mainOutputId = 101L;
- long sideOutputId = 102L;
+ long additionalOutputId = 102L;
DoFnInfo<?, ?> doFnInfo = DoFnInfo.forFn(
new TestDoFn(),
@@ -339,7 +339,7 @@ public class ProcessBundleHandlerTest {
mainOutputId,
ImmutableMap.of(
mainOutputId, TestDoFn.mainOutput,
- sideOutputId, TestDoFn.sideOutput));
+ additionalOutputId, TestDoFn.additionalOutput));
BeamFnApi.FunctionSpec functionSpec = BeamFnApi.FunctionSpec.newBuilder()
.setId("1L")
.setUrn(JAVA_DO_FN_URN)
@@ -372,25 +372,25 @@ public class ProcessBundleHandlerTest {
.putOutputs(Long.toString(mainOutputId), BeamFnApi.PCollection.newBuilder()
.setCoderReference(STRING_CODER_SPEC_ID)
.build())
- .putOutputs(Long.toString(sideOutputId), BeamFnApi.PCollection.newBuilder()
+ .putOutputs(Long.toString(additionalOutputId), BeamFnApi.PCollection.newBuilder()
.setCoderReference(STRING_CODER_SPEC_ID)
.build())
.build();
List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
- List<WindowedValue<String>> sideOutputValues = new ArrayList<>();
+ List<WindowedValue<String>> additionalOutputValues = new ArrayList<>();
BeamFnApi.Target mainOutputTarget = BeamFnApi.Target.newBuilder()
.setPrimitiveTransformReference(primitiveTransformId)
.setName(Long.toString(mainOutputId))
.build();
- BeamFnApi.Target sideOutputTarget = BeamFnApi.Target.newBuilder()
+ BeamFnApi.Target additionalOutputTarget = BeamFnApi.Target.newBuilder()
.setPrimitiveTransformReference(primitiveTransformId)
- .setName(Long.toString(sideOutputId))
+ .setName(Long.toString(additionalOutputId))
.build();
Multimap<BeamFnApi.Target, ThrowingConsumer<WindowedValue<String>>> existingConsumers =
ImmutableMultimap.of(
mainOutputTarget, mainOutputValues::add,
- sideOutputTarget, sideOutputValues::add);
+ additionalOutputTarget, additionalOutputValues::add);
Multimap<BeamFnApi.Target, ThrowingConsumer<WindowedValue<String>>> newConsumers =
HashMultimap.create();
List<ThrowingRunnable> startFunctions = new ArrayList<>();
@@ -422,12 +422,12 @@ public class ProcessBundleHandlerTest {
valueInGlobalWindow("MainOutputA1"),
valueInGlobalWindow("MainOutputA2"),
valueInGlobalWindow("MainOutputB")));
- assertThat(sideOutputValues, contains(
- valueInGlobalWindow("SideOutputA1"),
- valueInGlobalWindow("SideOutputA2"),
- valueInGlobalWindow("SideOutputB")));
+ assertThat(additionalOutputValues, contains(
+ valueInGlobalWindow("AdditionalOutputA1"),
+ valueInGlobalWindow("AdditionalOutputA2"),
+ valueInGlobalWindow("AdditionalOutputB")));
mainOutputValues.clear();
- sideOutputValues.clear();
+ additionalOutputValues.clear();
Iterables.getOnlyElement(finishFunctions).run();
assertThat(mainOutputValues, contains(valueInGlobalWindow("FinishBundle")));
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
index f4bf198..1b6492e 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WritePartition.java
@@ -61,7 +61,7 @@ class WritePartition extends DoFn<String, KV<Long, List<String>>> {
KV<String, Long> fileResult = results.get(i);
if (currNumFiles + 1 > Write.MAX_NUM_FILES
|| currSizeBytes + fileResult.getValue() > Write.MAX_SIZE_BYTES) {
- c.sideOutput(multiPartitionsTag, KV.of(++partitionId, currResults));
+ c.output(multiPartitionsTag, KV.of(++partitionId, currResults));
currResults = Lists.newArrayList();
currNumFiles = 0;
currSizeBytes = 0;
@@ -71,9 +71,9 @@ class WritePartition extends DoFn<String, KV<Long, List<String>>> {
currResults.add(fileResult.getKey());
}
if (partitionId == 0) {
- c.sideOutput(singlePartitionTag, KV.of(++partitionId, currResults));
+ c.output(singlePartitionTag, KV.of(++partitionId, currResults));
} else {
- c.sideOutput(multiPartitionsTag, KV.of(++partitionId, currResults));
+ c.output(multiPartitionsTag, KV.of(++partitionId, currResults));
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/113471d6/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
index 92ab204..2a2bf91 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTest.java
@@ -2107,9 +2107,9 @@ public class BigQueryIOTest implements Serializable {
List<KV<Long, List<String>>> partitions;
if (expectedNumPartitions > 1) {
- partitions = tester.takeSideOutputElements(multiPartitionsTag);
+ partitions = tester.takeOutputElements(multiPartitionsTag);
} else {
- partitions = tester.takeSideOutputElements(singlePartitionTag);
+ partitions = tester.takeOutputElements(singlePartitionTag);
}
List<Long> partitionIds = Lists.newArrayList();
List<String> partitionFileNames = Lists.newArrayList();