weiqingy commented on code in PR #28881:
URL: https://github.com/apache/flink/pull/28881#discussion_r3890477490
##########
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:
https://github.com/apache/flink/pull/29046 has been merged. The rebase is
done.
--
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]
