riteshghorse commented on code in PR #29306:
URL: https://github.com/apache/beam/pull/29306#discussion_r1383886457
##########
sdks/go/pkg/beam/runners/prism/internal/preprocess.go:
##########
@@ -384,10 +445,149 @@ func prepareStage(stg *stage, comps *pipepb.Components,
pipelineConsumers map[st
if l := len(mainInputs); l == 1 {
stg.primaryInput = getOnlyValue(mainInputs)
} else if l > 1 {
- // Quick check that this is a lone flatten node, which is
handled runner side anyway
- // and only sent SDK side as part of a fused stage.
- if !(len(stg.transforms) == 1 &&
comps.GetTransforms()[stg.transforms[0]].GetSpec().GetUrn() ==
urns.TransformFlatten) {
- panic("expected flatten node, but wasn't")
+ // Quick check that this is lead by a flatten node, and that
it's handled runner side.
+ t := comps.GetTransforms()[stg.transforms[0]]
+ if !(t.GetSpec().GetUrn() == urns.TransformFlatten &&
t.GetEnvironmentId() == "") {
+ panic("expected runner flatten node, but wasn't")
+ }
+ }
+}
+
+// greedyFusion produces a pipeline as tightly fused as possible.
+//
+// Fusion is a critical optimization for performance of pipeline execution.
+// Thus it's important for SDKs to be capable of executing transforms in a
fused state.
+//
+// However, not all transforms can be fused into the same stage together.
+// Further, some transforms must be at the root of a stage.
+//
+// # Fusion Restrictions
+//
+// Environments: Transforms that aren't in the same environment can't b
Review Comment:
```suggestion
// Environments: Transforms that aren't in the same environment can't be
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
