dawidwys commented on a change in pull request #6711: [FLINK-9377] [core, state backends] Remove serializers from checkpoints URL: https://github.com/apache/flink/pull/6711#discussion_r220909700
########## File path: flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java ########## @@ -21,35 +21,131 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.core.io.VersionedIOReadableWritable; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.util.Preconditions; +import java.io.IOException; + /** * A {@code TypeSerializerConfigSnapshot} is a point-in-time view of a {@link TypeSerializer's} configuration. - * The configuration snapshot of a serializer is persisted along with checkpoints of the managed state that the - * serializer is registered to. + * The configuration snapshot of a serializer is persisted within checkpoints + * as a single source of meta information about the schema of serialized data in the checkpoint. + * This serves three purposes: + * + * <ul> + * <li><strong>Capturing serializer parameters and schema:</strong> a serializer's configuration snapshot + * represents information about the parameters, state, and schema of a serializer. + * This is explained in more detail below.</li> * - * <p>The persisted configuration may later on be used by new serializers to ensure serialization compatibility - * for the same managed state. In order for new serializers to be able to ensure this, the configuration snapshot - * should encode sufficient information about: + * <li><strong>Compatibility checks for new serializers:</strong> when new serializers are available, + * they need to be checked whether or not they are compatible to read the data written by the previous serializer. + * This is performed by providing the new serializer to the correspondibng serializer configuration + * snapshots in checkpoints.</li> + * + * <li><strong>Factory for a read serializer when schema conversion is required:<strong> in the case that new + * serializers are not compatible to read previous data, a schema conversion process executed across all data + * is required before the new serializer can be continued to be used. This conversion process requires a compatible + * read serializer to restore serialized bytes as objects, and then written back again using the new serializer. + * In this scenario, the serializer configuration snapshots in checkpoints doubles as a factory for the read + * serializer of the conversion process.</li> + * </ul> + * + * <h2>Serializer Configuration and Schema</h2> + * + * <p>Since serializer configuration snapshots needs to be used to ensure serialization compatibility + * for the same managed state as well as serving as a factory for compatible read serializers, the configuration + * snapshot should encode sufficient information about: * * <ul> * <li><strong>Parameter settings of the serializer:</strong> parameters of the serializer include settings * required to setup the serializer, or the state of the serializer if it is stateful. If the serializer * has nested serializers, then the configuration snapshot should also contain the parameters of the nested * serializers.</li> * - * <li><strong>Serialization schema of the serializer:</strong> the data format used by the serializer.</li> + * <li><strong>Serialization schema of the serializer:</strong> the binary format used by the serializer, or + * in other words, the schema of data written by the serializer.</li> * </ul> * * <p>NOTE: Implementations must contain the default empty nullary constructor. This is required to be able to * deserialize the configuration snapshot from its binary form. + * + * @param <T> The data type that the originating serializer of this configuration serializes. */ @PublicEvolving -public abstract class TypeSerializerConfigSnapshot extends VersionedIOReadableWritable { +public abstract class TypeSerializerConfigSnapshot<T> extends VersionedIOReadableWritable { /** The user code class loader; only relevant if this configuration instance was deserialized from binary form. */ private ClassLoader userCodeClassLoader; + /** + * The originating serializer of this configuration snapshot. + * + * TODO to allow for incrementally adapting the implementation of serializer config snapshot subclasses, + * TODO we currently have a base implementation for the {@link #restoreSerializer()} + * TODO method which simply returns this serializer instance. The serializer is written + * TODO and read using Java serialization as part of reading / writing the config snapshot + */ + private TypeSerializer<T> serializer; + + /** + * Creates a serializer using this configuration, that is capable of reading data + * written by the serializer described by this configuration. + * + * @return the restored serializer. + */ + public TypeSerializer<T> restoreSerializer() { + // TODO this implementation is only a placeholder; the intention is to have no default implementation + return serializer; + } + + /** + * Set the originating serializer of this configuration snapshot. + * + * TODO this method is a temporary workaround to inject the serializer instance to + * TODO be returned by the restoreSerializer() method. + */ + @Internal + public final void setSerializer(TypeSerializer<T> serializer) { + this.serializer = Preconditions.checkNotNull(serializer); + } + + /** + * Checks whether a new serializer is compatible to read data written be the originating serializer of this + * config snapshot; i.e. whether or not a new serializer is compatible with the previous serializer. + * + * @param newSerializer the new serializer to check against for schema compatibility. + * + * @return the resolve schema compatibility result. + */ + public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(TypeSerializer<?> newSerializer) { + @SuppressWarnings("unchecked") + TypeSerializer<T> castedSerializer = ((TypeSerializer<T>) newSerializer); + + if (castedSerializer.ensureCompatibility(this).isRequiresMigration()) { + return TypeSerializerSchemaCompatibility.compatibleAfterMigration(); + } else { + return TypeSerializerSchemaCompatibility.compatibleAfterReconfiguration(castedSerializer); + } + } + + @Override + public void write(DataOutputView out) throws IOException { + // bump the version; we use this to know that there is a serializer to read as part of the config + out.writeInt(getVersion() + 1); Review comment: Ok, so the path will change to `1(pre-1.6)` -> `2 (1+1) (this PR)` -> `3(3 + 0) (final version)`, right? I am just worried that we will have to change the version very carefully, as we would have to increase version of each serializer by 2 for the final version. Also if the final version will be released separately to version in this PR, this should be very carefully explained to the end users. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services