You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by cz...@apache.org on 2023/02/23 07:04:12 UTC
[flink-table-store] branch master updated: [FLINK-31179] Make data structures serializable
This is an automated email from the ASF dual-hosted git repository.
czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 00d83ce1 [FLINK-31179] Make data structures serializable
00d83ce1 is described below
commit 00d83ce1602473e5b6e199380cd3b3474bb5a8a9
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Feb 23 15:04:06 2023 +0800
[FLINK-31179] Make data structures serializable
This closes #551.
---
.../apache/flink/table/store/data/BinaryArray.java | 10 +++++--
.../apache/flink/table/store/data/BinaryMap.java | 17 ++++++++++--
.../apache/flink/table/store/data/BinaryRow.java | 6 ++++
.../flink/table/store/data/BinarySection.java | 30 +++++++++++++++++---
.../flink/table/store/data/BinaryString.java | 2 ++
.../org/apache/flink/table/store/data/Decimal.java | 5 +++-
.../flink/table/store/data/GenericArray.java | 5 +++-
.../apache/flink/table/store/data/GenericMap.java | 5 +++-
.../apache/flink/table/store/data/GenericRow.java | 5 +++-
.../apache/flink/table/store/data/NestedRow.java | 2 ++
.../apache/flink/table/store/data/Timestamp.java | 5 +++-
.../table/store/data/columnar/ColumnarArray.java | 5 +++-
.../table/store/data/columnar/ColumnarMap.java | 6 +++-
.../table/store/data/columnar/ColumnarRow.java | 6 +++-
.../flink/table/store/memory/MemorySegment.java | 9 +++++-
.../table/store/memory/MemorySegmentSource.java | 9 +++++-
.../apache/flink/table/store/utils/IOUtils.java | 12 ++++++++
.../flink/table/store/data/BinaryRowTest.java | 32 ++++++++++++++--------
.../serializer/InternalArraySerializerTest.java | 18 ++++++++++++
.../data/serializer/InternalMapSerializerTest.java | 11 ++++++--
.../store/data/serializer/SerializerTestBase.java | 27 ++++++++++++++++++
21 files changed, 196 insertions(+), 31 deletions(-)
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryArray.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryArray.java
index 9f3ee57d..62e7b308 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryArray.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryArray.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.data;
+import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.memory.MemorySegmentUtils;
import org.apache.flink.table.store.types.DataType;
@@ -37,9 +38,14 @@ import static org.apache.flink.table.store.memory.MemorySegment.UNSAFE;
* <pre>
* [size(int)] + [null bits(4-byte word boundaries)] + [values or offset&length] + [variable length part].
* </pre>
+ *
+ * @since 0.4.0
*/
+@Experimental
public final class BinaryArray extends BinarySection implements InternalArray, DataSetters {
+ private static final long serialVersionUID = 1L;
+
/** Offset for Arrays. */
private static final int BYTE_ARRAY_BASE_OFFSET = UNSAFE.arrayBaseOffset(byte[].class);
@@ -94,10 +100,10 @@ public final class BinaryArray extends BinarySection implements InternalArray, D
}
// The number of elements in this array
- private int size;
+ private transient int size;
/** The position to start storing array elements. */
- private int elementOffset;
+ private transient int elementOffset;
public BinaryArray() {}
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryMap.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryMap.java
index 910c9577..fffe1ca4 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryMap.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryMap.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.data;
+import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.memory.MemorySegmentUtils;
import org.apache.flink.table.store.types.DataType;
@@ -31,11 +32,16 @@ import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
* [4 byte(keyArray size in bytes)] + [Key BinaryArray] + [Value BinaryArray].
*
* <p>{@code BinaryMap} are influenced by Apache Spark UnsafeMapData.
+ *
+ * @since 0.4.0
*/
+@Experimental
public final class BinaryMap extends BinarySection implements InternalMap {
- private final BinaryArray keys;
- private final BinaryArray values;
+ private static final long serialVersionUID = 1L;
+
+ private transient BinaryArray keys;
+ private transient BinaryArray values;
public BinaryMap() {
keys = new BinaryArray();
@@ -54,7 +60,14 @@ public final class BinaryMap extends BinarySection implements InternalMap {
final int valueArrayBytes = sizeInBytes - keyArrayBytes - 4;
assert valueArrayBytes >= 0 : "valueArraySize (" + valueArrayBytes + ") should >= 0";
+ // see BinarySection.readObject, on this call stack, keys and values are not initialized
+ if (keys == null) {
+ keys = new BinaryArray();
+ }
keys.pointTo(segments, offset + 4, keyArrayBytes);
+ if (values == null) {
+ values = new BinaryArray();
+ }
values.pointTo(segments, offset + 4 + keyArrayBytes, valueArrayBytes);
assert keys.size() == values.size();
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryRow.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryRow.java
index 6d54db49..dd1943ba 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryRow.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryRow.java
@@ -17,6 +17,7 @@
package org.apache.flink.table.store.data;
+import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.memory.MemorySegmentUtils;
import org.apache.flink.table.store.types.DataType;
@@ -49,9 +50,14 @@ import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
* bigger pageSize of MemorySegment.
*
* <p>Variable-length part may fall into multiple MemorySegments.
+ *
+ * @since 0.4.0
*/
+@Experimental
public final class BinaryRow extends BinarySection implements InternalRow, DataSetters {
+ private static final long serialVersionUID = 1L;
+
public static final boolean LITTLE_ENDIAN =
(ByteOrder.nativeOrder() == ByteOrder.LITTLE_ENDIAN);
private static final long FIRST_BYTE_ZERO = LITTLE_ENDIAN ? ~0xFFL : ~(0xFFL << 56L);
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinarySection.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinarySection.java
index f0f0e9c5..0b1ff78d 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinarySection.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinarySection.java
@@ -19,10 +19,18 @@ package org.apache.flink.table.store.data;
import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.memory.MemorySegmentUtils;
+import org.apache.flink.table.store.utils.IOUtils;
import org.apache.flink.table.store.utils.Preconditions;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
/** Describe a section of memory. */
-public abstract class BinarySection {
+public abstract class BinarySection implements Serializable {
+
+ private static final long serialVersionUID = 1L;
/**
* It decides whether to put data in FixLenPart or VarLenPart. See more in {@link BinaryRow}.
@@ -52,9 +60,9 @@ public abstract class BinarySection {
*/
public static final long HIGHEST_SECOND_TO_EIGHTH_BIT = 0x7FL << 56;
- protected MemorySegment[] segments;
- protected int offset;
- protected int sizeInBytes;
+ protected transient MemorySegment[] segments;
+ protected transient int offset;
+ protected transient int sizeInBytes;
public BinarySection() {}
@@ -92,6 +100,20 @@ public abstract class BinarySection {
return MemorySegmentUtils.getBytes(segments, offset, sizeInBytes);
}
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ byte[] bytes = toBytes();
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ byte[] bytes = new byte[in.readInt()];
+ IOUtils.readFully(in, bytes);
+ pointTo(MemorySegment.wrap(bytes), 0, bytes.length);
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryString.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryString.java
index fbba13f9..921a29d3 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryString.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/BinaryString.java
@@ -41,6 +41,8 @@ import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
@Experimental
public final class BinaryString extends BinarySection implements Comparable<BinaryString> {
+ private static final long serialVersionUID = 1L;
+
public static final BinaryString EMPTY_UTF8 = BinaryString.fromBytes(encodeUTF8(""));
public BinaryString(MemorySegment[] segments, int offset, int sizeInBytes) {
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Decimal.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Decimal.java
index 114678d4..983b12c8 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Decimal.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Decimal.java
@@ -24,6 +24,7 @@ import org.apache.flink.table.store.types.DecimalType;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.io.Serializable;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.math.RoundingMode;
@@ -39,7 +40,9 @@ import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
* @since 0.4.0
*/
@Experimental
-public final class Decimal implements Comparable<Decimal> {
+public final class Decimal implements Comparable<Decimal>, Serializable {
+
+ private static final long serialVersionUID = 1L;
// member fields and static fields are package-visible,
// in order to be accessible for DecimalUtils
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
index 71d57364..609a329d 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericArray.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.types.ArrayType;
import org.apache.flink.table.store.utils.ArrayUtils;
+import java.io.Serializable;
import java.util.Arrays;
import java.util.Objects;
@@ -37,7 +38,9 @@ import java.util.Objects;
* @since 0.4.0
*/
@Experimental
-public final class GenericArray implements InternalArray {
+public final class GenericArray implements InternalArray, Serializable {
+
+ private static final long serialVersionUID = 1L;
private final Object array;
private final int size;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericMap.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericMap.java
index 1bdf0a12..142d00e2 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericMap.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericMap.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.types.MapType;
import org.apache.flink.table.store.types.MultisetType;
+import java.io.Serializable;
import java.util.Map;
import java.util.Objects;
@@ -40,7 +41,9 @@ import java.util.Objects;
* @since 0.4.0
*/
@Experimental
-public final class GenericMap implements InternalMap {
+public final class GenericMap implements InternalMap, Serializable {
+
+ private static final long serialVersionUID = 1L;
private final Map<?, ?> map;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericRow.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericRow.java
index 8461386c..09c7db28 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericRow.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/GenericRow.java
@@ -22,6 +22,7 @@ import org.apache.flink.table.store.annotation.Experimental;
import org.apache.flink.table.store.types.RowKind;
import org.apache.flink.table.store.types.RowType;
+import java.io.Serializable;
import java.util.Arrays;
import java.util.Objects;
@@ -44,7 +45,9 @@ import static org.apache.flink.table.store.utils.Preconditions.checkNotNull;
* @since 0.4.0
*/
@Experimental
-public final class GenericRow implements InternalRow {
+public final class GenericRow implements InternalRow, Serializable {
+
+ private static final long serialVersionUID = 1L;
/** The array to store the actual internal format values. */
private final Object[] fields;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/NestedRow.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/NestedRow.java
index fb5b7dc8..8ea6a3af 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/NestedRow.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/NestedRow.java
@@ -33,6 +33,8 @@ import static org.apache.flink.table.store.utils.Preconditions.checkArgument;
*/
public final class NestedRow extends BinarySection implements InternalRow, DataSetters {
+ private static final long serialVersionUID = 1L;
+
private final int arity;
private final int nullBitsSizeInBytes;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Timestamp.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Timestamp.java
index 3990f410..80583c89 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Timestamp.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/Timestamp.java
@@ -23,6 +23,7 @@ import org.apache.flink.table.store.types.LocalZonedTimestampType;
import org.apache.flink.table.store.types.TimestampType;
import org.apache.flink.table.store.utils.Preconditions;
+import java.io.Serializable;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -39,7 +40,9 @@ import java.time.LocalTime;
* @since 0.4.0
*/
@Experimental
-public final class Timestamp implements Comparable<Timestamp> {
+public final class Timestamp implements Comparable<Timestamp>, Serializable {
+
+ private static final long serialVersionUID = 1L;
// the number of milliseconds in a day
private static final long MILLIS_PER_DAY = 86400000; // = 24 * 60 * 60 * 1000
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarArray.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarArray.java
index 1480f4f3..a56c461a 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarArray.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarArray.java
@@ -26,10 +26,13 @@ import org.apache.flink.table.store.data.InternalMap;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.data.Timestamp;
+import java.io.Serializable;
import java.util.Arrays;
/** Columnar array to support access to vector column data. */
-public final class ColumnarArray implements InternalArray, DataSetters {
+public final class ColumnarArray implements InternalArray, DataSetters, Serializable {
+
+ private static final long serialVersionUID = 1L;
private final ColumnVector data;
private final int offset;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarMap.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarMap.java
index 68a4592e..6402fd3b 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarMap.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarMap.java
@@ -21,8 +21,12 @@ package org.apache.flink.table.store.data.columnar;
import org.apache.flink.table.store.data.InternalArray;
import org.apache.flink.table.store.data.InternalMap;
+import java.io.Serializable;
+
/** Columnar map to support access to vector column data. */
-public final class ColumnarMap implements InternalMap {
+public final class ColumnarMap implements InternalMap, Serializable {
+
+ private static final long serialVersionUID = 1L;
private final ColumnVector keyColumnVector;
private final ColumnVector valueColumnVector;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarRow.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarRow.java
index 36d16d58..6d379bac 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarRow.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/data/columnar/ColumnarRow.java
@@ -28,11 +28,15 @@ import org.apache.flink.table.store.data.Timestamp;
import org.apache.flink.table.store.data.columnar.BytesColumnVector.Bytes;
import org.apache.flink.table.store.types.RowKind;
+import java.io.Serializable;
+
/**
* Columnar row to support access to vector column data. It is a row view in {@link
* VectorizedColumnBatch}.
*/
-public final class ColumnarRow implements InternalRow, DataSetters {
+public final class ColumnarRow implements InternalRow, DataSetters, Serializable {
+
+ private static final long serialVersionUID = 1L;
private RowKind rowKind = RowKind.INSERT;
private VectorizedColumnBatch vectorizedColumnBatch;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegment.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegment.java
index b254a07c..a7b72980 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegment.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegment.java
@@ -18,6 +18,8 @@
package org.apache.flink.table.store.memory;
+import org.apache.flink.table.store.annotation.Experimental;
+
import javax.annotation.Nullable;
import java.io.DataInput;
@@ -32,7 +34,12 @@ import java.util.Objects;
import static org.apache.flink.table.store.memory.MemoryUtils.getByteBufferAddress;
-/** This class represents a piece of memory. */
+/**
+ * This class represents a piece of memory.
+ *
+ * @since 0.4.0
+ */
+@Experimental
public final class MemorySegment {
public static final sun.misc.Unsafe UNSAFE = MemoryUtils.UNSAFE;
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegmentSource.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegmentSource.java
index a7d0f8a1..3485e8fa 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegmentSource.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/memory/MemorySegmentSource.java
@@ -18,7 +18,14 @@
package org.apache.flink.table.store.memory;
-/** Interface describing entities that can provide memory segments. */
+import org.apache.flink.table.store.annotation.Experimental;
+
+/**
+ * Interface describing entities that can provide memory segments.
+ *
+ * @since 0.4.0
+ */
+@Experimental
public interface MemorySegmentSource {
/**
diff --git a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
index 188c0fc3..f05aeb24 100644
--- a/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
+++ b/flink-table-store-common/src/main/java/org/apache/flink/table/store/utils/IOUtils.java
@@ -85,6 +85,18 @@ public final class IOUtils {
// Stream input skipping
// ------------------------------------------------------------------------
+ /**
+ * Reads len bytes in a loop.
+ *
+ * @param in The InputStream to read from
+ * @param buf The buffer to fill
+ * @throws IOException if it could not read requested number of bytes for any reason (including
+ * EOF)
+ */
+ public static void readFully(final InputStream in, final byte[] buf) throws IOException {
+ readFully(in, buf, 0, buf.length);
+ }
+
/**
* Reads len bytes in a loop.
*
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/BinaryRowTest.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/BinaryRowTest.java
index 1166fa6e..cf57cb70 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/BinaryRowTest.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/BinaryRowTest.java
@@ -31,6 +31,7 @@ import org.apache.flink.table.store.types.IntType;
import org.apache.flink.table.store.types.RowKind;
import org.apache.flink.table.store.types.RowType;
import org.apache.flink.table.store.types.VarCharType;
+import org.apache.flink.table.store.utils.InstantiationUtil;
import org.junit.jupiter.api.Test;
@@ -47,6 +48,7 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.function.Consumer;
import static org.apache.flink.table.store.data.BinaryString.fromBytes;
import static org.apache.flink.table.store.data.BinaryString.fromString;
@@ -73,10 +75,10 @@ public class BinaryRowTest {
}
@Test
- public void testSetAndGet() {
- MemorySegment segment = MemorySegment.wrap(new byte[80]);
+ public void testSetAndGet() throws IOException, ClassNotFoundException {
+ MemorySegment segment = MemorySegment.wrap(new byte[100]);
BinaryRow row = new BinaryRow(9);
- row.pointTo(segment, 0, 80);
+ row.pointTo(segment, 20, 80);
row.setNullAt(0);
row.setInt(1, 11);
row.setLong(2, 22);
@@ -86,14 +88,22 @@ public class BinaryRowTest {
row.setByte(6, (byte) 66);
row.setFloat(7, 77f);
- assertThat((long) row.getDouble(3)).isEqualTo(33L);
- assertThat(row.getInt(1)).isEqualTo(11);
- assertThat(row.isNullAt(0)).isTrue();
- assertThat(row.getShort(5)).isEqualTo((short) 55);
- assertThat(row.getLong(2)).isEqualTo(22L);
- assertThat(row.getBoolean(4)).isTrue();
- assertThat(row.getByte(6)).isEqualTo((byte) 66);
- assertThat(row.getFloat(7)).isEqualTo(77f);
+ Consumer<BinaryRow> assertConsumer =
+ assertRow -> {
+ assertThat((long) assertRow.getDouble(3)).isEqualTo(33L);
+ assertThat(assertRow.getInt(1)).isEqualTo(11);
+ assertThat(assertRow.isNullAt(0)).isTrue();
+ assertThat(assertRow.getShort(5)).isEqualTo((short) 55);
+ assertThat(assertRow.getLong(2)).isEqualTo(22L);
+ assertThat(assertRow.getBoolean(4)).isTrue();
+ assertThat(assertRow.getByte(6)).isEqualTo((byte) 66);
+ assertThat(assertRow.getFloat(7)).isEqualTo(77f);
+ };
+
+ assertConsumer.accept(row);
+
+ // test serializable
+ assertConsumer.accept(InstantiationUtil.clone(row));
}
@Test
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalArraySerializerTest.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalArraySerializerTest.java
index a19ad3c1..dd743b9e 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalArraySerializerTest.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalArraySerializerTest.java
@@ -25,10 +25,12 @@ import org.apache.flink.table.store.data.GenericArray;
import org.apache.flink.table.store.data.InternalArray;
import org.apache.flink.table.store.data.columnar.ColumnarArray;
import org.apache.flink.table.store.data.columnar.heap.HeapBytesVector;
+import org.apache.flink.table.store.memory.MemorySegment;
import org.apache.flink.table.store.types.DataTypes;
import java.lang.reflect.Proxy;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
/** A test for the {@link InternalArraySerializer}. */
class InternalArraySerializerTest extends SerializerTestBase<InternalArray> {
@@ -65,6 +67,7 @@ class InternalArraySerializerTest extends SerializerTestBase<InternalArray> {
BinaryString.fromString("11"), null, BinaryString.fromString("ke")
}),
createArray("11", "haa"),
+ copyNewOffset(createArray("11", "haa")),
createArray("11", "haa", "ke"),
createArray("11", "haa", "ke"),
createArray("11", "lele", "haa", "ke"),
@@ -73,6 +76,21 @@ class InternalArraySerializerTest extends SerializerTestBase<InternalArray> {
};
}
+ @Override
+ protected InternalArray[] getSerializableTestData() {
+ InternalArray[] testData = getTestData();
+ return Arrays.copyOfRange(testData, 0, testData.length - 1);
+ }
+
+ static BinaryArray copyNewOffset(BinaryArray array) {
+ BinaryArray newArray = new BinaryArray();
+ byte[] bytes = array.toBytes();
+ byte[] newBytes = new byte[bytes.length + 10];
+ System.arraycopy(bytes, 0, newBytes, 10, bytes.length);
+ newArray.pointTo(MemorySegment.wrap(newBytes), 10, bytes.length);
+ return newArray;
+ }
+
static BinaryArray createArray(String... vs) {
BinaryArray array = new BinaryArray();
BinaryArrayWriter writer = new BinaryArrayWriter(array, vs.length, 8);
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalMapSerializerTest.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalMapSerializerTest.java
index f2b25b62..d8373bd0 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalMapSerializerTest.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/InternalMapSerializerTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.store.data.GenericMap;
import org.apache.flink.table.store.data.InternalArray;
import org.apache.flink.table.store.data.InternalMap;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
@@ -64,7 +65,6 @@ public class InternalMapSerializerTest extends SerializerTestBase<InternalMap> {
first.put(1, BinaryString.fromString(""));
return new InternalMap[] {
new GenericMap(first),
- new CustomMapData(first),
BinaryMap.valueOf(
createArray(1, 2), InternalArraySerializerTest.createArray("11", "haa")),
BinaryMap.valueOf(
@@ -75,10 +75,17 @@ public class InternalMapSerializerTest extends SerializerTestBase<InternalMap> {
InternalArraySerializerTest.createArray("11", "haa", "ke")),
BinaryMap.valueOf(
createArray(1, 5, 6, 7),
- InternalArraySerializerTest.createArray("11", "lele", "haa", "ke"))
+ InternalArraySerializerTest.createArray("11", "lele", "haa", "ke")),
+ new CustomMapData(first)
};
}
+ @Override
+ protected InternalMap[] getSerializableTestData() {
+ InternalMap[] testData = getTestData();
+ return Arrays.copyOfRange(testData, 0, testData.length - 1);
+ }
+
private static BinaryArray createArray(int... vs) {
BinaryArray array = new BinaryArray();
BinaryArrayWriter writer = new BinaryArrayWriter(array, vs.length, 4);
diff --git a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
index 3ee86c5a..b6da7ddd 100644
--- a/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
+++ b/flink-table-store-common/src/test/java/org/apache/flink/table/store/data/serializer/SerializerTestBase.java
@@ -51,6 +51,10 @@ public abstract class SerializerTestBase<T> {
protected abstract T[] getTestData();
+ protected T[] getSerializableTestData() {
+ return getTestData();
+ }
+
// --------------------------------------------------------------------------------------------
@Test
@@ -185,6 +189,29 @@ public abstract class SerializerTestBase<T> {
}
}
+ @Test
+ void testJavaSerializable() {
+ try {
+ T[] testData = getSerializableTestData();
+
+ for (T value : testData) {
+ byte[] bytes = InstantiationUtil.serializeObject(value);
+ assertTrue(bytes.length > 0, "No data available during deserialization.");
+
+ T deserialized =
+ InstantiationUtil.deserializeObject(
+ bytes, this.getClass().getClassLoader());
+ checkToString(deserialized);
+
+ deepEquals("Deserialized value if wrong.", value, deserialized);
+ }
+ } catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ fail("Exception in test: " + e.getMessage());
+ }
+ }
+
// --------------------------------------------------------------------------------------------
private void deepEquals(String message, T should, T is) {