Github user StefanRRichter commented on a diff in the pull request: https://github.com/apache/flink/pull/6196#discussion_r199783587 --- Diff: flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeSerializer.java --- @@ -0,0 +1,275 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.common.typeutils; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.util.Preconditions; + +import javax.annotation.Nonnull; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.IntStream; + +/** + * Base class for composite serializers. + * + * <p>This class serializes a composite type using array of its field serializers. + * Fields are indexed the same way as their serializers. + * + * @param <T> type of custom serialized value + */ +public abstract class CompositeSerializer<T> extends TypeSerializer<T> { + private static final long serialVersionUID = 1L; + + /** Serializers for fields which constitute T. */ + protected final TypeSerializer<Object>[] fieldSerializers; + + /** Whether T is an immutable type. */ + final boolean immutableTargetType; + + /** Byte length of target object in serialized form. */ + private final int length; + + /** Whether any field serializer is stateful. */ + private final boolean stateful; + + private final int hashCode; + + @SuppressWarnings("unchecked") + protected CompositeSerializer(boolean immutableTargetType, TypeSerializer<?> ... fieldSerializers) { + Preconditions.checkNotNull(fieldSerializers); + Preconditions.checkArgument(Arrays.stream(fieldSerializers).allMatch(Objects::nonNull)); + this.immutableTargetType = immutableTargetType && + Arrays.stream(fieldSerializers).allMatch(TypeSerializer::isImmutableType); + this.fieldSerializers = (TypeSerializer<Object>[]) fieldSerializers; + this.length = calcLength(); + this.stateful = isStateful(); + this.hashCode = Arrays.hashCode(fieldSerializers); + } + + private boolean isStateful() { + TypeSerializer[] duplicatedSerializers = duplicateFieldSerializers(); + return IntStream.range(0, fieldSerializers.length) + .anyMatch(i -> fieldSerializers[i] != duplicatedSerializers[i]); + } + + /** Create new instance from its fields. */ + public abstract T createInstance(@Nonnull Object ... values); + + /** Modify field of existing instance. Supported only by mutable types. */ + protected abstract void setField(@Nonnull T value, int index, Object fieldValue); + + /** Get field of existing instance. */ + protected abstract Object getField(@Nonnull T value, int index); + + /** Factory for concrete serializer. */ + protected abstract CompositeSerializer<T> createSerializerInstance(TypeSerializer<?> ... originalSerializers); + + @Override + public CompositeSerializer<T> duplicate() { + return stateful ? createSerializerInstance(duplicateFieldSerializers()) : this; --- End diff -- Another small point here for `createSerializerInstance(...)`: we have no (non-public) constructor that can also take all boolean flags, length, and (maybe) hash directly. So if we copy the serializer, I guess it always goes through the whole process again to figure this out, but we could just copy it from the previous instance.
---