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:46:31 UTC
[GitHub] [beam] boyuanzz opened a new pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
boyuanzz opened a new pull request #13120:
URL: https://github.com/apache/beam/pull/13120
**Please** add a meaningful description for your change here
------------------------
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] boyuanzz commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r507919651
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
In that case, I would image that the buffering data should be encoded/decoded with grpc input coder, which has been lengthperfixed already. Would you like to elaborate more on this?
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r506633500
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
I would not expect it will break non-stateful pipelines since non-stateful transform doesn't look into this coder?
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r509441928
##########
File path: runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptorsTest.java
##########
@@ -151,6 +157,99 @@ public void onTimer() {}
ensureLengthPrefixed(timerKeyCoder, originalKeyCoder, pbsCoderMap);
}
+ @Test
+ public void testLengthPrefixingOfInputCoderExecutableStage() throws Exception {
+ Pipeline p = Pipeline.create();
+ Coder<Void> voidCoder = VoidCoder.of();
+ assertThat(ModelCoderRegistrar.isKnownCoder(voidCoder), is(false));
+ p.apply("impulse", Impulse.create())
+ .apply(
+ ParDo.of(
+ new DoFn<byte[], Void>() {
+ @ProcessElement
+ public void process(ProcessContext ctxt) {}
+ }))
+ .setCoder(voidCoder)
+ .apply(
+ ParDo.of(
+ new DoFn<Void, Void>() {
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<Void, Void> tracker) {}
+
+ @GetInitialRestriction
+ public Void getInitialRestriction() {
+ return null;
+ }
+
+ @NewTracker
+ public SomeTracker newTracker(@Restriction Void restriction) {
+ return null;
+ }
+ }))
+ .setCoder(voidCoder);
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);
+ RunnerApi.Pipeline pipelineWithSdfExpanded =
+ ProtoOverrides.updateTransform(
+ PTransformTranslation.PAR_DO_TRANSFORM_URN,
+ pipelineProto,
+ SplittableParDoExpander.createSizedReplacement());
+ FusedPipeline fused = GreedyPipelineFuser.fuse(pipelineWithSdfExpanded);
+ Optional<ExecutableStage> optionalStage =
+ Iterables.tryFind(
+ fused.getFusedStages(),
+ (ExecutableStage stage) ->
+ stage.getTransforms().stream()
+ .anyMatch(
+ transform ->
+ transform
+ .getTransform()
+ .getSpec()
+ .getUrn()
+ .equals(
+ PTransformTranslation
+ .SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN)));
+ checkState(
+ optionalStage.isPresent(),
+ "Expected a stage with SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN.");
+
+ ExecutableStage stage = optionalStage.get();
+ PipelineNode.PCollectionNode inputPCollection = stage.getInputPCollection();
+ Map<String, RunnerApi.Coder> stageCoderMap = stage.getComponents().getCodersMap();
+ RunnerApi.Coder originalMainInputCoder =
+ stageCoderMap.get(inputPCollection.getPCollection().getCoderId());
+
+ BeamFnApi.ProcessBundleDescriptor pbd =
+ ProcessBundleDescriptors.fromExecutableStage(
+ "test_stage", stage, Endpoints.ApiServiceDescriptor.getDefaultInstance())
+ .getProcessBundleDescriptor();
+ Map<String, RunnerApi.Coder> pbsCoderMap = pbd.getCodersMap();
+
+ RunnerApi.Coder pbsMainInputCoder =
+ pbsCoderMap.get(pbd.getPcollectionsOrThrow(inputPCollection.getId()).getCoderId());
+
+ RunnerApi.Coder kvCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(pbsMainInputCoder).keyCoderId());
+ RunnerApi.Coder elementCoder =
Review comment:
```suggestion
RunnerApi.Coder keyCoder =
```
##########
File path: runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptorsTest.java
##########
@@ -151,6 +157,99 @@ public void onTimer() {}
ensureLengthPrefixed(timerKeyCoder, originalKeyCoder, pbsCoderMap);
}
+ @Test
+ public void testLengthPrefixingOfInputCoderExecutableStage() throws Exception {
+ Pipeline p = Pipeline.create();
+ Coder<Void> voidCoder = VoidCoder.of();
+ assertThat(ModelCoderRegistrar.isKnownCoder(voidCoder), is(false));
+ p.apply("impulse", Impulse.create())
+ .apply(
+ ParDo.of(
+ new DoFn<byte[], Void>() {
+ @ProcessElement
+ public void process(ProcessContext ctxt) {}
+ }))
+ .setCoder(voidCoder)
+ .apply(
+ ParDo.of(
+ new DoFn<Void, Void>() {
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<Void, Void> tracker) {}
+
+ @GetInitialRestriction
+ public Void getInitialRestriction() {
+ return null;
+ }
+
+ @NewTracker
+ public SomeTracker newTracker(@Restriction Void restriction) {
+ return null;
+ }
+ }))
+ .setCoder(voidCoder);
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);
+ RunnerApi.Pipeline pipelineWithSdfExpanded =
+ ProtoOverrides.updateTransform(
+ PTransformTranslation.PAR_DO_TRANSFORM_URN,
+ pipelineProto,
+ SplittableParDoExpander.createSizedReplacement());
+ FusedPipeline fused = GreedyPipelineFuser.fuse(pipelineWithSdfExpanded);
+ Optional<ExecutableStage> optionalStage =
+ Iterables.tryFind(
+ fused.getFusedStages(),
+ (ExecutableStage stage) ->
+ stage.getTransforms().stream()
+ .anyMatch(
+ transform ->
+ transform
+ .getTransform()
+ .getSpec()
+ .getUrn()
+ .equals(
+ PTransformTranslation
+ .SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN)));
+ checkState(
+ optionalStage.isPresent(),
+ "Expected a stage with SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN.");
+
+ ExecutableStage stage = optionalStage.get();
+ PipelineNode.PCollectionNode inputPCollection = stage.getInputPCollection();
+ Map<String, RunnerApi.Coder> stageCoderMap = stage.getComponents().getCodersMap();
+ RunnerApi.Coder originalMainInputCoder =
+ stageCoderMap.get(inputPCollection.getPCollection().getCoderId());
+
+ BeamFnApi.ProcessBundleDescriptor pbd =
+ ProcessBundleDescriptors.fromExecutableStage(
+ "test_stage", stage, Endpoints.ApiServiceDescriptor.getDefaultInstance())
+ .getProcessBundleDescriptor();
+ Map<String, RunnerApi.Coder> pbsCoderMap = pbd.getCodersMap();
+
+ RunnerApi.Coder pbsMainInputCoder =
+ pbsCoderMap.get(pbd.getPcollectionsOrThrow(inputPCollection.getId()).getCoderId());
+
+ RunnerApi.Coder kvCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(pbsMainInputCoder).keyCoderId());
+ RunnerApi.Coder elementCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(kvCoder).keyCoderId());
+ RunnerApi.Coder valueKvCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(kvCoder).valueCoderId());
+ RunnerApi.Coder valueCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(valueKvCoder).keyCoderId());
+
+ RunnerApi.Coder originalKvCoder =
+ stageCoderMap.get(ModelCoders.getKvCoderComponents(originalMainInputCoder).keyCoderId());
+ RunnerApi.Coder originalElementCoder =
Review comment:
```suggestion
RunnerApi.Coder originalKeyCoder =
```
##########
File path: runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptorsTest.java
##########
@@ -151,6 +157,99 @@ public void onTimer() {}
ensureLengthPrefixed(timerKeyCoder, originalKeyCoder, pbsCoderMap);
}
+ @Test
+ public void testLengthPrefixingOfInputCoderExecutableStage() throws Exception {
+ Pipeline p = Pipeline.create();
+ Coder<Void> voidCoder = VoidCoder.of();
+ assertThat(ModelCoderRegistrar.isKnownCoder(voidCoder), is(false));
+ p.apply("impulse", Impulse.create())
+ .apply(
+ ParDo.of(
+ new DoFn<byte[], Void>() {
+ @ProcessElement
+ public void process(ProcessContext ctxt) {}
+ }))
+ .setCoder(voidCoder)
+ .apply(
+ ParDo.of(
+ new DoFn<Void, Void>() {
+ @ProcessElement
+ public void processElement(
+ ProcessContext context, RestrictionTracker<Void, Void> tracker) {}
+
+ @GetInitialRestriction
+ public Void getInitialRestriction() {
+ return null;
+ }
+
+ @NewTracker
+ public SomeTracker newTracker(@Restriction Void restriction) {
+ return null;
+ }
+ }))
+ .setCoder(voidCoder);
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);
+ RunnerApi.Pipeline pipelineWithSdfExpanded =
+ ProtoOverrides.updateTransform(
+ PTransformTranslation.PAR_DO_TRANSFORM_URN,
+ pipelineProto,
+ SplittableParDoExpander.createSizedReplacement());
+ FusedPipeline fused = GreedyPipelineFuser.fuse(pipelineWithSdfExpanded);
+ Optional<ExecutableStage> optionalStage =
+ Iterables.tryFind(
+ fused.getFusedStages(),
+ (ExecutableStage stage) ->
+ stage.getTransforms().stream()
+ .anyMatch(
+ transform ->
+ transform
+ .getTransform()
+ .getSpec()
+ .getUrn()
+ .equals(
+ PTransformTranslation
+ .SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN)));
+ checkState(
+ optionalStage.isPresent(),
+ "Expected a stage with SPLITTABLE_PROCESS_SIZED_ELEMENTS_AND_RESTRICTIONS_URN.");
+
+ ExecutableStage stage = optionalStage.get();
+ PipelineNode.PCollectionNode inputPCollection = stage.getInputPCollection();
+ Map<String, RunnerApi.Coder> stageCoderMap = stage.getComponents().getCodersMap();
+ RunnerApi.Coder originalMainInputCoder =
+ stageCoderMap.get(inputPCollection.getPCollection().getCoderId());
+
+ BeamFnApi.ProcessBundleDescriptor pbd =
+ ProcessBundleDescriptors.fromExecutableStage(
+ "test_stage", stage, Endpoints.ApiServiceDescriptor.getDefaultInstance())
+ .getProcessBundleDescriptor();
+ Map<String, RunnerApi.Coder> pbsCoderMap = pbd.getCodersMap();
+
+ RunnerApi.Coder pbsMainInputCoder =
+ pbsCoderMap.get(pbd.getPcollectionsOrThrow(inputPCollection.getId()).getCoderId());
+
+ RunnerApi.Coder kvCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(pbsMainInputCoder).keyCoderId());
+ RunnerApi.Coder elementCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(kvCoder).keyCoderId());
+ RunnerApi.Coder valueKvCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(kvCoder).valueCoderId());
+ RunnerApi.Coder valueCoder =
+ pbsCoderMap.get(ModelCoders.getKvCoderComponents(valueKvCoder).keyCoderId());
+
+ RunnerApi.Coder originalKvCoder =
+ stageCoderMap.get(ModelCoders.getKvCoderComponents(originalMainInputCoder).keyCoderId());
+ RunnerApi.Coder originalElementCoder =
+ stageCoderMap.get(ModelCoders.getKvCoderComponents(originalKvCoder).keyCoderId());
+ RunnerApi.Coder originalvalueKvCoder =
+ stageCoderMap.get(ModelCoders.getKvCoderComponents(originalKvCoder).valueCoderId());
+ RunnerApi.Coder originalvalueCoder =
+ stageCoderMap.get(ModelCoders.getKvCoderComponents(originalvalueKvCoder).keyCoderId());
+
+ ensureLengthPrefixed(elementCoder, originalElementCoder, pbsCoderMap);
Review comment:
```suggestion
ensureLengthPrefixed(keyCoder, originalKeyCoder, pbsCoderMap);
```
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r509440175
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -174,26 +170,18 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
}
/**
- * Patches the input coder of a stateful transform to ensure that the byte representation of a key
- * used to partition the input element at the Runner, matches the key byte representation received
- * for state requests and timers from the SDK Harness. Stateful transforms always have a KvCoder
- * as input.
+ * Patches the input coder of the transform to ensure that the byte representation of input used
+ * at the Runner, matches the byte representation received from the SDK Harness.
*/
- private static void lengthPrefixKeyCoder(
- String inputColId, Components.Builder componentsBuilder) {
- RunnerApi.PCollection pcollection = componentsBuilder.getPcollectionsOrThrow(inputColId);
- RunnerApi.Coder kvCoder = componentsBuilder.getCodersOrThrow(pcollection.getCoderId());
- Preconditions.checkState(
- ModelCoders.KV_CODER_URN.equals(kvCoder.getSpec().getUrn()),
- "Stateful executable stages must use a KV coder, but is: %s",
- kvCoder.getSpec().getUrn());
- String keyCoderId = ModelCoders.getKvCoderComponents(kvCoder).keyCoderId();
- // Retain the original coder, but wrap in LengthPrefixCoder
- String newKeyCoderId =
- LengthPrefixUnknownCoders.addLengthPrefixedCoder(keyCoderId, componentsBuilder, false);
- // Replace old key coder with LengthPrefixCoder<old_key_coder>
- kvCoder = kvCoder.toBuilder().setComponentCoderIds(0, newKeyCoderId).build();
- componentsBuilder.putCoders(pcollection.getCoderId(), kvCoder);
+ private static void lengthPrefixAnyInputCoder(
+ String inputPCollectionId, Components.Builder componentsBuilder) {
+ RunnerApi.PCollection pcollection =
+ componentsBuilder.getPcollectionsOrThrow(inputPCollectionId);
+ String newInputCoderId =
+ LengthPrefixUnknownCoders.addLengthPrefixedCoder(
+ pcollection.getCoderId(), componentsBuilder, false);
+ componentsBuilder.putPcollections(
+ inputPCollectionId, pcollection.toBuilder().setCoderId(newInputCoderId).build());
Review comment:
Yeah that's right. This logic is also verified by the unit test.
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r506265236
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
Worth mentioning this will break any existing non-stateful pipelines. We could add another exception here for SDFs, but it's probably better to have a consistent encoding which does not depend on DoFns using state or timers.
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r508553910
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -174,26 +170,18 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
}
/**
- * Patches the input coder of a stateful transform to ensure that the byte representation of a key
- * used to partition the input element at the Runner, matches the key byte representation received
- * for state requests and timers from the SDK Harness. Stateful transforms always have a KvCoder
- * as input.
+ * Patches the input coder of the transform to ensure that the byte representation of input used
+ * at the Runner, matches the byte representation received from the SDK Harness.
*/
- private static void lengthPrefixKeyCoder(
- String inputColId, Components.Builder componentsBuilder) {
- RunnerApi.PCollection pcollection = componentsBuilder.getPcollectionsOrThrow(inputColId);
- RunnerApi.Coder kvCoder = componentsBuilder.getCodersOrThrow(pcollection.getCoderId());
- Preconditions.checkState(
- ModelCoders.KV_CODER_URN.equals(kvCoder.getSpec().getUrn()),
- "Stateful executable stages must use a KV coder, but is: %s",
- kvCoder.getSpec().getUrn());
- String keyCoderId = ModelCoders.getKvCoderComponents(kvCoder).keyCoderId();
- // Retain the original coder, but wrap in LengthPrefixCoder
- String newKeyCoderId =
- LengthPrefixUnknownCoders.addLengthPrefixedCoder(keyCoderId, componentsBuilder, false);
- // Replace old key coder with LengthPrefixCoder<old_key_coder>
- kvCoder = kvCoder.toBuilder().setComponentCoderIds(0, newKeyCoderId).build();
- componentsBuilder.putCoders(pcollection.getCoderId(), kvCoder);
+ private static void lengthPrefixAnyInputCoder(
+ String inputPCollectionId, Components.Builder componentsBuilder) {
+ RunnerApi.PCollection pcollection =
+ componentsBuilder.getPcollectionsOrThrow(inputPCollectionId);
+ String newInputCoderId =
+ LengthPrefixUnknownCoders.addLengthPrefixedCoder(
+ pcollection.getCoderId(), componentsBuilder, false);
+ componentsBuilder.putPcollections(
+ inputPCollectionId, pcollection.toBuilder().setCoderId(newInputCoderId).build());
Review comment:
I'm wondering, why is length-prefixing the key coder not necessary anymore? Wouldn't the SDK Harness be able to extract a non-length-prefixed key coder even though the input coder has been legth-prefixed? This would then cause a regression like in https://github.com/apache/beam/pull/9997 if the SDK Harness didn't use the NESTED contex, which it currently does because we had fix this a while ago: https://github.com/apache/beam/blob/57d249704da0c7bf3fb4e98b087ced2a28605fb3/sdks/python/apache_beam/runners/worker/bundle_processor.py#L769 The idea was to always ensure keys are length-prefixed, so we never run into inconsistent key encodings between the Runner and the SDK Harness.
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r507826701
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
It does, e.g. for buffering data during checkpoints. However, there are usually also other changes in Beam which makes restoring state from previous Beam versions hard.
----------------------------------------------------------------
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] boyuanzz merged pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz merged pull request #13120:
URL: https://github.com/apache/beam/pull/13120
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r508538663
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
My bad, the length prefixing here only applies to the coder in the ProcessBundleDescriptor. Indeed, any buffering won't be affected by this because the elements have already been decoded when they are buffered.
For the test case, FlinkSavepointITCase should verify that.
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r507928115
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -141,9 +139,7 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
Map<String, Map<String, TimerSpec>> timerSpecs = forTimerSpecs(stage, components);
- if (bagUserStateSpecs.size() > 0 || timerSpecs.size() > 0) {
- lengthPrefixKeyCoder(stage.getInputPCollection().getId(), components);
- }
+ lengthPrefixAnyInputCoder(stage.getInputPCollection().getId(), components);
Review comment:
Or do we have some test cases that we can verify?
----------------------------------------------------------------
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] boyuanzz commented on pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #13120:
URL: https://github.com/apache/beam/pull/13120#issuecomment-713772376
Thanks for your help! I'll merge this PR once all tests pass.
----------------------------------------------------------------
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] boyuanzz commented on pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #13120:
URL: https://github.com/apache/beam/pull/13120#issuecomment-713758065
Run Python_PVR_Flink PreCommit
----------------------------------------------------------------
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] mxm commented on a change in pull request #13120: [BEAM-10940] Lengthprefix any input coder for an ProcessBundleDescriptor.
Posted by GitBox <gi...@apache.org>.
mxm commented on a change in pull request #13120:
URL: https://github.com/apache/beam/pull/13120#discussion_r508560805
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ProcessBundleDescriptors.java
##########
@@ -174,26 +170,18 @@ private static ExecutableProcessBundleDescriptor fromExecutableStageInternal(
}
/**
- * Patches the input coder of a stateful transform to ensure that the byte representation of a key
- * used to partition the input element at the Runner, matches the key byte representation received
- * for state requests and timers from the SDK Harness. Stateful transforms always have a KvCoder
- * as input.
+ * Patches the input coder of the transform to ensure that the byte representation of input used
+ * at the Runner, matches the byte representation received from the SDK Harness.
*/
- private static void lengthPrefixKeyCoder(
- String inputColId, Components.Builder componentsBuilder) {
- RunnerApi.PCollection pcollection = componentsBuilder.getPcollectionsOrThrow(inputColId);
- RunnerApi.Coder kvCoder = componentsBuilder.getCodersOrThrow(pcollection.getCoderId());
- Preconditions.checkState(
- ModelCoders.KV_CODER_URN.equals(kvCoder.getSpec().getUrn()),
- "Stateful executable stages must use a KV coder, but is: %s",
- kvCoder.getSpec().getUrn());
- String keyCoderId = ModelCoders.getKvCoderComponents(kvCoder).keyCoderId();
- // Retain the original coder, but wrap in LengthPrefixCoder
- String newKeyCoderId =
- LengthPrefixUnknownCoders.addLengthPrefixedCoder(keyCoderId, componentsBuilder, false);
- // Replace old key coder with LengthPrefixCoder<old_key_coder>
- kvCoder = kvCoder.toBuilder().setComponentCoderIds(0, newKeyCoderId).build();
- componentsBuilder.putCoders(pcollection.getCoderId(), kvCoder);
+ private static void lengthPrefixAnyInputCoder(
+ String inputPCollectionId, Components.Builder componentsBuilder) {
+ RunnerApi.PCollection pcollection =
+ componentsBuilder.getPcollectionsOrThrow(inputPCollectionId);
+ String newInputCoderId =
+ LengthPrefixUnknownCoders.addLengthPrefixedCoder(
+ pcollection.getCoderId(), componentsBuilder, false);
+ componentsBuilder.putPcollections(
+ inputPCollectionId, pcollection.toBuilder().setCoderId(newInputCoderId).build());
Review comment:
I guess this will still work because if we have a KV Coder we will always end up length-prefixing the key because the KV coder is a known coder and it will recurse into the key afterwards.
----------------------------------------------------------------
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