[Bug]: Pipeline fusion should break at @RequiresStableInput/determinism boundary

[je-ik]

What happened?

If a DoFn is annotated using @RequiresStableInput, a runner should break fusion to prevent fusing non-deterministic DoFns into the stable stage.

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[kennknowles]

Fusion is up to the runner, right? It is up to each runner how it makes the input stable. It could be stable because all the previous items in the stage are deterministic. It is not required to break fusion and checkpoint. And breaking fusion may not be sufficient to checkpoint, depending on runner.

[je-ik]

Sorry, I should've attached link to the ML thread.

TL;DR

It is runner-dependent how to achieve stable input, but every runner is able to ensure stability at the boundary of executable stage only. If we have two DoFns, say NonDeterminsticDoFn -> StableDoFn and these two get fused into single executable stage, then the contract of @RequiresStableInput do the StableDoFn is broken, because the runner can ensure stability only for the fused (NonDeterminsticDoFn + StableDoFn) DoFn, which is executed in the harness. Retrying such stage leads to non-stable input to the StableDoFn (due to non-determinism of the leading DoFn). Therefore, fusion needs to be broken at the boundary of @RequiresStableInput to make sure, that the executable stage starts with the stable DoFn.

[kennknowles]

Yes, I agree with most of this. The runner makes all fusion decisions and the GreedyPipelineFuser (and all of runners-core and runners-core-construction) is utility code provided to make it easier to build a runner.

But it is still not required for fusion to be broken at the boundary of @RequiresStableInput. You need to do an analysis to determine which PCollections are stable. Stable input -> deterministic transform -> stable output. It would probably be quite wasteful to always break fusion at the @RequiresStableInput.

[je-ik]

I think there is no way to determine if a DoFn is deterministic. Adding an annotation or something like that seems fragile. Users might easily forget to add it and testing the "stability of input" is non-trivial, so it will likely not be covered by user tests.

Stable input DoFns are quite rare, so hard-breaking fusion feels it should not add too much overhead.

[kennknowles]

Yea, it would have to be a user annotation. Users may make mistakes, but library authors can get it right. For example MapElements, Project, Filter, etc can all be marked deterministic. Even though stable DoFn might be rare, maybe pipelines only have 1-4 fusion breaks so this could be a huge performance hit.

[je-ik]

I don't think MapElements is required to be deterministic. Even in the test, that verifies @RequiresStableInput this assumption is broken (on purpose), see

beam/sdks/java/core/src/test/java/org/apache/beam/sdk/RequiresStableInputIT.java

Line 153 in e21b5b7

.apply("Single-PairWithRandomKey", MapElements.via(new PairWithRandomKeyFn()))

I'm afraid we might be trying to achieve the infamous "premature optimization" here.

[kennknowles]

Ah yes of course it is a property of the Fn. I was mistakenly considering primarily side effects. And even then, I am wrong since you can call out to services in your map fn.

It isn't premature, though. We very quickly saw major performance degradation when we automatically broke fusion in Dataflow for all stable inputs. The current PTransforms in the Beam SDK are designed in many cases to have some projection or trivial transformation in between the checkpoint and the @RequiresStableInput. I don't know specifically which ones it was, but it is common. For example a GBK then dropping keys and then writing to a file.