diff --git a/docs/layouts/shortcodes/generated/execution_config_configuration.html b/docs/layouts/shortcodes/generated/execution_config_configuration.html index e71bf3d7673a92..f58ab78893720c 100644 --- a/docs/layouts/shortcodes/generated/execution_config_configuration.html +++ b/docs/layouts/shortcodes/generated/execution_config_configuration.html @@ -356,6 +356,12 @@ Boolean Whether to compress spilled data. Currently we only support compress spilled data for sort and hash-agg and hash-join operators. + +
table.exec.state.schema-evolution.enabled

Streaming + false + Boolean + When enabled, a RowData keyed-state value whose schema changed in a backward-compatible way (adding nullable fields, reordering fields by name, evolving a nested ROW) is migrated when state is restored, instead of the restore failing. Only a state's own value serializer is covered: a RowData nested below a composite serializer, or a state whose serializer the operator pre-builds, is still rejected. Takes effect only on a state backend that migrates restored values at the object level, currently RocksDB; on other backends the restore still fails. Disabled by default. +
table.exec.state.ttl

Streaming 0 ms diff --git a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java index fe8eb1e8e8d96d..482960a1034ce8 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfig.java @@ -18,6 +18,7 @@ package org.apache.flink.api.common.serialization; +import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.typeinfo.TypeInfoFactory; import org.apache.flink.configuration.PipelineOptions; @@ -71,6 +72,21 @@ public interface SerializerConfig extends Serializable { /** Returns whether forces Flink to register Apache Avro classes in Kryo serializer. */ TernaryBoolean isForceKryoAvroEnabled(); + /** + * Whether state schema evolution for {@code RowData} state is enabled. When enabled, a {@code + * RowData} state value serializer admits a backward-compatible schema change and migrates the + * stored values when state is restored, instead of the restore failing. + * + *

This is a runtime hook rather than user-facing configuration. The option itself is owned + * by {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED} and only mirrored + * here by key, because this module cannot depend on the table API. Set the option there, not + * through this method. + */ + @Internal + default boolean isStateSchemaEvolutionEnabled() { + return false; + } + /** * Sets all relevant options contained in the {@link ReadableConfig} such as e.g. {@link * PipelineOptions#FORCE_KRYO}. diff --git a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java index 94bb6a0e7109db..453b2f175a3d76 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/serialization/SerializerConfigImpl.java @@ -23,6 +23,8 @@ import org.apache.flink.api.common.functions.InvalidTypesException; import org.apache.flink.api.common.typeinfo.TypeInfoFactory; import org.apache.flink.api.java.typeutils.TypeExtractor; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.ConfigurationUtils; import org.apache.flink.configuration.PipelineOptions; @@ -47,6 +49,14 @@ public final class SerializerConfigImpl implements SerializerConfig { private static final long serialVersionUID = 1L; + // Mirrors org.apache.flink.table.api.config.ExecutionConfigOptions + // .TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED by key string, because flink-core cannot depend + // on flink-table-api-java. The value travels in the shared job Configuration. + private static final ConfigOption STATE_SCHEMA_EVOLUTION_ENABLED = + ConfigOptions.key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false); + private final Configuration configuration; // ------------------------------- User code values -------------------------------------------- @@ -259,6 +269,11 @@ public void setForceKryo(boolean forceKryo) { configuration.set(PipelineOptions.FORCE_KRYO, forceKryo); } + @Override + public boolean isStateSchemaEvolutionEnabled() { + return configuration.get(STATE_SCHEMA_EVOLUTION_ENABLED); + } + /** Returns whether the Apache Avro is the serializer for POJOs. */ public boolean isForceAvroEnabled() { return configuration.get(PipelineOptions.FORCE_AVRO); @@ -357,6 +372,10 @@ public void configure(ReadableConfig configuration, ClassLoader classLoader) { configuration .getOptional(PipelineOptions.SERIALIZATION_CONFIG) .ifPresent(c -> parseSerializationConfigWithExceptionHandling(classLoader, c)); + configuration + .getOptional(STATE_SCHEMA_EVOLUTION_ENABLED) + .ifPresent( + enabled -> this.configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, enabled)); } @SuppressWarnings("unchecked") diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/StateSchemaEvolvingSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/StateSchemaEvolvingSerializer.java new file mode 100644 index 00000000000000..a50d5fd05f39c1 --- /dev/null +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/StateSchemaEvolvingSerializer.java @@ -0,0 +1,87 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.api.common.functions.SerializerFactory; +import org.apache.flink.api.common.typeinfo.TypeInformation; + +/** + * A {@link TypeSerializer} that can admit a backward-compatible change of the schema of the values + * it serializes, provided the state it belongs to will actually have those values migrated. + * + *

Implementations are inert by default: a serializer only starts admitting schema changes once + * {@link #withStateSchemaEvolution()} has been called on it, and only if the job configuration + * opted in. + * + * @param the type of the serialized values + */ +@Internal +public interface StateSchemaEvolvingSerializer { + + /** + * Returns a serializer that admits a backward-compatible schema change and migrates the stored + * values when state is restored, or {@code this} if the job configuration did not opt in. + * + *

This is called only on a state's own value serializer, never on an arbitrary serializer + * encountered while walking a type. + */ + TypeSerializer withStateSchemaEvolution(); + + /** + * Decorates a factory so that the serializer it produces for a state value is armed for schema + * evolution. + * + *

Only a caller whose backend migrates restored values through {@link + * TypeSerializerSnapshot#migrate} may use this. A caller that does not decorate its factory + * keeps today's behavior unchanged. + */ + static SerializerFactory arming(SerializerFactory delegate) { + // Not a lambda: SerializerFactory's single method is generic, which a lambda cannot + // implement. + return new SerializerFactory() { + @Override + public TypeSerializer createSerializer(TypeInformation typeInformation) { + return armStateValueSerializer(delegate.createSerializer(typeInformation)); + } + }; + } + + /** + * Arms the serializer a state holds for its values, if it supports schema evolution at all. + * + *

The serializer passed in is armed, and nothing below it: there is no descent into a + * composite, and no recursion. A serializer nested below the state value is not reached by + * {@link TypeSerializerSnapshot#migrate}, so arming one would let a compatibility check report + * {@code compatibleAfterMigration} for bytes that nothing ever migrates. + * + *

The serializers armed here are therefore a subset of those {@code + * TtlAwareSerializer#wrapTtlAwareSerializer} descends into, which are the ones some backend + * calls {@code migrate} on. Being a subset is what keeps this sound. Widening the descent to + * close the gap is only safe once every caller registering the widened shape is known to reach + * such a backend, which is not true of the seam as it stands: operator state and broadcast + * state register list and map descriptors through it and never migrate. + */ + @SuppressWarnings("unchecked") + static TypeSerializer armStateValueSerializer(TypeSerializer serializer) { + return serializer instanceof StateSchemaEvolvingSerializer + ? ((StateSchemaEvolvingSerializer) serializer).withStateSchemaEvolution() + : serializer; + } +} diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java index 1fe4134ee51c9a..45b56d9fa10076 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java @@ -134,6 +134,41 @@ void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLo TypeSerializerSchemaCompatibility resolveSchemaCompatibility( TypeSerializerSnapshot oldSerializerSnapshot); + /** + * Migrates a single state value from the schema described by {@code oldSerializerSnapshot} to + * the schema described by this (new) snapshot. Like {@link + * #resolveSchemaCompatibility(TypeSerializerSnapshot)}, this is invoked on the new snapshot and + * receives the old snapshot as its argument. + * + *

The default implementation returns the value unchanged: a value already deserialized with + * the prior serializer is structurally compatible with the current serializer, so the caller + * can re-serialize it as-is. A serializer whose in-memory representation is coupled to its + * schema should override this to transform the value into the new layout -- for example by + * inserting nulls for added fields or reordering fields by name. An implementation may return + * the given value or a new instance. + * + *

The migration is not applied recursively to nested serializers. The snapshot of a + * composite type returns its value unchanged unless it overrides this method to decompose the + * value and migrate each part, so a caller that needs a nested value migrated must reach the + * nested snapshot itself. An implementation that does so should not assume that the old and the + * new snapshot expose nested snapshots of the same type: restoring may have replaced those of + * the old snapshot with decorators, so nested snapshots are best matched by position or name + * rather than by class. + * + * @param oldSerializerSnapshot snapshot of the serializer that wrote the value. A caller that + * holds the snapshot persisted with the state should pass that one in preference to a + * snapshot re-derived from a serializer restored from it, because that round trip does not + * always reproduce the schema that was written. + * @param value the value, already deserialized with the prior serializer. It may be {@code + * null} wherever the prior serializer can produce {@code null}. An implementation that + * decomposes a composite value may likewise pass {@code null} to a nested snapshot for an + * absent part, even where that part's serializer would reject {@code null} at top level. + * @return the value adapted to the schema of the current serializer. + */ + default T migrate(TypeSerializerSnapshot oldSerializerSnapshot, T value) { + return value; + } + // ------------------------------------------------------------------------ // read / write utilities // ------------------------------------------------------------------------ diff --git a/flink-core/src/test/java/org/apache/flink/api/common/serialization/SerializerConfigImplTest.java b/flink-core/src/test/java/org/apache/flink/api/common/serialization/SerializerConfigImplTest.java index e150825e65992c..cfbf708d2b894b 100644 --- a/flink-core/src/test/java/org/apache/flink/api/common/serialization/SerializerConfigImplTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/common/serialization/SerializerConfigImplTest.java @@ -20,6 +20,8 @@ import org.apache.flink.api.common.typeinfo.TypeInfoFactory; import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.Configuration; import com.esotericsoftware.kryo.Kryo; @@ -42,6 +44,47 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; class SerializerConfigImplTest { + + /** + * Mirrors {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED}, which lives + * in a module this one cannot depend on. + */ + private static final ConfigOption STATE_SCHEMA_EVOLUTION_ENABLED = + ConfigOptions.key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false); + + @Test + void testStateSchemaEvolutionFlagSurvivesConstructionAndCopy() { + Configuration configuration = new Configuration(); + configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, true); + + SerializerConfigImpl serializerConfig = new SerializerConfigImpl(configuration); + + assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isTrue(); + assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isTrue(); + } + + @Test + void testStateSchemaEvolutionFlagDefaultsFalseThroughCopy() { + SerializerConfigImpl serializerConfig = new SerializerConfigImpl(); + + assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isFalse(); + assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isFalse(); + } + + @Test + void testStateSchemaEvolutionFlagSurvivesConfigure() { + Configuration configuration = new Configuration(); + configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, true); + + SerializerConfigImpl serializerConfig = new SerializerConfigImpl(); + serializerConfig.configure(configuration, SerializerConfigImplTest.class.getClassLoader()); + + assertThat(serializerConfig.isStateSchemaEvolutionEnabled()).isTrue(); + assertThat(serializerConfig.copy().isStateSchemaEvolutionEnabled()).isTrue(); + } + @Test void testReadingDefaultConfig() { SerializerConfig config = new SerializerConfigImpl(); diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java index b176adc482b36f..3dbace1bf58583 100644 --- a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java +++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshotTest.java @@ -54,6 +54,15 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( .isTrue(); } + @Test + void testMigrateReturnsValueUnchangedByDefault() { + TypeSerializerSnapshot oldSnapshot = new NotCompletedTypeSerializerSnapshot(); + TypeSerializerSnapshot newSnapshot = new NotCompletedTypeSerializerSnapshot(); + Integer value = 1000; + + assertThat(newSnapshot.migrate(oldSnapshot, value)).isSameAs(value); + } + private static class NotCompletedTypeSerializer extends TypeSerializer { @Override diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java index 1bec7b2218dc91..6e1ddeec98e569 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java @@ -20,9 +20,12 @@ import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.functions.SerializerFactory; import org.apache.flink.api.common.state.InternalCheckpointListener; import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.core.fs.CloseableRegistry; import org.apache.flink.runtime.checkpoint.CheckpointOptions; @@ -387,7 +390,7 @@ public S getOrCreateKeyedState( InternalKvState kvState = keyValueStatesByName.get(stateDescriptor.getName()); if (kvState == null) { if (!stateDescriptor.isSerializerInitialized()) { - stateDescriptor.initializeSerializerUnlessSet(executionConfig); + stateDescriptor.initializeSerializerUnlessSet(stateValueSerializerFactory()); } kvState = MetricsTrackingStateFactory.createStateAndWrapWithMetricsTrackingIfEnabled( @@ -403,6 +406,29 @@ public S getOrCreateKeyedState( return (S) kvState; } + /** + * The factory a state descriptor registered here uses for its own value serializer. It + * reproduces what {@link StateDescriptor#initializeSerializerUnlessSet(ExecutionConfig)} + * builds, and arms schema evolution on top of it only when this backend migrates restored + * values at the object level. + */ + private SerializerFactory stateValueSerializerFactory() { + SerializerFactory factory = + new SerializerFactory() { + @Override + public TypeSerializer createSerializer( + TypeInformation typeInformation) { + return typeInformation.createSerializer( + executionConfig == null + ? null + : executionConfig.getSerializerConfig()); + } + }; + return supportsObjectLevelValueMigration() + ? StateSchemaEvolvingSerializer.arming(factory) + : factory; + } + public void publishQueryableStateIfEnabled( StateDescriptor stateDescriptor, InternalKvState kvState) { if (stateDescriptor.isQueryable()) { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java index 82cdb849d99eee..3c902ee60204f5 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java @@ -33,6 +33,7 @@ import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; import org.apache.flink.util.Preconditions; import javax.annotation.Nonnull; @@ -52,6 +53,13 @@ public class DefaultKeyedStateStore implements KeyedStateStore { @Nullable protected final AsyncKeyedStateBackend asyncKeyedStateBackend; protected final SerializerFactory serializerFactory; + /** + * The factory used for a state's own value serializer. It arms schema evolution only on a + * backend that migrates restored values at the object level; on any other backend it is {@link + * #serializerFactory} itself, so those states keep today's behavior. + */ + private final SerializerFactory stateValueSerializerFactory; + protected SupportKeyedStateApiSet supportKeyedStateApiSet; public DefaultKeyedStateStore( @@ -71,6 +79,10 @@ public DefaultKeyedStateStore( this.keyedStateBackend = keyedStateBackend; this.asyncKeyedStateBackend = asyncKeyedStateBackend; this.serializerFactory = Preconditions.checkNotNull(serializerFactory); + this.stateValueSerializerFactory = + keyedStateBackend != null && keyedStateBackend.supportsObjectLevelValueMigration() + ? StateSchemaEvolvingSerializer.arming(this.serializerFactory) + : this.serializerFactory; if (keyedStateBackend != null) { // By default, we support state v1 this.supportKeyedStateApiSet = SupportKeyedStateApiSet.STATE_V1; @@ -85,7 +97,7 @@ public DefaultKeyedStateStore( public ValueState getState(ValueStateDescriptor stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { - stateProperties.initializeSerializerUnlessSet(serializerFactory); + stateProperties.initializeSerializerUnlessSet(stateValueSerializerFactory); return getPartitionedState(stateProperties); } catch (Exception e) { throw new RuntimeException("Error while getting state", e); @@ -96,7 +108,7 @@ public ValueState getState(ValueStateDescriptor stateProperties) { public ListState getListState(ListStateDescriptor stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { - stateProperties.initializeSerializerUnlessSet(serializerFactory); + stateProperties.initializeSerializerUnlessSet(stateValueSerializerFactory); ListState originalState = getPartitionedState(stateProperties); return new UserFacingListState<>(originalState); } catch (Exception e) { @@ -108,7 +120,7 @@ public ListState getListState(ListStateDescriptor stateProperties) { public ReducingState getReducingState(ReducingStateDescriptor stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { - stateProperties.initializeSerializerUnlessSet(serializerFactory); + stateProperties.initializeSerializerUnlessSet(stateValueSerializerFactory); return getPartitionedState(stateProperties); } catch (Exception e) { throw new RuntimeException("Error while getting state", e); @@ -120,7 +132,7 @@ public AggregatingState getAggregatingState( AggregatingStateDescriptor stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { - stateProperties.initializeSerializerUnlessSet(serializerFactory); + stateProperties.initializeSerializerUnlessSet(stateValueSerializerFactory); return getPartitionedState(stateProperties); } catch (Exception e) { throw new RuntimeException("Error while getting state", e); @@ -131,7 +143,7 @@ public AggregatingState getAggregatingState( public MapState getMapState(MapStateDescriptor stateProperties) { requireNonNull(stateProperties, "The state properties must not be null"); try { - stateProperties.initializeSerializerUnlessSet(serializerFactory); + stateProperties.initializeSerializerUnlessSet(stateValueSerializerFactory); MapState originalState = getPartitionedState(stateProperties); return new UserFacingMapState<>(originalState); } catch (Exception e) { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java index d773fdb4a0af1c..2592f432dfffae 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java @@ -19,6 +19,7 @@ package org.apache.flink.runtime.state; import org.apache.flink.annotation.Experimental; +import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.typeutils.TypeSerializer; @@ -167,6 +168,21 @@ default boolean isSafeToReuseKVState() { return false; } + /** + * Whether this backend migrates restored values at the object level. + * + *

A backend returning {@code true} guarantees that every value it restores under a {@code + * compatibleAfterMigration} verdict passes through {@link + * org.apache.flink.api.common.typeutils.TypeSerializerSnapshot#migrate} on the state's own + * value serializer. State schema evolution is armed only on such a backend: on any other one + * the verdict would be accepted and the stored bytes rewritten under the new schema with + * nothing ever converting them. + */ + @Internal + default boolean supportsObjectLevelValueMigration() { + return false; + } + /** * @return fixed lower-case string identifying the type of the underlying state backend, e.g. * rocksdb, hashmap, forst, batch. diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java index 2f50ba6c668337..0dabcc5c869b24 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializer.java @@ -26,6 +26,8 @@ import org.apache.flink.core.memory.DataOutputView; import org.apache.flink.util.function.SupplierWithException; +import javax.annotation.Nullable; + import java.io.IOException; import java.util.List; import java.util.Map; @@ -47,6 +49,13 @@ public class TtlAwareSerializer> extends TypeSeri private final S typeSerializer; + /** + * Snapshot of {@link #bareValueSerializer()}, computed on first use. {@link + * #migrateValueFromPriorSerializer} runs once per migrated state value while the serializer + * stays the same, and taking a snapshot allocates one object per nested serializer. + */ + private transient TypeSerializerSnapshot bareValueSerializerSnapshot; + public TtlAwareSerializer(S typeSerializer) { checkArgument( !(typeSerializer instanceof TtlAwareSerializer), @@ -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}. + * + *

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 priorTtlAwareSerializer, + @Nullable TypeSerializerSnapshot priorSerializerSnapshot, SupplierWithException 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) inputSupplier.get()).getUserValue() - : inputSupplier.get(); + outputRecord = (T) migratedValue; } this.serialize(outputRecord, target); } + /** + * The snapshot describing the schema the prior bare value was written with. + * + *

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; + + // Thrown rather than checked through Preconditions: this runs once per migrated state + // value, so the message must not be built while the check is passing. + boolean isTtlSnapshot = priorSnapshot instanceof TtlStateFactory.TtlSerializerSnapshot; + if (!priorSerializer.wrapsTtlValue()) { + if (isTtlSnapshot) { + throw new IllegalArgumentException( + "The prior serializer does not wrap values in TtlValue, but its persisted snapshot is a TtlSerializerSnapshot."); + } + return priorSnapshot; + } + if (!isTtlSnapshot) { + throw new IllegalArgumentException( + "The prior serializer wraps values in TtlValue, so its persisted snapshot should be a TtlSerializerSnapshot, but was " + + priorSnapshot.getClass().getName() + + "."); + } + // The persisted snapshot describes the TtlValue envelope, so descend to the user value + // the same way bareValueSerializer() descends the serializer. + return ((TtlStateFactory.TtlSerializerSnapshot) priorSnapshot) + .getValueSerializerSnapshot(); + } + + private TypeSerializerSnapshot bareValueSerializerSnapshot() { + if (bareValueSerializerSnapshot == null) { + bareValueSerializerSnapshot = bareValueSerializer().snapshotConfiguration(); + } + return bareValueSerializerSnapshot; + } + + /** + * The serializer of the bare (non-TTL) value: the user value serializer of a {@link + * TtlStateFactory.TtlSerializer}, otherwise the wrapped serializer itself. + */ + private TypeSerializer bareValueSerializer() { + return wrapsTtlValue() + ? ((TtlStateFactory.TtlSerializer) typeSerializer).getValueSerializer() + : typeSerializer; + } + + /** + * Whether the values this serializer reads and writes are {@link TtlValue} envelopes. Narrower + * than {@link #isTtlEnabled()}, which is also true for a list or map serializer whose element + * or value serializer is a {@link TtlStateFactory.TtlSerializer}: such a serializer wraps the + * collection, not a single {@code TtlValue}. + */ + private boolean wrapsTtlValue() { + return typeSerializer instanceof TtlStateFactory.TtlSerializer; + } + @Override public void copy(DataInputView source, DataOutputView target) throws IOException { typeSerializer.copy(source, target); diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java index 95867d16a7b041..f9b191f8cd8f3e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContext.java @@ -203,21 +203,18 @@ public C getBroadcastVariableWithInitializer( @Override public ValueState getState(ValueStateDescriptor stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); - stateProperties.initializeSerializerUnlessSet(this::createSerializer); return keyedStateStore.getState(stateProperties); } @Override public ListState getListState(ListStateDescriptor stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); - stateProperties.initializeSerializerUnlessSet(this::createSerializer); return keyedStateStore.getListState(stateProperties); } @Override public ReducingState getReducingState(ReducingStateDescriptor stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); - stateProperties.initializeSerializerUnlessSet(this::createSerializer); return keyedStateStore.getReducingState(stateProperties); } @@ -225,14 +222,12 @@ public ReducingState getReducingState(ReducingStateDescriptor statePro public AggregatingState getAggregatingState( AggregatingStateDescriptor stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); - stateProperties.initializeSerializerUnlessSet(this::createSerializer); return keyedStateStore.getAggregatingState(stateProperties); } @Override public MapState getMapState(MapStateDescriptor stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); - stateProperties.initializeSerializerUnlessSet(this::createSerializer); return keyedStateStore.getMapState(stateProperties); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolutionArmingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolutionArmingTest.java new file mode 100644 index 00000000000000..66a2466b89d693 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolutionArmingTest.java @@ -0,0 +1,374 @@ +/* + * 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.runtime.state; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.functions.SerializerFactory; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.state.State; +import org.apache.flink.api.common.state.StateDescriptor; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.fs.CloseableRegistry; +import org.apache.flink.runtime.asyncprocessing.StateExecutor; +import org.apache.flink.runtime.asyncprocessing.StateRequestHandler; +import org.apache.flink.runtime.checkpoint.CheckpointOptions; +import org.apache.flink.runtime.state.StateSchemaEvolvingTestSerializer.StateSchemaEvolvingTestTypeInfo; +import org.apache.flink.runtime.state.StateSnapshotTransformer.StateSnapshotTransformFactory; +import org.apache.flink.runtime.state.heap.HeapPriorityQueueElement; +import org.apache.flink.runtime.state.v2.internal.InternalKeyedState; + +import org.junit.jupiter.api.Test; + +import javax.annotation.Nonnull; + +import java.util.Collections; +import java.util.List; +import java.util.concurrent.RunnableFuture; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests the backend capability gate on state schema evolution arming: only a keyed backend that + * performs object-level value migration may hand a state its armed value serializer. + * + *

Assertions are made on the serializer object the state descriptor ends up holding, so a gate + * that is bypassed shows up as an armed serializer rather than having to be read out of the code + * path. + */ +class StateSchemaEvolutionArmingTest { + + /** + * Mirrors {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED}, which lives + * in a module this one cannot depend on. + */ + private static final ConfigOption STATE_SCHEMA_EVOLUTION_ENABLED = + ConfigOptions.key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false); + + @Test + void valueStateIsArmedOnAnObjectLevelMigratingBackend() { + ValueStateDescriptor descriptor = + new ValueStateDescriptor<>("value", new StateSchemaEvolvingTestTypeInfo()); + + keyedStateStore(true).getState(descriptor); + + assertThat(armedFlagOf(descriptor.getSerializer())).isTrue(); + } + + @Test + void valueStateIsNotArmedOnABackendWithoutObjectLevelMigration() { + ValueStateDescriptor descriptor = + new ValueStateDescriptor<>("value", new StateSchemaEvolvingTestTypeInfo()); + + keyedStateStore(false).getState(descriptor); + + assertThat(armedFlagOf(descriptor.getSerializer())).isFalse(); + } + + @Test + void v2ValueStateIsNotArmed() { + org.apache.flink.api.common.state.v2.ValueStateDescriptor descriptor = + new org.apache.flink.api.common.state.v2.ValueStateDescriptor<>( + "value", new StateSchemaEvolvingTestTypeInfo()); + + asyncKeyedStateStore().getValueState(descriptor); + + assertThat(armedFlagOf(descriptor.getSerializer())).isFalse(); + } + + @Test + void operatorListStateIsNotArmed() throws Exception { + ListStateDescriptor descriptor = + new ListStateDescriptor<>("list", new StateSchemaEvolvingTestTypeInfo()); + + operatorStateBackend().getListState(descriptor); + + assertThat(armedFlagOf(descriptor.getElementSerializer())).isFalse(); + } + + @Test + void broadcastStateIsNotArmed() throws Exception { + MapStateDescriptor descriptor = + new MapStateDescriptor<>( + "broadcast", Types.INT, new StateSchemaEvolvingTestTypeInfo()); + + operatorStateBackend().getBroadcastState(descriptor); + + assertThat(armedFlagOf(descriptor.getValueSerializer())).isFalse(); + } + + private static boolean armedFlagOf(TypeSerializer serializer) { + assertThat(serializer).isInstanceOf(StateSchemaEvolvingTestSerializer.class); + return ((StateSchemaEvolvingTestSerializer) serializer).isArmed(); + } + + private static DefaultKeyedStateStore keyedStateStore(boolean objectLevelValueMigration) { + return new DefaultKeyedStateStore( + new TestKeyedStateBackend(objectLevelValueMigration), serializerFactory()); + } + + private static DefaultKeyedStateStore asyncKeyedStateStore() { + DefaultKeyedStateStore store = + new DefaultKeyedStateStore( + new TestKeyedStateBackend(true), + new TestAsyncKeyedStateBackend(), + serializerFactory()); + store.setSupportKeyedStateApiSetV2(); + return store; + } + + private static DefaultOperatorStateBackend operatorStateBackend() throws Exception { + return new DefaultOperatorStateBackendBuilder( + StateSchemaEvolutionArmingTest.class.getClassLoader(), + new ExecutionConfig(configurationWithSchemaEvolutionEnabled()), + false, + Collections.emptyList(), + new CloseableRegistry()) + .build(); + } + + private static SerializerFactory serializerFactory() { + SerializerConfig config = + new SerializerConfigImpl(configurationWithSchemaEvolutionEnabled()); + return new SerializerFactory() { + @Override + public TypeSerializer createSerializer(TypeInformation typeInformation) { + return typeInformation.createSerializer(config); + } + }; + } + + private static Configuration configurationWithSchemaEvolutionEnabled() { + Configuration configuration = new Configuration(); + configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, true); + return configuration; + } + + /** + * A keyed backend whose object-level value migration capability is fixed per instance. {@link + * DefaultKeyedStateStore} only reads that capability and forwards the descriptor to {@link + * #getPartitionedState}, so every other method is left unsupported. + */ + private static final class TestKeyedStateBackend implements KeyedStateBackend { + + private final boolean objectLevelValueMigration; + + private TestKeyedStateBackend(boolean objectLevelValueMigration) { + this.objectLevelValueMigration = objectLevelValueMigration; + } + + @Override + public boolean supportsObjectLevelValueMigration() { + return objectLevelValueMigration; + } + + @Override + public S getPartitionedState( + N namespace, + TypeSerializer namespaceSerializer, + StateDescriptor stateDescriptor) { + return null; + } + + @Override + public String getBackendTypeIdentifier() { + return "test"; + } + + @Override + public void setCurrentKey(Integer newKey) { + throw new UnsupportedOperationException(); + } + + @Override + public Integer getCurrentKey() { + throw new UnsupportedOperationException(); + } + + @Override + public void setCurrentKeyAndKeyGroup(Integer newKey, int newKeyGroupIndex) { + throw new UnsupportedOperationException(); + } + + @Override + public TypeSerializer getKeySerializer() { + throw new UnsupportedOperationException(); + } + + @Override + public void applyToAllKeys( + N namespace, + TypeSerializer namespaceSerializer, + StateDescriptor stateDescriptor, + KeyedStateFunction function) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream getKeys(String state, N namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream getKeys(List states, N namespace) { + throw new UnsupportedOperationException(); + } + + @Override + public Stream> getKeysAndNamespaces(String state) { + throw new UnsupportedOperationException(); + } + + @Override + public S getOrCreateKeyedState( + TypeSerializer namespaceSerializer, StateDescriptor stateDescriptor) { + throw new UnsupportedOperationException(); + } + + @Override + public void dispose() { + throw new UnsupportedOperationException(); + } + + @Override + public void registerKeySelectionListener(KeySelectionListener listener) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean deregisterKeySelectionListener(KeySelectionListener listener) { + throw new UnsupportedOperationException(); + } + + @Nonnull + @Override + public IS createOrUpdateInternalState( + @Nonnull TypeSerializer namespaceSerializer, + @Nonnull StateDescriptor stateDesc, + @Nonnull StateSnapshotTransformFactory snapshotTransformFactory) { + throw new UnsupportedOperationException(); + } + + @Nonnull + @Override + public & Keyed> + KeyGroupedInternalPriorityQueue create( + @Nonnull String stateName, + @Nonnull TypeSerializer byteOrderedElementSerializer) { + throw new UnsupportedOperationException(); + } + } + + /** + * An async keyed backend that accepts a v2 descriptor and returns no state, so the v2 path + * through {@link DefaultKeyedStateStore} can be exercised without a real backend. + */ + private static final class TestAsyncKeyedStateBackend + implements AsyncKeyedStateBackend { + + @Override + public + S getOrCreateKeyedState( + N defaultNamespace, + TypeSerializer namespaceSerializer, + org.apache.flink.api.common.state.v2.StateDescriptor stateDesc) { + return null; + } + + @Override + public String getBackendTypeIdentifier() { + return "test"; + } + + @Override + public void setup(@Nonnull StateRequestHandler stateRequestHandler) { + throw new UnsupportedOperationException(); + } + + @Nonnull + @Override + public S createStateInternal( + @Nonnull N defaultNamespace, + @Nonnull TypeSerializer namespaceSerializer, + @Nonnull org.apache.flink.api.common.state.v2.StateDescriptor stateDesc) { + throw new UnsupportedOperationException(); + } + + @Nonnull + @Override + public StateExecutor createStateExecutor() { + throw new UnsupportedOperationException(); + } + + @Override + public KeyGroupRange getKeyGroupRange() { + throw new UnsupportedOperationException(); + } + + @Override + public void dispose() { + throw new UnsupportedOperationException(); + } + + @Override + public void close() { + throw new UnsupportedOperationException(); + } + + @Override + public void notifyCheckpointSubsumed(long checkpointId) { + throw new UnsupportedOperationException(); + } + + @Override + public void notifyCheckpointComplete(long checkpointId) { + throw new UnsupportedOperationException(); + } + + @Override + public RunnableFuture> snapshot( + long checkpointId, + long timestamp, + CheckpointStreamFactory streamFactory, + CheckpointOptions checkpointOptions) { + throw new UnsupportedOperationException(); + } + + @Nonnull + @Override + public & Keyed> + KeyGroupedInternalPriorityQueue create( + @Nonnull String stateName, + @Nonnull TypeSerializer byteOrderedElementSerializer) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolvingTestSerializer.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolvingTestSerializer.java new file mode 100644 index 00000000000000..de90195a1323ab --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSchemaEvolvingTestSerializer.java @@ -0,0 +1,198 @@ +/* + * 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.runtime.state; + +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.core.memory.DataInputView; +import org.apache.flink.core.memory.DataOutputView; + +import java.io.IOException; + +/** + * An integer serializer that records whether it was armed for state schema evolution, so a test can + * assert arming on the serializer object a state descriptor actually ends up holding. + * + *

Lives here because the production {@link StateSchemaEvolvingSerializer} implementation is a + * table-module serializer that flink-runtime cannot see. + */ +public class StateSchemaEvolvingTestSerializer extends TypeSerializer + implements StateSchemaEvolvingSerializer { + + private static final long serialVersionUID = 1L; + + private final boolean armed; + + public StateSchemaEvolvingTestSerializer() { + this(false); + } + + private StateSchemaEvolvingTestSerializer(boolean armed) { + this.armed = armed; + } + + public boolean isArmed() { + return armed; + } + + @Override + public TypeSerializer withStateSchemaEvolution() { + return armed ? this : new StateSchemaEvolvingTestSerializer(true); + } + + @Override + public boolean isImmutableType() { + return true; + } + + @Override + public TypeSerializer duplicate() { + return new StateSchemaEvolvingTestSerializer(armed); + } + + @Override + public Integer createInstance() { + return 0; + } + + @Override + public Integer copy(Integer from) { + return from; + } + + @Override + public Integer copy(Integer from, Integer reuse) { + return from; + } + + @Override + public int getLength() { + return Integer.BYTES; + } + + @Override + public void serialize(Integer record, DataOutputView target) throws IOException { + target.writeInt(record); + } + + @Override + public Integer deserialize(DataInputView source) throws IOException { + return source.readInt(); + } + + @Override + public Integer deserialize(Integer reuse, DataInputView source) throws IOException { + return source.readInt(); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + target.writeInt(source.readInt()); + } + + /** The armed flag is runtime-only state, so it must not perturb serializer equality. */ + @Override + public boolean equals(Object obj) { + return obj instanceof StateSchemaEvolvingTestSerializer; + } + + @Override + public int hashCode() { + return StateSchemaEvolvingTestSerializer.class.hashCode(); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new StateSchemaEvolvingTestSerializerSnapshot(); + } + + /** Snapshot for {@link StateSchemaEvolvingTestSerializer}. */ + public static final class StateSchemaEvolvingTestSerializerSnapshot + extends SimpleTypeSerializerSnapshot { + + public StateSchemaEvolvingTestSerializerSnapshot() { + super(StateSchemaEvolvingTestSerializer::new); + } + } + + /** Type information producing an unarmed {@link StateSchemaEvolvingTestSerializer}. */ + public static final class StateSchemaEvolvingTestTypeInfo extends TypeInformation { + + private static final long serialVersionUID = 1L; + + @Override + public boolean isBasicType() { + return false; + } + + @Override + public boolean isTupleType() { + return false; + } + + @Override + public int getArity() { + return 1; + } + + @Override + public int getTotalFields() { + return 1; + } + + @Override + public Class getTypeClass() { + return Integer.class; + } + + @Override + public boolean isKeyType() { + return false; + } + + @Override + public TypeSerializer createSerializer(SerializerConfig config) { + return new StateSchemaEvolvingTestSerializer(); + } + + @Override + public String toString() { + return StateSchemaEvolvingTestTypeInfo.class.getSimpleName(); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof StateSchemaEvolvingTestTypeInfo; + } + + @Override + public int hashCode() { + return StateSchemaEvolvingTestTypeInfo.class.hashCode(); + } + + @Override + public boolean canEqual(Object obj) { + return obj instanceof StateSchemaEvolvingTestTypeInfo; + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java index 65908573facd6c..97ba6b19793bef 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateSerializerProviderTest.java @@ -21,13 +21,18 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.StringSerializer; import org.apache.flink.core.memory.DataInputView; import org.apache.flink.core.memory.DataOutputView; +import org.apache.flink.runtime.state.ttl.TtlAwareSerializerSnapshot; import org.apache.flink.runtime.testutils.statemigration.TestType; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -311,6 +316,42 @@ void testEagerlyRegisterIncompatibleSerializer() { .isInstanceOf(IllegalStateException.class); } + // -------------------------------------------------------------------------------- + // Tests for the ttl-aware wrapping of the previous serializer snapshot + // -------------------------------------------------------------------------------- + + /** + * Registering a new serializer replaces the nested snapshot of the previous snapshot in place, + * so the snapshot a caller still holds no longer reports what the checkpoint wrote: its nested + * snapshot becomes a {@link TtlAwareSerializerSnapshot} around the original. Anything that + * descends a restored composite snapshot has to expect that layer. + */ + @Test + void testRegisterNewSerializerWrapsNestedSnapshotOfPreviousSnapshotInPlace() { + ListSerializerSnapshot previousSnapshot = + (ListSerializerSnapshot) + new ListSerializer<>(StringSerializer.INSTANCE).snapshotConfiguration(); + TypeSerializerSnapshot elementSnapshotAsWritten = + previousSnapshot.getElementSerializerSnapshot(); + + StateSerializerProvider> testProvider = + StateSerializerProvider.fromPreviousSerializerSnapshot(previousSnapshot); + testProvider.registerNewSerializerForRestoredState( + new ListSerializer<>(StringSerializer.INSTANCE)); + + // The same snapshot instance now reports a different element snapshot than it did above. + TypeSerializerSnapshot elementSnapshotAfterRestore = + previousSnapshot.getElementSerializerSnapshot(); + assertThat(elementSnapshotAfterRestore).isNotSameAs(elementSnapshotAsWritten); + assertThat(elementSnapshotAfterRestore).isInstanceOf(TtlAwareSerializerSnapshot.class); + // The original is carried inside the wrapper, not re-derived: a re-derived snapshot would + // be an equal instance of the same class but a different object. + assertThat( + ((TtlAwareSerializerSnapshot) elementSnapshotAfterRestore) + .getOrinalTypeSerializerSnapshot()) + .isSameAs(elementSnapshotAsWritten); + } + // -------------------------------------------------------------------------------- // Utilities // -------------------------------------------------------------------------------- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java index 90c7f70432c00b..b15cc46eb9b706 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerTest.java @@ -18,19 +18,36 @@ package org.apache.flink.runtime.state.ttl; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.IntSerializer; import org.apache.flink.api.common.typeutils.base.ListSerializer; import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.LongSerializer; import org.apache.flink.api.common.typeutils.base.MapSerializer; import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.api.java.typeutils.runtime.NullableSerializer; +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.junit.jupiter.api.Test; +import java.io.IOException; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class TtlAwareSerializerTest { + private static final String VALUE = "value"; + private static final long PRIOR_TIMESTAMP = 1_000L; + private static final long CURRENT_TIMESTAMP = 9_999L; + private static final TtlTimeProvider FIXED_TIME_PROVIDER = () -> CURRENT_TIMESTAMP; + @Test void testSerializerTtlEnabled() { IntSerializer intSerializer = IntSerializer.INSTANCE; @@ -147,4 +164,385 @@ void testSnapshotConfiguration() { .getValueSerializerSnapshot())) .isInstanceOf(TtlAwareSerializerSnapshot.class); } + + @Test + void testMigrateValueNoTtlToNoTtl() throws IOException { + TtlAwareSerializer prior = stringSerializer(false); + TtlAwareSerializer current = stringSerializer(false); + + assertThat(migrate(current, prior, VALUE)).isEqualTo(serialize(current, VALUE)); + } + + @Test + void testMigrateValueNoTtlToTtlStampsCurrentTime() throws IOException { + TtlAwareSerializer prior = stringSerializer(false); + TtlAwareSerializer current = stringSerializer(true); + + assertThat(migrate(current, prior, VALUE)) + .isEqualTo(serialize(current, new TtlValue<>(VALUE, CURRENT_TIMESTAMP))); + } + + @Test + void testMigrateValueTtlToNoTtlUnwraps() throws IOException { + TtlAwareSerializer prior = stringSerializer(true); + TtlAwareSerializer current = stringSerializer(false); + + assertThat(migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isEqualTo(serialize(current, VALUE)); + } + + @Test + void testMigrateValueTtlToTtlKeepsPriorTimestamp() throws IOException { + TtlAwareSerializer prior = stringSerializer(true); + TtlAwareSerializer current = stringSerializer(true); + TtlValue priorValue = new TtlValue<>(VALUE, PRIOR_TIMESTAMP); + + assertThat(migrate(current, prior, priorValue)).isEqualTo(serialize(current, priorValue)); + } + + @Test + void testMigrateValueInvokesHookOnNewSnapshotWithOldSnapshotAsArgument() throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=old|to=new"); + } + + @Test + void testMigrateValueInvokesHookOnUnwrappedValueWithInnerSnapshots() throws IOException { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + byte[] migrated = migrate(current, prior, new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=old|to=new"); + assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP); + } + + @Test + void testMigrateValueUsesPersistedPriorSnapshotNotOneRederivedFromPriorSerializer() + throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("rederived")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, new TaggedSnapshot("persisted"), VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=persisted|to=new"); + } + + @Test + void testMigrateValueDescendsPersistedTtlSnapshotToItsValueSnapshot() throws IOException { + TtlAwareSerializer prior = taggedTtlSerializer("rederived"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + // Tagged differently from the prior serializer, so both falling back to a re-derived + // snapshot and skipping the descent into the TtlValue envelope change the result. + TypeSerializerSnapshot persisted = + taggedTtlSerializer("persisted") + .getOriginalTypeSerializer() + .snapshotConfiguration(); + + byte[] migrated = + migrate(current, prior, persisted, new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new"); + } + + /** + * A list or map state persists its element or value snapshot wrapped in a {@link + * TtlAwareSerializerSnapshot}, which the descent has to see through to reach the user value + * snapshot. Value state persists the unwrapped shape covered by the test above. + */ + @Test + void testMigrateValueDescendsPersistedTtlAwareElementSnapshot() throws IOException { + ListSerializerSnapshot persistedListSnapshot = + (ListSerializerSnapshot) + TtlAwareSerializer.wrapTtlAwareSerializer( + new ListSerializer<>(rawTaggedTtlSerializer("persisted"))) + .snapshotConfiguration(); + TypeSerializerSnapshot persistedElementSnapshot = + persistedListSnapshot.getElementSerializerSnapshot(); + assertThat(persistedElementSnapshot).isInstanceOf(TtlAwareSerializerSnapshot.class); + + TtlAwareSerializer prior = taggedTtlSerializer("rederived"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + byte[] migrated = + migrate( + current, + prior, + persistedElementSnapshot, + new TtlValue<>(VALUE, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isEqualTo(VALUE + "|from=persisted|to=new"); + } + + @Test + void testMigrateValueRejectsNonTtlSnapshotForTtlPriorSerializer() { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + + assertThatThrownBy( + () -> + migrate( + current, + prior, + new TaggedSnapshot("mismatched"), + new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("should be a TtlSerializerSnapshot"); + } + + @Test + void testMigrateValueRejectsNonTtlSnapshotBehindTtlAwareLayer() { + TtlAwareSerializer prior = taggedTtlSerializer("old"); + TtlAwareSerializer current = taggedTtlSerializer("new"); + // The TtlAware layer is seen through, so a mismatch inside it is still caught. + TypeSerializerSnapshot wrappedMismatch = + new TtlAwareSerializerSnapshot<>(new TaggedSnapshot("mismatched")); + + assertThatThrownBy( + () -> + migrate( + current, + prior, + wrappedMismatch, + new TtlValue<>(VALUE, PRIOR_TIMESTAMP))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("should be a TtlSerializerSnapshot"); + } + + @Test + void testMigrateValueRejectsTtlSnapshotForNonTtlPriorSerializer() { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + TypeSerializerSnapshot ttlSnapshot = + rawTaggedTtlSerializer("mismatched").snapshotConfiguration(); + + assertThatThrownBy(() -> migrate(current, prior, ttlSnapshot, VALUE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("does not wrap values in TtlValue"); + } + + @Test + void testMigrateValueFallsBackToPriorSerializerSnapshotWhenNonePersisted() throws IOException { + TtlAwareSerializer prior = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("old")); + TtlAwareSerializer current = + TtlAwareSerializer.wrapTtlAwareSerializer(new TaggedStringSerializer("new")); + + byte[] migrated = migrate(current, prior, null, VALUE); + + assertThat(current.deserialize(new DataInputDeserializer(migrated))) + .isEqualTo(VALUE + "|from=old|to=new"); + } + + @Test + void testMigrateValueTtlToTtlWithNullUserValue() throws IOException { + TtlAwareSerializer prior = nullTolerantTtlSerializer(); + TtlAwareSerializer current = nullTolerantTtlSerializer(); + + byte[] migrated = migrate(current, prior, new TtlValue<>(null, PRIOR_TIMESTAMP)); + + TtlValue result = (TtlValue) current.deserialize(new DataInputDeserializer(migrated)); + assertThat(result.getUserValue()).isNull(); + assertThat(result.getLastAccessTimestamp()).isEqualTo(PRIOR_TIMESTAMP); + } + + /** Migrates with the snapshot the state backends would have persisted for {@code prior}. */ + private static byte[] migrate( + TtlAwareSerializer current, TtlAwareSerializer prior, Object priorValue) + throws IOException { + return migrate( + current, + prior, + prior.getOriginalTypeSerializer().snapshotConfiguration(), + priorValue); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static byte[] migrate( + TtlAwareSerializer current, + TtlAwareSerializer prior, + TypeSerializerSnapshot priorSnapshot, + Object priorValue) + throws IOException { + DataOutputSerializer output = new DataOutputSerializer(64); + ((TtlAwareSerializer) current) + .migrateValueFromPriorSerializer( + (TtlAwareSerializer) prior, + priorSnapshot, + () -> priorValue, + output, + FIXED_TIME_PROVIDER); + return output.getCopyOfBuffer(); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static byte[] serialize(TtlAwareSerializer serializer, Object value) + throws IOException { + DataOutputSerializer output = new DataOutputSerializer(64); + ((TtlAwareSerializer) serializer).serialize(value, output); + return output.getCopyOfBuffer(); + } + + private static TtlAwareSerializer stringSerializer(boolean ttlEnabled) { + return ttlEnabled + ? TtlAwareSerializer.wrapTtlAwareSerializer( + new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, StringSerializer.INSTANCE)) + : TtlAwareSerializer.wrapTtlAwareSerializer(StringSerializer.INSTANCE); + } + + private static TtlAwareSerializer taggedTtlSerializer(String tag) { + return TtlAwareSerializer.wrapTtlAwareSerializer(rawTaggedTtlSerializer(tag)); + } + + private static TtlStateFactory.TtlSerializer rawTaggedTtlSerializer(String tag) { + return new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, new TaggedStringSerializer(tag)); + } + + private static TtlAwareSerializer nullTolerantTtlSerializer() { + return TtlAwareSerializer.wrapTtlAwareSerializer( + new TtlStateFactory.TtlSerializer<>( + LongSerializer.INSTANCE, + NullableSerializer.wrapIfNullIsNotSupported( + StringSerializer.INSTANCE, false))); + } + + /** A string serializer whose snapshot records the schema it belongs to. */ + private static final class TaggedStringSerializer extends TypeSerializer { + + private final String tag; + + private TaggedStringSerializer(String tag) { + this.tag = tag; + } + + @Override + public boolean isImmutableType() { + return true; + } + + @Override + public TypeSerializer duplicate() { + return this; + } + + @Override + public String createInstance() { + return ""; + } + + @Override + public String copy(String from) { + return from; + } + + @Override + public String copy(String from, String reuse) { + return from; + } + + @Override + public int getLength() { + return -1; + } + + @Override + public void serialize(String record, DataOutputView target) throws IOException { + StringSerializer.INSTANCE.serialize(record, target); + } + + @Override + public String deserialize(DataInputView source) throws IOException { + return StringSerializer.INSTANCE.deserialize(source); + } + + @Override + public String deserialize(String reuse, DataInputView source) throws IOException { + return deserialize(source); + } + + @Override + public void copy(DataInputView source, DataOutputView target) throws IOException { + StringSerializer.INSTANCE.copy(source, target); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof TaggedStringSerializer + && tag.equals(((TaggedStringSerializer) obj).tag); + } + + @Override + public int hashCode() { + return tag.hashCode(); + } + + @Override + public TypeSerializerSnapshot snapshotConfiguration() { + return new TaggedSnapshot(tag); + } + } + + /** + * Appends both the argument snapshot's tag and its own tag to the migrated value, so a + * migration that swaps receiver and argument produces a different result. + */ + public static final class TaggedSnapshot implements TypeSerializerSnapshot { + + private String tag; + + public TaggedSnapshot() {} + + private TaggedSnapshot(String tag) { + this.tag = tag; + } + + @Override + public int getCurrentVersion() { + return 1; + } + + @Override + public void writeSnapshot(DataOutputView out) throws IOException { + out.writeUTF(tag); + } + + @Override + public void readSnapshot(int readVersion, DataInputView in, ClassLoader userCodeClassLoader) + throws IOException { + tag = in.readUTF(); + } + + @Override + public TypeSerializer restoreSerializer() { + return new TaggedStringSerializer(tag); + } + + @Override + public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( + TypeSerializerSnapshot oldSerializerSnapshot) { + return TypeSerializerSchemaCompatibility.compatibleAsIs(); + } + + @Override + public String migrate(TypeSerializerSnapshot oldSerializerSnapshot, String value) { + return value + "|from=" + ((TaggedSnapshot) oldSerializerSnapshot).tag + "|to=" + tag; + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java index d8b4234252c244..ffe2f41c03a4e7 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlAwareSerializerUpgradeTest.java @@ -161,6 +161,7 @@ public DataInputView readAndThenWriteData( DataOutputSerializer migratedOut = new DataOutputSerializer(INITIAL_OUTPUT_BUFFER_SIZE); writer.migrateValueFromPriorSerializer( reader, + reader.getOriginalTypeSerializer().snapshotConfiguration(), () -> reader.deserialize(originalDataInput), migratedOut, TtlTimeProvider.DEFAULT); diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java index b1b71029eed943..b10cd889e578e2 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/StreamingRuntimeContextTest.java @@ -57,6 +57,8 @@ import org.apache.flink.runtime.state.KeyGroupRange; import org.apache.flink.runtime.state.KeyedStateBackend; import org.apache.flink.runtime.state.KeyedStateBackendParametersImpl; +import org.apache.flink.runtime.state.StateSchemaEvolvingTestSerializer; +import org.apache.flink.runtime.state.StateSchemaEvolvingTestSerializer.StateSchemaEvolvingTestTypeInfo; import org.apache.flink.runtime.state.VoidNamespace; import org.apache.flink.runtime.state.VoidNamespaceSerializer; import org.apache.flink.runtime.state.hashmap.HashMapStateBackend; @@ -247,6 +249,26 @@ void testMapStateReturnsEmptyMapByDefault() throws Exception { assertThat(value.iterator()).isExhausted(); } + @Test + void testValueStateSerializerIsArmedForObjectLevelMigratingBackend() throws Exception { + final ExecutionConfig config = new ExecutionConfig(); + final AtomicReference descriptorCapture = new AtomicReference<>(); + + StreamingRuntimeContext context = + createRuntimeContext(descriptorCapture, config, false, true); + ValueStateDescriptor descr = + new ValueStateDescriptor<>("name", new StateSchemaEvolvingTestTypeInfo()); + context.getState(descr); + + StateDescriptor descrIntercepted = (StateDescriptor) descriptorCapture.get(); + TypeSerializer serializer = descrIntercepted.getSerializer(); + + // The keyed state store owns the arming; a serializer initialized before delegating to the + // store would win the descriptor's CAS and leave this unarmed. + assertThat(serializer).isInstanceOf(StateSchemaEvolvingTestSerializer.class); + assertThat(((StateSchemaEvolvingTestSerializer) serializer).isArmed()).isTrue(); + } + @Test void testV2ValueStateInstantiation() throws Exception { @@ -404,11 +426,21 @@ private StreamingRuntimeContext createRuntimeContext() throws Exception { private StreamingRuntimeContext createRuntimeContext( AtomicReference descriptorCapture, ExecutionConfig config, boolean stateV2) throws Exception { + return createRuntimeContext(descriptorCapture, config, stateV2, false); + } + + private StreamingRuntimeContext createRuntimeContext( + AtomicReference descriptorCapture, + ExecutionConfig config, + boolean stateV2, + boolean objectLevelValueMigration) + throws Exception { return createDescriptorCapturingMockOp( descriptorCapture, config, MockEnvironment.builder().setExecutionConfig(config).build(), - stateV2) + stateV2, + objectLevelValueMigration) .getRuntimeContext(); } @@ -428,7 +460,8 @@ private static AbstractStreamOperator createDescriptorCapturingMockOp( final AtomicReference ref, final ExecutionConfig config, Environment environment, - boolean stateV2) + boolean stateV2, + boolean objectLevelValueMigration) throws Exception { StreamConfig streamConfig = new StreamConfig(new Configuration()); @@ -457,6 +490,8 @@ protected void setup( new StreamTaskStateInitializerImpl(environment, new HashMapStateBackend()); KeyedStateBackend keyedStateBackend = mock(KeyedStateBackend.class); + when(keyedStateBackend.supportsObjectLevelValueMigration()) + .thenReturn(objectLevelValueMigration); AsyncKeyedStateBackend asyncKeyedStateBackend = mock(AsyncKeyedStateBackend.class); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java index b60347ff208128..758f71237933d4 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/AbstractRocksDBState.java @@ -19,6 +19,7 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -36,6 +37,8 @@ import org.rocksdb.RocksDBException; import org.rocksdb.WriteOptions; +import javax.annotation.Nullable; + import java.io.IOException; import static org.apache.flink.util.Preconditions.checkArgument; @@ -191,6 +194,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer priorSerializer, + @Nullable TypeSerializerSnapshot priorSerializerSnapshot, TypeSerializer newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -203,6 +207,7 @@ public void migrateSerializedValue( try { ttlAwareNewSerializer.migrateValueFromPriorSerializer( ttlAwarePriorSerializer, + priorSerializerSnapshot, () -> ttlAwarePriorSerializer.deserialize(serializedOldValueInput), serializedMigratedValueOutput, ttlTimeProvider); diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java index b77232cb067b3c..01c7d1a7508980 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java @@ -886,9 +886,12 @@ private void migrateStateValues( Tuple2> stateMetaInfo) throws Exception { + // The snapshot the state was written with. Preferred over one re-derived from the previous + // serializer, which is itself restored from this snapshot and does not always round trip. + TypeSerializerSnapshot previousSerializerSnapshot = + stateMetaInfo.f1.getPreviousStateSerializerSnapshot(); + if (stateDesc.getType() == StateDescriptor.Type.MAP) { - TypeSerializerSnapshot previousSerializerSnapshot = - stateMetaInfo.f1.getPreviousStateSerializerSnapshot(); checkState( previousSerializerSnapshot != null, "the previous serializer snapshot should exist."); @@ -973,6 +976,7 @@ private void migrateStateValues( serializedValueInput, migratedSerializedValueOutput, previousTtlAwareSerializer, + previousSerializerSnapshot, currentTtlAwareSerializer, this.ttlTimeProvider); @@ -1111,6 +1115,14 @@ public boolean isSafeToReuseKVState() { return !(priorityQueueFactory instanceof HeapPriorityQueueSetFactory); } + @Override + public boolean supportsObjectLevelValueMigration() { + // A compatibleAfterMigration verdict routes every restored entry through + // migrateStateValues, whose per-entry migrateSerializedValue call reaches the migrate hook + // on the state's own value serializer. + return true; + } + @Override public String getBackendTypeIdentifier() { return StateBackendLoader.ROCKSDB_STATE_BACKEND_NAME; diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java index 8c6088df23d539..66b273a5ecb085 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBListState.java @@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -200,6 +202,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer> priorSerializer, + @Nullable TypeSerializerSnapshot> priorSerializerSnapshot, TypeSerializer> newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -214,11 +217,29 @@ public void migrateSerializedValue( TtlAwareSerializer newTtlAwareElementSerializer = ((TtlAwareSerializer.TtlAwareListSerializer) newSerializer) .getElementSerializer(); + // Descend the persisted snapshot the same way as the serializer, so element migration + // sees the schema the elements were written with. A state that carries no persisted + // snapshot leaves this null, and the element migration re-derives one instead. + TypeSerializerSnapshot priorElementSerializerSnapshot = null; + if (priorSerializerSnapshot != null) { + // Thrown rather than checked through Preconditions: this method runs once per state + // entry, so the message must not be built while the check is passing. + if (!(priorSerializerSnapshot instanceof ListSerializerSnapshot)) { + throw new IllegalArgumentException( + "The previous serializer snapshot of a list state should be a ListSerializerSnapshot, but was " + + priorSerializerSnapshot.getClass().getName() + + "."); + } + priorElementSerializerSnapshot = + ((ListSerializerSnapshot) priorSerializerSnapshot) + .getElementSerializerSnapshot(); + } try { while (serializedOldValueInput.available() > 0) { newTtlAwareElementSerializer.migrateValueFromPriorSerializer( priorTtlAwareElementSerializer, + priorElementSerializerSnapshot, () -> ListDelimitedSerializer.deserializeNextElement( serializedOldValueInput, priorTtlAwareElementSerializer), diff --git a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java index c811fc8caac2ca..f87ad7d6b2fcfa 100644 --- a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java +++ b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBMapState.java @@ -22,7 +22,9 @@ import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; import org.apache.flink.api.common.typeutils.base.MapSerializer; +import org.apache.flink.api.common.typeutils.base.MapSerializerSnapshot; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.core.memory.DataInputDeserializer; import org.apache.flink.core.memory.DataOutputSerializer; @@ -227,6 +229,7 @@ public void migrateSerializedValue( DataInputDeserializer serializedOldValueInput, DataOutputSerializer serializedMigratedValueOutput, TypeSerializer> priorSerializer, + @Nullable TypeSerializerSnapshot> priorSerializerSnapshot, TypeSerializer> newSerializer, TtlTimeProvider ttlTimeProvider) throws StateMigrationException { @@ -240,6 +243,23 @@ public void migrateSerializedValue( TtlAwareSerializer newTtlAwareMapValueSerializer = ((TtlAwareSerializer.TtlAwareMapSerializer) newSerializer) .getValueSerializer(); + // Descend the persisted snapshot the same way as the serializer, so value migration sees + // the schema the map values were written with. A state that carries no persisted + // snapshot leaves this null, and the value migration re-derives one instead. + TypeSerializerSnapshot priorMapValueSerializerSnapshot = null; + if (priorSerializerSnapshot != null) { + // Thrown rather than checked through Preconditions: this method runs once per state + // entry, so the message must not be built while the check is passing. + if (!(priorSerializerSnapshot instanceof MapSerializerSnapshot)) { + throw new IllegalArgumentException( + "The previous serializer snapshot of a map state should be a MapSerializerSnapshot, but was " + + priorSerializerSnapshot.getClass().getName() + + "."); + } + priorMapValueSerializerSnapshot = + ((MapSerializerSnapshot) priorSerializerSnapshot) + .getValueSerializerSnapshot(); + } try { boolean isNull = serializedOldValueInput.readBoolean(); @@ -250,6 +270,7 @@ public void migrateSerializedValue( } else { newTtlAwareMapValueSerializer.migrateValueFromPriorSerializer( priorTtlAwareMapValueSerializer, + priorMapValueSerializerSnapshot, () -> priorTtlAwareMapValueSerializer.deserialize(serializedOldValueInput), serializedMigratedValueOutput, ttlTimeProvider); diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java index a1ed3ec97ff588..dcc4a304a398e3 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/ExecutionConfigOptions.java @@ -45,6 +45,23 @@ public class ExecutionConfigOptions { // State Options // ------------------------------------------------------------------------ + @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) + public static final ConfigOption TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED = + key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "When enabled, a RowData keyed-state value whose schema changed in a " + + "backward-compatible way (adding nullable fields, reordering " + + "fields by name, evolving a nested ROW) is migrated when " + + "state is restored, instead of the restore failing. Only a " + + "state's own value serializer is covered: a RowData nested " + + "below a composite serializer, or a state whose serializer " + + "the operator pre-builds, is still rejected. Takes effect " + + "only on a state backend that migrates restored values at " + + "the object level, currently RocksDB; on other backends the " + + "restore still fails. Disabled by default."); + @Documentation.TableOption(execMode = Documentation.ExecMode.STREAMING) public static final ConfigOption IDLE_STATE_RETENTION = key("table.exec.state.ttl") diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/config/ExecutionConfigOptionsTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/config/ExecutionConfigOptionsTest.java new file mode 100644 index 00000000000000..57a403c15d3a1f --- /dev/null +++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/config/ExecutionConfigOptionsTest.java @@ -0,0 +1,48 @@ +/* + * 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.table.api.config; + +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.configuration.Configuration; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ExecutionConfigOptions}. */ +class ExecutionConfigOptionsTest { + + /** + * {@link SerializerConfigImpl} reads the state schema evolution option by key string, because + * flink-core cannot depend on this module. Nothing else ties the two together: if the key here + * changed, the option would keep documenting and validating while the runtime read its own + * unset key and the feature stayed permanently off. This is the only place both sides are + * visible at once. + */ + @Test + void stateSchemaEvolutionOptionReachesTheSerializerConfig() { + Configuration configuration = new Configuration(); + configuration.set(ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED, true); + + assertThat(new SerializerConfigImpl(configuration).isStateSchemaEvolutionEnabled()) + .isTrue(); + assertThat(new SerializerConfigImpl(new Configuration()).isStateSchemaEvolutionEnabled()) + .isFalse(); + } +} diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalTypeInfo.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalTypeInfo.java index 63fb8e23d22a72..5bb76b40401eff 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalTypeInfo.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/InternalTypeInfo.java @@ -194,7 +194,19 @@ public boolean isKeyType() { } @Override + @SuppressWarnings("unchecked") public TypeSerializer createSerializer(SerializerConfig config) { + if (config != null + && config.isStateSchemaEvolutionEnabled() + && typeSerializer instanceof RowDataSerializer) { + // The cached instance is shared by every use of this type information, so the opt-in + // is recorded on a copy rather than in place. The copy goes to every caller, not only + // to state registration, so with the option on this method stops handing out the + // shared instance for row types; callers compare serializers by equality, which + // neither evolution flag affects. + return (TypeSerializer) + ((RowDataSerializer) typeSerializer).withSchemaEvolutionAllowed(); + } return typeSerializer; } diff --git a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java index 718fbc20be9209..8618d4abd6bfef 100644 --- a/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java +++ b/flink-table/flink-table-type-utils/src/main/java/org/apache/flink/table/runtime/typeutils/RowDataSerializer.java @@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.typeutils.CompositeTypeSerializerUtil; import org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.common.typeutils.TypeSerializerSchemaCompatibility; import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; @@ -38,18 +39,23 @@ import org.apache.flink.table.data.writer.BinaryRowWriter; import org.apache.flink.table.data.writer.BinaryWriter; import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; import javax.annotation.Nullable; import java.io.IOException; import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; import java.util.stream.IntStream; /** Serializer for {@link RowData}. */ @Internal -public class RowDataSerializer extends AbstractRowDataSerializer { +public class RowDataSerializer extends AbstractRowDataSerializer + implements StateSchemaEvolvingSerializer { private static final long serialVersionUID = 1L; private BinaryRowDataSerializer binarySerializer; @@ -58,6 +64,16 @@ public class RowDataSerializer extends AbstractRowDataSerializer { private final TypeSerializer[] fieldSerializers; private final RowData.FieldGetter[] fieldGetters; + /** Whether the job configuration opted in to state schema evolution. */ + private final transient boolean schemaEvolutionAllowed; + + /** + * Whether this serializer is the value serializer of a state whose backend migrates restored + * values, and may therefore admit a backward-compatible schema change. This is the bit the + * snapshot carries into a compatibility check. + */ + private final transient boolean stateSchemaEvolutionEnabled; + private transient BinaryRowData reuseRow; private transient BinaryRowWriter reuseWriter; @@ -86,9 +102,20 @@ private RowDataSerializer( LogicalType[] types, TypeSerializer[] fieldSerializers, @Nullable String[] fieldNames) { + this(types, fieldSerializers, fieldNames, false, false); + } + + private RowDataSerializer( + LogicalType[] types, + TypeSerializer[] fieldSerializers, + @Nullable String[] fieldNames, + boolean schemaEvolutionAllowed, + boolean stateSchemaEvolutionEnabled) { this.types = types; this.fieldNames = fieldNames; this.fieldSerializers = fieldSerializers; + this.schemaEvolutionAllowed = schemaEvolutionAllowed; + this.stateSchemaEvolutionEnabled = stateSchemaEvolutionEnabled; this.binarySerializer = new BinaryRowDataSerializer(types.length); this.fieldGetters = IntStream.range(0, types.length) @@ -107,13 +134,69 @@ TypeSerializer[] fieldSerializers() { return fieldSerializers; } + @VisibleForTesting + boolean isSchemaEvolutionAllowed() { + return schemaEvolutionAllowed; + } + + @VisibleForTesting + boolean isStateSchemaEvolutionEnabled() { + return stateSchemaEvolutionEnabled; + } + + /** Returns a copy that records the job configuration having opted in to schema evolution. */ + RowDataSerializer withSchemaEvolutionAllowed() { + return new RowDataSerializer( + types, fieldSerializers, fieldNames, true, stateSchemaEvolutionEnabled); + } + + @Override + public TypeSerializer withStateSchemaEvolution() { + return schemaEvolutionAllowed ? enableStateSchemaEvolution() : this; + } + + /** + * Returns a copy that admits a backward-compatible schema change, recursing down the nested + * {@code ROW} spine. + * + *

The opt-in is checked once, by the caller, and never again below: nested field serializers + * are built by {@link InternalSerializers} inside this class's constructor, so they never carry + * the opt-in flag. Re-checking it one level down would silently no-op and short-circuit every + * nested {@code ROW} evolution to incompatible, because an unarmed nested snapshot rejects the + * change and that rejection propagates to the whole row. + * + *

The descent stops at anything that is not a {@code ROW}: the migration remap does not + * reach inside an ARRAY, MAP, MULTISET, RAW or structured value, so an evolved one is rejected + * by the leaf type-equality check instead. A structured type's serializer is a {@code + * RowDataSerializer} too, but its type root is not {@code ROW}, so the guard excludes it. + */ + private RowDataSerializer enableStateSchemaEvolution() { + TypeSerializer[] evolvingFieldSerializers = new TypeSerializer[fieldSerializers.length]; + for (int i = 0; i < fieldSerializers.length; i++) { + evolvingFieldSerializers[i] = + types[i].getTypeRoot() == LogicalTypeRoot.ROW + && fieldSerializers[i] instanceof RowDataSerializer + ? ((RowDataSerializer) fieldSerializers[i]).enableStateSchemaEvolution() + : fieldSerializers[i]; + } + return new RowDataSerializer( + types, evolvingFieldSerializers, fieldNames, schemaEvolutionAllowed, true); + } + @Override public TypeSerializer duplicate() { TypeSerializer[] duplicateFieldSerializers = new TypeSerializer[fieldSerializers.length]; for (int i = 0; i < fieldSerializers.length; i++) { duplicateFieldSerializers[i] = fieldSerializers[i].duplicate(); } - return new RowDataSerializer(types, duplicateFieldSerializers, fieldNames); + // Both evolution flags travel with the copy: a state backend registers a duplicate of the + // serializer, so a flag dropped here would never reach the snapshot. + return new RowDataSerializer( + types, + duplicateFieldSerializers, + fieldNames, + schemaEvolutionAllowed, + stateSchemaEvolutionEnabled); } @Override @@ -298,7 +381,8 @@ public int getLength() { @Override public TypeSerializerSnapshot snapshotConfiguration() { - return new RowDataSerializerSnapshot(types, fieldSerializers, fieldNames); + return new RowDataSerializerSnapshot( + types, fieldSerializers, fieldNames, stateSchemaEvolutionEnabled); } /** {@link TypeSerializerSnapshot} for {@link BinaryRowDataSerializer}. */ @@ -309,15 +393,27 @@ public static final class RowDataSerializerSnapshot implements TypeSerializerSna private @Nullable String[] fieldNames; private NestedSerializersSnapshotDelegate nestedSerializersSnapshotDelegate; + /** + * Whether the serializer this snapshot was taken from admits a backward-compatible schema + * change. It is deliberately not part of the snapshot format: a snapshot read back from + * bytes describes stored state, not a running job's configuration, so it always resolves + * with evolution off. + */ + private boolean stateSchemaEvolutionEnabled; + @SuppressWarnings("unused") public RowDataSerializerSnapshot() { // this constructor is used when restoring from a checkpoint/savepoint. } RowDataSerializerSnapshot( - LogicalType[] types, TypeSerializer[] serializers, @Nullable String[] fieldNames) { + LogicalType[] types, + TypeSerializer[] serializers, + @Nullable String[] fieldNames, + boolean stateSchemaEvolutionEnabled) { this.types = types; this.fieldNames = fieldNames; + this.stateSchemaEvolutionEnabled = stateSchemaEvolutionEnabled; this.nestedSerializersSnapshotDelegate = new NestedSerializersSnapshotDelegate(serializers); } @@ -387,25 +483,224 @@ public TypeSerializerSchemaCompatibility resolveSchemaCompatibility( RowDataSerializerSnapshot oldRowDataSerializerSnapshot = (RowDataSerializerSnapshot) oldSerializerSnapshot; - if (!Arrays.equals(types, oldRowDataSerializerSnapshot.types)) { + // A side that carries no names cannot disagree with anything: without names a reorder + // is undetectable, so identical types keep meaning "compatible as is" there, exactly as + // they do unarmed. + boolean namesDisagree = + stateSchemaEvolutionEnabled + && fieldNames != null + && oldRowDataSerializerSnapshot.fieldNames != null + && !Arrays.equals(fieldNames, oldRowDataSerializerSnapshot.fieldNames); + + // Identical positional layout: the nested composite path. Equal types at equal + // positions say nothing about which field is which, so an armed resolution has to see + // the names agree as well before it can treat the layout as unchanged -- otherwise a + // reorder or a rename among same-typed fields resolves here as needing no migration + // and leaves every value sitting under a neighbour's name. + if (Arrays.equals(types, oldRowDataSerializerSnapshot.types) && !namesDisagree) { + CompositeTypeSerializerUtil.IntermediateCompatibilityResult + intermediateResult = + CompositeTypeSerializerUtil + .constructIntermediateCompatibilityResult( + nestedSerializersSnapshotDelegate + .getNestedSerializerSnapshots(), + oldRowDataSerializerSnapshot + .nestedSerializersSnapshotDelegate + .getNestedSerializerSnapshots()); + + if (intermediateResult.isCompatibleWithReconfiguredSerializer()) { + RowDataSerializer reconfiguredCompositeSerializer = restoreSerializer(); + return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer( + reconfiguredCompositeSerializer); + } + + return intermediateResult.getFinalResult(); + } + + if (!stateSchemaEvolutionEnabled) { + return TypeSerializerSchemaCompatibility.incompatible(); + } + + // The new side must carry field names. A name-less new serializer is reachable -- a + // structured type resolves to one -- and admitting it would open a permanent name-less + // evolution channel rather than a ramp for savepoints taken before names were stored. + if (fieldNames == null) { return TypeSerializerSchemaCompatibility.incompatible(); } - CompositeTypeSerializerUtil.IntermediateCompatibilityResult - intermediateResult = - CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult( - nestedSerializersSnapshotDelegate - .getNestedSerializerSnapshots(), - oldRowDataSerializerSnapshot.nestedSerializersSnapshotDelegate - .getNestedSerializerSnapshots()); - - if (intermediateResult.isCompatibleWithReconfiguredSerializer()) { - RowDataSerializer reconfiguredCompositeSerializer = restoreSerializer(); - return TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer( - reconfiguredCompositeSerializer); + return oldRowDataSerializerSnapshot.fieldNames != null + ? checkNameBasedEvolution(oldRowDataSerializerSnapshot) + : checkPositionalEvolution(oldRowDataSerializerSnapshot); + } + + private TypeSerializerSchemaCompatibility checkNameBasedEvolution( + RowDataSerializerSnapshot oldSnapshot) { + int[] oldToNew = buildNameMapping(oldSnapshot.fieldNames, this.fieldNames); + int[] newToOld = buildNameMapping(this.fieldNames, oldSnapshot.fieldNames); + + // (A) Every new-only field (no matching old field) must be nullable. + for (int newPos = 0; newPos < newToOld.length; newPos++) { + if (newToOld[newPos] == -1 && !types[newPos].isNullable()) { + return TypeSerializerSchemaCompatibility.incompatible(); + } } - return intermediateResult.getFinalResult(); + // (B) Every old field must survive with a compatible type, and nested snapshots are + // aligned old->new so nested ROW evolution can recurse. Leaf (non-ROW) fields + // require an exactly equal type; ROW fields defer to the nested recursion in (C). + TypeSerializerSnapshot[] newNested = + nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots(); + TypeSerializerSnapshot[] alignedNewNested = + new TypeSerializerSnapshot[oldSnapshot.types.length]; + for (int oldPos = 0; oldPos < oldToNew.length; oldPos++) { + int newPos = oldToNew[oldPos]; + if (newPos == -1) { + return TypeSerializerSchemaCompatibility.incompatible(); // field removed + } + LogicalType oldType = oldSnapshot.types[oldPos]; + LogicalType newType = types[newPos]; + if (!bothRow(oldType, newType) && !oldType.equals(newType)) { + return TypeSerializerSchemaCompatibility.incompatible(); // leaf type changed + } + alignedNewNested[oldPos] = newNested[newPos]; + } + + // (C) Recurse into the aligned nested snapshot pairs. + return resolveAlignedNested(alignedNewNested, oldSnapshot); + } + + /** + * Resolves against a prior snapshot that carries no field names, matching fields by + * position. + * + *

Position is a stable identity only for an append. An insertion in the middle is + * indistinguishable from a retype plus an append, and the two demand opposite migrations, + * so the old layout has to be a prefix of the new one. + */ + private TypeSerializerSchemaCompatibility checkPositionalEvolution( + RowDataSerializerSnapshot oldSnapshot) { + if (types.length < oldSnapshot.types.length) { + return TypeSerializerSchemaCompatibility.incompatible(); + } + for (int i = 0; i < oldSnapshot.types.length; i++) { + LogicalType oldType = oldSnapshot.types[i]; + LogicalType newType = types[i]; + if (!bothRow(oldType, newType) && !oldType.equals(newType)) { + return TypeSerializerSchemaCompatibility.incompatible(); + } + } + for (int i = oldSnapshot.types.length; i < types.length; i++) { + if (!types[i].isNullable()) { + return TypeSerializerSchemaCompatibility.incompatible(); + } + } + + // constructIntermediateCompatibilityResult requires both arrays to have the same + // length, so only the prefix the old layout covers is handed to it. + TypeSerializerSnapshot[] alignedNewNested = + Arrays.copyOf( + nestedSerializersSnapshotDelegate.getNestedSerializerSnapshots(), + oldSnapshot.types.length); + return resolveAlignedNested(alignedNewNested, oldSnapshot); + } + + private TypeSerializerSchemaCompatibility resolveAlignedNested( + TypeSerializerSnapshot[] alignedNewNested, + RowDataSerializerSnapshot oldSnapshot) { + CompositeTypeSerializerUtil.IntermediateCompatibilityResult nested = + CompositeTypeSerializerUtil.constructIntermediateCompatibilityResult( + alignedNewNested, + oldSnapshot.nestedSerializersSnapshotDelegate + .getNestedSerializerSnapshots()); + // A reconfigured nested serializer is deliberately not propagated here, unlike on the + // identical-layout path. Reconfiguration exists so a new serializer can read old bytes; + // once the values have been remapped there are no old bytes left, because the migrated + // row is re-encoded by the state's own new serializer. + return nested.isIncompatible() + ? TypeSerializerSchemaCompatibility.incompatible() + : TypeSerializerSchemaCompatibility.compatibleAfterMigration(); + } + + private static boolean bothRow(LogicalType oldType, LogicalType newType) { + return oldType.getTypeRoot() == LogicalTypeRoot.ROW + && newType.getTypeRoot() == LogicalTypeRoot.ROW; + } + + @Override + public RowData migrate( + TypeSerializerSnapshot oldSerializerSnapshot, RowData value) { + if (value == null) { + return null; + } + // Runs once per restored entry, so a mismatch would otherwise surface as a bare + // ClassCastException from deep inside a migration loop. + Preconditions.checkArgument( + oldSerializerSnapshot instanceof RowDataSerializerSnapshot, + "Cannot migrate RowData state from %s.", + oldSerializerSnapshot.getClass().getName()); + RowDataSerializerSnapshot oldSnapshot = + (RowDataSerializerSnapshot) oldSerializerSnapshot; + return getNewRowData(value, oldSnapshot.restoreSerializer(), restoreSerializer()); + } + + // Remaps oldData into the new layout. Name-based when both serializers carry field names; + // otherwise positions map 1:1 up to the common field count. RowKind preserved; added + // fields and null sources become null; nested ROW values are remapped recursively. + private static GenericRowData getNewRowData( + RowData oldData, RowDataSerializer oldSerializer, RowDataSerializer newSerializer) { + GenericRowData newData = new GenericRowData(newSerializer.getArity()); + newData.setRowKind(oldData.getRowKind()); + int[] positions = buildPositionMapping(oldSerializer, newSerializer); + for (int newPos = 0; newPos < newSerializer.getArity(); newPos++) { + int oldPos = positions[newPos]; + if (oldPos != -1 && !oldData.isNullAt(oldPos)) { + Object fieldValue = oldSerializer.fieldGetters[oldPos].getFieldOrNull(oldData); + if (fieldValue instanceof RowData) { + fieldValue = + getNewRowData( + (RowData) fieldValue, + (RowDataSerializer) oldSerializer.fieldSerializers[oldPos], + (RowDataSerializer) newSerializer.fieldSerializers[newPos]); + } + newData.setField(newPos, fieldValue); + } else { + newData.setField(newPos, null); + } + } + return newData; + } + + // positions[newPos] = matching old position, or -1 for an added field. + private static int[] buildPositionMapping( + RowDataSerializer oldSerializer, RowDataSerializer newSerializer) { + if (oldSerializer.getFieldNames() != null && newSerializer.getFieldNames() != null) { + // Already indexed by new position, one entry per new field. + return buildNameMapping( + newSerializer.getFieldNames(), oldSerializer.getFieldNames()); + } + // The old arity comes from the serializer, not from the record: the compatibility rule + // is a statement about the snapshot's layout, and reading it off the record would make + // the remap follow whatever turned up instead. Under the prefix bound the two agree, + // which is exactly what would make a divergence invisible. + int[] positions = new int[newSerializer.getArity()]; + int commonFields = Math.min(oldSerializer.getArity(), newSerializer.getArity()); + for (int i = 0; i < newSerializer.getArity(); i++) { + positions[i] = i < commonFields ? i : -1; + } + return positions; + } + + // mapping[i] = index in toNames of the field named fromNames[i], or -1 if absent. + private static int[] buildNameMapping(String[] fromNames, String[] toNames) { + Map toIndex = new HashMap<>(toNames.length); + for (int i = 0; i < toNames.length; i++) { + toIndex.put(toNames[i], i); + } + int[] mapping = new int[fromNames.length]; + for (int i = 0; i < fromNames.length; i++) { + mapping[i] = toIndex.getOrDefault(fromNames[i], -1); + } + return mapping; } } } diff --git a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerSchemaEvolutionTest.java b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerSchemaEvolutionTest.java new file mode 100644 index 00000000000000..afceaeb7e9c466 --- /dev/null +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataSerializerSchemaEvolutionTest.java @@ -0,0 +1,573 @@ +/* + * 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.table.runtime.typeutils; + +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.StructuredType; +import org.apache.flink.table.types.logical.VarCharType; +import org.apache.flink.types.RowKind; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Arrays; + +import static org.apache.flink.table.data.StringData.fromString; +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests the opt-in state schema evolution carried by {@link RowDataSerializer}: the two evolution + * bits on the serializer, and the compatibility verdict plus {@code migrate} remap performed by + * {@link RowDataSerializerSnapshot} once the state bit is set. + * + *

Complements {@link RowDataSerializerFieldNamesTest}, which covers the field-name metadata and + * the V4 snapshot format that this evolution builds on, and {@link RowDataSerializerTest}, which + * covers per-record serialization round trips. + */ +class RowDataSerializerSchemaEvolutionTest { + + /** + * Mirrors {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED}, which lives + * in a module this one cannot depend on. + */ + private static final ConfigOption STATE_SCHEMA_EVOLUTION_ENABLED = + ConfigOptions.key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false); + + // ------------------------------------------------------------------------ + // Name-based evolution + // ------------------------------------------------------------------------ + + @Test + void addedNullableFieldAtEndIsMigrated() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), VarCharType.STRING_TYPE)); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"a", "b", "c"}, + new IntType(), + VarCharType.STRING_TYPE, + new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + GenericRowData oldValue = GenericRowData.of(1, fromString("x")); + oldValue.setRowKind(RowKind.UPDATE_AFTER); + RowData migrated = newSnap.migrate(oldSnap, oldValue); + + assertThat(migrated.getArity()).isEqualTo(3); + assertThat(migrated.getInt(0)).isEqualTo(1); + assertThat(migrated.getString(1)).isEqualTo(fromString("x")); + assertThat(migrated.isNullAt(2)).isTrue(); + assertThat(migrated.getRowKind()).isEqualTo(RowKind.UPDATE_AFTER); + } + + @Test + void addedNullableFieldInMiddleIsMigrated() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "c"}, new IntType(), new BigIntType())); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"a", "b", "c"}, + new IntType(), + new IntType(), + new BigIntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + RowData migrated = newSnap.migrate(oldSnap, GenericRowData.of(1, 99L)); + + assertThat(migrated.getInt(0)).isEqualTo(1); + assertThat(migrated.isNullAt(1)).isTrue(); + assertThat(migrated.getLong(2)).isEqualTo(99L); + } + + @Test + void reorderedFieldsAreMigratedByName() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + RowDataSerializerSnapshot newSnap = + newSnapshot(row(new String[] {"b", "a"}, new BigIntType(), new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + RowData migrated = newSnap.migrate(oldSnap, GenericRowData.of(7, 42L)); + + assertThat(migrated.getLong(0)).isEqualTo(42L); + assertThat(migrated.getInt(1)).isEqualTo(7); + } + + /** + * The reorder that {@link #reorderedFieldsAreMigratedByName} cannot catch: with the same type + * at every position the {@code LogicalType[]} arrays are equal, so only a name comparison + * distinguishes this from an unchanged layout. Resolving it as needing no migration would leave + * every value under its neighbour's name. + */ + @Test + void sameTypedReorderIsMigratedByName() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new IntType())); + RowDataSerializerSnapshot newSnap = + newSnapshot(row(new String[] {"b", "a"}, new IntType(), new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + RowData migrated = newSnap.migrate(oldSnap, GenericRowData.of(7, 42)); + + assertThat(migrated.getInt(0)).isEqualTo(42); + assertThat(migrated.getInt(1)).isEqualTo(7); + } + + /** A rename among same-typed fields drops the old field, so it cannot be migrated. */ + @Test + void sameTypedRenameIsIncompatible() throws IOException { + RowDataSerializerSnapshot oldSnap = oldSnapshot(row(new String[] {"a"}, new IntType())); + RowDataSerializerSnapshot newSnap = newSnapshot(row(new String[] {"x"}, new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + /** + * With the option off, field names are not consulted at all: an identical positional layout + * resolves exactly as it does without this feature, whatever the names did. + */ + @Test + void sameTypedReorderIsCompatibleAsIsWhenNotOptedIn() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new IntType())); + RowDataSerializerSnapshot unarmedNewSnap = + (RowDataSerializerSnapshot) + InternalSerializers.create( + row(new String[] {"b", "a"}, new IntType(), new IntType())) + .snapshotConfiguration(); + + assertThat(unarmedNewSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAsIs()).isTrue(); + } + + /** + * A prior snapshot from before field names were persisted names nothing, so a reorder cannot be + * detected there and identical types must keep meaning "no migration needed". Reporting a + * migration instead would rewrite every entry of an unchanged state on the first restore after + * a user opts in, which is the most common path this option will meet. + */ + @Test + void nameLessOldSnapshotWithUnchangedLayoutIsCompatibleAsIs() { + RowDataSerializerSnapshot oldSnap = nameLessSnapshot(new IntType(), new BigIntType()); + RowDataSerializerSnapshot newSnap = + newSnapshot(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAsIs()).isTrue(); + } + + /** + * The mirror case: a name-less new side against a named prior snapshot with identical types. + * Opting in must never narrow a restore that succeeds with the option off. + */ + @Test + void nameLessNewSnapshotWithUnchangedLayoutIsCompatibleAsIs() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + RowDataSerializer nameLessArmed = + (RowDataSerializer) + new RowDataSerializer(new IntType(), new BigIntType()) + .withSchemaEvolutionAllowed() + .withStateSchemaEvolution(); + RowDataSerializerSnapshot newSnap = + (RowDataSerializerSnapshot) nameLessArmed.snapshotConfiguration(); + + assertThat(nameLessArmed.isStateSchemaEvolutionEnabled()).isTrue(); + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAsIs()).isTrue(); + } + + /** + * An unchanged layout can still reach {@code migrate}, because a nested serializer may report + * {@code compatibleAfterMigration} on its own. The remap must then reproduce the row exactly. + */ + @Test + void migrateReproducesAnUnchangedLayout() throws IOException { + RowType rowType = row(new String[] {"a", "b"}, new IntType(), VarCharType.STRING_TYPE); + + RowData migrated = + newSnapshot(rowType) + .migrate(oldSnapshot(rowType), GenericRowData.of(1, fromString("x"))); + + assertThat(migrated.getArity()).isEqualTo(2); + assertThat(migrated.getInt(0)).isEqualTo(1); + assertThat(migrated.getString(1)).isEqualTo(fromString("x")); + } + + @Test + void addedNotNullFieldIsIncompatible() throws IOException { + RowDataSerializerSnapshot oldSnap = oldSnapshot(row(new String[] {"a"}, new IntType())); + RowDataSerializerSnapshot newSnap = + newSnapshot(row(new String[] {"a", "b"}, new IntType(), new IntType(false))); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void droppedFieldIsIncompatible() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + RowDataSerializerSnapshot newSnap = newSnapshot(row(new String[] {"a"}, new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void changedLeafTypeIsIncompatible() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"a", "b"}, new IntType(), new IntType())); + RowDataSerializerSnapshot newSnap = + newSnapshot(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void evolvedNestedRowIsMigrated() throws IOException { + RowType oldNested = row(new String[] {"x", "y"}, new IntType(), VarCharType.STRING_TYPE); + RowType newNested = + row( + new String[] {"x", "y", "z"}, + new IntType(), + VarCharType.STRING_TYPE, + new IntType()); + RowType newType = row(new String[] {"id", "nested"}, new IntType(), newNested); + + RowDataSerializer stateSerializer = + (RowDataSerializer) + StateSchemaEvolvingSerializer.armStateValueSerializer(optedIn(newType)); + + // Nested field serializers are built by InternalSerializers inside the RowDataSerializer + // constructor, so they never carry the opt-in bit. The state bit therefore has to be set on + // them unconditionally down the ROW spine; re-checking the opt-in bit one level down would + // leave the nested snapshot unarmed and short-circuit the whole row to incompatible. + RowDataSerializer nestedSerializer = + (RowDataSerializer) stateSerializer.fieldSerializers()[1]; + assertThat(nestedSerializer.isSchemaEvolutionAllowed()).isFalse(); + assertThat(nestedSerializer.isStateSchemaEvolutionEnabled()).isTrue(); + + RowDataSerializerSnapshot newSnap = + (RowDataSerializerSnapshot) stateSerializer.snapshotConfiguration(); + RowDataSerializerSnapshot oldSnap = + oldSnapshot(row(new String[] {"id", "nested"}, new IntType(), oldNested)); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + RowData migrated = + newSnap.migrate( + oldSnap, GenericRowData.of(5, GenericRowData.of(8, fromString("hi")))); + + assertThat(migrated.getInt(0)).isEqualTo(5); + RowData migratedNested = migrated.getRow(1, 3); + assertThat(migratedNested.getInt(0)).isEqualTo(8); + assertThat(migratedNested.getString(1)).isEqualTo(fromString("hi")); + assertThat(migratedNested.isNullAt(2)).isTrue(); + } + + @Test + void incompatibleNestedRowChangeIsIncompatible() throws IOException { + RowDataSerializerSnapshot oldSnap = + oldSnapshot( + row( + new String[] {"id", "nested"}, + new IntType(), + row(new String[] {"a"}, new IntType()))); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"id", "nested"}, + new IntType(), + row(new String[] {"a"}, new BigIntType()))); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void changedArrayElementRowIsIncompatible() throws IOException { + RowType oldElement = row(new String[] {"a"}, new IntType()); + RowType newElement = row(new String[] {"a"}, new BigIntType()); + RowType newType = row(new String[] {"id", "arr"}, new IntType(), new ArrayType(newElement)); + + // The recursion descends only into ROW-typed fields, so the ROW nested under the ARRAY is + // never reached and the change has to be rejected by the leaf type-equality check instead. + RowDataSerializer stateSerializer = armed(newType); + ArrayDataSerializer arraySerializer = + (ArrayDataSerializer) stateSerializer.fieldSerializers()[1]; + RowDataSerializer elementSerializer = (RowDataSerializer) arraySerializer.getEleSer(); + assertThat(elementSerializer.isStateSchemaEvolutionEnabled()).isFalse(); + + RowDataSerializerSnapshot newSnap = + (RowDataSerializerSnapshot) stateSerializer.snapshotConfiguration(); + RowDataSerializerSnapshot oldSnap = + oldSnapshot( + row(new String[] {"id", "arr"}, new IntType(), new ArrayType(oldElement))); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + // ------------------------------------------------------------------------ + // Opt-in semantics + // ------------------------------------------------------------------------ + + @Test + void unarmedNewSnapshotIsIncompatible() throws IOException { + RowType newType = row(new String[] {"a", "b"}, new IntType(), new IntType()); + RowDataSerializer notOptedIn = + (RowDataSerializer) InternalTypeInfo.of(newType).createSerializer(config(false)); + RowDataSerializerSnapshot newSnap = + (RowDataSerializerSnapshot) + notOptedIn.withStateSchemaEvolution().snapshotConfiguration(); + + assertThat( + newSnap.resolveSchemaCompatibility( + oldSnapshot(row(new String[] {"a"}, new IntType()))) + .isIncompatible()) + .isTrue(); + } + + @Test + void armedBitIsNotPersisted() throws IOException { + RowType oldType = row(new String[] {"a"}, new IntType()); + RowType newType = row(new String[] {"a", "b"}, new IntType(), new IntType()); + + assertThat( + newSnapshot(newType) + .resolveSchemaCompatibility(oldSnapshot(oldType)) + .isCompatibleAfterMigration()) + .isTrue(); + + RowDataSerializerSnapshot restored = roundTrip(armed(newType).snapshotConfiguration()); + + assertThat(restored.resolveSchemaCompatibility(oldSnapshot(oldType)).isIncompatible()) + .isTrue(); + } + + @Test + void withStateSchemaEvolutionIsIdentityWhenNotOptedIn() { + RowDataSerializer serializer = + InternalSerializers.create( + row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + + assertThat(serializer.withStateSchemaEvolution()).isSameAs(serializer); + } + + @Test + void optingInDoesNotMutateTheCachedSerializer() { + InternalTypeInfo typeInfo = + InternalTypeInfo.of(row(new String[] {"a", "b"}, new IntType(), new BigIntType())); + RowDataSerializer cached = typeInfo.toRowSerializer(); + + RowDataSerializer optedIn = (RowDataSerializer) typeInfo.createSerializer(config(true)); + + assertThat(optedIn).isNotSameAs(cached); + assertThat(optedIn.isSchemaEvolutionAllowed()).isTrue(); + assertThat(cached.isSchemaEvolutionAllowed()).isFalse(); + assertThat(typeInfo.createSerializer(config(false))).isSameAs(cached); + } + + @Test + void duplicateCarriesBothEvolutionBits() { + RowType rowType = row(new String[] {"a", "b"}, new IntType(), new BigIntType()); + + RowDataSerializer armedDuplicate = (RowDataSerializer) armed(rowType).duplicate(); + assertThat(armedDuplicate.isSchemaEvolutionAllowed()).isTrue(); + assertThat(armedDuplicate.isStateSchemaEvolutionEnabled()).isTrue(); + + RowDataSerializer plainDuplicate = + (RowDataSerializer) InternalSerializers.create(rowType).duplicate(); + assertThat(plainDuplicate.isSchemaEvolutionAllowed()).isFalse(); + assertThat(plainDuplicate.isStateSchemaEvolutionEnabled()).isFalse(); + } + + @Test + void equalsAndHashCodeIgnoreEvolutionBits() { + RowType rowType = row(new String[] {"a", "b"}, new IntType(), new BigIntType()); + RowDataSerializer plain = InternalSerializers.create(rowType); + + assertThat(armed(rowType)).isEqualTo(plain); + assertThat(armed(rowType).hashCode()).isEqualTo(plain.hashCode()); + } + + // ------------------------------------------------------------------------ + // Positional fallback for a prior snapshot written without field names + // ------------------------------------------------------------------------ + + @Test + void positionalAppendedNullableFieldIsMigrated() { + RowDataSerializerSnapshot oldSnap = nameLessSnapshot(new IntType(), new BigIntType()); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"a", "b", "c"}, + new IntType(), + new BigIntType(), + new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isCompatibleAfterMigration()) + .isTrue(); + + RowData migrated = newSnap.migrate(oldSnap, GenericRowData.of(1, 2L)); + + assertThat(migrated.getArity()).isEqualTo(3); + assertThat(migrated.getInt(0)).isEqualTo(1); + assertThat(migrated.getLong(1)).isEqualTo(2L); + assertThat(migrated.isNullAt(2)).isTrue(); + } + + @Test + void positionalAppendedNotNullFieldIsIncompatible() { + RowDataSerializerSnapshot oldSnap = nameLessSnapshot(new IntType(), new BigIntType()); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"a", "b", "c"}, + new IntType(), + new BigIntType(), + new IntType(false))); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void positionalRetypedFieldIsIncompatible() { + RowDataSerializerSnapshot oldSnap = nameLessSnapshot(new IntType(), new BigIntType()); + RowDataSerializerSnapshot newSnap = + newSnapshot( + row( + new String[] {"a", "b", "c"}, + new IntType(), + new IntType(), + new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void positionalDroppedFieldIsIncompatible() { + RowDataSerializerSnapshot oldSnap = nameLessSnapshot(new IntType(), new BigIntType()); + RowDataSerializerSnapshot newSnap = newSnapshot(row(new String[] {"a"}, new IntType())); + + assertThat(newSnap.resolveSchemaCompatibility(oldSnap).isIncompatible()).isTrue(); + } + + @Test + void nameLessNewSnapshotIsIncompatible() throws IOException { + // A STRUCTURED_TYPE resolves to a RowDataSerializer built from field types only, so it can + // be opted in and armed while still carrying no field names. + StructuredType structuredType = + StructuredType.newBuilder("org.apache.flink.table.NameLess") + .attributes( + Arrays.asList( + new StructuredType.StructuredAttribute("a", new IntType()), + new StructuredType.StructuredAttribute( + "b", new BigIntType()))) + .build(); + RowDataSerializer nameLess = + (RowDataSerializer) + InternalTypeInfo.of(structuredType).createSerializer(config(true)); + RowDataSerializer stateSerializer = (RowDataSerializer) nameLess.withStateSchemaEvolution(); + + assertThat(stateSerializer.getFieldNames()).isNull(); + assertThat(stateSerializer.isStateSchemaEvolutionEnabled()).isTrue(); + + RowDataSerializerSnapshot newSnap = + (RowDataSerializerSnapshot) stateSerializer.snapshotConfiguration(); + + assertThat( + newSnap.resolveSchemaCompatibility( + oldSnapshot(row(new String[] {"a"}, new IntType()))) + .isIncompatible()) + .isTrue(); + } + + // ------------------------------------------------------------------------ + + private static RowType row(String[] names, LogicalType... types) { + return RowType.of(types, names); + } + + private static SerializerConfig config(boolean schemaEvolutionEnabled) { + Configuration configuration = new Configuration(); + configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, schemaEvolutionEnabled); + return new SerializerConfigImpl(configuration); + } + + /** A serializer the job opted in, but that is not yet a state's own value serializer. */ + private static RowDataSerializer optedIn(RowType rowType) { + return (RowDataSerializer) InternalTypeInfo.of(rowType).createSerializer(config(true)); + } + + private static RowDataSerializer armed(RowType rowType) { + return (RowDataSerializer) optedIn(rowType).withStateSchemaEvolution(); + } + + /** + * The new side of a resolution. Kept in memory rather than round-tripped, because the state bit + * is not part of the snapshot format and a round trip clears it. + */ + private static RowDataSerializerSnapshot newSnapshot(RowType rowType) { + return (RowDataSerializerSnapshot) armed(rowType).snapshotConfiguration(); + } + + /** The old side of a resolution, read back from bytes the way a restore produces it. */ + private static RowDataSerializerSnapshot oldSnapshot(RowType rowType) throws IOException { + return roundTrip(InternalSerializers.create(rowType).snapshotConfiguration()); + } + + /** A prior snapshot from before field names were persisted. */ + private static RowDataSerializerSnapshot nameLessSnapshot(LogicalType... types) { + return (RowDataSerializerSnapshot) new RowDataSerializer(types).snapshotConfiguration(); + } + + private static RowDataSerializerSnapshot roundTrip(TypeSerializerSnapshot snapshot) + throws IOException { + DataOutputSerializer out = new DataOutputSerializer(256); + TypeSerializerSnapshot.writeVersionedSnapshot(out, snapshot); + DataInputDeserializer in = new DataInputDeserializer(out.getCopyOfBuffer()); + return (RowDataSerializerSnapshot) + TypeSerializerSnapshot.readVersionedSnapshot( + in, RowDataSerializerSchemaEvolutionTest.class.getClassLoader()); + } +} diff --git a/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataStateSchemaEvolutionArmingTest.java b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataStateSchemaEvolutionArmingTest.java new file mode 100644 index 00000000000000..d23f03d5f58ab1 --- /dev/null +++ b/flink-table/flink-table-type-utils/src/test/java/org/apache/flink/table/runtime/typeutils/RowDataStateSchemaEvolutionArmingTest.java @@ -0,0 +1,160 @@ +/* + * 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.table.runtime.typeutils; + +import org.apache.flink.api.common.functions.SerializerFactory; +import org.apache.flink.api.common.serialization.SerializerConfig; +import org.apache.flink.api.common.serialization.SerializerConfigImpl; +import org.apache.flink.api.common.state.ListStateDescriptor; +import org.apache.flink.api.common.state.MapStateDescriptor; +import org.apache.flink.api.common.state.ValueStateDescriptor; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.common.typeutils.StateSchemaEvolvingSerializer; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.api.common.typeutils.base.ListSerializer; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.api.java.typeutils.ListTypeInfo; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.runtime.TupleSerializer; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.RowDataSerializer.RowDataSerializerSnapshot; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests which state shapes the arming {@link SerializerFactory} decorator reaches. It arms exactly + * one structural level, so a {@link RowDataSerializer} sitting below any composite serializer stays + * unarmed and its state fails closed at restore. + * + *

Every assertion is made on the serializer object actually reached through the state + * descriptor. {@link RowDataSerializerSchemaEvolutionTest} covers what an armed serializer then + * admits. + */ +class RowDataStateSchemaEvolutionArmingTest { + + /** + * Mirrors {@code ExecutionConfigOptions.TABLE_EXEC_STATE_SCHEMA_EVOLUTION_ENABLED}, which lives + * in a module this one cannot depend on. + */ + private static final ConfigOption STATE_SCHEMA_EVOLUTION_ENABLED = + ConfigOptions.key("table.exec.state.schema-evolution.enabled") + .booleanType() + .defaultValue(false); + + private static final RowType ROW_TYPE = + RowType.of( + new LogicalType[] {new IntType(), new BigIntType()}, new String[] {"a", "b"}); + + @Test + @SuppressWarnings("unchecked") + void intervalJoinShapedMapStateIsNotArmed() { + // MapState>>, the shape the interval join registers. + ListTypeInfo> valueTypeInfo = + new ListTypeInfo<>( + new TupleTypeInfo>( + InternalTypeInfo.of(ROW_TYPE), Types.BOOLEAN)); + MapStateDescriptor>> descriptor = + new MapStateDescriptor<>("cache", Types.LONG, valueTypeInfo); + + descriptor.initializeSerializerUnlessSet(armingFactory(true)); + + ListSerializer> listSerializer = + (ListSerializer>) descriptor.getValueSerializer(); + TupleSerializer> tupleSerializer = + (TupleSerializer>) listSerializer.getElementSerializer(); + TypeSerializer nestedField = tupleSerializer.getFieldSerializers()[0]; + RowDataSerializer nested = (RowDataSerializer) nestedField; + + // The job opted in, so the nested serializer carries the opt-in bit; what it must not carry + // is the state bit, because nothing will call migrate on a serializer two levels down. + assertThat(nested.isSchemaEvolutionAllowed()).isTrue(); + assertThat(nested.isStateSchemaEvolutionEnabled()).isFalse(); + + RowType priorRowType = RowType.of(new LogicalType[] {new IntType()}, new String[] {"a"}); + RowDataSerializerSnapshot nestedSnapshot = + (RowDataSerializerSnapshot) nested.snapshotConfiguration(); + RowDataSerializerSnapshot priorSnapshot = + (RowDataSerializerSnapshot) + InternalSerializers.create(priorRowType).snapshotConfiguration(); + + assertThat(nestedSnapshot.resolveSchemaCompatibility(priorSnapshot).isIncompatible()) + .isTrue(); + } + + @Test + void valueStateIsArmed() { + ValueStateDescriptor descriptor = + new ValueStateDescriptor<>("value", InternalTypeInfo.of(ROW_TYPE)); + + descriptor.initializeSerializerUnlessSet(armingFactory(true)); + + assertThat(((RowDataSerializer) descriptor.getSerializer()).isStateSchemaEvolutionEnabled()) + .isTrue(); + } + + @Test + void valueStateIsNotArmedWhenTheOptionIsOff() { + ValueStateDescriptor descriptor = + new ValueStateDescriptor<>("value", InternalTypeInfo.of(ROW_TYPE)); + + descriptor.initializeSerializerUnlessSet(armingFactory(false)); + + assertThat(((RowDataSerializer) descriptor.getSerializer()).isStateSchemaEvolutionEnabled()) + .isFalse(); + } + + @Test + void listStateElementIsNotArmed() { + ListStateDescriptor descriptor = + new ListStateDescriptor<>("list", InternalTypeInfo.of(ROW_TYPE)); + + descriptor.initializeSerializerUnlessSet(armingFactory(true)); + + assertThat( + ((RowDataSerializer) descriptor.getElementSerializer()) + .isStateSchemaEvolutionEnabled()) + .isFalse(); + } + + private static SerializerFactory armingFactory(boolean schemaEvolutionEnabled) { + Configuration configuration = new Configuration(); + configuration.set(STATE_SCHEMA_EVOLUTION_ENABLED, schemaEvolutionEnabled); + SerializerConfig config = new SerializerConfigImpl(configuration); + return StateSchemaEvolvingSerializer.arming( + new SerializerFactory() { + @Override + public TypeSerializer createSerializer( + TypeInformation typeInformation) { + return typeInformation.createSerializer(config); + } + }); + } +}