[FLINK-11329][core] Migrating CompositeSerializers #7590

igalshilman · 2019-01-28T20:01:47Z

What is the purpose of the change

This PR migrates the following serializers to the new TypeSerializerSnapshot abstraction:

ScalaOptionSerializer 4e1a755
StreamElementSerializer 9a0ddc1
BufferEntrySerializer 9a56f43
UnionSerializer 317e6b8
RowSerializer 3a1a867
NullableSerializer d93ff8a
CompositeSerializer ce25363

Brief change log

Each commit migrates a single serializer's TypeSerializerConfigSnapshot to the new TypeSerializerSnapsoht.

Verifying this change

Each migrated serializer has an associated StateMigrationTest that verifies snapshot backwards compatibility with previous versions.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

tzulitai

Review for commit 516d39a.

tzulitai · 2019-01-30T06:11:01Z

flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java

@@ -302,6 +299,7 @@ static PrecomputedParameters precompute(
 	}

 	/** Snapshot field serializers of composite type. */


nit: Add @deprecated message and direct to new snapshot class.

tzulitai · 2019-01-30T06:18:17Z

...stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/ValueWithTs.java

@@ -56,8 +57,8 @@ public String toString() {
 	/** Serializer for Serializer. */
 	public static class Serializer extends CompositeSerializer<ValueWithTs<?>> {

-		public Serializer(TypeSerializer<?> userValueSerializer) {
-			super(true, userValueSerializer, LongSerializer.INSTANCE);
+		public Serializer(TypeSerializer<?> valueSerializer, TypeSerializer<Long> timestampSerializer) {


I don't think we need a public constructor that accepts the timestamp serializer.
This should be a private constructor used only by the snapshot class.

We should still have a public constructor that accepts the user value serializer, and by default just uses LongSerializer.INSTANCE as the new timestamp serializer.

As far as I can tell this is not a user facing serializer, rather used internally by the TtlSerializer,
and I think it is important to make it explicit that this is a composite serializer and these are the nested serializers that define it.

As a way to reduce verbosity, we can add a static factory method with an explicit name.
Feel free to decide for yourself :-)

ok, that's a fair argument that I can agree with. Since I don't have a strong opinion on this, I'll leave the constructors as is in this PR.

tzulitai · 2019-01-30T06:18:38Z

...stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/ValueWithTs.java

+		@Override
+		protected Serializer createOuterSerializerWithNestedSerializers(TypeSerializer<?>[] nestedSerializers) {
+			TypeSerializer<?> valueSerializer = nestedSerializers[0];
+			TypeSerializer<Long> timeSerializer = (TypeSerializer<Long>) nestedSerializers[1];


nit: time --> timestamp for naming consistency

tzulitai · 2019-01-30T06:19:48Z

flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlStateFactory.java

@@ -126,15 +128,15 @@ private IS createState() throws Exception {
 	@SuppressWarnings("unchecked")
 	private IS createValueState() throws Exception {
 		ValueStateDescriptor<TtlValue<SV>> ttlDescriptor = new ValueStateDescriptor<>(
-			stateDesc.getName(), new TtlSerializer<>(stateDesc.getSerializer()));
+			stateDesc.getName(), new TtlSerializer<>(LongSerializer.INSTANCE, stateDesc.getSerializer()));


As mentioned above, having to pass in a LongSerializer.INSTANCE every time we're instantiating a TtlSerializer seems very redundant.

I disagree, same reasons as above.

tzulitai · 2019-01-30T06:31:09Z

...stream-state-ttl-test/src/main/java/org/apache/flink/streaming/tests/verify/ValueWithTs.java

@@ -56,8 +57,8 @@ public String toString() {
 	/** Serializer for Serializer. */
 	public static class Serializer extends CompositeSerializer<ValueWithTs<?>> {


This serializer class previously did not have a serialVersionUID defined.
Need to explicitly set it to what it was before, because I guess the serial version UID would have changed when adding the new constructors.

OTOH, there seems to be missing a migration test for this serializer, because that would have caught this problem.

Ah scratch that, just realized that this is only a serializer used in tests.

But still, I'd actually suggest adding that, to enforce good practices.

tzulitai

Review for 8bba092.
Changes here looks good 👍

tzulitai · 2019-01-30T06:53:50Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java

-		}
-
-		return CompatibilityResult.requiresMigration();
-	}

 	/**
 	 * Configuration snapshot for serializers of nullable types, containing the
 	 * configuration snapshot of its original serializer.


nit: Add @deprecated message.

tzulitai

d876fd0 LGTM 👍

tzulitai · 2019-01-30T07:19:07Z

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/RowSerializer.java

-					}
+	/**
+	 * A snapshot for {@link RowSerializer}.
+	 */


nit: Add @deprecated message.

tzulitai

58a3692 LGTM 👍

tzulitai · 2019-01-30T07:25:33Z

...streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/CoGroupedStreams.java

-
-			return CompatibilityResult.requiresMigration();
+		public TypeSerializerSnapshot<TaggedUnion<T1, T2>> snapshotConfiguration() {
+			return new UnionSerializerSnapshot<>(this);
 		}
 	}

 	/**
 	 * The {@link TypeSerializerConfigSnapshot} for the {@link UnionSerializer}.


nit: Add @deprecated message

tzulitai

c375ca0 LGTM 👍

tzulitai

0fa9adb LGTM 👍

tzulitai

One problem with 307f6ee.
Please see comment.

tzulitai · 2019-01-30T07:47:12Z

.../src/main/java/org/apache/flink/api/scala/typeutils/ScalaOptionSerializerConfigSnapshot.java

@@ -31,6 +33,7 @@
 * allow calling different base class constructors from subclasses, while we need that
 * for the default empty constructor.


nit: Add @deprecated message.

tzulitai · 2019-01-30T07:51:23Z

flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/OptionSerializer.scala

-          : ScalaOptionSerializerConfigSnapshot[A] =>
-        ensureCompatibilityInternal(optionSerializerConfigSnapshot)
-      case legacyOptionSerializerConfigSnapshot
-          : OptionSerializer.OptionSerializerConfigSnapshot[A] =>


Removing this path will lead to problems when restoring from Flink 1.3, because this snapshot class was used back in Flink 1.3.

OTOH, it should be possible to redirect OptionSerializerConfigSnapshot's compatibility check to the new ScalaOptionSerializerSnapshot.

…lizerConfigSnapshot to ScalaOptionSerializerSnapshot This closes apache#7590.

tzulitai · 2019-01-30T09:56:41Z

Thanks for the work @igalshilman!

For every serializer that is touched, I checked that:

The touched serializer has a serialVersionUID defined already so that we're not breaking backwards compatibility
The ensureCompatibility method is removed (except from CompositeSerializer)
Old config snapshot correctly delegates compatibility check to new serializer
Has migration tests that cover the change

LGTM. I'll merge https://github.com/tzulitai/flink/tree/igal-dawid-gordon-composite once Travis is green!

Igal Shilman added 8 commits January 29, 2019 13:54

[FLINK-11329] Migrating CompositeSerializer

516d39a

[FLINK-11329] Migrating the NullableSerializer

8bba092

[FLINK-11329] Migrating the RowSerializer

d876fd0

[FLINK-11329] Migrating the UnionSerializer

58a3692

[FLINK-11329] Migrate BufferEntrySerializer

c375ca0

[FLINK-11329] Migrate StreamElementSerializer

0fa9adb

[FLINK-11329] Migrating ScalaOptionSerializer

307f6ee

[FLINK-11329] Adjusting checkstyle

688b910

igalshilman force-pushed the composite-serializers-pr branch from 6c2eb62 to 688b910 Compare January 29, 2019 13:24

tzulitai reviewed Jan 30, 2019

View reviewed changes

tzulitai added a commit to tzulitai/flink that referenced this pull request Jan 30, 2019

[FLINK-11329] [scala] Delegate compatibility check on old OptionSeria…

a559bc8

…lizerConfigSnapshot to ScalaOptionSerializerSnapshot This closes apache#7590.

tzulitai added a commit to tzulitai/flink that referenced this pull request Jan 30, 2019

[FLINK-11329] [scala] Delegate compatibility check on old OptionSeria…

a6c0c2f

…lizerConfigSnapshot to ScalaOptionSerializerSnapshot This closes apache#7590.

asfgit closed this in 198ceb2 Jan 30, 2019

rmetzger added component=API/TypeSerializationSystem component=Runtime/StateBackends labels Mar 18, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-11329][core] Migrating CompositeSerializers #7590

[FLINK-11329][core] Migrating CompositeSerializers #7590

igalshilman commented Jan 28, 2019

tzulitai left a comment

tzulitai Jan 30, 2019

tzulitai Jan 30, 2019

igalshilman Jan 30, 2019

tzulitai Jan 30, 2019

tzulitai Jan 30, 2019

tzulitai Jan 30, 2019

igalshilman Jan 30, 2019

tzulitai Jan 30, 2019 •

edited

tzulitai Jan 30, 2019

igalshilman Jan 30, 2019

tzulitai left a comment

tzulitai Jan 30, 2019

tzulitai left a comment

tzulitai Jan 30, 2019

tzulitai left a comment

tzulitai Jan 30, 2019

tzulitai left a comment

tzulitai left a comment

tzulitai left a comment

tzulitai Jan 30, 2019

tzulitai Jan 30, 2019

tzulitai commented Jan 30, 2019

		@@ -302,6 +299,7 @@ static PrecomputedParameters precompute(
		}

		/** Snapshot field serializers of composite type. */

		@@ -56,8 +57,8 @@ public String toString() {
		/** Serializer for Serializer. */
		public static class Serializer extends CompositeSerializer<ValueWithTs<?>> {

		@@ -31,6 +33,7 @@
		* allow calling different base class constructors from subclasses, while we need that
		* for the default empty constructor.

[FLINK-11329][core] Migrating CompositeSerializers #7590

[FLINK-11329][core] Migrating CompositeSerializers #7590

Conversation

igalshilman commented Jan 28, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai Jan 30, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

tzulitai left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tzulitai commented Jan 30, 2019

tzulitai Jan 30, 2019 •

edited