You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/10/14 22:11:59 UTC
[GitHub] [beam] kennknowles opened a new pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
kennknowles opened a new pull request #13118:
URL: https://github.com/apache/beam/pull/13118
While developing #13053 I discovered that some output replacements were not being re-wired correctly, because the PValues were at different levels of expansion: since a PCollectionView is still a PValue it can sneak into some of these Maps that are only supposed to have PCollections.
In this PR, I make all the maps only allow PCollections to eliminate that source of error. To keep things reasonably small I am proposing this one separate from #13053
I also consolidate some pieces into the core SDK when they are tightly related, like `PTransformReplacement` and `PTransformReplacements` (note the plural, used to indicate a companion class). I also deleted some dead code in the files I touched.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r509407958
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformReplacements.java
##########
@@ -15,20 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.core.construction;
+package org.apache.beam.sdk.runners;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
import java.util.Map;
import java.util.Set;
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
/** */
+@Internal
Review comment:
We agree on replaceAll like functionality not existing in sdks core but I don't think having yet another layer on top of the protos would be much better then a few static methods that do common things that are necessary to easily walk and mutate a graph.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r509442928
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+
+/**
+ * <b><i>For internal use. No backwards compatibility guarantees.</i></b>
+ *
+ * <p>A primitive value within Beam.
+ */
+@Internal
+public class PValues {
+
+ // Do not instantiate
+ private PValues() {}
+
+ // For backwards-compatibility, PCollectionView is still a "PValue" to users, which occurs in
Review comment:
Done
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
Review comment:
I put it here because it is a [companion](https://docs.scala-lang.org/overviews/scala-book/companion-objects.html) to `PValue`. It is conventional for that to be the plural name, in the same package. If you still want me to move it after this explanation, I will. I don't care enough to block on it.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformReplacements.java
##########
@@ -15,20 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.core.construction;
+package org.apache.beam.sdk.runners;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
import java.util.Map;
import java.util.Set;
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
/** */
+@Internal
Review comment:
Resolved: I am removing the moves from the PR to avoid getting blocked on anything that could reopen discussions.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+
+/**
+ * <b><i>For internal use. No backwards compatibility guarantees.</i></b>
+ *
+ * <p>A primitive value within Beam.
+ */
+@Internal
+public class PValues {
+
+ // Do not instantiate
+ private PValues() {}
+
+ // For backwards-compatibility, PCollectionView is still a "PValue" to users, which occurs in
+ // three places:
+ //
+ // POutput#expand (users can write custom POutputs)
+ // PInput#expand (users can write custom PInputs)
+ // PTransform#getAdditionalInputs (users can have their composites report inputs not passed by
+ // apply())
+ //
+ // These all return Map<TupleTag<?> PValue>. A user's implementation of these methods is permitted
+ // to return
+ // either a PCollection or a PCollectionView for each PValue. PCollection's expand to themselves
+ // and
+ // PCollectionView expands to the PCollection that it is a view of.
+ public static Map<TupleTag<?>, PCollection<?>> fullyExpand(
+ Map<TupleTag<?>, PValue> partiallyExpanded) {
+ Map<TupleTag<?>, PCollection<?>> result = new LinkedHashMap<>();
+ for (Map.Entry<TupleTag<?>, PValue> pvalue : partiallyExpanded.entrySet()) {
+ if (pvalue.getValue() instanceof PCollection) {
+ PCollection<?> previous = result.put(pvalue.getKey(), (PCollection<?>) pvalue.getValue());
+ checkArgument(
+ previous == null,
+ "Found conflicting %ss in flattened expansion of %s: %s maps to %s and %s",
+ partiallyExpanded,
+ TupleTag.class.getSimpleName(),
+ pvalue.getKey(),
+ previous,
+ pvalue.getValue());
+ } else {
+ if (pvalue.getValue().expand().size() == 1
+ && Iterables.getOnlyElement(pvalue.getValue().expand().values())
+ .equals(pvalue.getValue())) {
+ throw new IllegalStateException(
+ String.format(
+ "Non %s %s that expands into itself %s",
+ PCollection.class.getSimpleName(),
+ PValue.class.getSimpleName(),
+ pvalue.getValue()));
+ }
+ // At this point we know it is a PCollectionView or some internal hacked PValue. To be
Review comment:
Done
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
##########
@@ -1403,4 +1427,55 @@ public int size() {
};
}
}
+
+ public static <InputT, ViewT> PCollectionView<ViewT> findPCollectionView(
Review comment:
I've moved it into the test class where it is used. I would still like to leave the alteration of `DataflowRunner` out of this PR, since my other PR that just adds a `checkState` illustrates that the `DataflowRunner` batch view overrides result in a corrupted graph that sort of works by luck. I don't want to disturb that potentially sensitive situation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #13118: [BEAM-11094] Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r514578489
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/runners/TransformHierarchy.java
##########
@@ -483,12 +402,12 @@ private void setOutput(POutput output) {
checkState(
this.outputs == null, "Tried to specify more than one output for %s", getFullName());
checkNotNull(output, "Tried to set the output of %s to null", getFullName());
- this.outputs = output.expand();
+ this.outputs = PValues.fullyExpand(output.expand());
Review comment:
I think this line and the similar one below below made it so a `PCollectionView` is not in `outputs`, so topological traversal does not eagerly process `PCollectionView` as a value, and PCollectionViews also do not have a producer (which is correct for the Beam model). This seems to affect only Dataflow and only because it has special logic which treats PCollectionView as a value in its v1beta3 translation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709628568
KinesisIO flake is not relevant.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-712487284
It would have been helpful to have made the main part of the change as a separate commit from all the renames / `PValues#expand` changes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r508101186
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+
+/**
+ * <b><i>For internal use. No backwards compatibility guarantees.</i></b>
+ *
+ * <p>A primitive value within Beam.
+ */
+@Internal
+public class PValues {
+
+ // Do not instantiate
+ private PValues() {}
+
+ // For backwards-compatibility, PCollectionView is still a "PValue" to users, which occurs in
Review comment:
Might as well and make this javadoc.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.apache.beam.sdk.annotations.Internal;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
+
+/**
+ * <b><i>For internal use. No backwards compatibility guarantees.</i></b>
+ *
+ * <p>A primitive value within Beam.
+ */
+@Internal
+public class PValues {
+
+ // Do not instantiate
+ private PValues() {}
+
+ // For backwards-compatibility, PCollectionView is still a "PValue" to users, which occurs in
+ // three places:
+ //
+ // POutput#expand (users can write custom POutputs)
+ // PInput#expand (users can write custom PInputs)
+ // PTransform#getAdditionalInputs (users can have their composites report inputs not passed by
+ // apply())
+ //
+ // These all return Map<TupleTag<?> PValue>. A user's implementation of these methods is permitted
+ // to return
+ // either a PCollection or a PCollectionView for each PValue. PCollection's expand to themselves
+ // and
+ // PCollectionView expands to the PCollection that it is a view of.
+ public static Map<TupleTag<?>, PCollection<?>> fullyExpand(
+ Map<TupleTag<?>, PValue> partiallyExpanded) {
+ Map<TupleTag<?>, PCollection<?>> result = new LinkedHashMap<>();
+ for (Map.Entry<TupleTag<?>, PValue> pvalue : partiallyExpanded.entrySet()) {
+ if (pvalue.getValue() instanceof PCollection) {
+ PCollection<?> previous = result.put(pvalue.getKey(), (PCollection<?>) pvalue.getValue());
+ checkArgument(
+ previous == null,
+ "Found conflicting %ss in flattened expansion of %s: %s maps to %s and %s",
+ partiallyExpanded,
+ TupleTag.class.getSimpleName(),
+ pvalue.getKey(),
+ previous,
+ pvalue.getValue());
+ } else {
+ if (pvalue.getValue().expand().size() == 1
+ && Iterables.getOnlyElement(pvalue.getValue().expand().values())
+ .equals(pvalue.getValue())) {
+ throw new IllegalStateException(
+ String.format(
+ "Non %s %s that expands into itself %s",
+ PCollection.class.getSimpleName(),
+ PValue.class.getSimpleName(),
+ pvalue.getValue()));
+ }
+ // At this point we know it is a PCollectionView or some internal hacked PValue. To be
Review comment:
nit: if you use `/* */` blocks they will get line formatted correctly
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformReplacements.java
##########
@@ -15,20 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.core.construction;
+package org.apache.beam.sdk.runners;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
import java.util.Map;
import java.util.Set;
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
/** */
+@Internal
Review comment:
Why did this (and related classes) have to move?
I could be convinced otherwise to keep the move but it looks like long term we would want to get rid of Pipeline#replaceAll some day since runners should only doing proto -> proto conversions.
The only place I could find it in was PipelineTest. Can we instead move that test to somewhere in runners core construction?
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
##########
@@ -1403,4 +1427,55 @@ public int size() {
};
}
}
+
+ public static <InputT, ViewT> PCollectionView<ViewT> findPCollectionView(
Review comment:
Can we choose not to expose this as it looks like it is only used within PipelineTest.java
If there is a future need we can move it again (in hopefully a much smaller change) so its easier to reason about visibility.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PValues.java
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.values;
Review comment:
This might make more sense in `org.apache.beam.sdk.runners`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708872269
run jet validatesrunner
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: [BEAM-11094] Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713914112
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709463928
Dataflow VR failed in Metrics. Flink VR timed out on CountingSourceTest.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708855469
Opened #13121 which demonstrated the `checkState` failure exists on `master`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713341055
Oh I should also say that `PValues#expand` change _is_ part of the core change, not extra. The change is: instead of `PValue#expand` which leaves things ambiguous, always use `PValues#expandXYZ` which leaves things unambiguous.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r508962544
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformReplacements.java
##########
@@ -15,20 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.beam.runners.core.construction;
+package org.apache.beam.sdk.runners;
import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
import java.util.Map;
import java.util.Set;
-import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Iterables;
/** */
+@Internal
Review comment:
I don't agree with this. There are many reasons
- Protos are about wire formats and often not preferred. One time they are very much not preferred is when the proto is a low-level representation of a higher-level concept, like a graph.
- The SDKs are specifically designed to be the best API for writing correct Beam subgraphs. Using them is better than direct graph manipulation (and graph manipulation is better than proto manipulation).
I may have misunderstood what you meant by "proto -> proto" conversions. But if it is to serve the same purpose as `Pipeline#replaceAll` then I think it should look more or less the same. We could remove `Pipeline#replaceAll` from the core SDK and have an enhanced version just for runners, but that is a lot of complexity.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709464007
run flink validatesrunner
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles merged pull request #13118: [BEAM-11094] Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles merged pull request #13118:
URL: https://github.com/apache/beam/pull/13118
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709491139
run java postcommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-712472678
R: @lukecwik
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713264134
The moves were necessary in order to use them in `PipelineTest`. The moves should be noops for review. I can undo them and move to some "ExtraPipelineTest" file in core construction if you like.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709490395
run dataflow validatesrunner
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on a change in pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #13118:
URL: https://github.com/apache/beam/pull/13118#discussion_r509018684
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
##########
@@ -1403,4 +1427,55 @@ public int size() {
};
}
}
+
+ public static <InputT, ViewT> PCollectionView<ViewT> findPCollectionView(
Review comment:
Yea. This is actually a simplified version of something that exists in `DataflowRunner.java`. I did not want to risk the stability of this change by also altering it there, but they should be merged.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: [BEAM-11094] Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713913712
Filed BEAM-11094
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708872308
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709490672
PTAL. I believe all test failures were known flakes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708720293
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708872202
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-710675974
Noting that all ValidatesRunner tests for all runners are unaffected by this change. I will rebase to get fixes that are now applied to master, but there should not be a need to re-run the ValidatesRunner tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: [BEAM-11094] Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713914434
Since it is approved, and I have to squash the commit history, and tests need re-running, I am going to flatten the commits now and force push.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708718665
Even though the raw LOC is large, it is mostly super trivial. It should just remove a source of errors. It simplifies runner code a little, removing casting all over the place.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-709628843
It seems the `MetricsTest` failure is present on `master`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708722860
It did uncover the existing problem (which exists at HEAD) because I added a `checkState`. This is generally not an issue because the new outputs of `View.AsList` are never used as such. But in my larger change it caused crashes because after two rounds of substitution the graph is really messed up.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] kennknowles commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
kennknowles commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-708718789
run flink validatesrunner
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #13118: Only work with fully expanded PCollections in TransformHierarchy and PTransformReplacements
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13118:
URL: https://github.com/apache/beam/pull/13118#issuecomment-713896444
Please associate PR with JIRA as this isn't a few lines being edited.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org