You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2022/01/13 07:40:40 UTC
[flink-table-store] 02/03: [FLINK-25627] Add KeyValue and KeyValueSerializer
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
commit bda887cec0b78f991663e140b26888c507964664
Author: JingsongLi <lz...@aliyun.com>
AuthorDate: Wed Jan 12 18:01:03 2022 +0800
[FLINK-25627] Add KeyValue and KeyValueSerializer
Co-authored-by: tsreaper <ts...@gmail.com>
---
.../apache/flink/table/store/file/KeyValue.java | 76 +++++++
.../flink/table/store/file/KeyValueSerializer.java | 84 ++++++++
.../apache/flink/table/store/file/ValueKind.java | 48 +++++
.../table/store/file/utils/ObjectSerializer.java | 78 ++++++++
.../table/store/file/KeyValueSerializerTest.java | 70 +++++++
.../table/store/file/TestKeyValueGenerator.java | 218 +++++++++++++++++++++
.../store/file/utils/ObjectSerializerTestBase.java | 64 ++++++
7 files changed, 638 insertions(+)
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
new file mode 100644
index 0000000..d5ff507
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
@@ -0,0 +1,76 @@
+/*
+ * 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.flink.table.store.file;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TinyIntType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A key value, including user key, sequence number, value kind and value. This object can be
+ * reused.
+ */
+public class KeyValue {
+
+ private RowData key;
+ private long sequenceNumber;
+ private ValueKind valueKind;
+ private RowData value;
+
+ public KeyValue setValue(RowData value) {
+ this.value = value;
+ return this;
+ }
+
+ public KeyValue replace(RowData key, long sequenceNumber, ValueKind valueKind, RowData value) {
+ this.key = key;
+ this.sequenceNumber = sequenceNumber;
+ this.valueKind = valueKind;
+ this.value = value;
+ return this;
+ }
+
+ public RowData key() {
+ return key;
+ }
+
+ public long sequenceNumber() {
+ return sequenceNumber;
+ }
+
+ public ValueKind valueKind() {
+ return valueKind;
+ }
+
+ public RowData value() {
+ return value;
+ }
+
+ public static RowType schema(RowType keyType, RowType valueType) {
+ List<RowType.RowField> fields = new ArrayList<>(keyType.getFields());
+ fields.add(new RowType.RowField("_SEQUENCE_NUMBER", new BigIntType(false)));
+ fields.add(new RowType.RowField("_VALUE_KIND", new TinyIntType(false)));
+ fields.addAll(valueType.getFields());
+ return new RowType(fields);
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueSerializer.java
new file mode 100644
index 0000000..788e87a
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValueSerializer.java
@@ -0,0 +1,84 @@
+/*
+ * 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.flink.table.store.file;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.utils.JoinedRowData;
+import org.apache.flink.table.store.file.utils.ObjectSerializer;
+import org.apache.flink.table.store.file.utils.OffsetRowData;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * Serializer for {@link KeyValue}.
+ *
+ * <p>NOTE: {@link RowData} and {@link KeyValue} produced by this serializer are reused.
+ */
+public class KeyValueSerializer extends ObjectSerializer<KeyValue> {
+
+ private final int keyArity;
+
+ private final GenericRowData reusedMeta;
+ private final JoinedRowData reusedKeyWithMeta;
+ private final JoinedRowData reusedRow;
+
+ private final OffsetRowData reusedKey;
+ private final OffsetRowData reusedValue;
+ private final KeyValue reusedKv;
+
+ public KeyValueSerializer(RowType keyType, RowType valueType) {
+ super(KeyValue.schema(keyType, valueType));
+
+ this.keyArity = keyType.getFieldCount();
+ int valueArity = valueType.getFieldCount();
+
+ this.reusedMeta = new GenericRowData(2);
+ this.reusedKeyWithMeta = new JoinedRowData();
+ this.reusedRow = new JoinedRowData();
+
+ this.reusedKey = new OffsetRowData(keyArity, 0);
+ this.reusedValue = new OffsetRowData(valueArity, keyArity + 2);
+ this.reusedKv = new KeyValue().replace(reusedKey, -1, null, reusedValue);
+ }
+
+ @Override
+ public RowData toRow(KeyValue record) {
+ return toRow(record.key(), record.sequenceNumber(), record.valueKind(), record.value());
+ }
+
+ public RowData toRow(RowData key, long sequenceNumber, ValueKind valueKind, RowData value) {
+ reusedMeta.setField(0, sequenceNumber);
+ reusedMeta.setField(1, valueKind.toByteValue());
+ return reusedRow.replace(reusedKeyWithMeta.replace(key, reusedMeta), value);
+ }
+
+ @Override
+ public KeyValue fromRow(RowData row) {
+ reusedKey.replace(row);
+ reusedValue.replace(row);
+ long sequenceNumber = row.getLong(keyArity);
+ ValueKind valueKind = ValueKind.fromByteValue(row.getByte(keyArity + 1));
+ reusedKv.replace(reusedKey, sequenceNumber, valueKind, reusedValue);
+ return reusedKv;
+ }
+
+ public KeyValue getReusedKv() {
+ return reusedKv;
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/ValueKind.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/ValueKind.java
new file mode 100644
index 0000000..685d3cb
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/ValueKind.java
@@ -0,0 +1,48 @@
+/*
+ * 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.flink.table.store.file;
+
+/** Value kind of a record. */
+public enum ValueKind {
+ ADD((byte) 0),
+
+ DELETE((byte) 1);
+
+ private final byte value;
+
+ ValueKind(byte value) {
+ this.value = value;
+ }
+
+ public byte toByteValue() {
+ return value;
+ }
+
+ public static ValueKind fromByteValue(byte value) {
+ switch (value) {
+ case 0:
+ return ADD;
+ case 1:
+ return DELETE;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported byte value '" + value + "' for value kind.");
+ }
+ }
+}
diff --git a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/ObjectSerializer.java b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/ObjectSerializer.java
new file mode 100644
index 0000000..ef9dc33
--- /dev/null
+++ b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/ObjectSerializer.java
@@ -0,0 +1,78 @@
+/*
+ * 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.flink.table.store.file.utils;
+
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/** A serializer to serialize object by {@link RowDataSerializer}. */
+public abstract class ObjectSerializer<T> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final RowDataSerializer rowSerializer;
+
+ public ObjectSerializer(RowType rowType) {
+ this.rowSerializer = InternalSerializers.create(rowType);
+ }
+
+ /** Get the number of fields. */
+ public int numFields() {
+ return rowSerializer.getArity();
+ }
+
+ /**
+ * Serializes the given record to the given target output view.
+ *
+ * @param record The record to serialize.
+ * @param target The output view to write the serialized data to.
+ * @throws IOException Thrown, if the serialization encountered an I/O related error. Typically
+ * raised by the output view, which may have an underlying I/O channel to which it
+ * delegates.
+ */
+ public final void serialize(T record, DataOutputView target) throws IOException {
+ rowSerializer.serialize(toRow(record), target);
+ }
+
+ /**
+ * De-serializes a record from the given source input view.
+ *
+ * @param source The input view from which to read the data.
+ * @return The deserialized element.
+ * @throws IOException Thrown, if the de-serialization encountered an I/O related error.
+ * Typically raised by the input view, which may have an underlying I/O channel from which
+ * it reads.
+ */
+ public final T deserialize(DataInputView source) throws IOException {
+ return fromRow(rowSerializer.deserialize(source));
+ }
+
+ /** Convert a {@link T} to {@link RowData}. */
+ public abstract RowData toRow(T record);
+
+ /** Convert a {@link RowData} to {@link T}. */
+ public abstract T fromRow(RowData rowData);
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/KeyValueSerializerTest.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/KeyValueSerializerTest.java
new file mode 100644
index 0000000..2756367
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/KeyValueSerializerTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.flink.table.store.file;
+
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.store.file.utils.ObjectSerializer;
+import org.apache.flink.table.store.file.utils.ObjectSerializerTestBase;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link KeyValueSerializer}. */
+public class KeyValueSerializerTest extends ObjectSerializerTestBase<KeyValue> {
+
+ private final TestKeyValueGenerator gen = new TestKeyValueGenerator();
+
+ @Override
+ protected ObjectSerializer<KeyValue> serializer() {
+ return new KeyValueSerializer(
+ TestKeyValueGenerator.KEY_TYPE, TestKeyValueGenerator.ROW_TYPE);
+ }
+
+ @Override
+ protected KeyValue object() {
+ return gen.next();
+ }
+
+ @Override
+ protected void checkResult(KeyValue expected, KeyValue actual) {
+ assertThat(
+ equals(
+ expected,
+ actual,
+ TestKeyValueGenerator.KEY_SERIALIZER,
+ TestKeyValueGenerator.ROW_SERIALIZER))
+ .isTrue();
+ }
+
+ public static boolean equals(
+ KeyValue kv1,
+ KeyValue kv2,
+ RowDataSerializer keySerializer,
+ RowDataSerializer valueSerializer) {
+ return kv1.sequenceNumber() == kv2.sequenceNumber()
+ && kv1.valueKind() == kv2.valueKind()
+ && keySerializer
+ .toBinaryRow(kv1.key())
+ .copy()
+ .equals(keySerializer.toBinaryRow(kv2.key()).copy())
+ && valueSerializer
+ .toBinaryRow(kv1.value())
+ .copy()
+ .equals(valueSerializer.toBinaryRow(kv2.value()).copy());
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestKeyValueGenerator.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestKeyValueGenerator.java
new file mode 100644
index 0000000..6a39165
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestKeyValueGenerator.java
@@ -0,0 +1,218 @@
+/*
+ * 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.flink.table.store.file;
+
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.generated.RecordComparator;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+/** Random {@link KeyValue} generator. */
+public class TestKeyValueGenerator {
+
+ // record structure
+ //
+ // * dt: string, 20201110 ~ 20201111 -------|
+ // * hr: int, 9 ~ 10 -------+----> partition
+ //
+ // * shopId: int, 0 ~ 9 -------|
+ // * orderId: long, any value -------+----> primary key
+ //
+ // * itemId: long, any value
+ // * price & amount: int[], [1 ~ 100, 1 ~ 10]
+ // * comment: string, length from 10 to 1000
+ public static final RowType ROW_TYPE =
+ RowType.of(
+ new LogicalType[] {
+ new VarCharType(false, 8),
+ new IntType(false),
+ new IntType(false),
+ new BigIntType(false),
+ new BigIntType(),
+ new ArrayType(new IntType()),
+ new VarCharType(Integer.MAX_VALUE)
+ },
+ new String[] {
+ "dt", "hr", "shopId", "orderId", "itemId", "priceAmount", "comment"
+ });
+ public static final RowType PARTITION_TYPE =
+ RowType.of(
+ new LogicalType[] {new VarCharType(false, 8), new IntType(false)},
+ new String[] {"dt", "hr"});
+ public static final RowType KEY_TYPE =
+ RowType.of(
+ new LogicalType[] {new IntType(false), new BigIntType(false)},
+ new String[] {"key_shopId", "key_orderId"});
+
+ public static final RowDataSerializer ROW_SERIALIZER = new RowDataSerializer(ROW_TYPE);
+ private static final RowDataSerializer PARTITION_SERIALIZER =
+ new RowDataSerializer(PARTITION_TYPE);
+ public static final RowDataSerializer KEY_SERIALIZER = new RowDataSerializer(KEY_TYPE);
+ private static final RecordComparator KEY_COMPARATOR =
+ (a, b) -> {
+ int firstResult = a.getInt(0) - b.getInt(0);
+ if (firstResult != 0) {
+ return firstResult;
+ }
+ return Long.compare(a.getLong(1), b.getLong(1));
+ };
+
+ private final Random random;
+
+ private final List<Order> addedOrders;
+ private final List<Order> deletedOrders;
+
+ private long sequenceNumber;
+
+ public TestKeyValueGenerator() {
+ this.random = new Random();
+
+ this.addedOrders = new ArrayList<>();
+ this.deletedOrders = new ArrayList<>();
+
+ this.sequenceNumber = 0;
+ }
+
+ public KeyValue next() {
+ int op = random.nextInt(5);
+ Order order = null;
+ ValueKind kind = ValueKind.ADD;
+ if (op == 0 && addedOrders.size() > 0) {
+ // delete order
+ order = pick(addedOrders);
+ deletedOrders.add(order);
+ kind = ValueKind.DELETE;
+ } else if (op == 1) {
+ // update order
+ if (random.nextBoolean() && deletedOrders.size() > 0) {
+ order = pick(deletedOrders);
+ } else if (addedOrders.size() > 0) {
+ order = pick(addedOrders);
+ }
+ if (order != null) {
+ order.update();
+ addedOrders.add(order);
+ kind = ValueKind.ADD;
+ }
+ }
+ if (order == null) {
+ // new order
+ order = new Order();
+ addedOrders.add(order);
+ kind = ValueKind.ADD;
+ }
+ return new KeyValue()
+ .replace(
+ KEY_SERIALIZER
+ .toBinaryRow(GenericRowData.of(order.shopId, order.orderId))
+ .copy(),
+ sequenceNumber++,
+ kind,
+ ROW_SERIALIZER
+ .toBinaryRow(
+ GenericRowData.of(
+ StringData.fromString(order.dt),
+ order.hr,
+ order.shopId,
+ order.orderId,
+ order.itemId,
+ order.priceAmount == null
+ ? null
+ : new GenericArrayData(order.priceAmount),
+ StringData.fromString(order.comment)))
+ .copy());
+ }
+
+ public BinaryRowData getPartition(KeyValue kv) {
+ return PARTITION_SERIALIZER
+ .toBinaryRow(GenericRowData.of(kv.value().getString(0), kv.value().getInt(1)))
+ .copy();
+ }
+
+ public void sort(List<KeyValue> kvs) {
+ kvs.sort(
+ (a, b) -> {
+ int keyCompareResult = KEY_COMPARATOR.compare(a.key(), b.key());
+ return keyCompareResult != 0
+ ? keyCompareResult
+ : Long.compare(a.sequenceNumber(), b.sequenceNumber());
+ });
+ }
+
+ public int compareKeys(BinaryRowData a, BinaryRowData b) {
+ return KEY_COMPARATOR.compare(a, b);
+ }
+
+ private Order pick(List<Order> list) {
+ int idx = random.nextInt(list.size());
+ Order tmp = list.get(idx);
+ list.set(idx, list.get(list.size() - 1));
+ list.remove(list.size() - 1);
+ return tmp;
+ }
+
+ private class Order {
+ private final String dt;
+ private final int hr;
+ private final int shopId;
+ private final long orderId;
+ @Nullable private Long itemId;
+ @Nullable private int[] priceAmount;
+ @Nullable private String comment;
+
+ private Order() {
+ dt = String.valueOf(random.nextInt(2) + 20211110);
+ hr = random.nextInt(2) + 8;
+ shopId = random.nextInt(10);
+ orderId = random.nextLong();
+ update();
+ }
+
+ private String randomString(int length) {
+ StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < length; i++) {
+ builder.append((char) (random.nextInt(127 - 32) + 32));
+ }
+ return builder.toString();
+ }
+
+ private void update() {
+ itemId = random.nextInt(10) == 0 ? null : random.nextLong();
+ priceAmount =
+ random.nextInt(10) == 0
+ ? null
+ : new int[] {random.nextInt(100) + 1, random.nextInt(100) + 1};
+ comment = random.nextInt(10) == 0 ? null : randomString(random.nextInt(1001 - 10) + 10);
+ }
+ }
+}
diff --git a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/ObjectSerializerTestBase.java b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/ObjectSerializerTestBase.java
new file mode 100644
index 0000000..ff7f994
--- /dev/null
+++ b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/utils/ObjectSerializerTestBase.java
@@ -0,0 +1,64 @@
+/*
+ * 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.flink.table.store.file.utils;
+
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ObjectSerializer}. */
+public abstract class ObjectSerializerTestBase<T> {
+
+ // we do not use @RepeatedTests in these tests to test the reuse of serializers
+ private static final int TRIES = 100;
+
+ @Test
+ public void testToFromRow() {
+ ObjectSerializer<T> serializer = serializer();
+ for (int i = 0; i < TRIES; i++) {
+ T object = object();
+ checkResult(object, serializer.fromRow(serializer.toRow(object)));
+ }
+ }
+
+ @Test
+ public void testSerialize() throws IOException {
+ ObjectSerializer<T> serializer = serializer();
+ for (int i = 0; i < TRIES; i++) {
+ T object = object();
+ DataOutputSerializer out = new DataOutputSerializer(128);
+ serializer.serialize(object, out);
+ T actual = serializer.deserialize(new DataInputDeserializer(out.getCopyOfBuffer()));
+ checkResult(object, actual);
+ }
+ }
+
+ protected abstract ObjectSerializer<T> serializer();
+
+ protected abstract T object();
+
+ protected void checkResult(T expected, T actual) {
+ assertThat(actual).isEqualTo(expected);
+ }
+}