weiqingy commented on code in PR #28881:
URL: https://github.com/apache/flink/pull/28881#discussion_r3888432394
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java:
##########
@@ -214,11 +217,29 @@ public void migrateSerializedValue(
TtlAwareSerializer<V, ?> newTtlAwareElementSerializer =
((TtlAwareSerializer.TtlAwareListSerializer<V>) 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<V> 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()
+ + ".");
+ }
Review Comment:
This guard is deliberately the same shape as the two
`Preconditions.checkArgument` calls immediately above it (lines 209-212), which
also throw `IllegalArgumentException` outside the try. All three assert the
same kind of thing: that the caller passed the expected serializer or snapshot
type.
That is an argument-contract violation rather than a restore failure.
Wrapping it in `StateMigrationException` would report a state migration problem
to the user when the fault is actually an internal miswiring, and it would
leave this method inconsistent with the guards next to it and with the
equivalent guards in `AbstractRocksDBState` and `RocksDBMapState`.
Keeping it as is.
##########
flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java:
##########
@@ -240,6 +243,23 @@ public void migrateSerializedValue(
TtlAwareSerializer<UV, ?> newTtlAwareMapValueSerializer =
((TtlAwareSerializer.TtlAwareMapSerializer<UK, UV>)
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<UV> 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()
+ + ".");
+ }
Review Comment:
This guard is deliberately the same shape as the two
`Preconditions.checkArgument` calls immediately above it (lines 237-238), which
also throw `IllegalArgumentException` outside the try. All three assert the
same kind of thing: that the caller passed the expected serializer or snapshot
type.
That is an argument-contract violation rather than a restore failure.
Wrapping it in `StateMigrationException` would report a state migration problem
to the user when the fault is actually an internal miswiring, and it would
leave this method inconsistent with the guards next to it and with the
equivalent guards in `AbstractRocksDBState` and `RocksDBListState`.
Keeping it as is.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java:
##########
@@ -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}.
+ *
+ * <p>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<T, ?> priorTtlAwareSerializer,
+ @Nullable TypeSerializerSnapshot<T> priorSerializerSnapshot,
SupplierWithException<T, IOException> 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<T>)
inputSupplier.get()).getUserValue()
- : inputSupplier.get();
+ outputRecord = (T) migratedValue;
}
this.serialize(outputRecord, target);
}
+ /**
+ * The snapshot describing the schema the prior bare value was written
with.
+ *
+ * <p>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;
Review Comment:
Good catch, although the typo is pre-existing:
`getOrinalTypeSerializerSnapshot` came in with FLINK-36521 and this PR only
calls it.
Rather than add an alias, I have opened #29046 to rename it to
`getOriginalTypeSerializerSnapshot`. That also restores symmetry with the
correctly spelled `TtlAwareSerializer#getOriginalTypeSerializer` that it
mirrors. `TtlAwareSerializerSnapshot` carries no API annotation and all eight
call sites are inside `flink-runtime`, so a straight rename is cleaner than
keeping a deprecated alias around permanently.
I will rebase this PR onto it once that merges.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
