image.png

    1. /*
    2.  * Licensed to the Apache Software Foundation (ASF) under one
    3.  * or more contributor license agreements.  See the NOTICE file
    4.  * distributed with this work for additional information
    5.  * regarding copyright ownership.  The ASF licenses this file
    6.  * to you under the Apache License, Version 2.0 (the
    7.  * "License"); you may not use this file except in compliance
    8.  * with the License.  You may obtain a copy of the License at
    9.  *
    10.  *     http://www.apache.org/licenses/LICENSE-2.0
    11.  *
    12.  * Unless required by applicable law or agreed to in writing, software
    13.  * distributed under the License is distributed on an "AS IS" BASIS,
    14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    15.  * See the License for the specific language governing permissions and
    16.  * limitations under the License.
    17.  */
    19. package org.apache.flink.api.common.typeutils;
    21. import org.apache.flink.annotation.PublicEvolving;
    22. import org.apache.flink.core.memory.DataInputView;
    23. import org.apache.flink.core.memory.DataOutputView;
    25. import java.io.IOException;
    26. import java.io.Serializable;
    28. /**
    29.  * This interface describes the methods that are required for a data type to be handled by the Flink
    30.  * runtime. Specifically, this interface contains the serialization and copying methods.
    31.  *
    32.  * <p>The methods in this class are not necessarily thread safe. To avoid unpredictable side effects,
    33.  * it is recommended to call {@code duplicate()} method and use one serializer instance per thread.
    34.  *
    35.  * <p><b>Upgrading TypeSerializers to the new TypeSerializerSnapshot model</b>
    36.  *
    37.  * <p>This section is relevant if you implemented a TypeSerializer in Flink versions up to 1.6 and want
    38.  * to adapt that implementation to the new interfaces that support proper state schema evolution, while maintaining
    39.  * backwards compatibility. Please follow these steps:
    40.  *
    41.  * <ul>
    42.  *     <li>Change the type serializer's config snapshot to implement {@link TypeSerializerSnapshot}, rather
    43.  *     than extending {@code TypeSerializerConfigSnapshot} (as previously).
    44.  *     <li>If the above step was completed, then the upgrade is done. Otherwise, if changing to implement
    45.  *     {@link TypeSerializerSnapshot} directly in-place as the same class isn't possible (perhaps because the new snapshot
    46.  *     is intended to have completely different written contents or intended to have a different class name),
    47.  *     retain the old serializer snapshot class (extending {@code TypeSerializerConfigSnapshot}) under
    48.  *     the same name and give the updated serializer snapshot class (the one extending {@code TypeSerializerSnapshot})
    49.  *     a new name.
    50.  *     <li>Override the {@code TypeSerializerConfigSnapshot#resolveSchemaCompatibility(TypeSerializer)}
    51.  *     method to perform the compatibility check based on configuration written by the old serializer snapshot class.
    52.  * </ul>
    53.  *
    54.  * @param <T> The data type that the serializer serializes.
    55.  */
    56. @PublicEvolving
    57. public abstract class TypeSerializer<T> implements Serializable {
    59.     private static final long serialVersionUID = 1L;
    61.     // --------------------------------------------------------------------------------------------
    62.     // General information about the type and the serializer
    63.     // --------------------------------------------------------------------------------------------
    65.     /**
    66.      * Gets whether the type is an immutable type.
    67.      * 
    68.      * @return True, if the type is immutable.
    69.      */
    70.     public abstract boolean isImmutableType();
    72.     /**
    73.      * Creates a deep copy of this serializer if it is necessary, i.e. if it is stateful. This
    74.      * can return itself if the serializer is not stateful.
    75.      *
    76.      * We need this because Serializers might be used in several threads. Stateless serializers
    77.      * are inherently thread-safe while stateful serializers might not be thread-safe.
    78.      */
    79.     public abstract TypeSerializer<T> duplicate();
    81.     // --------------------------------------------------------------------------------------------
    82.     // Instantiation & Cloning
    83.     // --------------------------------------------------------------------------------------------
    85.     /**
    86.      * Creates a new instance of the data type.
    87.      * 
    88.      * @return A new instance of the data type.
    89.      */
    90.     public abstract T createInstance();
    92.     /**
    93.      * Creates a deep copy of the given element in a new element.
    94.      * 
    95.      * @param from The element reuse be copied.
    96.      * @return A deep copy of the element.
    97.      */
    98.     public abstract T copy(T from);
    100.     /**
    101.      * Creates a copy from the given element.
    102.      * The method makes an attempt to store the copy in the given reuse element, if the type is mutable.
    103.      * This is, however, not guaranteed.
    104.      * 
    105.      * @param from The element to be copied.
    106.      * @param reuse The element to be reused. May or may not be used.
    107.      * @return A deep copy of the element.
    108.      */
    109.     public abstract T copy(T from, T reuse);
    111.     // --------------------------------------------------------------------------------------------
    113.     /**
    114.      * Gets the length of the data type, if it is a fix length data type.
    115.      * 
    116.      * @return The length of the data type, or <code>-1</code> for variable length data types.
    117.      */
    118.     public abstract int getLength();
    120.     // --------------------------------------------------------------------------------------------
    122.     /**
    123.      * Serializes the given record to the given target output view.
    124.      * 
    125.      * @param record The record to serialize.
    126.      * @param target The output view to write the serialized data to.
    127.      * 
    128.      * @throws IOException Thrown, if the serialization encountered an I/O related error. Typically raised by the
    129.      *                     output view, which may have an underlying I/O channel to which it delegates.
    130.      */
    131.     public abstract void serialize(T record, DataOutputView target) throws IOException;
    133.     /**
    134.      * De-serializes a record from the given source input view.
    135.      * 
    136.      * @param source The input view from which to read the data.
    137.      * @return The deserialized element.
    138.      * 
    139.      * @throws IOException Thrown, if the de-serialization encountered an I/O related error. Typically raised by the
    140.      *                     input view, which may have an underlying I/O channel from which it reads.
    141.      */
    142.     public abstract T deserialize(DataInputView source) throws IOException;
    144.     /**
    145.      * De-serializes a record from the given source input view into the given reuse record instance if mutable.
    146.      * 
    147.      * @param reuse The record instance into which to de-serialize the data.
    148.      * @param source The input view from which to read the data.
    149.      * @return The deserialized element.
    150.      * 
    151.      * @throws IOException Thrown, if the de-serialization encountered an I/O related error. Typically raised by the
    152.      *                     input view, which may have an underlying I/O channel from which it reads.
    153.      */
    154.     public abstract T deserialize(T reuse, DataInputView source) throws IOException;
    156.     /**
    157.      * Copies exactly one record from the source input view to the target output view. Whether this operation
    158.      * works on binary data or partially de-serializes the record to determine its length (such as for records
    159.      * of variable length) is up to the implementer. Binary copies are typically faster. A copy of a record containing
    160.      * two integer numbers (8 bytes total) is most efficiently implemented as
    161.      * {@code target.write(source, 8);}.
    162.      *  
    163.      * @param source The input view from which to read the record.
    164.      * @param target The target output view to which to write the record.
    165.      * 
    166.      * @throws IOException Thrown if any of the two views raises an exception.
    167.      */
    168.     public abstract void copy(DataInputView source, DataOutputView target) throws IOException;
    170.     public abstract boolean equals(Object obj);
    172.     public abstract int hashCode();
    174.     // --------------------------------------------------------------------------------------------
    175.     // Serializer configuration snapshot for checkpoints/savepoints
    176.     // --------------------------------------------------------------------------------------------
    178.     /**
    179.      * Snapshots the configuration of this TypeSerializer. This method is only relevant if the serializer is
    180.      * used to state stored in checkpoints/savepoints.
    181.      *
    182.      * <p>The snapshot of the TypeSerializer is supposed to contain all information that affects the serialization
    183.      * format of the serializer. The snapshot serves two purposes: First, to reproduce the serializer when the
    184.      * checkpoint/savepoint is restored, and second, to check whether the serialization format is compatible
    185.      * with the serializer used in the restored program.
    186.      *
    187.      * <p><b>IMPORTANT:</b> TypeSerializerSnapshots changed after Flink 1.6. Serializers implemented against
    188.      * Flink versions up to 1.6 should still work, but adjust to new model to enable state evolution and be
    189.      * future-proof.
    190.      * See the class-level comments, section "Upgrading TypeSerializers to the new TypeSerializerSnapshot model"
    191.      * for details.
    192.      *
    193.      * @see TypeSerializerSnapshot#resolveSchemaCompatibility(TypeSerializer)
    194.      *
    195.      * @return snapshot of the serializer's current configuration (cannot be {@code null}).
    196.      */
    197.     public abstract TypeSerializerSnapshot<T> snapshotConfiguration();
    198. }