Preserve pane index through reshuffle.

[claudevdm]

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @shunping for label python.
R: @jrmccluskey for label go.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[kennknowles]

This seems suspiciously out of place, being a magic string and also not a standard coder. What is the story behind it?

[claudevdm]

I placed it here because python switches between tuple coder depending on number of elements which cauuses python tests to fail with prism runner

beam/sdks/python/apache_beam/coders/coders.py

Line 1232 in 4fe34db

if self.is_kv_coder():

For example, with the pane index change there ends up being a tuple coder with 3 elements (because pane info is now included)

coders {
    key: "ref_Coder_TupleCoder_6"
    value {
      spec {
        urn: "beam:coder:tuple:v1"
      }
      component_coder_ids: "ref_Coder_BytesCoder_1"
      component_coder_ids: "ref_Coder_NullableCoder_7"
      component_coder_ids: "ref_Coder_FastPrimitivesCoder_8"
    }
  }

But without pane info it falls back to using KV coder which is supported by prism

coders {
    key: "ref_Coder_TupleCoder_6"
    value {
      spec {
        urn: "beam:coder:kv:v1"
      }
      component_coder_ids: "ref_Coder_BytesCoder_1"
      component_coder_ids: "ref_Coder_NullableCoder_7"
    }
  }

Prism error without this change:
ERROR:root:prism error building stage stage-002:
unknown coder urn key: beam:coder:tuple:v1

[kennknowles]

This seems like something we probably need to fix another way. The actual shuffle needs KVs, with everything we want to preserve reified into the value component.

[claudevdm]

The actual shuffle needs KVs

What do you mean by this? The way I understand it is

• Reshuffle adds random keys (k, v)
• ReifyMetadata maintains a kv, with a nested tuple as a value (value, timestamp, pane_info)
  

  beam/sdks/python/apache_beam/transforms/util.py

  Line 966 in db0aa82

  return key, (value, timestamp, pane_info)

I guess if we want to avoid this we can have a nested kv in ReifyMetedata so it is

• key, (value, (timestamp, pane_info))
  Then the regular kv coder should work?
  Or can we also use windowed_value as the value in the reify output instead of a tuple with the medatada?

The original reify just used a kv as the value in the reify function

beam/sdks/python/apache_beam/transforms/util.py

Line 972 in 57d1c35

return key, (value, timestamp)

We are now including pane info as mentioned above so it becomes a tuple

This only happens for global window case, in the custom window case the value for reify is a windowed value

beam/sdks/python/apache_beam/transforms/util.py

Line 996 in 57d1c35

return key, windowed_value.WindowedValue(value, timestamp, [window])

[kennknowles]

I see. Yea KV(key, (value, timestamp, pane_info) should be fine. The runner should not need to have any understanding of the coder for (value, timestamp, pane_info) since that is in general user type / coder.

[kennknowles]

I'm curious why it came up vs "just working" like you'd expect with other KV(key, whatever) objects. The process of submitting portably should result in a length-prefix coder wrapper if needed so the runner can just pass the bytes through.

[claudevdm]

It seems unkown coders are only legth prefixed if they are "leaf" and no sub-components coders. The tuple coder has sub components, so it is not length prefixed

beam/sdks/go/pkg/beam/runners/prism/internal/coders.go

Line 126 in 23f27df

// If we don't know this coder, and it has no sub components,

[claudevdm]

I chatted with Rebo who said componentizable coders do not play nicely with runner side length prefixing. He said adding a tuple coder is the right move here, and since Python isn't planning on getting rid of it (?) we might want to make it documented as a standard coder.