diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java index 1fe4134ee51c9a..45b56d9fa10076 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java @@ -134,6 +134,41 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo TypeSerializerSchemaCompatibility resolveSchemaCompatibility( TypeSerializerSnapshot oldSerializerSnapshot); + /** + * Migrates a single state value from the schema described by {@code oldSerializerSnapshot} to + * the schema described by this (new) snapshot. Like {@link + * #resolveSchemaCompatibility(TypeSerializerSnapshot)}, this is invoked on the new snapshot and + * receives the old snapshot as its argument. + * + *

The default implementation returns the value unchanged: a value already deserialized with + * the prior serializer is structurally compatible with the current serializer, so the caller + * can re-serialize it as-is. A serializer whose in-memory representation is coupled to its + * schema should override this to transform the value into the new layout -- for example by + * inserting nulls for added fields or reordering fields by name. An implementation may return + * the given value or a new instance. + * + *

The migration is not applied recursively to nested serializers. The snapshot of a + * composite type returns its value unchanged unless it overrides this method to decompose the + * value and migrate each part, so a caller that needs a nested value migrated must reach the + * nested snapshot itself. An implementation that does so should not assume that the old and the + * new snapshot expose nested snapshots of the same type: restoring may have replaced those of + * the old snapshot with decorators, so nested snapshots are best matched by position or name + * rather than by class. + * + * @param oldSerializerSnapshot snapshot of the serializer that wrote the value. A caller that + * holds the snapshot persisted with the state should pass that one in preference to a + * snapshot re-derived from a serializer restored from it, because that round trip does not + * always reproduce the schema that was written. + * @param value the value, already deserialized with the prior serializer. It may be {@code + * null} wherever the prior serializer can produce {@code null}. An implementation that + * decomposes a composite value may likewise pass {@code null} to a nested snapshot for an + * absent part, even where that part's serializer would reject {@code null} at top level. + * @return the value adapted to the schema of the current serializer. + */ + default T migrate(TypeSerializerSnapshot oldSerializerSnapshot, T value) { + return value; + } + // ------------------------------------------------------------------------ // read / write utilities // ------------------------------------------------------------------------ diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java index b176adc482b36f..3dbace1bf58583 100644 --- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java @@ -54,6 +54,15 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( .isTrue(); } + @Test + void testMigrateReturnsValueUnchangedByDefault() { + TypeSerializerSnapshot oldSnapshot = new NotCompletedTypeSerializerSnapshot(); + TypeSerializerSnapshot newSnapshot = new NotCompletedTypeSerializerSnapshot(); + Integer value = 1000; + + assertThat(newSnapshot.migrate(oldSnapshot, value)).isSameAs(value); + } + private static class NotCompletedTypeSerializer extends TypeSerializer { @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java index 2f50ba6c668337..0dabcc5c869b24 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java @@ -26,6 +26,8 @@ import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.util.function.SupplierWithException; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.List; import java.util.Map; @@ -47,6 +49,13 @@ public class TtlAwareSerializer> extends TypeSeri private final S typeSerializer; + /** + * Snapshot of {@link #bareValueSerializer()}, computed on first use. {@link + * #migrateValueFromPriorSerializer} runs once per migrated state value while the serializer + * stays the same, and taking a snapshot allocates one object per nested serializer. + */ + private transient TypeSerializerSnapshot bareValueSerializerSnapshot; + public TtlAwareSerializer(S typeSerializer) { checkArgument( !(typeSerializer instanceof TtlAwareSerializer), @@ -128,31 +137,129 @@ public int hashCode() { return Objects.hash(isTtlEnabled, typeSerializer); } - @SuppressWarnings("unchecked") + /** + * Reads one state value written by {@code priorTtlAwareSerializer}, adapts it to this + * serializer's TTL setting and value schema, and writes it to {@code target}. + * + *

The value is unwrapped to its bare form, passed through {@link + * TypeSerializerSnapshot#migrate}, and re-wrapped. The hook returns the value unchanged unless + * the value serializer overrides it, so a value whose schema did not change is written back + * byte for byte. + * + * @param priorSerializerSnapshot the snapshot persisted with the state for {@code + * priorTtlAwareSerializer}, or {@code null} for a state that carries none. + */ + @SuppressWarnings({"unchecked", "rawtypes"}) public void migrateValueFromPriorSerializer( TtlAwareSerializer priorTtlAwareSerializer, + @Nullable TypeSerializerSnapshot priorSerializerSnapshot, SupplierWithException inputSupplier, DataOutputView target, TtlTimeProvider ttlTimeProvider) throws IOException { + T priorValue = inputSupplier.get(); + Object bareValue = + priorTtlAwareSerializer.wrapsTtlValue() + ? ((TtlValue) priorValue).getUserValue() + : priorValue; + + TypeSerializerSnapshot newSnapshot = bareValueSerializerSnapshot(); + Object migratedValue = + newSnapshot.migrate( + priorBareValueSerializerSnapshot( + priorTtlAwareSerializer, priorSerializerSnapshot), + bareValue); + T outputRecord; - if (this.isTtlEnabled()) { - outputRecord = - priorTtlAwareSerializer.isTtlEnabled - ? inputSupplier.get() - : (T) - new TtlValue<>( - inputSupplier.get(), - ttlTimeProvider.currentTimestamp()); + if (this.wrapsTtlValue()) { + // Carrying the prior timestamp over keeps the value's expiry where it was; migration + // is not a state access. + long lastAccessTimestamp = + priorTtlAwareSerializer.wrapsTtlValue() + ? ((TtlValue) priorValue).getLastAccessTimestamp() + : ttlTimeProvider.currentTimestamp(); + outputRecord = (T) new TtlValue<>(migratedValue, lastAccessTimestamp); } else { - outputRecord = - priorTtlAwareSerializer.isTtlEnabled - ? ((TtlValue) inputSupplier.get()).getUserValue() - : inputSupplier.get(); + outputRecord = (T) migratedValue; } this.serialize(outputRecord, target); } + /** + * The snapshot describing the schema the prior bare value was written with. + * + *

The snapshot persisted with the state is preferred over one re-derived from the prior + * serializer, because the prior serializer is itself restored from that snapshot and the round + * trip back to a snapshot is not always lossless: a POJO field that no longer exists on the + * class returns under a generated placeholder name, which would present a schema that was never + * written. Only the absence of a persisted snapshot falls back to the re-derived one: a + * persisted snapshot that does not match the prior serializer is an error, not a second reason + * to fall back, because re-deriving there would silently reintroduce that lossy round trip. + */ + private static TypeSerializerSnapshot priorBareValueSerializerSnapshot( + TtlAwareSerializer priorSerializer, + @Nullable TypeSerializerSnapshot priorSerializerSnapshot) { + if (priorSerializerSnapshot == null) { + return priorSerializer.bareValueSerializerSnapshot(); + } + // TtlAwareSerializerSnapshot is the snapshot counterpart of this class, so the persisted + // snapshot carries that layer wherever the serializer carries the wrapper: for a list or + // map state it is the element or value snapshot, for a value state the whole snapshot. + TypeSerializerSnapshot priorSnapshot = + priorSerializerSnapshot instanceof TtlAwareSerializerSnapshot + ? ((TtlAwareSerializerSnapshot) priorSerializerSnapshot) + .getOrinalTypeSerializerSnapshot() + : priorSerializerSnapshot; + + // Thrown rather than checked through Preconditions: this runs once per migrated state + // value, so the message must not be built while the check is passing. + boolean isTtlSnapshot = priorSnapshot instanceof TtlStateFactory.TtlSerializerSnapshot; + if (!priorSerializer.wrapsTtlValue()) { + if (isTtlSnapshot) { + throw new IllegalArgumentException( + "The prior serializer does not wrap values in TtlValue, but its persisted snapshot is a TtlSerializerSnapshot."); + } + return priorSnapshot; + } + if (!isTtlSnapshot) { + throw new IllegalArgumentException( + "The prior serializer wraps values in TtlValue, so its persisted snapshot should be a TtlSerializerSnapshot, but was " + + priorSnapshot.getClass().getName() + + "."); + } + // The persisted snapshot describes the TtlValue envelope, so descend to the user value + // the same way bareValueSerializer() descends the serializer. + return ((TtlStateFactory.TtlSerializerSnapshot) priorSnapshot) + .getValueSerializerSnapshot(); + } + + private TypeSerializerSnapshot bareValueSerializerSnapshot() { + if (bareValueSerializerSnapshot == null) { + bareValueSerializerSnapshot = bareValueSerializer().snapshotConfiguration(); + } + return bareValueSerializerSnapshot; + } + + /** + * The serializer of the bare (non-TTL) value: the user value serializer of a {@link + * TtlStateFactory.TtlSerializer}, otherwise the wrapped serializer itself. + */ + private TypeSerializer bareValueSerializer() { + return wrapsTtlValue() + ? ((TtlStateFactory.TtlSerializer) typeSerializer).getValueSerializer() + : typeSerializer; + } + + /** + * Whether the values this serializer reads and writes are {@link TtlValue} envelopes. Narrower + * than {@link #isTtlEnabled()}, which is also true for a list or map serializer whose element + * or value serializer is a {@link TtlStateFactory.TtlSerializer}: such a serializer wraps the + * collection, not a single {@code TtlValue}. + */ + private boolean wrapsTtlValue() { + return typeSerializer instanceof TtlStateFactory.TtlSerializer; + } + @Override public void copy(DataInputView source, DataOutputView target) throws IOException { typeSerializer.copy(source, target); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java index 65908573facd6c..97ba6b19793bef 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java @@ -21,13 +21,18 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.StringSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.runtime.state.ttl.TtlAwareSerializerSnapshot; import org.apache.flink.runtime.testutils.statemigration.TestType; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -311,6 +316,42 @@ void testEagerlyRegisterIncompatibleSerializer() { .isInstanceOf(IllegalStateException.class); } + // -------------------------------------------------------------------------------- + // Tests for the ttl-aware wrapping of the previous serializer snapshot + // -------------------------------------------------------------------------------- + + /** + * Registering a new serializer replaces the nested snapshot of the previous snapshot in place, + * so the snapshot a caller still holds no longer reports what the checkpoint wrote: its nested + * snapshot becomes a {@link TtlAwareSerializerSnapshot} around the original. Anything that + * descends a restored composite snapshot has to expect that layer. + */ + @Test + void testRegisterNewSerializerWrapsNestedSnapshotOfPreviousSnapshotInPlace() { + ListSerializerSnapshot previousSnapshot = + (ListSerializerSnapshot) + new ListSerializer<>(StringSerializer.INSTANCE).snapshotConfiguration(); + TypeSerializerSnapshot elementSnapshotAsWritten = + previousSnapshot.getElementSerializerSnapshot(); + + StateSerializerProvider> testProvider = + StateSerializerProvider.fromPreviousSerializerSnapshot(previousSnapshot); + testProvider.registerNewSerializerForRestoredState( + new ListSerializer<>(StringSerializer.INSTANCE)); + + // The same snapshot instance now reports a different element snapshot than it did above. + TypeSerializerSnapshot elementSnapshotAfterRestore = + previousSnapshot.getElementSerializerSnapshot(); + assertThat(elementSnapshotAfterRestore).isNotSameAs(elementSnapshotAsWritten); + assertThat(elementSnapshotAfterRestore).isInstanceOf(TtlAwareSerializerSnapshot.class); + // The original is carried inside the wrapper, not re-derived: a re-derived snapshot would + // be an equal instance of the same class but a different object. + assertThat( + ((TtlAwareSerializerSnapshot) elementSnapshotAfterRestore) + .getOrinalTypeSerializerSnapshot()) + .isSameAs(elementSnapshotAsWritten); + } + // -------------------------------------------------------------------------------- // Utilities // -------------------------------------------------------------------------------- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java index 90c7f70432c00b..b15cc46eb9b706 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java @@ -18,19 +18,36 @@ package org.apache.flink.runtime.state.ttl; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.IntSerializer; import org.apache.flink.api.common.typeutils.base.ListSerializer; import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.LongSerializer; import org.apache.flink.api.common.typeutils.base.MapSerializer; import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.java.typeutils.runtime.NullableSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.core.memory.DataOutputView; import org.junit.jupiter.api.Test; +import java.io.IOException; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class TtlAwareSerializerTest { + private static final String VALUE = "value"; + private static final long PRIOR_TIMESTAMP = 1_000L; + private static final long CURRENT_TIMESTAMP = 9_999L; + private static final TtlTimeProvider FIXED_TIME_PROVIDER = () -> CURRENT_TIMESTAMP; + @Test void testSerializerTtlEnabled() { IntSerializer intSerializer = IntSerializer.INSTANCE; @@ -147,4 +164,385 @@ void testSnapshotConfiguration() { .getValueSerializerSnapshot())) .isInstanceOf(TtlAwareSerializerSnapshot.class); } + + @Test + void testMigrateValueNoTtlToNoTtl() throws IOException { + TtlAwareSerializer prior = stringSerializer(false); + TtlAwareSerializer current = stringSerializer(false); + + assertThat(migrate(current, prior, VALUE)).isEqualTo(serialize(current, VALUE)); + } + + @Test + void testMigrateValueNoTtlToTtlStampsCurrentTime() throws IOException { + TtlAwareSerializer prior = stringSerializer(false); + TtlAwareSerializer current = stringSerializer(true); + + assertThat(migrate(current, prior, VALUE)) + .isEqualTo(serialize(current, new TtlValue<>(VALUE, CURRENT_TIMESTAMP))); + } + + @Test + void testMigrateValueTtlToNoTtlUnwraps() throws IOException { + TtlAwareSerializer prior = stringSerializer(true); + TtlAwareSerializer current = stringSerializer(false); + + assertThat(migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isEqualTo(serialize(current, VALUE)); + } + + @Test + void testMigrateValueTtlToTtlKeepsPriorTimestamp() throws IOException { + TtlAwareSerializer prior = stringSerializer(true); + TtlAwareSerializer current = stringSerializer(true); + TtlValue priorValue = new TtlValue<>(VALUE, PRIOR_TIMESTAMP); + + assertThat(migrate(current, prior, priorValue)).isEqualTo(serialize(current, priorValue)); + } + + @Test + void testMigrateValueInvokesHookOnNewSnapshotWithOldSnapshotAsArgument() throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=old|to=new"); + } + + @Test + void testMigrateValueInvokesHookOnUnwrappedValueWithInnerSnapshots() throws IOException { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + byte[] migrated = migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=old|to=new"); + assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP); + } + + @Test + void testMigrateValueUsesPersistedPriorSnapshotNotOneRederivedFromPriorSerializer() + throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("rederived")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, new TaggedSnapshot("persisted"), VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=persisted|to=new"); + } + + @Test + void testMigrateValueDescendsPersistedTtlSnapshotToItsValueSnapshot() throws IOException { + TtlAwareSerializer prior = taggedTtlSerializer("rederived"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + // Tagged differently from the prior serializer, so both falling back to a re-derived + // snapshot and skipping the descent into the TtlValue envelope change the result. + TypeSerializerSnapshot persisted = + taggedTtlSerializer("persisted") + .getOriginalTypeSerializer() + .snapshotConfiguration(); + + byte[] migrated = + migrate(current, prior, persisted, new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new"); + } + + /** + * A list or map state persists its element or value snapshot wrapped in a {@link + * TtlAwareSerializerSnapshot}, which the descent has to see through to reach the user value + * snapshot. Value state persists the unwrapped shape covered by the test above. + */ + @Test + void testMigrateValueDescendsPersistedTtlAwareElementSnapshot() throws IOException { + ListSerializerSnapshot persistedListSnapshot = + (ListSerializerSnapshot) + TtlAwareSerializer.wrapTtlAwareSerializer( + new ListSerializer<>(rawTaggedTtlSerializer("persisted"))) + .snapshotConfiguration(); + TypeSerializerSnapshot persistedElementSnapshot = + persistedListSnapshot.getElementSerializerSnapshot(); + assertThat(persistedElementSnapshot).isInstanceOf(TtlAwareSerializerSnapshot.class); + + TtlAwareSerializer prior = taggedTtlSerializer("rederived"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + byte[] migrated = + migrate( + current, + prior, + persistedElementSnapshot, + new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new"); + } + + @Test + void testMigrateValueRejectsNonTtlSnapshotForTtlPriorSerializer() { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + assertThatThrownBy( + () -> + migrate( + current, + prior, + new TaggedSnapshot("mismatched"), + new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("should be a TtlSerializerSnapshot"); + } + + @Test + void testMigrateValueRejectsNonTtlSnapshotBehindTtlAwareLayer() { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + // The TtlAware layer is seen through, so a mismatch inside it is still caught. + TypeSerializerSnapshot wrappedMismatch = + new TtlAwareSerializerSnapshot<>(new TaggedSnapshot("mismatched")); + + assertThatThrownBy( + () -> + migrate( + current, + prior, + wrappedMismatch, + new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("should be a TtlSerializerSnapshot"); + } + + @Test + void testMigrateValueRejectsTtlSnapshotForNonTtlPriorSerializer() { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + TypeSerializerSnapshot ttlSnapshot = + rawTaggedTtlSerializer("mismatched").snapshotConfiguration(); + + assertThatThrownBy(() -> migrate(current, prior, ttlSnapshot, VALUE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not wrap values in TtlValue"); + } + + @Test + void testMigrateValueFallsBackToPriorSerializerSnapshotWhenNonePersisted() throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, null, VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=old|to=new"); + } + + @Test + void testMigrateValueTtlToTtlWithNullUserValue() throws IOException { + TtlAwareSerializer prior = nullTolerantTtlSerializer(); + TtlAwareSerializer current = nullTolerantTtlSerializer(); + + byte[] migrated = migrate(current, prior, new TtlValue<>(null, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isNull(); + assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP); + } + + /** Migrates with the snapshot the state backends would have persisted for {@code prior}. */ + private static byte[] migrate( + TtlAwareSerializer current, TtlAwareSerializer prior, Object priorValue) + throws IOException { + return migrate( + current, + prior, + prior.getOriginalTypeSerializer().snapshotConfiguration(), + priorValue); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static byte[] migrate( + TtlAwareSerializer current, + TtlAwareSerializer prior, + TypeSerializerSnapshot priorSnapshot, + Object priorValue) + throws IOException { + DataOutputSerializer output = new DataOutputSerializer(64); + ((TtlAwareSerializer) current) + .migrateValueFromPriorSerializer( + (TtlAwareSerializer) prior, + priorSnapshot, + () -> priorValue, + output, + FIXED_TIME_PROVIDER); + return output.getCopyOfBuffer(); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static byte[] serialize(TtlAwareSerializer serializer, Object value) + throws IOException { + DataOutputSerializer output = new DataOutputSerializer(64); + ((TtlAwareSerializer) serializer).serialize(value, output); + return output.getCopyOfBuffer(); + } + + private static TtlAwareSerializer stringSerializer(boolean ttlEnabled) { + return ttlEnabled + ? TtlAwareSerializer.wrapTtlAwareSerializer( + new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, StringSerializer.INSTANCE)) + : TtlAwareSerializer.wrapTtlAwareSerializer(StringSerializer.INSTANCE); + } + + private static TtlAwareSerializer taggedTtlSerializer(String tag) { + return TtlAwareSerializer.wrapTtlAwareSerializer(rawTaggedTtlSerializer(tag)); + } + + private static TtlStateFactory.TtlSerializer rawTaggedTtlSerializer(String tag) { + return new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, new TaggedStringSerializer(tag)); + } + + private static TtlAwareSerializer nullTolerantTtlSerializer() { + return TtlAwareSerializer.wrapTtlAwareSerializer( + new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, + NullableSerializer.wrapIfNullIsNotSupported( + StringSerializer.INSTANCE, false))); + } + + /** A string serializer whose snapshot records the schema it belongs to. */ + private static final class TaggedStringSerializer extends TypeSerializer { + + private final String tag; + + private TaggedStringSerializer(String tag) { + this.tag = tag; + } + + @Override + public boolean isImmutableType() { + return true; + } + + @Override + public TypeSerializer duplicate() { + return this; + } + + @Override + public String createInstance() { + return ""; + } + + @Override + public String copy(String from) { + return from; + } + + @Override + public String copy(String from, String reuse) { + return from; + } + + @Override + public int getLength() { + return -1; + } + + @Override + public void serialize(String record, DataOutputView target) throws IOException { + StringSerializer.INSTANCE.serialize(record, target); + } + + @Override + public String deserialize(DataInputView source) throws IOException { + return StringSerializer.INSTANCE.deserialize(source); + } + + @Override + public String deserialize(String reuse, DataInputView source) throws IOException { + return deserialize(source); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + StringSerializer.INSTANCE.copy(source, target); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof TaggedStringSerializer + && tag.equals(((TaggedStringSerializer) obj).tag); + } + + @Override + public int hashCode() { + return tag.hashCode(); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new TaggedSnapshot(tag); + } + } + + /** + * Appends both the argument snapshot's tag and its own tag to the migrated value, so a + * migration that swaps receiver and argument produces a different result. + */ + public static final class TaggedSnapshot implements TypeSerializerSnapshot { + + private String tag; + + public TaggedSnapshot() {} + + private TaggedSnapshot(String tag) { + this.tag = tag; + } + + @Override + public int getCurrentVersion() { + return 1; + } + + @Override + public void writeSnapshot(DataOutputView out) throws IOException { + out.writeUTF(tag); + } + + @Override + public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException { + tag = in.readUTF(); + } + + @Override + public TypeSerializer restoreSerializer() { + return new TaggedStringSerializer(tag); + } + + @Override + public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( + TypeSerializerSnapshot oldSerializerSnapshot) { + return TypeSerializerSchemaCompatibility.compatibleAsIs(); + } + + @Override + public String migrate(TypeSerializerSnapshot oldSerializerSnapshot, String value) { + return value + "|from=" + ((TaggedSnapshot) oldSerializerSnapshot).tag + "|to=" + tag; + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java index d8b4234252c244..ffe2f41c03a4e7 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java @@ -161,6 +161,7 @@ public DataInputView readAndThenWriteData( DataOutputSerializer migratedOut = new DataOutputSerializer(INITIAL_OUTPUT_BUFFER_SIZE); writer.migrateValueFromPriorSerializer( reader, + reader.getOriginalTypeSerializer().snapshotConfiguration(), () -> reader.deserialize(originalDataInput), migratedOut, TtlTimeProvider.DEFAULT); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java index b60347ff208128..758f71237933d4 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java @@ -19,6 +19,7 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -36,6 +37,8 @@ import org.rocksdb.RocksDBException; import org.rocksdb.WriteOptions; +import javax.annotation.Nullable; + import java.io.IOException; import static org.apache.flink.util.Preconditions.checkArgument; @@ -191,6 +194,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer priorSerializer, + @Nullable TypeSerializerSnapshot priorSerializerSnapshot, TypeSerializer newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -203,6 +207,7 @@ public void migrateSerializedValue( try { ttlAwareNewSerializer.migrateValueFromPriorSerializer( ttlAwarePriorSerializer, + priorSerializerSnapshot, () -> ttlAwarePriorSerializer.deserialize(serializedOldValueInput), serializedMigratedValueOutput, ttlTimeProvider); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java index b77232cb067b3c..d76c672fba8505 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java @@ -886,9 +886,12 @@ private void migrateStateValues( Tuple2> stateMetaInfo) throws Exception { + // The snapshot the state was written with. Preferred over one re-derived from the previous + // serializer, which is itself restored from this snapshot and does not always round trip. + TypeSerializerSnapshot previousSerializerSnapshot = + stateMetaInfo.f1.getPreviousStateSerializerSnapshot(); + if (stateDesc.getType() == StateDescriptor.Type.MAP) { - TypeSerializerSnapshot previousSerializerSnapshot = - stateMetaInfo.f1.getPreviousStateSerializerSnapshot(); checkState( previousSerializerSnapshot != null, "the previous serializer snapshot should exist."); @@ -973,6 +976,7 @@ private void migrateStateValues( serializedValueInput, migratedSerializedValueOutput, previousTtlAwareSerializer, + previousSerializerSnapshot, currentTtlAwareSerializer, this.ttlTimeProvider); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java index 8c6088df23d539..66b273a5ecb085 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java @@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -200,6 +202,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer> priorSerializer, + @Nullable TypeSerializerSnapshot> priorSerializerSnapshot, TypeSerializer> newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -214,11 +217,29 @@ public void migrateSerializedValue( TtlAwareSerializer newTtlAwareElementSerializer = ((TtlAwareSerializer.TtlAwareListSerializer) newSerializer) .getElementSerializer(); + // Descend the persisted snapshot the same way as the serializer, so element migration + // sees the schema the elements were written with. A state that carries no persisted + // snapshot leaves this null, and the element migration re-derives one instead. + TypeSerializerSnapshot priorElementSerializerSnapshot = null; + if (priorSerializerSnapshot != null) { + // Thrown rather than checked through Preconditions: this method runs once per state + // entry, so the message must not be built while the check is passing. + if (!(priorSerializerSnapshot instanceof ListSerializerSnapshot)) { + throw new IllegalArgumentException( + "The previous serializer snapshot of a list state should be a ListSerializerSnapshot, but was " + + priorSerializerSnapshot.getClass().getName() + + "."); + } + priorElementSerializerSnapshot = + ((ListSerializerSnapshot) priorSerializerSnapshot) + .getElementSerializerSnapshot(); + } try { while (serializedOldValueInput.available() > 0) { newTtlAwareElementSerializer.migrateValueFromPriorSerializer( priorTtlAwareElementSerializer, + priorElementSerializerSnapshot, () -> ListDelimitedSerializer.deserializeNextElement( serializedOldValueInput, priorTtlAwareElementSerializer), diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java index c811fc8caac2ca..f87ad7d6b2fcfa 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java @@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.MapSerializer; +import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -227,6 +229,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer> priorSerializer, + @Nullable TypeSerializerSnapshot> priorSerializerSnapshot, TypeSerializer> newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -240,6 +243,23 @@ public void migrateSerializedValue( TtlAwareSerializer newTtlAwareMapValueSerializer = ((TtlAwareSerializer.TtlAwareMapSerializer) newSerializer) .getValueSerializer(); + // Descend the persisted snapshot the same way as the serializer, so value migration sees + // the schema the map values were written with. A state that carries no persisted + // snapshot leaves this null, and the value migration re-derives one instead. + TypeSerializerSnapshot priorMapValueSerializerSnapshot = null; + if (priorSerializerSnapshot != null) { + // Thrown rather than checked through Preconditions: this method runs once per state + // entry, so the message must not be built while the check is passing. + if (!(priorSerializerSnapshot instanceof MapSerializerSnapshot)) { + throw new IllegalArgumentException( + "The previous serializer snapshot of a map state should be a MapSerializerSnapshot, but was " + + priorSerializerSnapshot.getClass().getName() + + "."); + } + priorMapValueSerializerSnapshot = + ((MapSerializerSnapshot) priorSerializerSnapshot) + .getValueSerializerSnapshot(); + } try { boolean isNull = serializedOldValueInput.readBoolean(); @@ -250,6 +270,7 @@ public void migrateSerializedValue( } else { newTtlAwareMapValueSerializer.migrateValueFromPriorSerializer( priorTtlAwareMapValueSerializer, + priorMapValueSerializerSnapshot, () -> priorTtlAwareMapValueSerializer.deserialize(serializedOldValueInput), serializedMigratedValueOutput, ttlTimeProvider);