You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/01/30 14:54:12 UTC
[flink] 07/18: [FLINK-11329] [table] Migrating the RowSerializer to
use new compatibility API
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b434b32c0875d31acf30ab156d94d11bd1379975
Author: Igal Shilman <ig...@data-artisans.com>
AuthorDate: Sun Jan 27 15:41:49 2019 +0100
[FLINK-11329] [table] Migrating the RowSerializer to use new compatibility API
---
.../api/java/typeutils/runtime/RowSerializer.java | 138 +++++++++++----------
.../runtime/RowSerializerMigrationTest.java | 64 ++++++++++
.../test/resources/flink-1.6-row-serializer-data | Bin 0 -> 240 bytes
.../resources/flink-1.6-row-serializer-snapshot | Bin 0 -> 1444 bytes
.../test/resources/flink-1.7-row-serializer-data | Bin 0 -> 240 bytes
.../resources/flink-1.7-row-serializer-snapshot | Bin 0 -> 1454 bytes
6 files changed, 139 insertions(+), 63 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
index e29f681..df33402 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java
@@ -15,18 +15,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.api.java.typeutils.runtime;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.CompatibilityResult;
-import org.apache.flink.api.common.typeutils.CompatibilityUtil;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
-import org.apache.flink.api.common.typeutils.TypeDeserializerAdapter;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility;
import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
-import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.types.Row;
@@ -34,7 +32,6 @@ import org.apache.flink.types.Row;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.util.Arrays;
-import java.util.List;
import static org.apache.flink.api.java.typeutils.runtime.NullMaskUtils.readIntoAndCopyNullMask;
import static org.apache.flink.api.java.typeutils.runtime.NullMaskUtils.readIntoNullMask;
@@ -95,7 +92,8 @@ public final class RowSerializer extends TypeSerializer<Row> {
if (fromField != null) {
Object copy = fieldSerializers[i].copy(fromField);
result.setField(i, copy);
- } else {
+ }
+ else {
result.setField(i, null);
}
}
@@ -123,11 +121,13 @@ public final class RowSerializer extends TypeSerializer<Row> {
if (reuseField != null) {
Object copy = fieldSerializers[i].copy(fromField, reuseField);
reuse.setField(i, copy);
- } else {
+ }
+ else {
Object copy = fieldSerializers[i].copy(fromField);
reuse.setField(i, copy);
}
- } else {
+ }
+ else {
reuse.setField(i, null);
}
}
@@ -163,7 +163,6 @@ public final class RowSerializer extends TypeSerializer<Row> {
}
}
-
@Override
public Row deserialize(DataInputView source) throws IOException {
int len = fieldSerializers.length;
@@ -176,7 +175,8 @@ public final class RowSerializer extends TypeSerializer<Row> {
for (int i = 0; i < len; i++) {
if (nullMask[i]) {
result.setField(i, null);
- } else {
+ }
+ else {
result.setField(i, fieldSerializers[i].deserialize(source));
}
}
@@ -198,11 +198,13 @@ public final class RowSerializer extends TypeSerializer<Row> {
for (int i = 0; i < len; i++) {
if (nullMask[i]) {
reuse.setField(i, null);
- } else {
+ }
+ else {
Object reuseField = reuse.getField(i);
if (reuseField != null) {
reuse.setField(i, fieldSerializers[i].deserialize(reuseField, source));
- } else {
+ }
+ else {
reuse.setField(i, fieldSerializers[i].deserialize(source));
}
}
@@ -260,73 +262,83 @@ public final class RowSerializer extends TypeSerializer<Row> {
}
// --------------------------------------------------------------------------------------------
- // Serializer configuration snapshotting & compatibility
+ // Serializer configuration snapshoting & compatibility
// --------------------------------------------------------------------------------------------
@Override
- public RowSerializerConfigSnapshot snapshotConfiguration() {
- return new RowSerializerConfigSnapshot(fieldSerializers);
+ public TypeSerializerSnapshot<Row> snapshotConfiguration() {
+ return new RowSerializerSnapshot(this);
}
- @Override
- public CompatibilityResult<Row> ensureCompatibility(TypeSerializerConfigSnapshot<?> configSnapshot) {
- if (configSnapshot instanceof RowSerializerConfigSnapshot) {
- List<Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>>> previousFieldSerializersAndConfigs =
- ((RowSerializerConfigSnapshot) configSnapshot).getNestedSerializersAndConfigs();
-
- if (previousFieldSerializersAndConfigs.size() == fieldSerializers.length) {
- boolean requireMigration = false;
- TypeSerializer<?>[] convertDeserializers = new TypeSerializer<?>[fieldSerializers.length];
-
- CompatibilityResult<?> compatResult;
- int i = 0;
- for (Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>> f : previousFieldSerializersAndConfigs) {
- compatResult = CompatibilityUtil.resolveCompatibilityResult(
- f.f0,
- UnloadableDummyTypeSerializer.class,
- f.f1,
- fieldSerializers[i]);
-
- if (compatResult.isRequiresMigration()) {
- requireMigration = true;
-
- if (compatResult.getConvertDeserializer() == null) {
- // one of the field serializers cannot provide a fallback deserializer
- return CompatibilityResult.requiresMigration();
- } else {
- convertDeserializers[i] =
- new TypeDeserializerAdapter<>(compatResult.getConvertDeserializer());
- }
- }
+ /**
+ * A snapshot for {@link RowSerializer}.
+ *
+ * @deprecated this snapshot class is no longer in use, and is maintained only for backwards compatibility.
+ * It is fully replaced by {@link RowSerializerSnapshot}.
+ */
+ @Deprecated
+ public static final class RowSerializerConfigSnapshot extends CompositeTypeSerializerConfigSnapshot<Row> {
- i++;
- }
+ private static final int VERSION = 1;
- if (requireMigration) {
- return CompatibilityResult.requiresMigration(new RowSerializer(convertDeserializers));
- } else {
- return CompatibilityResult.compatible();
- }
- }
+ /**
+ * This empty nullary constructor is required for deserializing the configuration.
+ */
+ public RowSerializerConfigSnapshot() {
+ }
+
+ public RowSerializerConfigSnapshot(TypeSerializer[] fieldSerializers) {
+ super(fieldSerializers);
}
- return CompatibilityResult.requiresMigration();
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<Row> resolveSchemaCompatibility(TypeSerializer<Row> newSerializer) {
+ TypeSerializerSnapshot<?>[] nestedSnapshots = getNestedSerializersAndConfigs()
+ .stream()
+ .map(t -> t.f1)
+ .toArray(TypeSerializerSnapshot[]::new);
+
+ return CompositeTypeSerializerUtil.delegateCompatibilityCheckToNewSnapshot(
+ newSerializer,
+ new RowSerializerSnapshot(),
+ nestedSnapshots);
+ }
}
- public static final class RowSerializerConfigSnapshot extends CompositeTypeSerializerConfigSnapshot<Row> {
+ /**
+ * A {@link TypeSerializerSnapshot} for RowSerializer.
+ */
+ public static final class RowSerializerSnapshot extends CompositeTypeSerializerSnapshot<Row, RowSerializer> {
- private static final int VERSION = 1;
+ private static final int VERSION = 2;
- /** This empty nullary constructor is required for deserializing the configuration. */
- public RowSerializerConfigSnapshot() {}
+ @SuppressWarnings("WeakerAccess")
+ public RowSerializerSnapshot() {
+ super(RowSerializer.class);
+ }
- public RowSerializerConfigSnapshot(TypeSerializer[] fieldSerializers) {
- super(fieldSerializers);
+ RowSerializerSnapshot(RowSerializer serializerInstance) {
+ super(serializerInstance);
}
@Override
- public int getVersion() {
+ protected int getCurrentOuterSnapshotVersion() {
return VERSION;
}
+
+ @Override
+ protected TypeSerializer<?>[] getNestedSerializers(RowSerializer outerSerializer) {
+ return outerSerializer.fieldSerializers;
+ }
+
+ @Override
+ protected RowSerializer createOuterSerializerWithNestedSerializers(TypeSerializer<?>[] nestedSerializers) {
+ return new RowSerializer(nestedSerializers);
+ }
}
}
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/RowSerializerMigrationTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/RowSerializerMigrationTest.java
new file mode 100644
index 0000000..7a7888a
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/runtime/RowSerializerMigrationTest.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.api.java.typeutils.runtime;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshotMigrationTestBase;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.api.java.typeutils.runtime.RowSerializer.RowSerializerSnapshot;
+import org.apache.flink.testutils.migration.MigrationVersion;
+import org.apache.flink.types.Row;
+
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.Collection;
+
+/**
+ * State migration test for {@link RowSerializer}.
+ */
+@RunWith(Parameterized.class)
+public class RowSerializerMigrationTest extends TypeSerializerSnapshotMigrationTestBase<Row> {
+
+ public RowSerializerMigrationTest(TestSpecification<Row> testSpecification) {
+ super(testSpecification);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Parameterized.Parameters(name = "Test Specification = {0}")
+ public static Collection<TestSpecification<?>> testSpecifications() {
+
+ final TestSpecifications testSpecifications = new TestSpecifications(MigrationVersion.v1_6, MigrationVersion.v1_7);
+
+ testSpecifications.add(
+ "row-serializer",
+ RowSerializer.class,
+ RowSerializerSnapshot.class,
+ RowSerializerMigrationTest::stringLongRowSupplier);
+
+ return testSpecifications.get();
+ }
+
+ private static TypeSerializer<Row> stringLongRowSupplier() {
+ RowTypeInfo rowTypeInfo = new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
+ return rowTypeInfo.createSerializer(new ExecutionConfig());
+ }
+}
diff --git a/flink-core/src/test/resources/flink-1.6-row-serializer-data b/flink-core/src/test/resources/flink-1.6-row-serializer-data
new file mode 100644
index 0000000..2cd0bd3
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-row-serializer-data differ
diff --git a/flink-core/src/test/resources/flink-1.6-row-serializer-snapshot b/flink-core/src/test/resources/flink-1.6-row-serializer-snapshot
new file mode 100644
index 0000000..9200e23
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.6-row-serializer-snapshot differ
diff --git a/flink-core/src/test/resources/flink-1.7-row-serializer-data b/flink-core/src/test/resources/flink-1.7-row-serializer-data
new file mode 100644
index 0000000..41aa078
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-row-serializer-data differ
diff --git a/flink-core/src/test/resources/flink-1.7-row-serializer-snapshot b/flink-core/src/test/resources/flink-1.7-row-serializer-snapshot
new file mode 100644
index 0000000..f632dc4
Binary files /dev/null and b/flink-core/src/test/resources/flink-1.7-row-serializer-snapshot differ