flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/NullableSerializer.java - flink - Git at Google

 /*
  * 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.java.typeutils.runtime;

 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.typeutils.CompatibilityResult;
 import org.apache.flink.api.common.typeutils.CompatibilityUtil;
 import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
 import org.apache.flink.api.common.typeutils.TypeDeserializerAdapter;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
 import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.core.memory.DataInputDeserializer;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputSerializer;
 import org.apache.flink.core.memory.DataOutputView;
 import org.apache.flink.util.Preconditions;

 import javax.annotation.Nonnull;

 import java.io.IOException;
 import java.util.List;

 /**
  * Serializer wrapper to add support of {@code null} value serialization.
  *
  * <p>If the target serializer does not support {@code null} values of its type,
  * you can use this class to wrap this serializer.
  * This is a generic treatment of {@code null} value serialization
  * which comes with the cost of additional byte in the final serialized value.
  * The {@code NullableSerializer} will intercept {@code null} value serialization case
  * and prepend the target serialized value with a boolean flag marking whether it is {@code null} or not.
  * <pre> {@code
  * TypeSerializer<T> originalSerializer = ...;
  * TypeSerializer<T> serializerWithNullValueSupport = NullableSerializer.wrap(originalSerializer);
  * // or
  * TypeSerializer<T> serializerWithNullValueSupport = NullableSerializer.wrapIfNullIsNotSupported(originalSerializer);
  * }}</pre>
  *
  * @param <T> type to serialize
  */
 public class NullableSerializer<T> extends TypeSerializer<T> {
 	private static final long serialVersionUID = 3335569358214720033L;
 	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

 	@Nonnull
 	private final TypeSerializer<T> originalSerializer;
 	private final byte[] padding;

 	private NullableSerializer(@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
 		this.originalSerializer = originalSerializer;
 		this.padding = createPadding(originalSerializer.getLength(), padNullValueIfFixedLen);

 	}

 	private static <T> byte[] createPadding(int originalSerializerLength, boolean padNullValueIfFixedLen) {
 		boolean padNullValue = originalSerializerLength > 0 && padNullValueIfFixedLen;
 		return padNullValue ? new byte[originalSerializerLength] : EMPTY_BYTE_ARRAY;
 	}

 	/**
 	 * This method tries to serialize {@code null} value with the {@code originalSerializer}
 	 * and wraps it in case of {@link NullPointerException}, otherwise it returns the {@code originalSerializer}.
 	 *
 	 * @param originalSerializer serializer to wrap and add {@code null} support
 	 * @param padNullValueIfFixedLen pad null value to preserve the fixed length of original serializer
 	 * @return serializer which supports {@code null} values
 	 */
 	public static <T> TypeSerializer<T> wrapIfNullIsNotSupported(
 		@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
 		return checkIfNullSupported(originalSerializer) ?
 			originalSerializer : wrap(originalSerializer, padNullValueIfFixedLen);
 	}

 	/**
 	 * This method checks if {@code serializer} supports {@code null} value.
 	 *
 	 * @param serializer serializer to check
 	 */
 	public static <T> boolean checkIfNullSupported(@Nonnull TypeSerializer<T> serializer) {
 		int length = serializer.getLength() > 0 ? serializer.getLength() : 1;
 		DataOutputSerializer dos = new DataOutputSerializer(length);
 		try {
 			serializer.serialize(null, dos);
 		} catch (IOException | RuntimeException e) {
 			return false;
 		}
 		Preconditions.checkArgument(
 			serializer.getLength() < 0 || serializer.getLength() == dos.getCopyOfBuffer().length,
 			"The serialized form of the null value should have the same length " +
 				"as any other if the length is fixed in the serializer");
 		DataInputDeserializer dis = new DataInputDeserializer(dos.getSharedBuffer());
 		try {
 			Preconditions.checkArgument(serializer.deserialize(dis) == null);
 		} catch (IOException e) {
 			throw new RuntimeException(
 				String.format("Unexpected failure to deserialize just serialized null value with %s",
 					serializer.getClass().getName()), e);
 		}
 		Preconditions.checkArgument(
 			serializer.copy(null) == null,
 			"Serializer %s has to be able properly copy null value if it can serialize it",
 			serializer.getClass().getName());
 		return true;
 	}

 	private boolean padNullValue() {
 		return padding.length > 0;
 	}

 	/**
 	 * This method wraps the {@code originalSerializer} with the {@code NullableSerializer} if not already wrapped.
 	 *
 	 * @param originalSerializer serializer to wrap and add {@code null} support
 	 * @param padNullValueIfFixedLen pad null value to preserve the fixed length of original serializer
 	 * @return wrapped serializer which supports {@code null} values
 	 */
 	public static <T> TypeSerializer<T> wrap(
 		@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
 		return originalSerializer instanceof NullableSerializer ?
 			originalSerializer : new NullableSerializer<>(originalSerializer, padNullValueIfFixedLen);
 	}

 	@Override
 	public boolean isImmutableType() {
 		return originalSerializer.isImmutableType();
 	}

 	@Override
 	public TypeSerializer<T> duplicate() {
 		TypeSerializer<T> duplicateOriginalSerializer = originalSerializer.duplicate();
 		return duplicateOriginalSerializer == originalSerializer ?
 			this : new NullableSerializer<>(originalSerializer.duplicate(), padNullValue());
 	}

 	@Override
 	public T createInstance() {
 		return originalSerializer.createInstance();
 	}

 	@Override
 	public T copy(T from) {
 		return from == null ? null : originalSerializer.copy(from);
 	}

 	@Override
 	public T copy(T from, T reuse) {
 		return from == null ? null :
 			(reuse == null ? originalSerializer.copy(from) : originalSerializer.copy(from, reuse));
 	}

 	@Override
 	public int getLength() {
 		return padNullValue() ? 1 + padding.length : -1;
 	}

 	@Override
 	public void serialize(T record, DataOutputView target) throws IOException {
 		if (record == null) {
 			target.writeBoolean(true);
 			target.write(padding);
 		} else {
 			target.writeBoolean(false);
 			originalSerializer.serialize(record, target);
 		}
 	}

 	@Override
 	public T deserialize(DataInputView source) throws IOException {
 		boolean isNull = deserializeNull(source);
 		return isNull ? null : originalSerializer.deserialize(source);
 	}

 	@Override
 	public T deserialize(T reuse, DataInputView source) throws IOException {
 		boolean isNull = deserializeNull(source);
 		return isNull ? null : (reuse == null ?
 			originalSerializer.deserialize(source) : originalSerializer.deserialize(reuse, source));
 	}

 	private boolean deserializeNull(DataInputView source) throws IOException {
 		boolean isNull = source.readBoolean();
 		if (isNull) {
 			source.skipBytesToRead(padding.length);
 		}
 		return isNull;
 	}

 	@Override
 	public void copy(DataInputView source, DataOutputView target) throws IOException {
 		boolean isNull = source.readBoolean();
 		target.writeBoolean(isNull);
 		if (isNull) {
 			target.write(padding);
 		} else {
 			originalSerializer.copy(source, target);
 		}
 	}

 	@Override
 	public boolean equals(Object obj) {
 		return obj == this ||
 			(obj != null && obj.getClass() == getClass() &&
 				originalSerializer.equals(((NullableSerializer) obj).originalSerializer));
 	}

 	@Override
 	public boolean canEqual(Object obj) {
 		return (obj != null && obj.getClass() == getClass() &&
 			originalSerializer.canEqual(((NullableSerializer) obj).originalSerializer));
 	}

 	@Override
 	public int hashCode() {
 		return originalSerializer.hashCode();
 	}

 	@Override
 	public NullableSerializerConfigSnapshot<T> snapshotConfiguration() {
 		return new NullableSerializerConfigSnapshot<>(originalSerializer);
 	}

 	@Override
 	public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
 		if (configSnapshot instanceof NullableSerializerConfigSnapshot) {
 			List<Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>> previousKvSerializersAndConfigs =
 				((NullableSerializerConfigSnapshot) configSnapshot).getNestedSerializersAndConfigs();

 			CompatibilityResult<T> compatResult = CompatibilityUtil.resolveCompatibilityResult(
 				previousKvSerializersAndConfigs.get(0).f0,
 				UnloadableDummyTypeSerializer.class,
 				previousKvSerializersAndConfigs.get(0).f1,
 				originalSerializer);

 			if (!compatResult.isRequiresMigration()) {
 				return CompatibilityResult.compatible();
 			} else if (compatResult.getConvertDeserializer() != null) {
 				return CompatibilityResult.requiresMigration(
 					new NullableSerializer<>(
 						new TypeDeserializerAdapter<>(compatResult.getConvertDeserializer()), padNullValue()));
 			}
 		}

 		return CompatibilityResult.requiresMigration();
 	}

 	/**
 	 * Configuration snapshot for serializers of nullable types, containing the
 	 * configuration snapshot of its original serializer.
 	 */
 	@Internal
 	public static class NullableSerializerConfigSnapshot<T> extends CompositeTypeSerializerConfigSnapshot {
 		private static final int VERSION = 1;

 		/** This empty nullary constructor is required for deserializing the configuration. */
 		@SuppressWarnings("unused")
 		public NullableSerializerConfigSnapshot() {}

 		NullableSerializerConfigSnapshot(TypeSerializer<T> originalSerializer) {
 			super(originalSerializer);
 		}

 		@Override
 		public int getVersion() {
 			return VERSION;
 		}
 	}
 }
	/*
	* 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.java.typeutils.runtime;

	import org.apache.flink.annotation.Internal;
	import org.apache.flink.api.common.typeutils.CompatibilityResult;
	import org.apache.flink.api.common.typeutils.CompatibilityUtil;
	import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
	import org.apache.flink.api.common.typeutils.TypeDeserializerAdapter;
	import org.apache.flink.api.common.typeutils.TypeSerializer;
	import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
	import org.apache.flink.api.common.typeutils.UnloadableDummyTypeSerializer;
	import org.apache.flink.api.java.tuple.Tuple2;
	import org.apache.flink.core.memory.DataInputDeserializer;
	import org.apache.flink.core.memory.DataInputView;
	import org.apache.flink.core.memory.DataOutputSerializer;
	import org.apache.flink.core.memory.DataOutputView;
	import org.apache.flink.util.Preconditions;

	import javax.annotation.Nonnull;

	import java.io.IOException;
	import java.util.List;

	/**
	* Serializer wrapper to add support of {@code null} value serialization.
	*
	* <p>If the target serializer does not support {@code null} values of its type,
	* you can use this class to wrap this serializer.
	* This is a generic treatment of {@code null} value serialization
	* which comes with the cost of additional byte in the final serialized value.
	* The {@code NullableSerializer} will intercept {@code null} value serialization case
	* and prepend the target serialized value with a boolean flag marking whether it is {@code null} or not.
	* <pre> {@code
	* TypeSerializer<T> originalSerializer = ...;
	* TypeSerializer<T> serializerWithNullValueSupport = NullableSerializer.wrap(originalSerializer);
	* // or
	* TypeSerializer<T> serializerWithNullValueSupport = NullableSerializer.wrapIfNullIsNotSupported(originalSerializer);
	* }}</pre>
	*
	* @param <T> type to serialize
	*/
	public class NullableSerializer<T> extends TypeSerializer<T> {
	private static final long serialVersionUID = 3335569358214720033L;
	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

	@Nonnull
	private final TypeSerializer<T> originalSerializer;
	private final byte[] padding;

	private NullableSerializer(@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
	this.originalSerializer = originalSerializer;
	this.padding = createPadding(originalSerializer.getLength(), padNullValueIfFixedLen);

	}

	private static <T> byte[] createPadding(int originalSerializerLength, boolean padNullValueIfFixedLen) {
	boolean padNullValue = originalSerializerLength > 0 && padNullValueIfFixedLen;
	return padNullValue ? new byte[originalSerializerLength] : EMPTY_BYTE_ARRAY;
	}

	/**
	* This method tries to serialize {@code null} value with the {@code originalSerializer}
	* and wraps it in case of {@link NullPointerException}, otherwise it returns the {@code originalSerializer}.
	*
	* @param originalSerializer serializer to wrap and add {@code null} support
	* @param padNullValueIfFixedLen pad null value to preserve the fixed length of original serializer
	* @return serializer which supports {@code null} values
	*/
	public static <T> TypeSerializer<T> wrapIfNullIsNotSupported(
	@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
	return checkIfNullSupported(originalSerializer) ?
	originalSerializer : wrap(originalSerializer, padNullValueIfFixedLen);
	}

	/**
	* This method checks if {@code serializer} supports {@code null} value.
	*
	* @param serializer serializer to check
	*/
	public static <T> boolean checkIfNullSupported(@Nonnull TypeSerializer<T> serializer) {
	int length = serializer.getLength() > 0 ? serializer.getLength() : 1;
	DataOutputSerializer dos = new DataOutputSerializer(length);
	try {
	serializer.serialize(null, dos);
	} catch (IOException \| RuntimeException e) {
	return false;
	}
	Preconditions.checkArgument(
	serializer.getLength() < 0 \|\| serializer.getLength() == dos.getCopyOfBuffer().length,
	"The serialized form of the null value should have the same length " +
	"as any other if the length is fixed in the serializer");
	DataInputDeserializer dis = new DataInputDeserializer(dos.getSharedBuffer());
	try {
	Preconditions.checkArgument(serializer.deserialize(dis) == null);
	} catch (IOException e) {
	throw new RuntimeException(
	String.format("Unexpected failure to deserialize just serialized null value with %s",
	serializer.getClass().getName()), e);
	}
	Preconditions.checkArgument(
	serializer.copy(null) == null,
	"Serializer %s has to be able properly copy null value if it can serialize it",
	serializer.getClass().getName());
	return true;
	}

	private boolean padNullValue() {
	return padding.length > 0;
	}

	/**
	* This method wraps the {@code originalSerializer} with the {@code NullableSerializer} if not already wrapped.
	*
	* @param originalSerializer serializer to wrap and add {@code null} support
	* @param padNullValueIfFixedLen pad null value to preserve the fixed length of original serializer
	* @return wrapped serializer which supports {@code null} values
	*/
	public static <T> TypeSerializer<T> wrap(
	@Nonnull TypeSerializer<T> originalSerializer, boolean padNullValueIfFixedLen) {
	return originalSerializer instanceof NullableSerializer ?
	originalSerializer : new NullableSerializer<>(originalSerializer, padNullValueIfFixedLen);
	}

	@Override
	public boolean isImmutableType() {
	return originalSerializer.isImmutableType();
	}

	@Override
	public TypeSerializer<T> duplicate() {
	TypeSerializer<T> duplicateOriginalSerializer = originalSerializer.duplicate();
	return duplicateOriginalSerializer == originalSerializer ?
	this : new NullableSerializer<>(originalSerializer.duplicate(), padNullValue());
	}

	@Override
	public T createInstance() {
	return originalSerializer.createInstance();
	}

	@Override
	public T copy(T from) {
	return from == null ? null : originalSerializer.copy(from);
	}

	@Override
	public T copy(T from, T reuse) {
	return from == null ? null :
	(reuse == null ? originalSerializer.copy(from) : originalSerializer.copy(from, reuse));
	}

	@Override
	public int getLength() {
	return padNullValue() ? 1 + padding.length : -1;
	}

	@Override
	public void serialize(T record, DataOutputView target) throws IOException {
	if (record == null) {
	target.writeBoolean(true);
	target.write(padding);
	} else {
	target.writeBoolean(false);
	originalSerializer.serialize(record, target);
	}
	}

	@Override
	public T deserialize(DataInputView source) throws IOException {
	boolean isNull = deserializeNull(source);
	return isNull ? null : originalSerializer.deserialize(source);
	}

	@Override
	public T deserialize(T reuse, DataInputView source) throws IOException {
	boolean isNull = deserializeNull(source);
	return isNull ? null : (reuse == null ?
	originalSerializer.deserialize(source) : originalSerializer.deserialize(reuse, source));
	}

	private boolean deserializeNull(DataInputView source) throws IOException {
	boolean isNull = source.readBoolean();
	if (isNull) {
	source.skipBytesToRead(padding.length);
	}
	return isNull;
	}

	@Override
	public void copy(DataInputView source, DataOutputView target) throws IOException {
	boolean isNull = source.readBoolean();
	target.writeBoolean(isNull);
	if (isNull) {
	target.write(padding);
	} else {
	originalSerializer.copy(source, target);
	}
	}

	@Override
	public boolean equals(Object obj) {
	return obj == this \|\|
	(obj != null && obj.getClass() == getClass() &&
	originalSerializer.equals(((NullableSerializer) obj).originalSerializer));
	}

	@Override
	public boolean canEqual(Object obj) {
	return (obj != null && obj.getClass() == getClass() &&
	originalSerializer.canEqual(((NullableSerializer) obj).originalSerializer));
	}

	@Override
	public int hashCode() {
	return originalSerializer.hashCode();
	}

	@Override
	public NullableSerializerConfigSnapshot<T> snapshotConfiguration() {
	return new NullableSerializerConfigSnapshot<>(originalSerializer);
	}

	@Override
	public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
	if (configSnapshot instanceof NullableSerializerConfigSnapshot) {
	List<Tuple2<TypeSerializer<?>, TypeSerializerConfigSnapshot>> previousKvSerializersAndConfigs =
	((NullableSerializerConfigSnapshot) configSnapshot).getNestedSerializersAndConfigs();

	CompatibilityResult<T> compatResult = CompatibilityUtil.resolveCompatibilityResult(
	previousKvSerializersAndConfigs.get(0).f0,
	UnloadableDummyTypeSerializer.class,
	previousKvSerializersAndConfigs.get(0).f1,
	originalSerializer);

	if (!compatResult.isRequiresMigration()) {
	return CompatibilityResult.compatible();
	} else if (compatResult.getConvertDeserializer() != null) {
	return CompatibilityResult.requiresMigration(
	new NullableSerializer<>(
	new TypeDeserializerAdapter<>(compatResult.getConvertDeserializer()), padNullValue()));
	}
	}

	return CompatibilityResult.requiresMigration();
	}

	/**
	* Configuration snapshot for serializers of nullable types, containing the
	* configuration snapshot of its original serializer.
	*/
	@Internal
	public static class NullableSerializerConfigSnapshot<T> extends CompositeTypeSerializerConfigSnapshot {
	private static final int VERSION = 1;

	/** This empty nullary constructor is required for deserializing the configuration. */
	@SuppressWarnings("unused")
	public NullableSerializerConfigSnapshot() {}

	NullableSerializerConfigSnapshot(TypeSerializer<T> originalSerializer) {
	super(originalSerializer);
	}

	@Override
	public int getVersion() {
	return VERSION;
	}
	}
	}