[flink] 01/02: [FLINK-23875][io] Add utils to read and write versioned list data

[ar...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 6df130f71e6bec3dafcc3a986e0444855717a8e7
Author: Fabian Paul <fa...@ververica.com>
AuthorDate: Mon Aug 30 11:57:33 2021 +0200

    [FLINK-23875][io] Add utils to read and write versioned list data
---
 .../core/io/SimpleVersionedSerialization.java      | 62 ++++++++++++++++++++++
 .../core/io/SimpleVersionedSerializationTest.java  | 18 +++++++
 2 files changed, 80 insertions(+)

diff --git a/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerialization.java b/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerialization.java
index f4d8e1b..47ac388 100644
--- a/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerialization.java
+++ b/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerialization.java
@@ -23,7 +23,9 @@ import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -62,6 +64,37 @@ public class SimpleVersionedSerialization {
     }
 
     /**
+     * Serializes the version and data into a stream.
+     *
+     * <p>Data serialized via this method can be deserialized via {@link
+     * #readVersionAndDeserializeList(SimpleVersionedSerializer, DataInputView)}.
+     *
+     * <p>The first eight bytes will be occupied by the version, as returned by {@link
+     * SimpleVersionedSerializer#getVersion()} and the length of the list. The remaining bytes will
+     * be the serialized data, as produced by {@link SimpleVersionedSerializer#serialize(Object)},
+     * plus its length.
+     *
+     * @param serializer The serializer to serialize the datum with.
+     * @param data list of datum to serialize.
+     * @param out The stream to serialize to.
+     */
+    public static <T> void writeVersionAndSerializeList(
+            SimpleVersionedSerializer<T> serializer, List<T> data, DataOutputView out)
+            throws IOException {
+        checkNotNull(serializer);
+        checkNotNull(data);
+        checkNotNull(out);
+
+        out.writeInt(serializer.getVersion());
+        out.writeInt(data.size());
+        for (final T datum : data) {
+            final byte[] serializedDatum = serializer.serialize(datum);
+            out.writeInt(serializedDatum.length);
+            out.write(serializer.serialize(datum));
+        }
+    }
+
+    /**
      * Deserializes the version and datum from a stream.
      *
      * <p>This method deserializes data serialized via {@link
@@ -89,6 +122,35 @@ public class SimpleVersionedSerialization {
     }
 
     /**
+     * Deserializes the version and data from a stream.
+     *
+     * <p>This method deserializes data serialized via {@link
+     * #writeVersionAndSerializeList(SimpleVersionedSerializer, List, DataOutputView)} .
+     *
+     * <p>The first four bytes will be interpreted as the version. The next four bytes will be
+     * interpreted as the length of the list, then length-many data will be read and deserialized
+     * via the {@link SimpleVersionedSerializer#deserialize(int, byte[])} method.
+     *
+     * @param serializer The serializer to serialize the datum with.
+     * @param in The stream to deserialize from.
+     */
+    public static <T> List<T> readVersionAndDeserializeList(
+            SimpleVersionedSerializer<T> serializer, DataInputView in) throws IOException {
+        checkNotNull(serializer);
+        checkNotNull(in);
+        final int serializerVersion = in.readInt();
+        final int dataSize = in.readInt();
+        final List<T> data = new ArrayList<>();
+        for (int ignored = 0; ignored < dataSize; ignored++) {
+            final int datumSize = in.readInt();
+            final byte[] datum = new byte[datumSize];
+            in.readFully(datum);
+            data.add(serializer.deserialize(serializerVersion, datum));
+        }
+        return data;
+    }
+
+    /**
      * Serializes the version and datum into a byte array. The first four bytes will be occupied by
      * the version (as returned by {@link SimpleVersionedSerializer#getVersion()}), written in
      * <i>big-endian</i> encoding. The remaining bytes will be the serialized datum, as produced by
diff --git a/flink-core/src/test/java/org/apache/flink/core/io/SimpleVersionedSerializationTest.java b/flink-core/src/test/java/org/apache/flink/core/io/SimpleVersionedSerializationTest.java
index 929632c..3362dab 100644
--- a/flink-core/src/test/java/org/apache/flink/core/io/SimpleVersionedSerializationTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/io/SimpleVersionedSerializationTest.java
@@ -21,10 +21,13 @@ package org.apache.flink.core.io;
 import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.core.memory.DataOutputSerializer;
 
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+
 import org.junit.Test;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
@@ -96,6 +99,21 @@ public class SimpleVersionedSerializationTest {
         assertEquals(testString, deserializedFromBytes);
     }
 
+    @Test
+    public void testListSerializationRoundTrip() throws IOException {
+        final SimpleVersionedSerializer<String> utfEncoder = new TestStringSerializer();
+        final List<String> datums = ImmutableList.of("beeep!", "beep!!!");
+
+        final DataOutputSerializer out = new DataOutputSerializer(32);
+        SimpleVersionedSerialization.writeVersionAndSerializeList(utfEncoder, datums, out);
+        final byte[] outBytes = out.getCopyOfBuffer();
+
+        final DataInputDeserializer in = new DataInputDeserializer(outBytes);
+        final List<String> deserialized =
+                SimpleVersionedSerialization.readVersionAndDeserializeList(utfEncoder, in);
+        assertEquals(datums, deserialized);
+    }
+
     @Test(expected = IllegalArgumentException.class)
     public void testUnderflow() throws Exception {
         SimpleVersionedSerialization.readVersionAndDeSerialize(