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/10 00:50:27 UTC
[GitHub] [beam] nehsyc opened a new pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
nehsyc opened a new pull request #13069:
URL: https://github.com/apache/beam/pull/13069
Add a well-known coder for ShardedKey in Java SDK
------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505804089
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
Review comment:
Ack.
----------------------------------------------------------------
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] nehsyc edited a comment on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc edited a comment on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707393502
> > > Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
> >
> >
> > Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
>
> Note that Go might also fail since it has some coder tests that rely on standard_coder.yaml. One PR can be fine.
>
> > > As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
> >
> >
> > Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
>
> I think always being non-null will be easier for users and runner authors. Having the default be "" is fine as it is expected that runners with their own implementation may choose other values. Runners can also choose "" if they so choose if they can differentiate internally as to whether something is re-using an existing shard id or defining a new one. All the runner / user cares about is that they are distinct.
Sounds reasonable. It wouldn't add complexity in code to avoid shard id being empty explicitly, unless the runner/user really wants to use empty bytes as a meaningful shard id in which case they need to be responsible for differentiating set/unset shard ids.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-709557723
> > I don't think the runner can know, given a ShardedKey, that the shard id was one that it chose (and, related, this would impose hidden restrictions on what the SDK could choose here). However, we're not making this a magic type, rather we're making GroupIntoBatches a magic transform (which seems a more sane direction). So I'm OK with making it required, with no special meanings for empty (or other) bytes.
> > The other question I'd like to bring up is the visibility of the ShardID. If it's not opaque, is there any reason to not just use a KV here?
>
> It is not opaque since the coder needs to access it - this is not a problem in Java but probably in Python since the coder impl is defined in a separate place. Maybe there is a workaround? Can we put the coder impl inside ShardedKey or?
> My understanding is that we make the ShardedKey a known type because the runner needs to be able to parse the shard id and replace it with something else. I am not sure how runner would replace K or V without knowing the type.
Done. I ended up not using AutoValue in Java and making shardId private. Also removed the getter for shardId in Python (although users can still access the member directly).
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503689424
##########
File path: model/pipeline/src/main/proto/beam_runner_api.proto
##########
@@ -1215,23 +1243,23 @@ message StandardArtifacts {
enum Types {
// A URN for locally-accessible artifact files.
// payload: ArtifactFilePayload
- FILE = 0 [(beam_urn) = "beam:artifact:type:file:v1"];
+ FILE = 0 [(beam_urn) = "beam:artifact:type:file:v1"];
Review comment:
Removed.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-715634829
@robertwb Hey Robert could you help merge this? I have squashed the commits.
----------------------------------------------------------------
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] nehsyc edited a comment on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc edited a comment on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707515750
> > > > Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
> > >
> > >
> > > Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
> >
> >
> > Note that Go might also fail since it has some coder tests that rely on standard_coder.yaml. One PR can be fine.
Thanks for pointing it out. The standard coder test in Go now skips ShardedKey coder test. Also added Python coder in this PR.
> > > > As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
> > >
> > >
> > > Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
> >
> >
> > I think always being non-null will be easier for users and runner authors. Having the default be "" is fine as it is expected that runners with their own implementation may choose other values. Runners can also choose "" if they so choose if they can differentiate internally as to whether something is re-using an existing shard id or defining a new one. All the runner / user cares about is that they are distinct.
>
> Sounds reasonable. It wouldn't add complexity in code to avoid shard id being empty explicitly, unless the runner/user really wants to use empty bytes as a meaningful shard id in which case they need to be responsible for differentiating set/unset shard ids.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503688064
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
Cool! Thanks for confirming.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
Review comment:
Done.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-710510945
@robertwb @lukecwik Please take a look. Unaddressed comments are left open.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707357737
> Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
> As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
----------------------------------------------------------------
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] lukecwik commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707242853
> R: @robertwb @lukecwik
>
> PTAL and see if the changes make sense and if there is anything I am missing.
>
> I am not sure what to do with the .yaml files... The locally modified file `sdks/python/apache_beam/portability/api/standard_coders.yaml` caused some python tests fail somehow. I wasn't intended to include that file in this PR since it is for Java SDk but I ended up reverting the changes and added that file here. Any guidance?
I don't think you should be creating a copy of standard_coders.yaml inside the python sdk. I was under the impression that the build process somehow moved the single standard_coders.yaml file around and made it available to the Python test but that might only be occurring during import into google.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505803746
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
+ keyCoder.verifyDeterministic();
Review comment:
Done.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505804582
##########
File path: sdks/python/apache_beam/coders/coders_test_common.py
##########
@@ -570,6 +571,37 @@ def test_map_coder(self):
}, {}, {i: str(i)
for i in range(5000)})
+ def test_sharded_key_coder(self):
+ coder = coders.ShardedKeyCoder(coders.BytesCoder())
+ # Verify cloud object representation
+ self.assertEqual({
+ '@type': 'kind:sharded_key',
+ 'component_encodings': [coders.BytesCoder().as_cloud_object()]
+ },
+ coder.as_cloud_object())
+
+ # Test binary representation
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'')))
Review comment:
Done.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-708092225
> > On empty key vs. marker, sorry for going back and forth--let's settle this before you change the code again.
> > Taking a step back, the question we want to answer is "do all ShardedKeys have a shard id" or is the shard id optional. If the former, I think representing it as a `(bytes, K)` tuple makes sense, but if we ever want to talk about "not having" a shard id than it should, logically, be `(Optional[bytes], K)` and we should use an explicit bit rather than have a "special" value of bytes that means "not there".
>
> Since we expect ShardedKey to be produced by the runner we don't expect the SDK/user to care so making it required simplifies user & SDK code. The runner is the only one that may care to differentiate unsharded vs sharded and can use any representation it wants to do so (e.g. by encoding Optional[bytes] into the bytes blob). The runners type doesn't have to match the user/SDK type.
> > On empty key vs. marker, sorry for going back and forth--let's settle this before you change the code again.
> > Taking a step back, the question we want to answer is "do all ShardedKeys have a shard id" or is the shard id optional. If the former, I think representing it as a `(bytes, K)` tuple makes sense, but if we ever want to talk about "not having" a shard id than it should, logically, be `(Optional[bytes], K)` and we should use an explicit bit rather than have a "special" value of bytes that means "not there".
>
> Since we expect ShardedKey to be produced by the runner we don't expect the SDK/user to care so making it required simplifies user & SDK code. The runner is the only one that may care to differentiate unsharded vs sharded and can use any representation it wants to do so (e.g. by encoding Optional[bytes] into the bytes blob). The runners type doesn't have to match the user/SDK type.
There doesn't seem to be a reason for end users to not have shard id set if they want to use a ShardedKey (otherwise why not just using the key?) For runners, shard id being non-existing could be meaningful so they can differentiate whether the shard has been set (during auto-sharding for example). But I think Luke made a good point that it's up to the runner to use any marker within the shard id bytes to do such checks.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504885576
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
Review comment:
I see. It requires a nested context which is always true for shard id. The `StreamCoderImpl` that the `ShardedKeyImpl` inherits seems to always use an [unnested](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/coders/coder_impl.py#L222) context so the encoded key won't have a length prefix if it is a string or bytes. Should I make the `ShardedKeyImpl` inherit `CoderImpl` directly to allow different encoding for different context? Also I am confused by the Java documentation for [encoding/decoding context](https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/Coder.java#L130) in Coder - seems like it's preferred to have a default context, either nested or unnested but not both.
----------------------------------------------------------------
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] lukecwik commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-715779039
rerun tests till green
----------------------------------------------------------------
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] lukecwik merged pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #13069:
URL: https://github.com/apache/beam/pull/13069
----------------------------------------------------------------
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] robertwb commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707899116
On empty key vs. marker, sorry for going back and forth--let's settle this before you change the code again.
Taking a step back, the question we want to answer is "do all ShardedKeys have a shard id" or is the shard id optional. If the former, I think representing it as a `(bytes, K)` tuple makes sense, but if we ever want to talk about "not having" a shard id than it should, logically, be `(Optional[bytes], K)` and we should use an explicit bit rather than have a "special" value of bytes that means "not there".
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-709475698
> I don't think the runner can know, given a ShardedKey, that the shard id was one that it chose (and, related, this would impose hidden restrictions on what the SDK could choose here). However, we're not making this a magic type, rather we're making GroupIntoBatches a magic transform (which seems a more sane direction). So I'm OK with making it required, with no special meanings for empty (or other) bytes.
>
> The other question I'd like to bring up is the visibility of the ShardID. If it's not opaque, is there any reason to not just use a KV here?
It is not opaque since the coder needs to access it - this is not a problem in Java but probably in Python since the coder impl is defined in a separate place. Maybe there is a workaround? Can we put the coder impl inside ShardedKey or?
My understanding is that we make the ShardedKey a known type because the runner needs to be able to parse the shard id and replace it with something else. I am not sure how runner would replace K or V without knowing the type.
----------------------------------------------------------------
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] lukecwik removed a comment on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik removed a comment on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-715779039
rerun tests till green
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r510504730
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == this) {
+ return true;
+ }
+ if (o instanceof ShardedKey) {
+ ShardedKey<?> that = (ShardedKey<?>) o;
+ return this.key.equals(that.key) && Arrays.equals(this.shardId, that.shardId);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ int h$ = 1;
Review comment:
Done.
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503430730
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
Review comment:
I thought we were going to use the empty byte array as the default and not differentiate null.
It will remove an if from encode/decode logic and also save one byte in the encoded representation.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ verifyDeterministic(this, "Key coder must be deterministic", keyCoder);
+ }
+
Review comment:
Please also implement structuralValue, verifyDeterministic, consistentWithEquals, isRegisterByteSizeObserverCheap.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
Review comment:
This is already implemented in StructuredCoder.java, no point in duplicating.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503688259
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ verifyDeterministic(this, "Key coder must be deterministic", keyCoder);
+ }
+
Review comment:
Done. Did not use the `StructuralByteArray` for shard id. Seems fine since AutoValue does compare values of bytes array as opposed to object identity.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
Review comment:
Removed.
----------------------------------------------------------------
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] lukecwik commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707376964
> > Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
>
> Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
Note that Go might also fail since it has some coder tests that rely on standard_coder.yaml. One PR can be fine.
>
> > As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
>
> Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
I think always being non-null will be easier for users and runner authors. Having the default be "" is fine as it is expected that runners with their own implementation may choose other values. Runners can also choose "" if they so choose if they can differentiate internally as to whether something is re-using an existing shard id or defining a new one. All the runner / user cares about is that they are distinct.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-706626478
R: @robertwb @lukecwik
PTAL and see if the changes make sense and if there is anything I am missing.
I am not sure what to do with the .yaml files... The locally modified file `sdks/python/apache_beam/portability/api/standard_coders.yaml` caused some python tests fail somehow. I wasn't intended to include that file in this PR since it is for Java SDk but I ended up reverting the changes and added that file here. Any guidance?
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-706626623
retest this please
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r510456957
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review comment:
I don't want you to change it, its just an FYI.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r506773486
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """
+ A sharded key consisting of a user key and an opaque shard id represented by
+ bytes.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ def __repr__(self):
+ return '(%s, %s)' % (repr(self.key), self._shard_id)
+
+ def __eq__(self, other):
+ return (
+ type(self) == type(other) and self.key == other.key and
+ self._shard_id == other._shard_id)
+
+ def __ne__(self, other):
+ # TODO(BEAM-5949): Needed for Python 2 compatibility.
Review comment:
Done.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r510504797
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review comment:
Ack.
----------------------------------------------------------------
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] robertwb commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505739346
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
Review comment:
We can simply access _shard_id directly from the coder and get rid of this "public" accessor.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505803886
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
+ keyCoder.verifyDeterministic();
+ }
+
+ @Override
+ public boolean consistentWithEquals() {
+ return shardCoder.consistentWithEquals() && keyCoder.consistentWithEquals();
Review comment:
Done.
----------------------------------------------------------------
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] robertwb commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504132029
##########
File path: sdks/python/apache_beam/coders/coders_test_common.py
##########
@@ -570,6 +571,37 @@ def test_map_coder(self):
}, {}, {i: str(i)
for i in range(5000)})
+ def test_sharded_key_coder(self):
+ coder = coders.ShardedKeyCoder(coders.BytesCoder())
+ # Verify cloud object representation
+ self.assertEqual({
+ '@type': 'kind:sharded_key',
+ 'component_encodings': [coders.BytesCoder().as_cloud_object()]
+ },
+ coder.as_cloud_object())
+
+ # Test binary representation
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'')))
Review comment:
Maybe move these to the yaml file so they can be shared among all languages.
##########
File path: sdks/python/apache_beam/coders/coders_test_common.py
##########
@@ -570,6 +571,37 @@ def test_map_coder(self):
}, {}, {i: str(i)
for i in range(5000)})
+ def test_sharded_key_coder(self):
+ coder = coders.ShardedKeyCoder(coders.BytesCoder())
+ # Verify cloud object representation
+ self.assertEqual({
+ '@type': 'kind:sharded_key',
+ 'component_encodings': [coders.BytesCoder().as_cloud_object()]
+ },
+ coder.as_cloud_object())
+
+ # Test binary representation
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'')))
+ self.assertEqual(b'\x00\x03key', coder.encode(ShardedKey(b'key')))
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'', b'')))
+ self.assertEqual(b'\x03123\00', coder.encode(ShardedKey(b'', b'123')))
+ self.assertEqual(b'\x03123\03key', coder.encode(ShardedKey(b'key', b'123')))
+
+ test_values = [
+ ShardedKey(b''),
Review comment:
Have at least one example with a non-bytes key type?
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
+ return self._shard_id
+
+ def __repr__(self):
+ return '(%s, %s)' % (repr(self.key), self.shard_id)
+
+ def __eq__(self, other):
+ return (
+ type(self) == type(other) and self.key == other.key and
+ self.shard_id == other.shard_id)
+
+ def __ne__(self, other):
+ # TODO(BEAM-5949): Needed for Python 2 compatibility.
+ return not self == other
+
+ def __hash__(self):
+ return ((hash(self.key) & 0xFFFFFFFFFFFFFFF) + 3 *
Review comment:
You can drop the anding with 0xFFFF. You could also just return `hash((self.key, self.shard_key))` (i.e. has the tuple).
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
Review comment:
Do we intend this to be public? IMHO, it'd be better to not have it part of the public API (and we can always add it later if needed).
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
Review comment:
Same for Java.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503688483
##########
File path: sdks/python/apache_beam/portability/api/standard_coders.yaml
##########
@@ -0,0 +1,410 @@
+#
Review comment:
Deleted.
##########
File path: runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
##########
@@ -89,6 +90,7 @@
/** Tests that Java SDK coders standardized by the Fn API meet the common spec. */
@RunWith(Parameterized.class)
public class CommonCoderTest {
+
Review comment:
Oops. Happened to apply IntelliJ's auto format. Reverted.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503689224
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
Review comment:
Yes. AutoValue seems to be doing a reasonable job for byte[]. It uses `Arrays.hashCode` and `Arrays.equals` for has and equality.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505803429
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
Review comment:
Done
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-712991740
@lukecwik Hey Luke, could you please take a look again on this PR? Thanks!
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505804431
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
+ return self._shard_id
+
+ def __repr__(self):
+ return '(%s, %s)' % (repr(self.key), self.shard_id)
+
+ def __eq__(self, other):
+ return (
+ type(self) == type(other) and self.key == other.key and
+ self.shard_id == other.shard_id)
+
+ def __ne__(self, other):
+ # TODO(BEAM-5949): Needed for Python 2 compatibility.
+ return not self == other
+
+ def __hash__(self):
+ return ((hash(self.key) & 0xFFFFFFFFFFFFFFF) + 3 *
Review comment:
Done.
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504123726
##########
File path: model/pipeline/src/main/proto/beam_runner_api.proto
##########
@@ -879,6 +879,29 @@ message StandardCoders {
// Components: None
// Experimental.
ROW = 13 [(beam_urn) = "beam:coder:row:v1"];
+
+ // Encodes a use key and a shard id which is an opaque byte string.
+ //
+ // The encoding for a sharded key consists of the length prefixed shard id
+ // and the encoded user key in the following order:
+ //
+ // varInt(len(shard id))
Review comment:
Instead of referencing varInt(len(shard id)) shard id we should just say that the shard id is encoded using `beam:coder:bytes:v1`
Other coders descriptions should be more like row where we refer to other well known encodings.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
Review comment:
In the long run we should aim to replace the existing implementation and its usage within WriteFiles.
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
+ self._key_coder_impl = key_coder_impl
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (ShardedKey, create_OutputStream, bool) -> None
+ self._shard_id_coder_impl.encode_to_stream(value.shard_id, out, True)
+ self._key_coder_impl.encode_to_stream(value.key, out, True)
+
+ def decode_from_stream(self, in_stream, nested):
+ # type: (create_InputStream, bool) -> ShardedKey
+ shard_id = self._shard_id_coder_impl.decode_from_stream(in_stream, True)
+ key = self._key_coder_impl.decode_from_stream(in_stream, True)
+ return ShardedKey(key=key, shard_id=shard_id)
+
+ def estimate_size(self, value, nested=False):
+ # type: (Any, bool) -> int
+ estimated_size = 0
+ estimated_size += (
+ self._shard_id_coder_impl.estimate_size(value.shard_id, nested=True))
+ estimated_size += (
+ self._key_coder_impl.estimate_size(value.key, nested=True))
Review comment:
```suggestion
self._key_coder_impl.estimate_size(value.key, nested=nested))
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
+ keyCoder.verifyDeterministic();
+ }
+
+ @Override
+ public boolean consistentWithEquals() {
+ return shardCoder.consistentWithEquals() && keyCoder.consistentWithEquals();
Review comment:
the bytearraycoder isn't consistent with equals yet the AutoValue ShardedKey is so we should state that ShardedKey is consistent with equals iff keyCoder is.
```suggestion
return keyCoder.consistentWithEquals();
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
Review comment:
No point in checking something that is required to be deterministic
```suggestion
```
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
+ self._key_coder_impl = key_coder_impl
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (ShardedKey, create_OutputStream, bool) -> None
+ self._shard_id_coder_impl.encode_to_stream(value.shard_id, out, True)
+ self._key_coder_impl.encode_to_stream(value.key, out, True)
Review comment:
```suggestion
self._key_coder_impl.encode_to_stream(value.key, out, nested)
```
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
Review comment:
Why do you need the LengthPrefixCoderImpl here?
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
+ self._key_coder_impl = key_coder_impl
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (ShardedKey, create_OutputStream, bool) -> None
+ self._shard_id_coder_impl.encode_to_stream(value.shard_id, out, True)
+ self._key_coder_impl.encode_to_stream(value.key, out, True)
+
+ def decode_from_stream(self, in_stream, nested):
+ # type: (create_InputStream, bool) -> ShardedKey
+ shard_id = self._shard_id_coder_impl.decode_from_stream(in_stream, True)
+ key = self._key_coder_impl.decode_from_stream(in_stream, True)
Review comment:
```suggestion
key = self._key_coder_impl.decode_from_stream(in_stream, nested)
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
+ keyCoder.verifyDeterministic();
Review comment:
See KvCoder for an example:
```suggestion
verifyDeterministic(this, "Key coder must be deterministic", keyCoder);
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
Review comment:
Please make `byte[0]` a constant. Please add comment saying that the default shard identifier is the empty byte string.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r502843307
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
Currently the shard id is encoded using a ByteArrayCoder which uses a VarInt coder to encode/decode the length in 32bits ([link](https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ByteArrayCoder.java#L103)), which is consistent with the parsing logic in the unified worker.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r502843307
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
Currently the shard id is encoded using a [`ByteArrayCoder`](https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ByteArrayCoder.java#L103) which uses a `VarInt` coder to encode/decode the length in 32bits. That is consistent with the parsing logic in the unified worker. But in Python we seem to be using 64bits (e.g., [`VarIntCoder`](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/coders/coder_impl.py#L785)). Should I make it consistent by explicitly encoding / decoding a long value here?
----------------------------------------------------------------
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] robertwb commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-709463485
I don't think the runner can know, given a ShardedKey, that the shard id was one that it chose (and, related, this would impose hidden restrictions on what the SDK could choose here). However, we're not making this a magic type, rather we're making GroupIntoBatches a magic transform (which seems a more sane direction). So I'm OK with making it required, with no special meanings for empty (or other) bytes.
The other question I'd like to bring up is the visibility of the ShardID. If it's not opaque, is there any reason to not just use a KV here?
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505825569
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
+ self._key_coder_impl = key_coder_impl
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (ShardedKey, create_OutputStream, bool) -> None
+ self._shard_id_coder_impl.encode_to_stream(value.shard_id, out, True)
+ self._key_coder_impl.encode_to_stream(value.key, out, True)
Review comment:
Didn't change this in the latest commit since it would by default apply an unnested context which doesn't seem to be consistent with the Java encoding. Please see my other comment above.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504339524
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
Review comment:
`BytesCoder` simply copies the value to the stream and it doesn't automatically prepend a length prefix ([code](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/coders/coder_impl.py#L491)). While in Java the `ByteArrayCoder` does prepend the length.
----------------------------------------------------------------
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] robertwb commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r506739408
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """
+ A sharded key consisting of a user key and an opaque shard id represented by
+ bytes.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
Review comment:
Any reason not to make this required?
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """
+ A sharded key consisting of a user key and an opaque shard id represented by
+ bytes.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ def __repr__(self):
+ return '(%s, %s)' % (repr(self.key), self._shard_id)
+
+ def __eq__(self, other):
+ return (
+ type(self) == type(other) and self.key == other.key and
+ self._shard_id == other._shard_id)
+
+ def __ne__(self, other):
+ # TODO(BEAM-5949): Needed for Python 2 compatibility.
Review comment:
We don't need Python 2 compatibility anymore.
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
+ self._key_coder_impl = key_coder_impl
+
+ def encode_to_stream(self, value, out, nested):
+ # type: (ShardedKey, create_OutputStream, bool) -> None
+ self._shard_id_coder_impl.encode_to_stream(value.shard_id, out, True)
+ self._key_coder_impl.encode_to_stream(value.key, out, True)
Review comment:
True is correct and matches the description and Java. Let's not create more coders that have different nested/unnested encodings.
##########
File path: sdks/python/apache_beam/coders/coders.py
##########
@@ -312,11 +312,12 @@ def register_urn(
@classmethod
@overload
- def register_urn(cls,
- urn, # type: str
- parameter_type, # type: Optional[Type[T]]
- fn # type: Callable[[T, List[Coder], PipelineContext], Any]
- ):
+ def register_urn(
Review comment:
Is this yapf or some other auto-formatter?
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707393502
> > > Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
> >
> >
> > Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
>
> Note that Go might also fail since it has some coder tests that rely on standard_coder.yaml. One PR can be fine.
>
> > > As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
> >
> >
> > Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
>
> I think always being non-null will be easier for users and runner authors. Having the default be "" is fine as it is expected that runners with their own implementation may choose other values. Runners can also choose "" if they so choose if they can differentiate internally as to whether something is re-using an existing shard id or defining a new one. All the runner / user cares about is that they are distinct.
Sounds reasonable. It wouldn't add complexity in code to avoid shard id being empty explicitly, unless the runner/user really wants to use empty bytes as a meaningful shard id in which case they need to be responsible for differentiating set/unset shard ids.
----------------------------------------------------------------
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] lukecwik commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707911475
> On empty key vs. marker, sorry for going back and forth--let's settle this before you change the code again.
>
> Taking a step back, the question we want to answer is "do all ShardedKeys have a shard id" or is the shard id optional. If the former, I think representing it as a `(bytes, K)` tuple makes sense, but if we ever want to talk about "not having" a shard id than it should, logically, be `(Optional[bytes], K)` and we should use an explicit bit rather than have a "special" value of bytes that means "not there".
Since we expect ShardedKey to be produced by the runner we don't expect the SDK/user to care so making it required simplifies user & SDK code. The runner is the only one that may care to differentiate unsharded vs sharded and can use any representation it wants to do so (e.g. by encoding Optional[bytes] into the bytes blob). The runners coder/type doesn't have to match the user/SDK type.
----------------------------------------------------------------
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] robertwb commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503424858
##########
File path: model/pipeline/src/main/proto/beam_runner_api.proto
##########
@@ -1215,23 +1243,23 @@ message StandardArtifacts {
enum Types {
// A URN for locally-accessible artifact files.
// payload: ArtifactFilePayload
- FILE = 0 [(beam_urn) = "beam:artifact:type:file:v1"];
+ FILE = 0 [(beam_urn) = "beam:artifact:type:file:v1"];
Review comment:
Can you remove the irrelevant whitespace changes?
##########
File path: sdks/python/apache_beam/portability/api/standard_coders.yaml
##########
@@ -0,0 +1,410 @@
+#
Review comment:
We shouldn't add this, it should be copied on demand. Otherwise things may diverge which defeats the purpose of standard coders.
##########
File path: runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
##########
@@ -89,6 +90,7 @@
/** Tests that Java SDK coders standardized by the Fn API meet the common spec. */
@RunWith(Parameterized.class)
public class CommonCoderTest {
+
Review comment:
There seems to be lots of whitespace changes in this file (and elsewhere).
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
The two encodings are the same for all values in the range we'll be dealing with (and neither Java nor GRPC can handle >4G sized byte arrays) so we should be fine here.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
Review comment:
Are we leveraging AutoValue for hash and equality? If so, I don't think that'll work with byte[] here.
----------------------------------------------------------------
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] nehsyc commented on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707515750
> > > > Regarding the yaml issues, I would like to see what the Python failures were. If it's just that this coder is not implemented, perhaps simply implementing it in Python as well would be the simplest.
> > >
> > >
> > > Yeah the failure is that the coder is not implemented. I have a separate workspace for python. If it's okay to put python and java changes together I can merge those changes to this PR.
> >
> >
> > Note that Go might also fail since it has some coder tests that rely on standard_coder.yaml. One PR can be fine.
Thanks for pointing it out. The standard coder test in Go now skips ShardedKey coder test. Also added Python coder in this PR. (Sorry for the mess of commits and force pushes. The last commit was for formatting.)
> > > > As for using the empty byte string vs. an explicit marker, I don't think the extra byte in savings is significant. I think the bigger question is whether logically we would want to allow the empty byte string as a shard key (and if it would complicate code/the contract to explicitly avoid it). If it's OK, it should be included as an example and test.
> > >
> > >
> > > Empty is different from null. We could 1) allow null value and add a constraint on non-null shard id to be non-empty, or 2) not allow null value and assume empty bytes to be non-existing/default shard id. Adding an indicator would avoid the the constraint or assumption on the user side I think.
> >
> >
> > I think always being non-null will be easier for users and runner authors. Having the default be "" is fine as it is expected that runners with their own implementation may choose other values. Runners can also choose "" if they so choose if they can differentiate internally as to whether something is re-using an existing shard id or defining a new one. All the runner / user cares about is that they are distinct.
>
> Sounds reasonable. It wouldn't add complexity in code to avoid shard id being empty explicitly, unless the runner/user really wants to use empty bytes as a meaningful shard id in which case they need to be responsible for differentiating set/unset shard ids.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505803650
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key) {
+ return new AutoValue_ShardedKey(new byte[0], key);
+ }
+
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length of shard id
+ // shard id
+ // encoded user key
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ shardCoder.verifyDeterministic();
Review comment:
Done.
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r510437586
##########
File path: model/pipeline/src/main/proto/beam_runner_api.proto
##########
@@ -879,6 +879,28 @@ message StandardCoders {
// Components: None
// Experimental.
ROW = 13 [(beam_urn) = "beam:coder:row:v1"];
+
+ // Encodes a use key and a shard id which is an opaque byte string.
Review comment:
```suggestion
// Encodes a user key and a shard id which is an opaque byte string.
```
##########
File path: model/pipeline/src/main/proto/beam_runner_api.proto
##########
@@ -879,6 +879,28 @@ message StandardCoders {
// Components: None
// Experimental.
ROW = 13 [(beam_urn) = "beam:coder:row:v1"];
+
+ // Encodes a use key and a shard id which is an opaque byte string.
+ //
+ // The encoding for a sharded key consists of a shard id byte string and the
+ // encoded user key in the following order:
+ //
+ // - length prefixed shard id using beam:coder:bytes:v1
Review comment:
```suggestion
// - bytes shard id using beam:coder:bytes:v1
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == this) {
+ return true;
+ }
+ if (o instanceof ShardedKey) {
+ ShardedKey<?> that = (ShardedKey<?>) o;
+ return this.key.equals(that.key) && Arrays.equals(this.shardId, that.shardId);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ int h$ = 1;
Review comment:
rename `h$` to `hash`
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == this) {
+ return true;
+ }
+ if (o instanceof ShardedKey) {
+ ShardedKey<?> that = (ShardedKey<?>) o;
+ return this.key.equals(that.key) && Arrays.equals(this.shardId, that.shardId);
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ int h$ = 1;
+ h$ *= 1000003;
+ h$ ^= key.hashCode();
+ h$ *= 1000003;
+ h$ ^= Arrays.hashCode(shardId);
+ return h$;
+ }
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // length prefixed shard id byte string
Review comment:
```suggestion
// shard id byte string
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
Review comment:
```suggestion
return "ShardedKey{key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
Review comment:
```suggestion
/** Creates a ShardedKey with given key and shard id. Shard id must not be null and must not be mutated. */
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/** A sharded key consisting of a user key and an opaque shard id represented by bytes. */
+public class ShardedKey<K> {
+ private static final byte[] EMPTY_SHARD_ID = new byte[0];
+
+ private final K key;
+ private final byte[] shardId;
+
+ protected ShardedKey(K key, byte[] shardId) {
+ this.key = key;
+ this.shardId = shardId;
+ }
+
+ /** Creates a ShardedKey with given key and shard id. Shard id must not be null. */
+ public static <K> ShardedKey<K> of(K key, byte[] shardId) {
+ checkArgument(key != null, "Key should not be null!");
+ checkArgument(shardId != null, "Shard id should not be null!");
+ return new ShardedKey<K>(key, shardId);
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ @Override
+ public String toString() {
+ return "ShardedKey{" + "key=" + key + ", shardId=" + Arrays.toString(shardId) + "}";
+ }
+
+ @Override
+ public boolean equals(Object o) {
Review comment:
FYI, this equals method assumes the key type has a valid equals method and generally shouldn't be used. We should use structuralValue anywhere for comparisons.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r502843307
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
Currently the shard id is encoded using a [`ByteArrayCoder`](https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ByteArrayCoder.java#L103) which uses a VarInt coder to encode/decode the length in 32bits. That seems to be consistent with the parsing logic in the unified worker.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r506773880
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """
+ A sharded key consisting of a user key and an opaque shard id represented by
+ bytes.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
Review comment:
I think no? Removed.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504339464
##########
File path: sdks/python/apache_beam/utils/sharded_key.py
##########
@@ -0,0 +1,65 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# pytype: skip-file
+
+from __future__ import absolute_import
+
+
+class ShardedKey(object):
+ """A sharded key consisting of a user key and a shard id.
+
+ Attributes:
+ key: The user key.
+ shard_id: An opaque byte string that uniquely represents a shard of the key.
+ """
+ def __init__(
+ self,
+ key,
+ shard_id=b'', # type: bytes
+ ):
+ # type: (...) -> None
+ assert shard_id is not None
+ self._key = key
+ self._shard_id = shard_id
+
+ @property
+ def key(self):
+ return self._key
+
+ @property
+ def shard_id(self):
Review comment:
At least the coder impl needs to access the shard id for encoding. In Java, the coder is defined inside of the ShardedKey class so we could make the shard id protected (not private for the AutoValue to work). But in Python the coder is defined separately.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503697080
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public void verifyDeterministic() throws NonDeterministicException {
+ verifyDeterministic(this, "Key coder must be deterministic", keyCoder);
+ }
+
Review comment:
Done.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
+ }
+ keyCoder.encode(shardedKey.getKey(), outStream);
+ }
+
+ @Override
+ public ShardedKey<K> decode(InputStream inStream) throws IOException {
+ if (!BooleanCoder.of().decode(inStream)) {
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, null);
+ }
+ byte[] shardId = shardCoder.decode(inStream);
+ K key = keyCoder.decode(inStream);
+ return ShardedKey.of(key, shardId);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getCoderArguments() {
+ return Collections.singletonList(keyCoder);
+ }
+
+ @Override
+ public List<? extends org.apache.beam.sdk.coders.Coder<?>> getComponents() {
+ return Collections.singletonList(keyCoder);
Review comment:
Done.
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r503427722
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/util/ShardedKey.java
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.StructuredCoder;
+import org.apache.beam.sdk.util.common.ElementByteSizeObserver;
+
+/**
+ * A sharded key consisting of a user key and a shard id represented by bytes.
+ *
+ * <p>This is a more generic definition of {@link org.apache.beam.sdk.values.ShardedKey}.
+ */
+@AutoValue
+public abstract class ShardedKey<K> {
+
+ public static <K> ShardedKey<K> of(K key, @Nullable byte[] shardId) {
+ return new AutoValue_ShardedKey(shardId, key);
+ }
+
+ @SuppressWarnings("mutable")
+ @Nullable
+ public abstract byte[] getShardId();
+
+ public abstract K getKey();
+
+ public static class Coder<K> extends StructuredCoder<ShardedKey<K>> {
+
+ private final ByteArrayCoder shardCoder = ByteArrayCoder.of();
+ private final org.apache.beam.sdk.coders.Coder<K> keyCoder;
+
+ private Coder(org.apache.beam.sdk.coders.Coder<K> coder) {
+ keyCoder = coder;
+ }
+
+ public static <K> ShardedKey.Coder<K> of(org.apache.beam.sdk.coders.Coder<K> keyCoder) {
+ return new ShardedKey.Coder<K>(keyCoder);
+ }
+
+ public org.apache.beam.sdk.coders.Coder<K> getKeyCoder() {
+ return keyCoder;
+ }
+
+ @Override
+ public void encode(ShardedKey<K> shardedKey, OutputStream outStream) throws IOException {
+ // The encoding should follow the order:
+ // null indicator
+ // length of shard id
+ // shard id
+ // encoded user key
+ BooleanCoder.of().encode(shardedKey.getShardId() != null, outStream);
+ if (shardedKey.getShardId() != null) {
+ shardCoder.encode(shardedKey.getShardId(), outStream);
Review comment:
The coder specification is underspecified as to how large the var int can be:
https://github.com/apache/beam/blob/eb75286af4c308f4fd85393b4a2bf98a98d6d73e/model/pipeline/src/main/proto/beam_runner_api.proto#L673
Java can't support `byte[]` greater then `2^31-1` in size. Also note that a varint64 coder and varint32 coder share the same encoding for all values from `[0, 2^31-1]` so this is unlikely to become an issue for shard id encoding so can you can ignore this for now.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r505804803
##########
File path: sdks/python/apache_beam/coders/coders_test_common.py
##########
@@ -570,6 +571,37 @@ def test_map_coder(self):
}, {}, {i: str(i)
for i in range(5000)})
+ def test_sharded_key_coder(self):
+ coder = coders.ShardedKeyCoder(coders.BytesCoder())
+ # Verify cloud object representation
+ self.assertEqual({
+ '@type': 'kind:sharded_key',
+ 'component_encodings': [coders.BytesCoder().as_cloud_object()]
+ },
+ coder.as_cloud_object())
+
+ # Test binary representation
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'')))
+ self.assertEqual(b'\x00\x03key', coder.encode(ShardedKey(b'key')))
+ self.assertEqual(b'\x00\x00', coder.encode(ShardedKey(b'', b'')))
+ self.assertEqual(b'\x03123\00', coder.encode(ShardedKey(b'', b'123')))
+ self.assertEqual(b'\x03123\03key', coder.encode(ShardedKey(b'key', b'123')))
+
+ test_values = [
+ ShardedKey(b''),
Review comment:
Added a string type and a tuple type.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r506762494
##########
File path: sdks/python/apache_beam/coders/coders.py
##########
@@ -312,11 +312,12 @@ def register_urn(
@classmethod
@overload
- def register_urn(cls,
- urn, # type: str
- parameter_type, # type: Optional[Type[T]]
- fn # type: Callable[[T, List[Coder], PipelineContext], Any]
- ):
+ def register_urn(
Review comment:
Yes this is yapf.
----------------------------------------------------------------
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] lukecwik edited a comment on pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik edited a comment on pull request #13069:
URL: https://github.com/apache/beam/pull/13069#issuecomment-707911475
> On empty key vs. marker, sorry for going back and forth--let's settle this before you change the code again.
>
> Taking a step back, the question we want to answer is "do all ShardedKeys have a shard id" or is the shard id optional. If the former, I think representing it as a `(bytes, K)` tuple makes sense, but if we ever want to talk about "not having" a shard id than it should, logically, be `(Optional[bytes], K)` and we should use an explicit bit rather than have a "special" value of bytes that means "not there".
Since we expect ShardedKey to be produced by the runner we don't expect the SDK/user to care so making it required simplifies user & SDK code. The runner is the only one that may care to differentiate unsharded vs sharded and can use any representation it wants to do so (e.g. by encoding Optional[bytes] into the bytes blob). The runners type doesn't have to match the user/SDK type.
----------------------------------------------------------------
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] lukecwik commented on a change in pull request #13069: [BEAM-10475] Add a well-known coder for ShardedKey in Java/Python SDK
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #13069:
URL: https://github.com/apache/beam/pull/13069#discussion_r504841758
##########
File path: sdks/python/apache_beam/coders/coder_impl.py
##########
@@ -1365,3 +1366,38 @@ def estimate_size(self, value, nested=False):
# type: (Any, bool) -> int
value_size = self._value_coder.estimate_size(value)
return get_varint_size(value_size) + value_size
+
+
+class ShardedKeyCoderImpl(StreamCoderImpl):
+ """For internal use only; no backwards-compatibility guarantees.
+
+ A coder for sharded user keys.
+
+ The encoding and decoding should follow the order:
+ length of shard id byte string
+ shard id byte string
+ encoded user key
+ """
+ def __init__(self, key_coder_impl):
+ self._shard_id_coder_impl = LengthPrefixCoderImpl(BytesCoderImpl())
Review comment:
I was under the impression that nested=True would have the BytesCoderImpl have the length written by the OutputStream:
https://github.com/apache/beam/blob/c921b0c8384809dbcab3edcd5007e92da25067c6/sdks/python/apache_beam/coders/slow_stream.py#L45
----------------------------------------------------------------
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