[BEAM-1337] Infer state coders #2133

aviemzur · 2017-03-01T05:28:37Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

aviemzur · 2017-03-01T05:28:50Z

coveralls · 2017-03-01T06:04:51Z

Coverage decreased (-0.1%) to 69.049% when pulling e5d4539 on aviemzur:infer-state-coders into d66029c on apache:master.

asfbot · 2017-03-01T06:05:30Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7999/
--none--

kennknowles · 2017-03-07T03:38:00Z

Sorry for the delay - this is on my radar and something I'm quite excited about.

kennknowles · 2017-03-08T04:57:52Z

...ct-java/src/main/java/org/apache/beam/runners/direct/CopyOnAccessInMemoryStateInternals.java

@@ -355,6 +379,12 @@ private boolean containedInUnderlying(StateNamespace namespace, StateTag<? super
 public <KeyT, ValueT> MapState<KeyT, ValueT> bindMap(
 StateTag<? super K, MapState<KeyT, ValueT>> address,
 Coder<KeyT> mapKeyCoder, Coder<ValueT> mapValueCoder) {
+ if (mapKeyCoder == null || mapValueCoder == null) {


I think that these parameters should not be @Nullable but instead the spec/tag should own this exception and throw it when it discovers that it has to provide these parameters but cannot.

Are the coders needed in all runners though?

My inclination is to say no, which is why I specifically throw exceptions in direct runner since it seems to try to force coder checks wherever encoding can happen (and this was the way to make this change testable).

Runners might handle state in a way which does not need encoding (For example, direct runner works without these coders at all, everything is in-mem. I added these exceptions to make it acknowledge the coders' very existence).

So actually I am going to have to say "yes". Just like every PCollection needs a coder, every pieces of state needs a coder. The portability goals of Beam include a runner written in Java running a pipeline written in Python, and vice-versa. Data in the underlying model (the Runner API / Fn API) is always just bytes with a coder (spec) to interpret it.

Ah, I See. I'll move the exceptions from CopyOnAccessInMemoryStateInternals to StateSpecs.

kennknowles · 2017-03-08T04:58:23Z

sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java

@@ -121,8 +156,23 @@ private StateSpecs() {}
 * Create a state spec that is optimized for adding values frequently, and occasionally retrieving
 * all the values that have been added.
 */
+ public static <T> StateSpec<Object, BagState<T>> bag() {
+ return bag(null);


Please do edit the classes and put @Nullable on the fields that are now nullable.

kennknowles · 2017-03-08T04:59:36Z

sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpec.java

+ * Given {code coders} are inferred from type arguments defined for this class.
+ * @param coders Array of coders indexed by the type arguments order.
+ */
+ void offerCoders(Coder[] coders);


This is nice, I think. I want to spend a minute pondering it.

What is the contract if it is called more than once? What if some of the coders cannot be provided but other can? These are issues that are very similar to TypedPValue.finishSpecifying() and the coder registry's internal logics.

So, as I saw it, coders that are already set take precedence, which is why implementations of this method check for coder references which are null and only set them. This is why I named the method offerCoders and not something like setCoders. I can improve the javadoc to explain this (Is there a better name for this method?).

If some coders cannot be provided and the coder is needed, the client which uses the coder will throw an exception notifying the user that a coder should have been set but was not. The user should then change their code to specify a coder explicitly using the StateSpec factory method which takes Coder arguments.
I'm not sure all clients will throw an exception which explains this (I made sure that the exceptions I added to direct runner explain why you are receiving this exception) but I think this holds true for many references to coders across the project.

That sounds about right. We should document like you have described here. I guess in the end each StateSpec implementation can decide what to do on its own. And for my other question, I think document that some entries in the array might be null. My favorite approach is to provide a Map<TypeVariable, Coder> but that might be a pain here.

I agree that would be better. The thing is that the StateSpec implementations do not hold a reference to the type variable.

kennknowles · 2017-03-08T05:01:17Z

sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java

+ public void offerCoders(Coder[] coders) {
+ if (this.coder == null) {
+ if (coders[0] != null) {
+ //noinspection unchecked


What system is this comment for? For javac you would use @SuppressWarnings("unchecked") and many tools grok that.

You're right, this works in IntelliJ IDEa, I'll change to an annotation on the method.

On a side note, findbugs doesn't seem to check for this,

Yea, we have a lot of things disabled that it would be nice to enable one module at a time until we can stop suppressing warnings. This one is actually pretty valuable to have enabled for code health IMO.

kennknowles · 2017-03-08T05:03:53Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

@@ -1815,6 +2051,69 @@ public void processElement(

 @Test
 @Category({RunnableOnService.class, UsesStatefulParDo.class})
+ public void testCombiningStateCoderInference() {


Nice test coverage for this.

Since coders are often very hard to debug, I think we could use test coverage also for failure cases to make sure the error message is actionable.

aviemzur · 2017-03-08T20:07:24Z

@kennknowles addressed your comments in my latest commits.
PTAL

asfbot · 2017-03-08T21:03:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8221/
--none--

coveralls · 2017-03-08T22:26:35Z

Coverage decreased (-0.2%) to 69.994% when pulling bfe1545 on aviemzur:infer-state-coders into 29d9bd3 on apache:master.

asfbot · 2017-03-08T22:32:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8222/
--none--

kennknowles · 2017-03-10T21:26:56Z

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

+ try {
+ Type typeArgument = typeArguments[i];
+ if (typeArgument instanceof Class) {
+ coders[i] = coderRegistry.getDefaultCoder((Class) typeArgument);


I think here you are going to want to pass a full TypeDescriptor, otherwise you can't infer a coder for List<Integer> for example. This is worth having a test for ValueState<List<String>> or some such.

kennknowles · 2017-03-10T21:27:42Z

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

+ if (typeArgument instanceof Class) {
+ coders[i] = coderRegistry.getDefaultCoder((Class) typeArgument);
+ }
+ } catch (CannotProvideCoderException ignored) {


Add a comment about why this is being ignored / what it means to ignore it.

kennknowles · 2017-03-10T21:30:33Z

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

@@ -739,6 +780,7 @@ public Unbound withSideInputs(
 @Override
 public PCollection<OutputT> expand(PCollection<? extends InputT> input) {
 validateWindowType(input, fn);
+ inferStateCodersIfNeeded(fn, input.getPipeline().getCoderRegistry());


There are going to be some limitations here that you actually could address.

class ListProcessorDoFn<T> extends DoFn<List<T>, List<T>> { ... ValueState<T> ... ... MapState<String, T> ... ... etc ... }

In these cases, you cannot infer a coder without using the coder from the input PCollection. But if you have the coder from the input, then it should work, in the same way we will infer an output coder for this DoFn.

aviemzur · 2017-03-11T07:54:38Z

@kennknowles addressed your comments in my latest 2 commits.
PTAL

coveralls · 2017-03-11T08:33:29Z

Coverage decreased (-0.2%) to 69.958% when pulling 2decabd on aviemzur:infer-state-coders into 29d9bd3 on apache:master.

asfbot · 2017-03-11T08:34:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8342/
--none--

aviemzur · 2017-03-11T08:46:18Z

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

@@ -930,6 +983,10 @@ public void populateDisplayData(Builder builder) {
 public PCollectionTuple expand(PCollection<? extends InputT> input) {
 // SplittableDoFn should be forbidden on the runner-side.
 validateWindowType(input, fn);
+
+ // Use coder registry to determine coders for all StateSpec defined in the fn signature.
+ inferStateCodersIfNeeded(fn, input.getPipeline().getCoderRegistry(), input.getCoder());


@kennknowles - Calling input.getCoder() here will throw an exception if the user did not set a coder or registered a default coder in coder registry.
I was thinking if I should do this inside of the inference logic instead and catch this exception. However, our previous discussion about forcing coders to be set regardless of runner makes me think we want this to fail these cases?

By the time an input is passed to expand it will have a coder.

kennknowles · 2017-03-23T02:42:07Z

sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/StateSpecs.java

@@ -44,6 +45,11 @@
 private StateSpecs() {}

 /** Create a simple state spec for values of type {@code T}. */
+ public static <T> StateSpec<Object, ValueState<T>> value() {
+ return value(null);


I would say that the contract on the value(Coder) factory method is that the coder is non-null (and it could have a checkArgument that enforces it). Here, you should just directly call the constructor with null instead.

kennknowles · 2017-03-23T02:46:25Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+ .apply(ParDo.of(fn)).setCoder(myIntegerCoder);
+
+ thrown.expect(PipelineExecutionException.class);
+ thrown.expectMessage("Unable to infer a coder for ValueState and no Coder was specified.");


I think that you should get the message at construction time in ParDo.expand(). Throwing the exception in bind() is a little too late. Or, you can make sure that ParDo.expand() calls bind() with some dummy visitor just to enforce that all the coders are ready.

Alternatively, you could be very explicit and have some kind of validate() / verifyCodersPresent() method on StateSpec.

coveralls · 2017-03-23T07:09:52Z

Coverage decreased (-0.2%) to 69.966% when pulling de69835 on aviemzur:infer-state-coders into 5e1be9f on apache:master.

asfbot · 2017-03-23T07:10:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8702/
--none--

aviemzur · 2017-03-23T08:25:31Z

@kennknowles Rebased on master and addressed your comments in my latest 2 commits.
PTAL

kennknowles

LGTM other than just a couple little comments! Really nice for users, and really nice code. You'll need to rebase again, and it might be good to run a quick ValidatesRunner suite like Flink, and it has the most extensive state support anyhow.

kennknowles · 2017-03-28T05:07:00Z

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

+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ } catch (Exception e) {
+ throw new Pipeline.PipelineExecutionException(e);


PipelineExecutionException is intended for "execution time" which is distinct from "construction time". This exception should always be at "construction time". We don't have a special wrapper exception for this, so perhaps just RuntimeException is appropriate here as well. But I notice that StateSpec.finishSpecifying() doesn't throw checked exceptions, so where are they coming from?

You're right, there are no checked exceptions. Implementations of StateSpec.finishSpecifying may throw IllegalStateException, but since we don't need to wrap them in Pipeline.PipelineExecutionException I can remove the second catch clause.

kennknowles · 2017-03-28T14:26:44Z

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

+ CoderRegistry coderRegistry,
+ Coder<InputT> inputCoder) {
+ Type stateType = stateDeclaration.stateType().getType();
+ if (stateType instanceof ParameterizedType) {


readability nit: I'd reverse the branches here to put the short one first, and just have a short circuit if not ParamterizedType return new Coder[0]. Also saves some indentation.

aviemzur · 2017-03-28T19:11:41Z

Rebased on top of master and made changes suggested in last review.

asfbot · 2017-03-28T19:33:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/2671/
--none--

asfbot · 2017-03-28T19:35:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_RunnableOnService_Flink/2100/
--none--

kennknowles · 2017-03-28T19:50:08Z

LGTM! And runners are happy.

coveralls · 2017-03-28T20:04:16Z

Coverage decreased (-0.2%) to 70.003% when pulling b21e931 on aviemzur:infer-state-coders into 0582eb9 on apache:master.

asfbot · 2017-03-28T20:05:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8864/
--none--

asfbot · 2017-03-30T06:24:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/2118/

Failed Tests: 9

beam_PostCommit_Java_ValidatesRunner_Flink/org.apache.beam:beam-runners-flink_2.10: 9

--none--

aviemzur · 2017-03-30T06:32:35Z

@aljoscha looks Flink runner fails with these new tests in ParDoTest. Direct runner and Dataflow runner pass, so I'm wondering if it could be an issue with Flink runner. Could you PTAL?

asfbot · 2017-03-30T08:38:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/2688/
--none--

aljoscha · 2017-03-31T20:49:41Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+ @Override
+ public MyInteger decode(InputStream inStream, Context context) throws CoderException,
+ IOException {
+ if (inStream.available() > 0) {


When this if is removed the test that fails on Flink with a NPE passed.

I think not having the if is valid but I'm not 100 % sure. What do you think, @kennknowles ?

I hadn't noticed this. Having the if is definitely incorrect, since this is checking whether or not it will block, which is irrelevant.

A simpler way to make this foolproof might be to just internally delegate to a VarIntCoder anyhow. It would also make it more obvious that the implementation of the coder doesn't matter.

It will indeed be simpler to delegate to VarIntCoder. I'll make the change.

aviemzur · 2017-04-01T04:20:57Z

Run Flink ValidatesRunner

aviemzur · 2017-04-01T04:21:03Z

Run Dataflow ValidatesRunner

asfbot · 2017-04-01T04:34:26Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/2150/
--none--

coveralls · 2017-04-01T05:10:14Z

Coverage decreased (-0.06%) to 70.132% when pulling 6595e88 on aviemzur:infer-state-coders into 0582eb9 on apache:master.

asfbot · 2017-04-01T05:11:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/9040/

Failed Tests: 2

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-examples-java: 1

org.apache.beam.examples.WindowedWordCountIT.testWindowedWordCountInStreaming

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-spark: 1

org.apache.beam.runners.spark.translation.streaming.CreateStreamTest.testLateDataAccumulating

--none--

asfbot · 2017-04-01T05:40:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/2710/
--none--

aviemzur · 2017-04-01T06:37:57Z

Retest this please

coveralls · 2017-04-01T07:23:57Z

Coverage decreased (-0.05%) to 70.145% when pulling 6595e88 on aviemzur:infer-state-coders into 0582eb9 on apache:master.

asfbot · 2017-04-01T07:25:14Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/9044/
--none--

kennknowles reviewed Mar 8, 2017

View reviewed changes

aviemzur force-pushed the infer-state-coders branch 2 times, most recently from 5efb262 to bfe1545 Compare March 8, 2017 20:06

kennknowles reviewed Mar 10, 2017

View reviewed changes

aviemzur commented Mar 11, 2017

View reviewed changes

kennknowles reviewed Mar 23, 2017

View reviewed changes

aviemzur force-pushed the infer-state-coders branch from 2decabd to de69835 Compare March 23, 2017 06:28

kennknowles approved these changes Mar 28, 2017

View reviewed changes

[BEAM-1337] Infer state coders

b21e931

aviemzur force-pushed the infer-state-coders branch from de69835 to b21e931 Compare March 28, 2017 19:10

aljoscha reviewed Mar 31, 2017

View reviewed changes

[BEAM-1337] Fix test coder bug.

6595e88

asfgit closed this in e31ca8b Apr 1, 2017

aviemzur deleted the infer-state-coders branch April 13, 2017 04:37

[BEAM-1337] Infer state coders #2133

[BEAM-1337] Infer state coders #2133

Conversation

aviemzur commented Mar 1, 2017 • edited Loading

aviemzur commented Mar 1, 2017

coveralls commented Mar 1, 2017

asfbot commented Mar 1, 2017

kennknowles commented Mar 7, 2017

Choose a reason for hiding this comment

aviemzur Mar 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aviemzur Mar 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aviemzur commented Mar 8, 2017

asfbot commented Mar 8, 2017

coveralls commented Mar 8, 2017

asfbot commented Mar 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aviemzur commented Mar 11, 2017

coveralls commented Mar 11, 2017

asfbot commented Mar 11, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Mar 23, 2017

asfbot commented Mar 23, 2017

aviemzur commented Mar 23, 2017

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aviemzur commented Mar 28, 2017

asfbot commented Mar 28, 2017

asfbot commented Mar 28, 2017

kennknowles commented Mar 28, 2017

coveralls commented Mar 28, 2017

asfbot commented Mar 28, 2017

asfbot commented Mar 30, 2017

Failed Tests: 9

beam_PostCommit_Java_ValidatesRunner_Flink/org.apache.beam:beam-runners-flink_2.10: 9

aviemzur commented Mar 30, 2017

asfbot commented Mar 30, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aviemzur commented Apr 1, 2017

aviemzur commented Apr 1, 2017

asfbot commented Apr 1, 2017

coveralls commented Apr 1, 2017

asfbot commented Apr 1, 2017

Failed Tests: 2

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-examples-java: 1

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-spark: 1

asfbot commented Apr 1, 2017

aviemzur commented Apr 1, 2017

coveralls commented Apr 1, 2017

asfbot commented Apr 1, 2017

aviemzur commented Mar 1, 2017 •

edited

Loading

aviemzur Mar 8, 2017 •

edited

Loading

aviemzur Mar 8, 2017 •

edited

Loading