Merge pull request #839: [proxima-beam-tools] explicitly add coder fo…

…r joins
datadrivencz · Oct 16, 2023 · e10f1d5 · e10f1d5
2 parents 7a45266 + 89d7f90
commit e10f1d5
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/beam/core/src/main/java/cz/o2/proxima/beam/core/io/PairCoder.java b/beam/core/src/main/java/cz/o2/proxima/beam/core/io/PairCoder.java
@@ -54,7 +54,6 @@ private PairCoder(Coder<K> keyCoder, Coder<V> valueCoder) {
 
   @Override
   public void encode(Pair<K, V> value, OutputStream outStream) throws IOException {
-
     keyCoder.encode(value.getFirst(), outStream);
     valueCoder.encode(value.getSecond(), outStream);
   }

diff --git a/beam/tools/src/main/java/cz/o2/proxima/beam/tools/groovy/BeamWindowedStream.java b/beam/tools/src/main/java/cz/o2/proxima/beam/tools/groovy/BeamWindowedStream.java
@@ -393,9 +393,11 @@ public <K, RIGHT> WindowedStream<Pair<T, RIGHT>> join(
               name == null
                   ? Join.innerJoin(leftKv, rightKv)
                   : Join.innerJoin(name + ".join", leftKv, rightKv);
-          return joined.apply(
-              MapElements.into(resultCoder.getEncodedTypeDescriptor())
-                  .via(kv -> Pair.of(kv.getValue().getKey(), kv.getValue().getValue())));
+          return joined
+              .apply(
+                  MapElements.into(resultCoder.getEncodedTypeDescriptor())
+                      .via(kv -> Pair.of(kv.getValue().getKey(), kv.getValue().getValue())))
+              .setCoder(resultCoder);
         });
   }