From 9a3e578ff0908c1741ec30e2d179b1defe1f9921 Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 2 Aug 2026 17:23:24 -0700 Subject: [PATCH 1/2] [FLINK-40296][core] Add object-level migrate hook to TypeSerializerSnapshot Add a default method that lets a serializer snapshot transform an already deserialized state value from the schema it was written with into the schema the current serializer expects: default T migrate(TypeSerializerSnapshot oldSerializerSnapshot, T value) Like resolveSchemaCompatibility, it is invoked on the new snapshot and receives the old snapshot as its argument. The default returns the value unchanged, so behavior is unaffected for every existing serializer: a value deserialized with the prior serializer is structurally compatible with the current one and can be re-serialized as is. The javadoc states that migration is not applied recursively to nested serializers. Unlike resolveSchemaCompatibility, which CompositeTypeSerializer- Snapshot delegates to the nested snapshots, migrate has no delegating override, so a composite returns its value unmigrated unless it decomposes the value itself. That asymmetry is invisible at the call site and would otherwise fail silently. Generated-by: Claude Code (Opus 5) --- .../typeutils/TypeSerializerSnapshot.java | 26 +++++++++++++++++++ .../typeutils/TypeSerializerSnapshotTest.java | 9 +++++++ 2 files changed, 35 insertions(+) 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..e4788736a68a1f 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,32 @@ 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. + * + * @param oldSerializerSnapshot snapshot of the serializer that wrote the value. + * @param value the value, already deserialized with the prior serializer. + * @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 From 9103aec00961cb235d947104f887899f68c9997b Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Sun, 2 Aug 2026 19:57:36 -0700 Subject: [PATCH 2/2] [FLINK-40297][runtime] Route TTL-aware value migration through the migrate hook TtlAwareSerializer.migrateValueFromPriorSerializer is the single entry point through which the RocksDB state backend migrates state values on restore: AbstractRocksDBState, RocksDBListState and RocksDBMapState all call it after unwrapping the state shape they own. It deserialized with the prior serializer and re-serialized with the new one, leaving a serializer no opportunity to adapt the value in between. Route it through TypeSerializerSnapshot.migrate: unwrap the prior value to its bare, non-TTL form, migrate it, then re-wrap when this serializer is TTL-enabled, preserving the prior timestamp when the prior value carried one. Behavior is unchanged, because no serializer overrides the hook yet and its default returns the value unchanged. The hook receives the persisted prior snapshot rather than one re-derived by calling snapshotConfiguration() on the restored prior serializer. That round trip is lossy: PojoSerializerSnapshot substitutes a synthetic name for a field that no longer exists on the class, so a migrate override reconciling fields by name would see a fabricated schema. The backend already holds the persisted snapshot above the migration loop, and each caller now descends it alongside its serializer. The descent unwraps the TtlAware decorator first. Registering a new serializer mutates the previous snapshot's nested snapshots in place, so a list or map state's persisted element or value snapshot is a TtlAwareSerializerSnapshot rather than the snapshot the checkpoint wrote. A snapshot of an unexpected type now fails rather than falling back to a re-derived one; only an absent snapshot falls back. Generated-by: Claude Code (Opus 5) --- .../typeutils/TypeSerializerSnapshot.java | 15 +- .../runtime/state/ttl/TtlAwareSerializer.java | 133 +++++- .../state/StateSerializerProviderTest.java | 41 ++ .../state/ttl/TtlAwareSerializerTest.java | 398 ++++++++++++++++++ .../ttl/TtlAwareSerializerUpgradeTest.java | 1 + .../state/rocksdb/AbstractRocksDBState.java | 5 + .../rocksdb/RocksDBKeyedStateBackend.java | 8 +- .../flink/state/rocksdb/RocksDBListState.java | 21 + .../flink/state/rocksdb/RocksDBMapState.java | 21 + 9 files changed, 625 insertions(+), 18 deletions(-) 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 e4788736a68a1f..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 @@ -150,10 +150,19 @@ TypeSerializerSchemaCompatibility resolveSchemaCompatibility( *

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. + * 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. - * @param value the value, already deserialized with the prior serializer. + * @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) { 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);